From 87b3db2a215c48e52b40cee3a4285757a8ef13c6 Mon Sep 17 00:00:00 2001 From: Yuta Saito Date: Thu, 8 Jan 2026 09:33:05 +0900 Subject: [PATCH] fix: OpenTelemetry keep parent spans alive with multiple integrations --- litellm/integrations/opentelemetry.py | 58 +++++++----- litellm/litellm_core_utils/litellm_logging.py | 42 ++++++++- .../integrations/test_opentelemetry.py | 1 + .../test_litellm_logging.py | 93 ++++++++++++++++++- 4 files changed, 166 insertions(+), 28 deletions(-) diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index a223925d59a..d708c4f73b1 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast import litellm from litellm._logging import verbose_logger from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.core_helpers import _get_parent_otel_span_from_kwargs from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.secret_managers.main import get_secret_bool from litellm.types.services import ServiceLoggerPayload @@ -270,7 +271,6 @@ class OpenTelemetry(CustomLogger): def create_tracer_provider(): provider = TracerProvider(resource=self._get_litellm_resource(self.config)) - provider.add_span_processor(self._get_span_processor()) return provider tracer_provider = self._get_or_create_provider( @@ -281,6 +281,7 @@ class OpenTelemetry(CustomLogger): create_new_provider_fn=create_tracer_provider, set_provider_fn=trace.set_tracer_provider, ) + tracer_provider.add_span_processor(self._get_span_processor()) # Grab our tracer from the TracerProvider (not from global context) # This ensures we use the provided TracerProvider (e.g., for testing) @@ -306,7 +307,7 @@ class OpenTelemetry(CustomLogger): metric_readers=[metric_reader], resource=self._get_litellm_resource(self.config), ) - + meter_provider = self._get_or_create_provider( provider=meter_provider, provider_name="MeterProvider", @@ -358,17 +359,12 @@ class OpenTelemetry(CustomLogger): from opentelemetry.sdk._logs import LoggerProvider as OTLoggerProvider from opentelemetry.sdk._logs.export import BatchLogRecordProcessor - def create_logger_provider(): - provider = OTLoggerProvider( - resource=self._get_litellm_resource(self.config) - ) - log_exporter = self._get_log_exporter() - provider.add_log_record_processor( - BatchLogRecordProcessor(log_exporter) # type: ignore[arg-type] - ) - return provider + log_exporter = self._get_log_exporter() - self._get_or_create_provider( + def create_logger_provider(): + return OTLoggerProvider(resource=self._get_litellm_resource(self.config)) + + logger_provider = self._get_or_create_provider( provider=logger_provider, provider_name="LoggerProvider", get_existing_provider_fn=get_logger_provider, @@ -376,6 +372,10 @@ class OpenTelemetry(CustomLogger): create_new_provider_fn=create_logger_provider, set_provider_fn=set_logger_provider, ) + logger_provider.add_log_record_processor( + BatchLogRecordProcessor(log_exporter) # type: ignore[arg-type] + ) + def log_success_event(self, kwargs, response_obj, start_time, end_time): self._handle_success(kwargs, response_obj, start_time, end_time) @@ -688,12 +688,6 @@ class OpenTelemetry(CustomLogger): # 6. Do NOT end parent span - it should be managed by its creator # External spans (from Langfuse, user code, HTTP headers, global context) must not be closed by LiteLLM - # However, proxy-created spans should be closed here - if ( - parent_span is not None - and parent_span.name == LITELLM_PROXY_REQUEST_SPAN_NAME - ): - parent_span.end(end_time=self._to_ns(end_time)) def _start_primary_span( self, @@ -1176,12 +1170,6 @@ class OpenTelemetry(CustomLogger): # Do NOT end parent span - it should be managed by its creator # External spans (from Langfuse, user code, HTTP headers, global context) must not be closed by LiteLLM - # However, proxy-created spans should be closed here - if ( - parent_otel_span is not None - and parent_otel_span.name == LITELLM_PROXY_REQUEST_SPAN_NAME - ): - parent_otel_span.end(end_time=self._to_ns(end_time)) def _record_exception_on_span(self, span: Span, kwargs: dict): """ @@ -2246,3 +2234,25 @@ class OpenTelemetry(CustomLogger): context=self.get_traceparent_from_header(headers=headers), kind=self.span_kind.SERVER, ) + + def close_litellm_proxy_request_span( + self, + kwargs: Optional[dict], + end_time: datetime, + ) -> None: + """End the proxy-created parent span after all callbacks finish.""" + + if kwargs is None: + return + + try: + parent_span = _get_parent_otel_span_from_kwargs(kwargs) + if parent_span is None: + return + if getattr(parent_span, "name", None) != LITELLM_PROXY_REQUEST_SPAN_NAME: + return + parent_span.end(end_time=self._to_ns(end_time)) + except Exception as exc: + verbose_logger.debug( + "OpenTelemetry: Failed to end proxy parent span: %s", str(exc) + ) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 7d036615f59..1bdc0574d5e 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -1853,6 +1853,8 @@ class Logging(LiteLLMLoggingBaseClass): cache_hit=cache_hit, standard_logging_object=kwargs.get("standard_logging_object", None), ) + + callbacks: List[Any] = [] litellm_params = self.model_call_details.get("litellm_params", {}) is_sync_request = ( litellm_params.get(CallTypes.acompletion.value, False) is not True @@ -1925,7 +1927,7 @@ class Logging(LiteLLMLoggingBaseClass): try: should_run = self.should_run_callback( callback=callback, - litellm_params=litellm_params, + litellm_params=litellm_params_local, event_hook="success_handler", ) if not should_run: @@ -2288,6 +2290,32 @@ class Logging(LiteLLMLoggingBaseClass): ), ) + if (is_sync_request + and self.call_type + != CallTypes.pass_through.value # pass-through endpoints call async_log_success_event + ): + self._finalize_proxy_parent_span(end_time=end_time) + + def _finalize_proxy_parent_span(self, end_time: Optional[dt_object]) -> None: + """Close the proxy-created parent span via the proxy OpenTelemetry logger.""" + + if end_time is None: + return + + from litellm.proxy.proxy_server import open_telemetry_logger + + if open_telemetry_logger is not None: + try: + open_telemetry_logger.close_litellm_proxy_request_span( + kwargs=self.model_call_details, + end_time=end_time, + ) + except Exception: + verbose_logger.debug( + "LiteLLM.Logging: Unable to close proxy parent span via OpenTelemetry", + exc_info=True, + ) + async def async_success_handler( # noqa: PLR0915 self, result=None, start_time=None, end_time=None, cache_hit=None, **kwargs ): @@ -2348,6 +2376,7 @@ class Logging(LiteLLMLoggingBaseClass): cache_hit=cache_hit, standard_logging_object=kwargs.get("standard_logging_object", None), ) + callbacks: List[Any] = [] ## BUILD COMPLETE STREAMED RESPONSE if "async_complete_streaming_response" in self.model_call_details: @@ -2589,6 +2618,8 @@ class Logging(LiteLLMLoggingBaseClass): self._handle_callback_failure(callback=callback) pass + self._finalize_proxy_parent_span(end_time=end_time) + def _handle_callback_failure(self, callback: Any): """ Handle callback logging failures by incrementing Prometheus metrics. @@ -2699,6 +2730,7 @@ class Logging(LiteLLMLoggingBaseClass): event_type="sync_failure" ): # prevent double logging return + litellm_params = self.model_call_details.get("litellm_params", {}) is_sync_request = ( litellm_params.get(CallTypes.acompletion.value, False) is not True @@ -2788,7 +2820,9 @@ class Logging(LiteLLMLoggingBaseClass): print_verbose=print_verbose, ) if ( - callable(callback) and customLogger is not None + callable(callback) + and customLogger is not None + and is_sync_request ): # custom logger functions customLogger.log_event( kwargs=self.model_call_details, @@ -2885,6 +2919,8 @@ class Logging(LiteLLMLoggingBaseClass): str(e) ) ) + if is_sync_request: + self._finalize_proxy_parent_span(end_time=end_time) async def async_failure_handler( self, exception, traceback_exception, start_time=None, end_time=None @@ -2950,6 +2986,8 @@ class Logging(LiteLLMLoggingBaseClass): # Track callback logging failures in Prometheus self._handle_callback_failure(callback=callback) + self._finalize_proxy_parent_span(end_time=end_time) + def _get_trace_id(self, service_name: Literal["langfuse"]) -> Optional[str]: """ For the given service (e.g. langfuse), return the trace_id actually logged. diff --git a/tests/test_litellm/integrations/test_opentelemetry.py b/tests/test_litellm/integrations/test_opentelemetry.py index 95fa6ed8f60..f35f96c9a03 100644 --- a/tests/test_litellm/integrations/test_opentelemetry.py +++ b/tests/test_litellm/integrations/test_opentelemetry.py @@ -266,6 +266,7 @@ class TestOpenTelemetry(unittest.TestCase): self.assertEqual(config.deployment_environment, "production") self.assertEqual(config.model_id, "litellm") + @patch.dict(os.environ, {}, clear=True) def test_open_telemetry_config_custom_service_name(self): """Model ID should inherit provided service name when not explicitly set.""" diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py index a4d3206fdc7..f525cf6f2fe 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py @@ -1,6 +1,8 @@ import json import os import sys +import time +from datetime import datetime from unittest.mock import MagicMock, patch import pytest @@ -9,8 +11,6 @@ sys.path.insert( 0, os.path.abspath("../../..") ) # Adds the parent directory to the system path -import time - from litellm.constants import SENTRY_DENYLIST, SENTRY_PII_DENYLIST from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging from litellm.litellm_core_utils.litellm_logging import set_callbacks @@ -162,6 +162,95 @@ def test_logging_prevent_double_logging(logging_obj): assert logging_obj.should_run_logging(event_type="async_failure") == True +def _set_base_model_call_details(logging_obj, litellm_params=None): + logging_obj.model_call_details = { + "litellm_params": litellm_params or {}, + "messages": [], + "input": [], + "optional_params": {}, + "model": logging_obj.model, + } + logging_obj.sync_streaming_chunks = [] + logging_obj.dynamic_success_callbacks = [] + + +def test_success_handler_finalizes_parent_span_for_async_requests(logging_obj): + logging_obj.stream = False + _set_base_model_call_details(logging_obj, litellm_params={}) + start = datetime.utcnow() + result_payload = {"id": "resp"} + + with patch.object( + LitellmLogging, + "_success_handler_helper_fn", + return_value=(start, start, result_payload), + ), patch.object( + LitellmLogging, "_get_assembled_streaming_response", return_value=None + ), patch.object( + LitellmLogging, "get_combined_callback_list", return_value=[] + ), patch.object( + LitellmLogging, "_finalize_proxy_parent_span" + ) as finalize_mock: + logging_obj.success_handler(result=result_payload) + finalize_mock.assert_called_once_with(end_time=start) + + +def test_success_handler_skips_parent_span_for_override_requests(logging_obj): + logging_obj.stream = False + _set_base_model_call_details(logging_obj, litellm_params={"acompletion": True}) + start = datetime.utcnow() + + with patch.object( + LitellmLogging, + "_success_handler_helper_fn", + return_value=(start, start, {}), + ), patch.object( + LitellmLogging, "_get_assembled_streaming_response", return_value=None + ), patch.object( + LitellmLogging, "get_combined_callback_list", return_value=[] + ), patch.object( + LitellmLogging, "_finalize_proxy_parent_span" + ) as finalize_mock: + logging_obj.success_handler(result={}) + finalize_mock.assert_not_called() + + +def test_failure_handler_finalizes_parent_span_for_async_requests(logging_obj): + _set_base_model_call_details(logging_obj, litellm_params={}) + start = datetime.utcnow() + exception = Exception("boom") + + with patch.object( + LitellmLogging, + "_failure_handler_helper_fn", + return_value=(start, start), + ), patch.object( + LitellmLogging, "get_combined_callback_list", return_value=[] + ), patch.object( + LitellmLogging, "_finalize_proxy_parent_span" + ) as finalize_mock: + logging_obj.failure_handler(exception=exception, traceback_exception="trace") + finalize_mock.assert_called_once_with(end_time=start) + + +def test_failure_handler_skips_parent_span_for_override_requests(logging_obj): + _set_base_model_call_details(logging_obj, litellm_params={"aembedding": True}) + start = datetime.utcnow() + exception = Exception("boom") + + with patch.object( + LitellmLogging, + "_failure_handler_helper_fn", + return_value=(start, start), + ), patch.object( + LitellmLogging, "get_combined_callback_list", return_value=[] + ), patch.object( + LitellmLogging, "_finalize_proxy_parent_span" + ) as finalize_mock: + logging_obj.failure_handler(exception=exception, traceback_exception="trace") + finalize_mock.assert_not_called() + + @pytest.mark.asyncio async def test_datadog_logger_not_shadowed_by_llm_obs(monkeypatch): """Ensure DataDog logger instantiates even when LLM Obs logger already cached."""