diff --git a/litellm/_service_logger.py b/litellm/_service_logger.py index 3128f02f40..b67d0d8606 100644 --- a/litellm/_service_logger.py +++ b/litellm/_service_logger.py @@ -145,16 +145,19 @@ class ServiceLogging(CustomLogger): event_metadata=event_metadata, ) elif callback == "otel" or isinstance(callback, OpenTelemetry): - from litellm.proxy.proxy_server import open_telemetry_logger + _otel_logger_to_use: Optional[OpenTelemetry] = None + if isinstance(callback, OpenTelemetry): + _otel_logger_to_use = callback + else: + from litellm.proxy.proxy_server import open_telemetry_logger - await self.init_otel_logger_if_none() + if open_telemetry_logger is not None and isinstance( + open_telemetry_logger, OpenTelemetry + ): + _otel_logger_to_use = open_telemetry_logger - if ( - parent_otel_span is not None - and open_telemetry_logger is not None - and isinstance(open_telemetry_logger, OpenTelemetry) - ): - await self.otel_logger.async_service_success_hook( + if _otel_logger_to_use is not None and parent_otel_span is not None: + await _otel_logger_to_use.async_service_success_hook( payload=payload, parent_otel_span=parent_otel_span, start_time=start_time, @@ -253,20 +256,24 @@ class ServiceLogging(CustomLogger): event_metadata=event_metadata, ) elif callback == "otel" or isinstance(callback, OpenTelemetry): - from litellm.proxy.proxy_server import open_telemetry_logger + _otel_logger_to_use: Optional[OpenTelemetry] = None + if isinstance(callback, OpenTelemetry): + _otel_logger_to_use = callback + else: + from litellm.proxy.proxy_server import open_telemetry_logger - await self.init_otel_logger_if_none() + if open_telemetry_logger is not None and isinstance( + open_telemetry_logger, OpenTelemetry + ): + _otel_logger_to_use = open_telemetry_logger if not isinstance(error, str): error = str(error) - if ( - parent_otel_span is not None - and open_telemetry_logger is not None - and isinstance(open_telemetry_logger, OpenTelemetry) - ): - await self.otel_logger.async_service_success_hook( + if _otel_logger_to_use is not None and parent_otel_span is not None: + await _otel_logger_to_use.async_service_failure_hook( payload=payload, + error=error, parent_otel_span=parent_otel_span, start_time=start_time, end_time=end_time, diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 6992ea17cc..08493a0e8e 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -156,7 +156,11 @@ class LangfuseOtelLogger(OpenTelemetry): "arguments": arguments_obj, } transformed_tool_calls.append(langfuse_tool_call) - safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, safe_dumps(transformed_tool_calls)) + safe_set_attribute( + span, + LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, + safe_dumps(transformed_tool_calls), + ) else: output_data = {} if message.get("role"): @@ -164,7 +168,11 @@ class LangfuseOtelLogger(OpenTelemetry): if message.get("content") is not None: output_data["content"] = message.get("content") if output_data: - safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, safe_dumps(output_data)) + safe_set_attribute( + span, + LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, + safe_dumps(output_data), + ) output = response_obj.get("output", []) if output: @@ -175,15 +183,28 @@ class LangfuseOtelLogger(OpenTelemetry): if item_type == "reasoning" and hasattr(item, "summary"): for summary in item.summary: if hasattr(summary, "text"): - output_items_data.append({"role": "reasoning_summary", "content": summary.text}) + output_items_data.append( + { + "role": "reasoning_summary", + "content": summary.text, + } + ) elif item_type == "message": - output_items_data.append({ - "role": getattr(item, "role", "assistant"), - "content": getattr(getattr(item, "content", [{}])[0], "text", "") - }) + output_items_data.append( + { + "role": getattr(item, "role", "assistant"), + "content": getattr( + getattr(item, "content", [{}])[0], "text", "" + ), + } + ) elif item_type == "function_call": arguments_str = getattr(item, "arguments", "{}") - arguments_obj = json.loads(arguments_str) if isinstance(arguments_str, str) else arguments_str + arguments_obj = ( + json.loads(arguments_str) + if isinstance(arguments_str, str) + else arguments_str + ) langfuse_tool_call = { "id": getattr(item, "id", ""), "name": getattr(item, "name", ""), @@ -193,7 +214,11 @@ class LangfuseOtelLogger(OpenTelemetry): } output_items_data.append(langfuse_tool_call) if output_items_data: - safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, safe_dumps(output_items_data)) + safe_set_attribute( + span, + LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, + safe_dumps(output_items_data), + ) @staticmethod def _set_langfuse_specific_attributes(span: Span, kwargs, response_obj): @@ -210,14 +235,22 @@ class LangfuseOtelLogger(OpenTelemetry): langfuse_environment = os.environ.get("LANGFUSE_TRACING_ENVIRONMENT") if langfuse_environment: - safe_set_attribute(span, LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value, langfuse_environment) + safe_set_attribute( + span, + LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value, + langfuse_environment, + ) metadata = LangfuseOtelLogger._extract_langfuse_metadata(kwargs) LangfuseOtelLogger._set_metadata_attributes(span=span, metadata=metadata) messages = kwargs.get("messages") if messages: - safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_INPUT.value, safe_dumps(messages)) + safe_set_attribute( + span, + LangfuseSpanAttributes.OBSERVATION_INPUT.value, + safe_dumps(messages), + ) LangfuseOtelLogger._set_observation_output(span=span, response_obj=response_obj) @@ -319,3 +352,15 @@ class LangfuseOtelLogger(OpenTelemetry): dynamic_headers["Authorization"] = auth_header return dynamic_headers + + async def async_service_success_hook(self, *args, **kwargs): + """ + Langfuse should not receive service success logs. + """ + pass + + async def async_service_failure_hook(self, *args, **kwargs): + """ + Langfuse should not receive service failure logs. + """ + pass diff --git a/tests/test_service_logger_otel.py b/tests/test_service_logger_otel.py new file mode 100644 index 0000000000..5cb21dadea --- /dev/null +++ b/tests/test_service_logger_otel.py @@ -0,0 +1,85 @@ +import os +import sys +import unittest +from unittest.mock import patch, AsyncMock, MagicMock + +# Add the project root to sys.path +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) + +import litellm +from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger +from litellm.integrations.opentelemetry import OpenTelemetry +from litellm.types.services import ServiceTypes +from litellm._service_logger import ServiceLogging + + +class TestServiceLoggerOTEL(unittest.IsolatedAsyncioTestCase): + def setUp(self): + # Reset callbacks before each test + litellm.service_callback = [] + os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-123" + os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-123" + + @patch("litellm.integrations.opentelemetry.OpenTelemetry._init_tracing") + @patch("litellm.integrations.opentelemetry.OpenTelemetry._init_metrics") + @patch("litellm.integrations.opentelemetry.OpenTelemetry._init_logs") + async def test_langfuse_otel_ignores_service_logs( + self, mock_logs, mock_metrics, mock_tracing + ): + """ + Test that LangfuseOtelLogger overrides the service logging hooks with 'pass'. + """ + logger = LangfuseOtelLogger() + + # Verify hooks are overriden + self.assertEqual( + logger.async_service_success_hook.__qualname__, + "LangfuseOtelLogger.async_service_success_hook", + ) + self.assertEqual( + logger.async_service_failure_hook.__qualname__, + "LangfuseOtelLogger.async_service_failure_hook", + ) + + @patch("litellm.integrations.opentelemetry.OpenTelemetry._init_tracing") + @patch("litellm.integrations.opentelemetry.OpenTelemetry._init_metrics") + @patch("litellm.integrations.opentelemetry.OpenTelemetry._init_logs") + async def test_service_logging_shadowing_fix( + self, mock_logs, mock_metrics, mock_tracing + ): + """ + Test the architectural fix: multiple OTEL loggers should receive logs independently. + """ + # 1. Initialize two loggers + langfuse_logger = LangfuseOtelLogger() + otel_logger = OpenTelemetry() + + # 2. Setup service_callback list + litellm.service_callback = [langfuse_logger, otel_logger] + + service_logging = ServiceLogging() + + # 3. Mock the base OpenTelemetry hook + with patch.object( + OpenTelemetry, "async_service_success_hook", new_callable=AsyncMock + ) as mock_base_hook: + # Trigger a service event + await service_logging.async_service_success_hook( + service=ServiceTypes.DB, + call_type="success", + duration=0.1, + parent_otel_span=MagicMock(), + start_time=0.0, + end_time=1.0, + ) + + # The architectural fix ensures we call each correctly. + self.assertEqual( + mock_base_hook.call_count, + 1, + "Generic OTEL logger should have received the log exactly once.", + ) + + +if __name__ == "__main__": + unittest.main()