tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-07-14 18:20:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #27018 from BerriAI/litellm_otelDualHandlerSpans

Browse Source 
[Fix] Isolate dual OTEL handlers
This commit is contained in:
Yassin Kortam
2026-05-01 17:59:15 -07:00
committed by GitHub
parent 502ad94538 10659e725f
commit 9ed2fc24bf
2 changed files with 218 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/integrations/opentelemetry.py
+35 -18
View File
@@ -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(
tests/test_litellm/integrations/test_opentelemetry.py
+183
View File
@@ -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