diff --git a/litellm/_service_logger.py b/litellm/_service_logger.py index fd14b3cdeb4..f2edb403e95 100644 --- a/litellm/_service_logger.py +++ b/litellm/_service_logger.py @@ -91,6 +91,9 @@ class ServiceLogging(CustomLogger): duration: float, error: Union[str, Exception], call_type: str, + parent_otel_span: Optional[Span] = None, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, ): """ - For counting if the redis, postgres call is unsuccessful @@ -119,6 +122,16 @@ class ServiceLogging(CustomLogger): payload=payload ) + from litellm.proxy.proxy_server import open_telemetry_logger + + if parent_otel_span is not None and open_telemetry_logger is not None: + await open_telemetry_logger.async_service_failure_hook( + payload=payload, + parent_otel_span=parent_otel_span, + start_time=start_time, + end_time=end_time, + ) + async def async_post_call_failure_hook( self, original_exception: Exception, user_api_key_dict: UserAPIKeyAuth ): diff --git a/litellm/caching.py b/litellm/caching.py index d1f3387ee43..497dd2371c5 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -26,6 +26,16 @@ def print_verbose(print_statement): pass +def _get_parent_otel_span_from_kwargs(kwargs: Optional[dict] = None): + try: + if kwargs is None: + return None + _metadata = kwargs.get("metadata") or {} + return _metadata.get("litellm_parent_otel_span") + except: + return None + + class BaseCache: def set_cache(self, key, value, **kwargs): raise NotImplementedError @@ -233,6 +243,9 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="increment_cache", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) return result @@ -246,6 +259,9 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="increment_cache", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) verbose_logger.error( @@ -276,6 +292,8 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_scan_iter", + start_time=start_time, + end_time=end_time, ) ) # DO NOT SLOW DOWN CALL B/C OF THIS return keys @@ -290,6 +308,8 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_scan_iter", + start_time=start_time, + end_time=end_time, ) ) raise e @@ -303,7 +323,12 @@ class RedisCache(BaseCache): _duration = end_time - start_time asyncio.create_task( self.service_logger_obj.async_service_failure_hook( - service=ServiceTypes.REDIS, duration=_duration, error=e + service=ServiceTypes.REDIS, + duration=_duration, + error=e, + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) # NON blocking - notify users Redis is throwing an exception @@ -331,6 +356,9 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_set_cache", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) except Exception as e: @@ -342,6 +370,9 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_set_cache", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) # NON blocking - notify users Redis is throwing an exception @@ -351,7 +382,7 @@ class RedisCache(BaseCache): value, ) - async def async_set_cache_pipeline(self, cache_list, ttl=None): + async def async_set_cache_pipeline(self, cache_list, ttl=None, **kwargs): """ Use Redis Pipelines for bulk write operations """ @@ -389,6 +420,9 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_set_cache_pipeline", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) return results @@ -402,6 +436,9 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_set_cache_pipeline", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) @@ -434,6 +471,9 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_increment", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) return result @@ -447,6 +487,9 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_increment", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) verbose_logger.error( @@ -540,6 +583,9 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_get_cache", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) return response @@ -553,6 +599,9 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_get_cache", + start_time=start_time, + end_time=end_time, + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) # NON blocking - notify users Redis is throwing an exception @@ -583,6 +632,8 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_batch_get_cache", + start_time=start_time, + end_time=end_time, ) ) @@ -608,6 +659,8 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_batch_get_cache", + start_time=start_time, + end_time=end_time, ) ) print_verbose(f"Error occurred in pipeline read - {str(e)}") @@ -1395,7 +1448,7 @@ class DualCache(BaseCache): if self.redis_cache is not None and local_only == False: await self.redis_cache.async_set_cache_pipeline( - cache_list=cache_list, ttl=kwargs.get("ttl", None) + cache_list=cache_list, ttl=kwargs.get("ttl", None), **kwargs ) except Exception as e: verbose_logger.error(f"LiteLLM Cache: Excepton async add_cache: {str(e)}") diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index 089b67368a4..84b2f88c65e 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -100,19 +100,70 @@ class OpenTelemetry(CustomLogger): from datetime import datetime from opentelemetry.trace import Status, StatusCode + _start_time_ns = start_time + _end_time_ns = end_time + + if isinstance(start_time, float): + _start_time_ns = int(int(start_time) * 1e9) + else: + _start_time_ns = self._to_ns(start_time) + + if isinstance(end_time, float): + _end_time_ns = int(int(end_time) * 1e9) + else: + _end_time_ns = self._to_ns(end_time) + if parent_otel_span is not None: _span_name = payload.service service_logging_span = self.tracer.start_span( name=_span_name, context=trace.set_span_in_context(parent_otel_span), - start_time=self._to_ns(start_time), + start_time=_start_time_ns, ) service_logging_span.set_attribute(key="call_type", value=payload.call_type) service_logging_span.set_attribute( key="service", value=payload.service.value ) service_logging_span.set_status(Status(StatusCode.OK)) - service_logging_span.end(end_time=self._to_ns(end_time)) + service_logging_span.end(end_time=_end_time_ns) + + async def async_service_failure_hook( + self, + payload: ServiceLoggerPayload, + parent_otel_span: Optional[Span] = None, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + ): + from opentelemetry import trace + from datetime import datetime + from opentelemetry.trace import Status, StatusCode + + _start_time_ns = start_time + _end_time_ns = end_time + + if isinstance(start_time, float): + _start_time_ns = int(int(start_time) * 1e9) + else: + _start_time_ns = self._to_ns(start_time) + + if isinstance(end_time, float): + _end_time_ns = int(int(end_time) * 1e9) + else: + _end_time_ns = self._to_ns(end_time) + + if parent_otel_span is not None: + _span_name = payload.service + service_logging_span = self.tracer.start_span( + name=_span_name, + context=trace.set_span_in_context(parent_otel_span), + start_time=_start_time_ns, + ) + service_logging_span.set_attribute(key="call_type", value=payload.call_type) + service_logging_span.set_attribute( + key="service", value=payload.service.value + ) + service_logging_span.set_status(Status(StatusCode.ERROR)) + service_logging_span.end(end_time=_end_time_ns) async def async_post_call_failure_hook( self, original_exception: Exception, user_api_key_dict: UserAPIKeyAuth diff --git a/litellm/proxy/proxy_config.yaml b/litellm/proxy/proxy_config.yaml index df6dfd13945..3c6b2f2017f 100644 --- a/litellm/proxy/proxy_config.yaml +++ b/litellm/proxy/proxy_config.yaml @@ -21,7 +21,6 @@ model_list: general_settings: master_key: sk-1234 - alerting: ["slack"] litellm_settings: callbacks: ["otel"] @@ -31,4 +30,6 @@ litellm_settings: - user - metadata - metadata.generation_name + cache: True +