This commit is contained in:
Vivek Raj 2026-11-05 00:45:55 +00:00 • committed by GitHub
commit 91fbe96be0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 44 additions and 3 deletions

View file

@ -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__,

View file

@ -5298,7 +5298,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()

View file

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