diff --git a/enterprise/litellm_enterprise/integrations/prometheus.py b/enterprise/litellm_enterprise/integrations/prometheus.py index d3b599edff..57db14fec4 100644 --- a/enterprise/litellm_enterprise/integrations/prometheus.py +++ b/enterprise/litellm_enterprise/integrations/prometheus.py @@ -298,6 +298,13 @@ class PrometheusLogger(CustomLogger): self.get_labels_for_metric("litellm_deployment_failed_fallbacks"), ) + # Callback Logging Failure Metrics + self.litellm_callback_logging_failures_metric = self._counter_factory( + name="litellm_callback_logging_failures_metric", + documentation="Total number of failures when emitting logs to callbacks (e.g. s3_v2, langfuse, etc)", + labelnames=["callback_name"], + ) + self.litellm_llm_api_failed_requests_metric = self._counter_factory( name="litellm_llm_api_failed_requests_metric", documentation="deprecated - use litellm_proxy_failed_requests_metric", @@ -1723,6 +1730,17 @@ class PrometheusLogger(CustomLogger): litellm_model_name, model_id, api_base, api_provider, exception_status ).inc() + def increment_callback_logging_failure( + self, + callback_name: str, + ): + """ + Increment metric when logging to a callback fails (e.g., s3_v2, langfuse, etc.) + """ + self.litellm_callback_logging_failures_metric.labels( + callback_name=callback_name + ).inc() + def track_provider_remaining_budget( self, provider: str, spend: float, budget_limit: float ): diff --git a/litellm/integrations/custom_logger.py b/litellm/integrations/custom_logger.py index 537507256a..6c613b6c14 100644 --- a/litellm/integrations/custom_logger.py +++ b/litellm/integrations/custom_logger.py @@ -577,6 +577,32 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac """ pass + def handle_callback_failure(self, callback_name: str): + """ + Handle callback logging failures by incrementing Prometheus metrics. + + Call this method in exception handlers within your callback when logging fails. + """ + try: + import litellm + from litellm._logging import verbose_logger + + all_callbacks = litellm.logging_callback_manager._get_all_callbacks() + + for callback_obj in all_callbacks: + if hasattr(callback_obj, 'increment_callback_logging_failure'): + verbose_logger.debug(f"Incrementing callback failure metric for {callback_name}") + callback_obj.increment_callback_logging_failure(callback_name=callback_name) # type: ignore + return + + verbose_logger.debug( + f"No callback with increment_callback_logging_failure method found for {callback_name}. " + "Ensure 'prometheus' is in your callbacks config." + ) + + except Exception as e: + from litellm._logging import verbose_logger + verbose_logger.debug(f"Error in handle_callback_failure for {callback_name}: {str(e)}") async def _strip_base64_from_messages( self, payload: "StandardLoggingPayload", max_depth: int = DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER diff --git a/litellm/integrations/s3_v2.py b/litellm/integrations/s3_v2.py index 37fe4fbb81..fcfdd112a5 100644 --- a/litellm/integrations/s3_v2.py +++ b/litellm/integrations/s3_v2.py @@ -254,7 +254,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): ) except Exception as e: verbose_logger.exception(f"s3 Layer Error - {str(e)}") - pass + self.handle_callback_failure(callback_name="S3Logger") async def async_upload_data_to_s3( self, batch_logging_element: s3BatchLoggingElement @@ -341,6 +341,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): response.raise_for_status() except Exception as e: verbose_logger.exception(f"Error uploading to s3: {str(e)}") + self.handle_callback_failure(callback_name="S3Logger") async def async_send_batch(self): """ @@ -495,6 +496,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): response.raise_for_status() except Exception as e: verbose_logger.exception(f"Error uploading to s3: {str(e)}") + self.handle_callback_failure(callback_name="S3Logger") async def _download_object_from_s3(self, s3_object_key: str) -> Optional[dict]: """ diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index fac57e038f..bcd7379429 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -1,6 +1,7 @@ # What is this? ## Common Utility file for Logging handler # Logging function -> log the exact model details + what's being sent | Non-Blocking +import asyncio import copy import datetime import json @@ -119,6 +120,7 @@ from litellm.types.utils import ( Usage, ) from litellm.types.videos.main import VideoObject +from litellm.types.containers.main import ContainerObject from litellm.utils import _get_base_model_from_metadata, executor, print_verbose from ..integrations.argilla import ArgillaLogger @@ -2162,6 +2164,11 @@ class Logging(LiteLLMLoggingBaseClass): ) if capture_exception: # log this error to sentry for debugging capture_exception(e) + # Track callback logging failures in Prometheus + try: + self._handle_callback_failure(callback=callback) + except Exception: + pass except Exception as e: verbose_logger.exception( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while success logging {}".format( @@ -2467,8 +2474,33 @@ class Logging(LiteLLMLoggingBaseClass): verbose_logger.error( f"LiteLLM.LoggingError: [Non-Blocking] Exception occurred while success logging {traceback.format_exc()}" ) + self._handle_callback_failure(callback=callback) pass + def _handle_callback_failure(self, callback: Any): + """ + Handle callback logging failures by incrementing Prometheus metrics. + + Works for both sync and async contexts since Prometheus counter increment is synchronous. + + Args: + callback: The callback that failed + """ + try: + callback_name = self._get_callback_name(callback) + + all_callbacks = litellm.logging_callback_manager._get_all_callbacks() + + for callback_obj in all_callbacks: + if hasattr(callback_obj, 'increment_callback_logging_failure'): + callback_obj.increment_callback_logging_failure(callback_name=callback_name) # type: ignore + break # Only increment once + + except Exception as e: + verbose_logger.debug( + f"Error in _handle_callback_failure: {str(e)}" + ) + def _failure_handler_helper_fn( self, exception, traceback_exception, start_time=None, end_time=None ): @@ -2804,6 +2836,10 @@ class Logging(LiteLLMLoggingBaseClass): str(e), callback ) ) + # Track callback logging failures in Prometheus + asyncio.create_task( + self._handle_callback_failure(callback=callback) + ) def _get_trace_id(self, service_name: Literal["langfuse"]) -> Optional[str]: """ @@ -2936,11 +2972,15 @@ class Logging(LiteLLMLoggingBaseClass): Helper to get the name of a callback function Args: - cb: The callback function/string to get the name of + cb: The callback object/function/string to get the name of Returns: The name of the callback """ + if isinstance(cb, str): + return cb + if hasattr(cb, "__class__"): + return cb.__class__.__name__ if hasattr(cb, "__name__"): return cb.__name__ if hasattr(cb, "__func__"): diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 1cd6333532..d2aab6adf1 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -6359,4 +6359,4 @@ class BaseLLMHTTPHandler: model=model, raw_response=response, logging_obj=logging_obj, - ) + ) \ No newline at end of file diff --git a/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py b/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py index 82f4c4bd29..7e01bda1f2 100644 --- a/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py +++ b/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py @@ -838,3 +838,86 @@ async def test_spend_counter_semantics(mock_prometheus_logger): # ============================================================================== # END SEMANTIC VALIDATION TESTS # ============================================================================== + + +# ============================================================================== +# CALLBACK FAILURE METRICS TESTS +# ============================================================================== + +def test_callback_failure_metric_increments(prometheus_logger): + """ + Test that the callback logging failure metric can be incremented. + + This tests the litellm_callback_logging_failures_metric counter. + """ + # Get initial value + initial_value = 0 + try: + initial_value = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name="S3Logger" + )._value.get() + except Exception: + initial_value = 0 + + # Increment the metric + prometheus_logger.increment_callback_logging_failure(callback_name="S3Logger") + + # Verify it incremented by 1 + current_value = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name="S3Logger" + )._value.get() + + assert current_value == initial_value + 1, \ + f"Expected callback failure metric to increment by 1, got {current_value - initial_value}" + + # Increment again for different callback + prometheus_logger.increment_callback_logging_failure(callback_name="LangFuseLogger") + + langfuse_value = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name="LangFuseLogger" + )._value.get() + + assert langfuse_value == 1, "LangFuseLogger metric should be 1" + + # S3Logger should still be initial + 1 + s3_value = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name="S3Logger" + )._value.get() + assert s3_value == initial_value + 1, "S3Logger metric should not change" + + print(f"✓ Callback failure metric test passed: S3Logger={s3_value}, LangFuseLogger={langfuse_value}") + + +def test_callback_failure_metric_different_callbacks(prometheus_logger): + """ + Test that different callbacks are tracked separately with their own labels. + """ + callbacks_to_test = ["S3Logger", "LangFuseLogger", "DataDogLogger", "CustomCallback"] + + for callback_name in callbacks_to_test: + # Get initial value + initial = 0 + try: + initial = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name=callback_name + )._value.get() + except Exception: + initial = 0 + + # Increment + prometheus_logger.increment_callback_logging_failure(callback_name=callback_name) + + # Verify incremented + current = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name=callback_name + )._value.get() + + assert current == initial + 1, \ + f"{callback_name} should increment by 1" + + print(f"✓ Multiple callback tracking test passed for {len(callbacks_to_test)} callbacks") + + +# ============================================================================== +# END CALLBACK FAILURE METRICS TESTS +# ==============================================================================