diff --git a/docs/my-website/docs/proxy/prometheus.md b/docs/my-website/docs/proxy/prometheus.md index d5a3466a84..93a0675f09 100644 --- a/docs/my-website/docs/proxy/prometheus.md +++ b/docs/my-website/docs/proxy/prometheus.md @@ -130,7 +130,12 @@ Monitor failures while shipping logs to downstream callbacks like `s3_v3` cold s | Metric Name | Description | |----------------------|--------------------------------------| -| `litellm_callback_logging_failures_metric` | Total number of failed attempts to emit logs to a configured callback. Labels: `"callback_name"`. Use this to alert on callback delivery issues such as repeated failures when writing to `s3_v3`. | +| `litellm_callback_logging_failures_metric` | Total number of failed attempts to emit logs to a configured callback. Labels: `"callback_name"`. Use this to alert on callback delivery issues such as repeated failures when writing to `s3_v3`, `langfuse`, or `langfuse_otel` and other otel providers | + +**Supported Callbacks:** +- `S3Logger` - S3 v2 cold storage failures +- `langfuse` - Langfuse logging failures +- `otel` - OpenTelemetry logging failures ## LLM Provider Metrics diff --git a/litellm/integrations/langfuse/langfuse_prompt_management.py b/litellm/integrations/langfuse/langfuse_prompt_management.py index 8f73eabad4..3986fc6a6e 100644 --- a/litellm/integrations/langfuse/langfuse_prompt_management.py +++ b/litellm/integrations/langfuse/langfuse_prompt_management.py @@ -300,43 +300,59 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge ) async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): - standard_callback_dynamic_params = kwargs.get( - "standard_callback_dynamic_params" - ) - langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request( - globalLangfuseLogger=self, - standard_callback_dynamic_params=standard_callback_dynamic_params, - in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, - ) - langfuse_logger_to_use.log_event_on_langfuse( - kwargs=kwargs, - response_obj=response_obj, - start_time=start_time, - end_time=end_time, - user_id=kwargs.get("user", None), - ) + try: + standard_callback_dynamic_params = kwargs.get( + "standard_callback_dynamic_params" + ) + langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request( + globalLangfuseLogger=self, + standard_callback_dynamic_params=standard_callback_dynamic_params, + in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, + ) + langfuse_logger_to_use.log_event_on_langfuse( + kwargs=kwargs, + response_obj=response_obj, + start_time=start_time, + end_time=end_time, + user_id=kwargs.get("user", None), + ) + except Exception as e: + from litellm._logging import verbose_logger + + verbose_logger.exception( + f"Langfuse Layer Error - Exception occurred while logging success event: {str(e)}" + ) + self.handle_callback_failure(callback_name="langfuse") async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): - standard_callback_dynamic_params = kwargs.get( - "standard_callback_dynamic_params" - ) - langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request( - globalLangfuseLogger=self, - standard_callback_dynamic_params=standard_callback_dynamic_params, - in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, - ) - standard_logging_object = cast( - Optional[StandardLoggingPayload], - kwargs.get("standard_logging_object", None), - ) - if standard_logging_object is None: - return - langfuse_logger_to_use.log_event_on_langfuse( - start_time=start_time, - end_time=end_time, - response_obj=None, - user_id=kwargs.get("user", None), - status_message=standard_logging_object["error_str"], - level="ERROR", - kwargs=kwargs, - ) + try: + standard_callback_dynamic_params = kwargs.get( + "standard_callback_dynamic_params" + ) + langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request( + globalLangfuseLogger=self, + standard_callback_dynamic_params=standard_callback_dynamic_params, + in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, + ) + standard_logging_object = cast( + Optional[StandardLoggingPayload], + kwargs.get("standard_logging_object", None), + ) + if standard_logging_object is None: + return + langfuse_logger_to_use.log_event_on_langfuse( + start_time=start_time, + end_time=end_time, + response_obj=None, + user_id=kwargs.get("user", None), + status_message=standard_logging_object["error_str"], + level="ERROR", + kwargs=kwargs, + ) + except Exception as e: + from litellm._logging import verbose_logger + + verbose_logger.exception( + f"Langfuse Layer Error - Exception occurred while logging failure event: {str(e)}" + ) + self.handle_callback_failure(callback_name="langfuse") diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index a8a7fa77b3..7aea333ee9 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -995,9 +995,13 @@ class OpenTelemetry(CustomLogger): from opentelemetry._logs import SeverityNumber, get_logger, get_logger_provider try: - from opentelemetry.sdk._logs import LogRecord as SdkLogRecord # type: ignore[attr-defined] # OTEL < 1.39.0 + from opentelemetry.sdk._logs import ( + LogRecord as SdkLogRecord, # type: ignore[attr-defined] # OTEL < 1.39.0 + ) except ImportError: - from opentelemetry.sdk._logs._internal import LogRecord as SdkLogRecord # OTEL >= 1.39.0 + from opentelemetry.sdk._logs._internal import ( + LogRecord as SdkLogRecord, # OTEL >= 1.39.0 + ) otel_logger = get_logger(LITELLM_LOGGER_NAME) @@ -1618,6 +1622,7 @@ class OpenTelemetry(CustomLogger): ) except Exception as e: + self.handle_callback_failure(callback_name= self.callback_name) verbose_logger.exception( "OpenTelemetry logging error in set_attributes %s", str(e) ) diff --git a/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py b/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py index 53757390a4..4a08c208fd 100644 --- a/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py +++ b/tests/enterprise/litellm_enterprise/integrations/test_prometheus.py @@ -937,6 +937,124 @@ def test_callback_failure_metric_different_callbacks(prometheus_logger): ) +@pytest.mark.asyncio +async def test_langfuse_callback_failure_metric(prometheus_logger): + """ + Test that Langfuse callback failures are properly tracked in Prometheus metrics. + + This test verifies that when Langfuse logging fails, the + litellm_callback_logging_failures_metric is incremented with callback_name="langfuse". + """ + from unittest.mock import MagicMock, patch + + from litellm.integrations.langfuse.langfuse_prompt_management import ( + LangfusePromptManagement, + ) + + # Get initial value + initial_value = 0 + try: + initial_value = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name="langfuse" + )._value.get() + except Exception: + initial_value = 0 + + # Create Langfuse logger with mocked initialization + with patch("litellm.integrations.langfuse.langfuse_prompt_management.langfuse_client_init"): + langfuse_logger = LangfusePromptManagement() + + # Mock the log_event_on_langfuse to raise an exception + with patch( + "litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler.get_langfuse_logger_for_request" + ) as mock_get_logger: + mock_logger = MagicMock() + mock_logger.log_event_on_langfuse.side_effect = Exception("Langfuse API error") + mock_get_logger.return_value = mock_logger + + # Mock handle_callback_failure to track calls + with patch.object(prometheus_logger, "increment_callback_logging_failure") as mock_increment: + # Inject prometheus logger into the langfuse logger + langfuse_logger.handle_callback_failure = lambda callback_name: mock_increment( + callback_name=callback_name + ) + + # Call async_log_success_event - should catch exception and increment metric + await langfuse_logger.async_log_success_event( + kwargs={}, + response_obj={}, + start_time=None, + end_time=None, + ) + + # Verify that increment was called with correct callback name + mock_increment.assert_called_once_with(callback_name="langfuse") + + print("✓ Langfuse callback failure metric test passed") + + +@pytest.mark.asyncio +async def test_langfuse_otel_callback_failure_metric(prometheus_logger): + """ + Test that Langfuse OTEL callback failures are properly tracked in Prometheus metrics. + + This test verifies that when Langfuse OTEL logging fails, the + litellm_callback_logging_failures_metric is incremented with callback_name="langfuse_otel". + """ + from unittest.mock import MagicMock, patch + + from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger + + # Get initial value + initial_value = 0 + try: + initial_value = prometheus_logger.litellm_callback_logging_failures_metric.labels( + callback_name="langfuse_otel" + )._value.get() + except Exception: + initial_value = 0 + + # Create Langfuse OTEL logger with mocked initialization + with patch("litellm.integrations.opentelemetry.OpenTelemetry.__init__", return_value=None): + langfuse_otel_logger = LangfuseOtelLogger(callback_name="langfuse_otel") + langfuse_otel_logger.callback_name = "langfuse_otel" + + # Mock handle_callback_failure to track calls + with patch.object(prometheus_logger, "increment_callback_logging_failure") as mock_increment: + # Inject prometheus logger into the langfuse otel logger + langfuse_otel_logger.handle_callback_failure = lambda callback_name: mock_increment( + callback_name=callback_name + ) + + # Test that the OpenTelemetry base class set_attributes exception handler works + # This is where langfuse_otel failures are caught and tracked + with patch.object(langfuse_otel_logger, "set_attributes") as mock_set_attributes: + # Simulate the exception handling in set_attributes + def set_attributes_with_error(*args, **kwargs): + # This simulates what happens in the real set_attributes method + try: + raise Exception("Attribute error") + except Exception as e: + langfuse_otel_logger.handle_callback_failure(callback_name=langfuse_otel_logger.callback_name) + + mock_set_attributes.side_effect = set_attributes_with_error + + # Call set_attributes + try: + langfuse_otel_logger.set_attributes( + span=MagicMock(), + kwargs={}, + response_obj={} + ) + except Exception: + pass + + # Verify that increment was called with correct callback name + mock_increment.assert_called_with(callback_name="langfuse_otel") + + print("✓ Langfuse OTEL callback failure metric test passed") + + # ============================================================================== # END CALLBACK FAILURE METRICS TESTS # ==============================================================================