From 74c992758950412672d515f43715597137ea658d Mon Sep 17 00:00:00 2001 From: Vivek Raj Date: Thu, 29 Oct 2026 08:51:57 +0530 Subject: [PATCH] fix(otel): key the success/failure span dedupe marker by litellm_call_id The once-per-request marker lives in litellm_params["metadata"], which is the caller's dict. Concurrent acompletion() calls that share one metadata dict therefore shared the marker, and every call after the first skipped its litellm_request span (1 of 20 emitted under asyncio.gather). Include litellm_call_id in the success and failure scopes so each request dedupes only its own sync/async dual-fire. --- litellm/integrations/opentelemetry.py | 4 +- tests/unit/integrations/test_opentelemetry.py | 2 +- .../test_opentelemetry_request_spans.py | 41 +++++++++++++++++++ 3 files changed, 44 insertions(+), 3 deletions(-) diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index 0d0e309034a..b2d53c05923 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -1405,7 +1405,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): # request (notably in streaming code paths). Guard against duplicate # span writes — but still close the proxy span on the skip path so # the trace doesn't leak an open root span. - if not self._emit_once(kwargs, "success"): + if not self._emit_once(kwargs, kwargs.get("litellm_call_id"), "success"): verbose_logger.debug( "OpenTelemetry: skipping duplicate success span for handler=%s", self.__class__.__name__, @@ -2191,7 +2191,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): # ``async_post_call_failure_hook`` can append the # "Failed Proxy Server Request" child span before closing it — # there is no proxy-span side-effect to preserve on the skip path. - if not self._emit_once(kwargs, "failure"): + if not self._emit_once(kwargs, kwargs.get("litellm_call_id"), "failure"): verbose_logger.debug( "OpenTelemetry: skipping duplicate failure span for handler=%s", self.__class__.__name__, diff --git a/tests/unit/integrations/test_opentelemetry.py b/tests/unit/integrations/test_opentelemetry.py index f0fca94c370..00523020492 100644 --- a/tests/unit/integrations/test_opentelemetry.py +++ b/tests/unit/integrations/test_opentelemetry.py @@ -5293,7 +5293,7 @@ class TestOpenTelemetrySpanDedupe(unittest.TestCase): kwargs = self._build_kwargs() kwargs["litellm_params"]["metadata"]["litellm_parent_otel_span"] = proxy_span - otel._emit_once(kwargs, "success") # pre-mark to force dedupe-skip branch + otel._emit_once(kwargs, kwargs.get("litellm_call_id"), "success") # pre-mark to force dedupe-skip branch self.assertTrue(proxy_span.is_recording()) start = datetime.utcnow() diff --git a/tests/unit/integrations/test_opentelemetry_request_spans.py b/tests/unit/integrations/test_opentelemetry_request_spans.py index 9ec40c00cb4..e72b53ba411 100644 --- a/tests/unit/integrations/test_opentelemetry_request_spans.py +++ b/tests/unit/integrations/test_opentelemetry_request_spans.py @@ -11,7 +11,9 @@ from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter import litellm +from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER _OPENAI_URL: Final = "https://api.openai.com/v1/chat/completions" _EXPECTED_SPAN_NAMES: Final = ("litellm_request", "raw_gen_ai_request") @@ -155,3 +157,42 @@ async def test_otel_callback_emits_the_request_and_raw_provider_spans( raw_attributes: Final = spans["raw_gen_ai_request"].attributes or {} expected_raw: Final = _RAW_STREAMING_ATTRIBUTES if streaming else _RAW_NON_STREAMING_ATTRIBUTES assert all(raw_attributes.get(name) is not None for name in expected_raw) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("logger_class", [OpenTelemetry, LangfuseOtelLogger]) +async def test_concurrent_calls_sharing_one_metadata_dict_each_emit_their_request_span( + logger_class: type[OpenTelemetry], monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("OTEL_SEMCONV_STABILITY_OPT_IN", raising=False) + exporter: Final = InMemorySpanExporter() + tracer_provider: Final = TracerProvider() + tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr( + litellm, + "callbacks", + [logger_class(config=OpenTelemetryConfig(exporter=exporter), tracer_provider=tracer_provider)], + ) + shared_metadata: Final = {"session_id": "batch"} + users: Final = tuple(f"{_USER}_{index}" for index in range(3)) + + await asyncio.gather( + *( + litellm.acompletion( + model="gpt-4.1-mini", + messages=[{"role": "user", "content": "hi"}], + mock_response="hello", + user=user, + metadata=shared_metadata, + ) + for user in users + ) + ) + await GLOBAL_LOGGING_WORKER.flush() + + request_users: Final = sorted( + str((span.attributes or {}).get("llm.user")) + for span in exporter.get_finished_spans() + if span.name == "litellm_request" + ) + assert request_users == sorted(users)