fix: OpenTelemetry keep parent spans alive with multiple integrations

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

litellm/litellm_core_utils/litellm_logging.py

@@ -1853,6 +1853,14 @@ class Logging(LiteLLMLoggingBaseClass):
             cache_hit=cache_hit,
             standard_logging_object=kwargs.get("standard_logging_object", None),
         )
+        callbacks: List[Any] = []
+        litellm_params_local = self.model_call_details.get("litellm_params", {})
+        is_async_logging_request = (
+            litellm_params_local.get("acompletion", False) is not True
+            and litellm_params_local.get("aembedding", False) is not True
+            and litellm_params_local.get("aimage_generation", False) is not True
+            and litellm_params_local.get("atranscription", False) is not True
+        )
         try:
             ## BUILD COMPLETE STREAMED RESPONSE
             complete_streaming_response: Optional[
@@ -1915,10 +1923,9 @@ class Logging(LiteLLMLoggingBaseClass):
             self.has_run_logging(event_type="sync_success")
             for callback in callbacks:
                 try:
-                    litellm_params = self.model_call_details.get("litellm_params", {})
                     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:
@@ -2229,22 +2236,7 @@ class Logging(LiteLLMLoggingBaseClass):
                             )
                     if (
                         isinstance(callback, CustomLogger)
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "acompletion", False
-                        )
-                        is not True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "aembedding", False
-                        )
-                        is not True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "aimage_generation", False
-                        )
-                        is not True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "atranscription", False
-                        )
-                        is not True
+                        and is_async_logging_request
                         and self.call_type
                         != CallTypes.pass_through.value  # pass-through endpoints call async_log_success_event
                     ):  # custom logger class
@@ -2272,22 +2264,7 @@ class Logging(LiteLLMLoggingBaseClass):
                             )
                     if (
                         callable(callback) is True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "acompletion", False
-                        )
-                        is not True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "aembedding", False
-                        )
-                        is not True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "aimage_generation", False
-                        )
-                        is not True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "atranscription", False
-                        )
-                        is not True
+                        and is_async_logging_request
                         and customLogger is not None
                     ):  # custom logger functions
                         print_verbose(
@@ -2326,6 +2303,32 @@ class Logging(LiteLLMLoggingBaseClass):
                 ),
             )
 
+        if (is_async_logging_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
     ):
@@ -2386,6 +2389,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:
@@ -2627,6 +2631,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.
@@ -2737,6 +2743,11 @@ class Logging(LiteLLMLoggingBaseClass):
             event_type="sync_failure"
         ):  # prevent double logging
             return
+        litellm_params_local = self.model_call_details.get("litellm_params", {})
+        is_async_logging_request = (
+            litellm_params_local.get("acompletion", False) is not True
+            and litellm_params_local.get("aembedding", False) is not True
+        )
         try:
             start_time, end_time = self._failure_handler_helper_fn(
                 exception=exception,
@@ -2762,10 +2773,9 @@ class Logging(LiteLLMLoggingBaseClass):
             self.has_run_logging(event_type="sync_failure")
             for callback in callbacks:
                 try:
-                    litellm_params = self.model_call_details.get("litellm_params", {})
                     should_run = self.should_run_callback(
                         callback=callback,
-                        litellm_params=litellm_params,
+                        litellm_params=litellm_params_local,
                         event_hook="failure_handler",
                     )
                     if not should_run:
@@ -2818,7 +2828,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_async_logging_request
                     ):  # custom logger functions
                         customLogger.log_event(
                             kwargs=self.model_call_details,
@@ -2830,14 +2842,7 @@ class Logging(LiteLLMLoggingBaseClass):
                         )
                     if (
                         isinstance(callback, CustomLogger)
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "acompletion", False
-                        )
-                        is not True
-                        and self.model_call_details.get("litellm_params", {}).get(
-                            "aembedding", False
-                        )
-                        is not True
+                        and is_async_logging_request
                     ):  # custom logger class
                         callback.log_failure_event(
                             start_time=start_time,
@@ -2922,6 +2927,8 @@ class Logging(LiteLLMLoggingBaseClass):
                     str(e)
                 )
             )
+        if is_async_logging_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
@@ -2987,6 +2994,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.

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."""

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
@@ -161,6 +161,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."""