From 10659e725f3c31820ab135020722d03ac29cec1c Mon Sep 17 00:00:00 2001 From: Michael Riad Zaky Date: Fri, 1 May 2026 17:50:39 -0700 Subject: [PATCH] isolate dual OTEL handlers --- litellm/integrations/opentelemetry.py | 53 +++-- .../integrations/test_opentelemetry.py | 183 ++++++++++++++++++ 2 files changed, 218 insertions(+), 18 deletions(-) diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index b6d91d0b76..77833e5de0 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -69,6 +69,8 @@ class OpenTelemetryConfig: deployment_environment: Optional[str] = None model_id: Optional[str] = None ignore_context_propagation: Optional[bool] = None + # When True, create a private TracerProvider instead of reusing or setting the global one. + skip_set_global: bool = False def __post_init__(self) -> None: # If endpoint is specified but exporter is still the default "console", @@ -259,16 +261,21 @@ class OpenTelemetry(CustomLogger): try: existing_provider = get_existing_provider_fn() - # If a real SDK provider exists (set by another SDK like Langfuse), use it - # This uses a positive check for SDK providers instead of a negative check for proxy providers if isinstance(existing_provider, sdk_provider_class): - verbose_logger.debug( - "OpenTelemetry: Using existing %s: %s", - provider_name, - type(existing_provider).__name__, - ) - provider = existing_provider - # Don't call set_provider to preserve existing context + if skip_set_global: + verbose_logger.debug( + "OpenTelemetry: existing %s found but skip_set_global=True; creating private %s for isolation", + provider_name, + provider_name, + ) + provider = create_new_provider_fn() + else: + verbose_logger.debug( + "OpenTelemetry: Using existing %s: %s", + provider_name, + type(existing_provider).__name__, + ) + provider = existing_provider else: # Default proxy provider or unknown type, create our own verbose_logger.debug("OpenTelemetry: Creating new %s", provider_name) @@ -293,6 +300,12 @@ class OpenTelemetry(CustomLogger): return provider + def _skip_set_global(self) -> bool: + # langfuse_otel relies on the Langfuse SDK's providers; don't overwrite them. + return self.config.skip_set_global or ( + hasattr(self, "callback_name") and self.callback_name == "langfuse_otel" + ) + def _init_tracing(self, tracer_provider): from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider @@ -303,11 +316,6 @@ class OpenTelemetry(CustomLogger): provider.add_span_processor(self._get_span_processor()) return provider - # CRITICAL FIX: For Langfuse OTEL, skip setting global provider to prevent interference - skip_global = ( - hasattr(self, "callback_name") and self.callback_name == "langfuse_otel" - ) - tracer_provider = self._get_or_create_provider( provider=tracer_provider, provider_name="TracerProvider", @@ -315,16 +323,18 @@ class OpenTelemetry(CustomLogger): sdk_provider_class=TracerProvider, create_new_provider_fn=create_tracer_provider, set_provider_fn=trace.set_tracer_provider, - skip_set_global=skip_global, + skip_set_global=self._skip_set_global(), ) # Grab our tracer from the TracerProvider (not from global context) # This ensures we use the provided TracerProvider (e.g., for testing) self.tracer = tracer_provider.get_tracer(LITELLM_TRACER_NAME) + self._tracer_provider = tracer_provider self.span_kind = SpanKind def _init_metrics(self, meter_provider): if not self.config.enable_metrics: + self._meter_provider = None self._operation_duration_histogram = None self._token_usage_histogram = None self._cost_histogram = None @@ -350,7 +360,9 @@ class OpenTelemetry(CustomLogger): sdk_provider_class=MeterProvider, create_new_provider_fn=create_meter_provider, set_provider_fn=metrics.set_meter_provider, + skip_set_global=self._skip_set_global(), ) + self._meter_provider = meter_provider meter = meter_provider.get_meter(__name__) @@ -388,6 +400,7 @@ class OpenTelemetry(CustomLogger): def _init_logs(self, logger_provider): # nothing to do if events disabled if not self.config.enable_events: + self._logger_provider = None return from opentelemetry._logs import get_logger_provider, set_logger_provider @@ -404,13 +417,14 @@ class OpenTelemetry(CustomLogger): ) return provider - self._get_or_create_provider( + self._logger_provider = self._get_or_create_provider( provider=logger_provider, provider_name="LoggerProvider", get_existing_provider_fn=get_logger_provider, sdk_provider_class=OTLoggerProvider, create_new_provider_fn=create_logger_provider, set_provider_fn=set_logger_provider, + skip_set_global=self._skip_set_global(), ) def log_success_event(self, kwargs, response_obj, start_time, end_time): @@ -1073,7 +1087,7 @@ class OpenTelemetry(CustomLogger): # See: https://github.com/open-telemetry/opentelemetry-python/pull/4676 # TODO: Refactor to use the proper OTEL Logs API instead of directly creating SDK LogRecords - from opentelemetry._logs import SeverityNumber, get_logger + from opentelemetry._logs import SeverityNumber try: from opentelemetry.sdk._logs import ( # type: ignore[attr-defined] # OTEL < 1.39.0 @@ -1084,7 +1098,10 @@ class OpenTelemetry(CustomLogger): LogRecord as SdkLogRecord, # type: ignore[attr-defined] # OTEL >= 1.39.0 ) - otel_logger = get_logger(LITELLM_LOGGER_NAME) + # Resolve through the handler's own LoggerProvider (which may be a + # private one when skip_set_global=True) rather than the module-level + # get_logger() which always goes through the global provider. + otel_logger = self._logger_provider.get_logger(LITELLM_LOGGER_NAME) parent_ctx = span.get_span_context() provider = (kwargs.get("litellm_params") or {}).get( diff --git a/tests/test_litellm/integrations/test_opentelemetry.py b/tests/test_litellm/integrations/test_opentelemetry.py index b31bbca889..962806c5b5 100644 --- a/tests/test_litellm/integrations/test_opentelemetry.py +++ b/tests/test_litellm/integrations/test_opentelemetry.py @@ -259,6 +259,189 @@ class TestOpenTelemetryProviderInitialization(unittest.TestCase): ), "Existing LoggerProvider should be respected and not overridden" +class TestOpenTelemetryDualHandlerIsolation(unittest.TestCase): + """Two OpenTelemetry handlers coexisting via skip_set_global=True + must each get their own provider for every signal (tracer/meter/logger).""" + + @staticmethod + def _wire_span_processor(exporter): + """Context manager: while active, the next OpenTelemetry instance + wires its TracerProvider to `exporter`.""" + return patch.object( + OpenTelemetry, + "_get_span_processor", + lambda self, dynamic_headers=None: SimpleSpanProcessor(exporter), + ) + + def test_skip_set_global_creates_isolated_tracer_provider(self): + from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider + + fake_existing = SDKTracerProvider() + own_exporter = InMemorySpanExporter() + cfg = OpenTelemetryConfig( + exporter="console", service_name="iso-test", skip_set_global=True + ) + with ( + patch.object(trace, "get_tracer_provider", return_value=fake_existing), + patch.object(trace, "set_tracer_provider") as mock_set, + self._wire_span_processor(own_exporter), + ): + handler = OpenTelemetry(config=cfg) + + self.assertIsNot(handler._tracer_provider, fake_existing) + mock_set.assert_not_called() + + handler.tracer.start_span("isolation_check").end() + handler._tracer_provider.force_flush(2000) + self.assertEqual( + [s.name for s in own_exporter.get_finished_spans()], + ["isolation_check"], + ) + + def test_skip_set_global_via_callback_name_back_compat(self): + from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider + + fake_existing = SDKTracerProvider() + cfg = OpenTelemetryConfig(exporter="console", service_name="lf-back-compat") + with ( + patch.object(trace, "get_tracer_provider", return_value=fake_existing), + patch.object(trace, "set_tracer_provider"), + self._wire_span_processor(InMemorySpanExporter()), + ): + handler = OpenTelemetry(config=cfg, callback_name="langfuse_otel") + + self.assertIsNot(handler._tracer_provider, fake_existing) + + def test_default_behavior_reuses_existing_sdk_tracer_provider(self): + from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider + + fake_existing = SDKTracerProvider() + with patch.object(trace, "get_tracer_provider", return_value=fake_existing): + handler = OpenTelemetry(config=OpenTelemetryConfig(service_name="shared")) + self.assertIs(handler._tracer_provider, fake_existing) + + def test_skip_set_global_creates_isolated_meter_provider(self): + from opentelemetry import metrics + from opentelemetry.sdk.metrics import MeterProvider as SDKMeterProvider + + fake_existing = SDKMeterProvider() + cfg = OpenTelemetryConfig( + exporter="console", + service_name="meter-iso-test", + enable_metrics=True, + skip_set_global=True, + ) + with ( + patch.object(metrics, "get_meter_provider", return_value=fake_existing), + patch.object(metrics, "set_meter_provider") as mock_set, + self._wire_span_processor(InMemorySpanExporter()), + ): + handler = OpenTelemetry(config=cfg) + + self.assertIsNot(handler._meter_provider, fake_existing) + mock_set.assert_not_called() + + def test_skip_set_global_creates_isolated_logger_provider(self): + from opentelemetry import _logs + from opentelemetry.sdk._logs import LoggerProvider as SDKLoggerProvider + + fake_existing = SDKLoggerProvider() + cfg = OpenTelemetryConfig( + exporter="console", + service_name="logger-iso-test", + enable_events=True, + skip_set_global=True, + ) + with ( + patch.object(_logs, "get_logger_provider", return_value=fake_existing), + patch.object(_logs, "set_logger_provider") as mock_set, + self._wire_span_processor(InMemorySpanExporter()), + ): + handler = OpenTelemetry(config=cfg) + + self.assertIsNot(handler._logger_provider, fake_existing) + mock_set.assert_not_called() + + def test_emitted_logs_route_to_isolated_logger_provider(self): + # End-to-end: emitted logs land in the handler's private LoggerProvider, + # not the global one. Guards against get_logger() bypassing self._logger_provider. + from opentelemetry import _logs + from opentelemetry.sdk._logs import LoggerProvider as SDKLoggerProvider + + global_exporter = InMemoryLogExporter() + fake_existing = SDKLoggerProvider() + fake_existing.add_log_record_processor( + SimpleLogRecordProcessor(global_exporter) + ) + + private_exporter = InMemoryLogExporter() + cfg = OpenTelemetryConfig( + exporter="console", + service_name="logger-emit-test", + enable_events=True, + skip_set_global=True, + ) + with ( + patch.object(_logs, "get_logger_provider", return_value=fake_existing), + patch.object(_logs, "set_logger_provider"), + patch.object( + OpenTelemetry, "_get_log_exporter", return_value=private_exporter + ), + self._wire_span_processor(InMemorySpanExporter()), + ): + handler = OpenTelemetry(config=cfg) + + span = handler.tracer.start_span("emit-test") + handler._emit_semantic_logs( + kwargs={"messages": [{"role": "user", "content": "hi"}]}, + response_obj={"choices": []}, + span=span, + ) + span.end() + handler._logger_provider.force_flush(2000) + + self.assertGreater(len(private_exporter.get_finished_logs()), 0) + self.assertEqual(len(global_exporter.get_finished_logs()), 0) + + def test_two_handlers_each_receive_their_own_spans(self): + # Handler A gets explicit injection (production-ish: claims the global). + exporter_a = InMemorySpanExporter() + provider_a = TracerProvider() + provider_a.add_span_processor(SimpleSpanProcessor(exporter_a)) + handler_a = OpenTelemetry( + config=OpenTelemetryConfig(service_name="handler-a"), + tracer_provider=provider_a, + ) + + # Handler B comes along with the global appearing to be A's provider. + exporter_b = InMemorySpanExporter() + cfg_b = OpenTelemetryConfig( + exporter="console", service_name="handler-b", skip_set_global=True + ) + with ( + patch.object(trace, "get_tracer_provider", return_value=provider_a), + patch.object(trace, "set_tracer_provider"), + self._wire_span_processor(exporter_b), + ): + handler_b = OpenTelemetry(config=cfg_b) + + self.assertIsNot(handler_a._tracer_provider, handler_b._tracer_provider) + + handler_a.tracer.start_span("from_handler_a").end() + handler_b.tracer.start_span("from_handler_b").end() + provider_a.force_flush(2000) + handler_b._tracer_provider.force_flush(2000) + + self.assertEqual( + sorted(s.name for s in exporter_a.get_finished_spans()), + ["from_handler_a"], + ) + self.assertEqual( + sorted(s.name for s in exporter_b.get_finished_spans()), + ["from_handler_b"], + ) + + class TestOpenTelemetry(unittest.TestCase): POLL_INTERVAL = 0.05 POLL_TIMEOUT = 2.0