From e86fa192570722fa1f766d9cade3e399ea5e8445 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Fri, 7 Jun 2024 16:20:23 -0700 Subject: [PATCH 1/4] fix - basic success logging for redis cache --- litellm/caching.py | 31 +++++++++++++++++++++++++++ litellm/integrations/opentelemetry.py | 17 +++++++++++++-- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/litellm/caching.py b/litellm/caching.py index d1f3387ee43..c1a4aa9b6ad 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 @@ -276,6 +289,9 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_scan_iter", + start_time=start_time, + end_time=end_time, + # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs) ) ) # DO NOT SLOW DOWN CALL B/C OF THIS return keys @@ -331,6 +347,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: @@ -389,6 +408,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 @@ -434,6 +456,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 @@ -540,6 +565,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 @@ -583,6 +611,9 @@ class RedisCache(BaseCache): service=ServiceTypes.REDIS, duration=_duration, call_type="async_batch_get_cache", + start_time=start_time, + end_time=end_time, + # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs) ) ) diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index 089b67368a4..deee0a2f266 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -100,19 +100,32 @@ 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_post_call_failure_hook( self, original_exception: Exception, user_api_key_dict: UserAPIKeyAuth From 3f59ee200aed56d67f12f3a0d07452b96bce22ab Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Fri, 7 Jun 2024 16:21:23 -0700 Subject: [PATCH 2/4] fix - cache - true --- litellm/proxy/proxy_config.yaml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 + From 5a5dd33b24eeb5d9ed2bdbd1ad2b98c4f3a5fa86 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Fri, 7 Jun 2024 16:30:29 -0700 Subject: [PATCH 3/4] feat - working exception logs for Redis errors --- litellm/_service_logger.py | 13 +++++++++ litellm/caching.py | 28 +++++++++++++++++++- litellm/integrations/opentelemetry.py | 38 +++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 1 deletion(-) 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 c1a4aa9b6ad..7fc535507af 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -259,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( @@ -306,6 +309,9 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_scan_iter", + start_time=start_time, + end_time=end_time, + # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) raise e @@ -319,7 +325,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 @@ -361,6 +372,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 @@ -424,6 +438,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), ) ) @@ -472,6 +489,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( @@ -581,6 +601,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 @@ -639,6 +662,9 @@ class RedisCache(BaseCache): duration=_duration, error=e, call_type="async_batch_get_cache", + start_time=start_time, + end_time=end_time, + # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) print_verbose(f"Error occurred in pipeline read - {str(e)}") diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index deee0a2f266..84b2f88c65e 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -127,6 +127,44 @@ class OpenTelemetry(CustomLogger): service_logging_span.set_status(Status(StatusCode.OK)) 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 ): From f152b5eb1d64980127447331cfb0562af2475e73 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Fri, 7 Jun 2024 16:36:04 -0700 Subject: [PATCH 4/4] feat - final working redis cache otel --- litellm/caching.py | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/litellm/caching.py b/litellm/caching.py index 7fc535507af..497dd2371c5 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -294,7 +294,6 @@ class RedisCache(BaseCache): call_type="async_scan_iter", start_time=start_time, end_time=end_time, - # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs) ) ) # DO NOT SLOW DOWN CALL B/C OF THIS return keys @@ -311,7 +310,6 @@ class RedisCache(BaseCache): call_type="async_scan_iter", start_time=start_time, end_time=end_time, - # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) raise e @@ -384,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 """ @@ -424,7 +422,7 @@ class RedisCache(BaseCache): call_type="async_set_cache_pipeline", start_time=start_time, end_time=end_time, - # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs) + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) return results @@ -440,7 +438,7 @@ class RedisCache(BaseCache): call_type="async_set_cache_pipeline", start_time=start_time, end_time=end_time, - # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), + parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) @@ -636,7 +634,6 @@ class RedisCache(BaseCache): call_type="async_batch_get_cache", start_time=start_time, end_time=end_time, - # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs) ) ) @@ -664,7 +661,6 @@ class RedisCache(BaseCache): call_type="async_batch_get_cache", start_time=start_time, end_time=end_time, - # parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) print_verbose(f"Error occurred in pipeline read - {str(e)}") @@ -1452,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)}")