Add Prometheus metric to track callback logging failures in S3 (#16209)

* Add v1 cut of container api

* fix lint errors

* Add proxy support to container apis & logging support (#16049)

* Add proxy support to container apis

* Add logging support

* Add cost tracking support for containers and documentation

* Add new constant documentation

* Add container cost in model map

* fix failing azure tests

* Update tests based on model map changes

* fix model map tests

* fix model map tests

* Container modeshould be container

* Container tests fix

* Merge branch 'main' into litellm_sameer_oct_staging_2

* Add Prometheus metric to track callback logging failures in S3 (#16102)

* Add proxy support to container apis

* Add logging support

* prometheus metric  measures how often s3_v2 is failing

* remove not needed files

* remove not needed files

* remove not needed files

* fix mypy errors

* Use logging_callback_manager to get all the callbacks

---------

Co-authored-by: Ishaan Jaffer <ishaanjaffer0324@gmail.com>

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
     ):

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

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]:
         """

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__"):

litellm/llms/custom_httpx/llm_http_handler.py

@@ -6359,4 +6359,4 @@ class BaseLLMHTTPHandler:
             model=model,
             raw_response=response,
             logging_obj=logging_obj,
-        )
+        )

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
+# ==============================================================================