From 95c1d5b0a6a4e1d36afc779adfd1c8a17dd486cc Mon Sep 17 00:00:00 2001 From: yucheng Date: Sat, 19 Sep 2026 00:33:46 +0000 Subject: [PATCH] feat(otel v2): name and re-root the kept generation under llm_only, widening to the account's widest scope Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../integrations/otel/plumbing/providers.py | 62 +++++++-- .../otel/test_otel_v2_destinations.py | 124 +++++++++++++++++- 2 files changed, 170 insertions(+), 16 deletions(-) diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index c43c49141e5..2c4375ce5f7 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -35,12 +35,13 @@ from opentelemetry.sdk.trace.export import ( from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, ) -from opentelemetry.trace import Span, SpanKind, Status, Tracer +from opentelemetry.trace import Span, SpanContext, SpanKind, Status, Tracer from opentelemetry.util.re import parse_env_headers from opentelemetry.util.types import Attributes, AttributeValue from litellm._logging import verbose_logger from litellm._version import version as litellm_version +from litellm.integrations.otel.mappers.langfuse import LANGFUSE_TRACE_NAME from litellm.integrations.otel.model.config import ExporterOwner, ExporterSpec, OpenTelemetryV2Config from litellm.integrations.otel.model.semconv import ( DB, @@ -380,8 +381,8 @@ _URL_KEYS: Final = frozenset({"http.url", "http.target", "url.full"}) _URL_QUERY_KEY: Final = "url.query" -class _TenantSpanView(ReadableSpan): - """A ``ReadableSpan`` view for one destination, leaving the operator's own span alone.""" +class _SpanView(ReadableSpan): + """A ``ReadableSpan`` view for one exporter, leaving the span every other exporter sees alone.""" def __init__( self, @@ -390,11 +391,12 @@ class _TenantSpanView(ReadableSpan): attributes: Attributes, events: Sequence[Event], status: Status, + parent: SpanContext | None, ) -> None: super().__init__( name=inner.name, context=inner.context, - parent=inner.parent, + parent=parent, resource=resource, attributes=attributes, events=events, @@ -431,6 +433,23 @@ def _in_scope(span: ReadableSpan, scope: "OtelSpanScope") -> bool: return scope == "full" or is_llm_call_span(span) +def _scoped(span: ReadableSpan, scope: "OtelSpanScope") -> ReadableSpan: + """Under ``llm_only`` the model call is the only span the exporter gets, so it goes out as the + trace's root (its parent is the request span that is held back) and, unless the caller named the + trace, its own name doubles as ``langfuse.trace.name`` so Langfuse does not show "Unnamed trace".""" + if scope == "full": + return span + attributes: Final = span.attributes or _NO_ATTRIBUTES + named: Final = ( + attributes + if LANGFUSE_TRACE_NAME in attributes + else MappingProxyType({**attributes, LANGFUSE_TRACE_NAME: span.name}) + ) + if span.parent is None and named is attributes: + return span + return _SpanView(span, span.resource, named, span.events, span.status, parent=None) + + def _guardrail_unreachable(attributes: Mapping[str, AttributeValue]) -> bool: return attributes.get(LiteLLM.GUARDRAIL_STATUS) in _GUARDRAIL_UNREACHABLE_STATUSES @@ -501,7 +520,7 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read return span resource: Final = span.resource.merge(Resource(extra)) if extra else span.resource status: Final = span.status if owned else Status(span.status.status_code) - return _TenantSpanView(span, resource, kept, events, status) + return _SpanView(span, resource, kept, events, status, parent=span.parent) class TenantFanOutSpanProcessor(SpanProcessor): @@ -552,7 +571,7 @@ class TenantFanOutSpanProcessor(SpanProcessor): if processor is None: continue try: - processor.on_end(_for_destination(span, destination)) + processor.on_end(_scoped(_for_destination(span, destination), destination.span_scope)) except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span verbose_logger.debug("OTel V2 fan-out: forwarding to %s failed: %s", destination.endpoint, exc) finally: @@ -779,13 +798,23 @@ class _OverriddenBackendFilter(SpanProcessor): straight through and the operator keeps its copy. ``scope`` narrows what the exporter receives independently of that: under - ``llm_only`` the model-call spans go through and the rest of the tree is held back. + ``llm_only`` the model-call spans go through as trace roots and the rest of the + tree is held back, unless a destination of the request names ``sink``, the account + this exporter writes to, with a wider scope: the fan-out then delivers the rest of + the tree there and the model call keeps its place in it. """ - def __init__(self, inner: SpanProcessor, owner: str | None, scope: "OtelSpanScope" = "full") -> None: + def __init__( + self, + inner: SpanProcessor, + owner: str | None, + scope: "OtelSpanScope" = "full", + sink: _SinkKey | None = None, + ) -> None: self._inner: Final = inner self._owner: Final = owner self._scope: Final = scope + self._sink: Final = sink def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: self._inner.on_start(span, parent_context) @@ -793,7 +822,17 @@ class _OverriddenBackendFilter(SpanProcessor): def on_end(self, span: ReadableSpan) -> None: if self._owner in suppressed_backends() or not _in_scope(span, self._scope): return - self._inner.on_end(span) + self._inner.on_end(_scoped(span, self._account_scope())) + + def _account_scope(self) -> "OtelSpanScope": + if self._scope == "full" or self._sink is None: + return self._scope + shared: Final = tuple( + destination.span_scope + for destination in request_destinations() + if _sink_key(destination.endpoint, destination.headers) == self._sink + ) + return _widest((self._scope, *shared)) def shutdown(self) -> None: self._inner.shutdown() @@ -1093,8 +1132,11 @@ def build_tracer_provider( ) owner = spec.owner.value if tenant_overrides and spec.owner is not None else None scope = _operator_scope(config, spec) + sink = _sink_key(spec.endpoint, parse_headers(spec.headers)) if _exports_to_the_wire(spec) else None provider.add_span_processor( - _OverriddenBackendFilter(processor, owner, scope) if owner is not None or scope != "full" else processor + _OverriddenBackendFilter(processor, owner, scope, sink) + if owner is not None or scope != "full" + else processor ) return provider diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py index f4c06c4647e..17f21f28adf 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py @@ -1559,7 +1559,9 @@ class TestSpanScope: def _same_account_provider(shared, operator_scope): provider = TracerProvider() provider.add_span_processor( - _OverriddenBackendFilter(SimpleSpanProcessor(shared), "langfuse_otel", operator_scope) + _OverriddenBackendFilter( + SimpleSpanProcessor(shared), "langfuse_otel", operator_scope, TestRoutingMode.OPERATOR_SINK + ) ) provider.add_span_processor( TenantFanOutSpanProcessor( @@ -1599,17 +1601,127 @@ class TestSpanScope: assert frozenset(finished) == expected assert len(finished) == len(expected), "the same account received a span twice" - def test_a_kept_generation_still_hangs_off_the_request_trace_with_its_trace_controls(self, monkeypatch): + def test_a_full_team_on_the_operators_llm_only_project_gets_one_whole_tree(self, monkeypatch): + """The operator's exporter writes the model call, the fan-out the rest, and Langfuse + upserts by span id: a re-rooted, self-named generation there would replace the one + parented under the request span and rename the whole trace after itself.""" + self._additive(monkeypatch) + shared = InMemorySpanExporter() + + self._run(self._same_account_provider(shared, "llm_only"), (self._same_account_destination("full"),)) + + whole = {s.name: s for s in shared.get_finished_spans()} + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes + + def test_an_llm_only_team_on_the_operators_llm_only_project_gets_re_rooted_generations(self, monkeypatch): + self._additive(monkeypatch) + shared = InMemorySpanExporter() + + self._run(self._same_account_provider(shared, "llm_only"), (self._same_account_destination("llm_only"),)) + + kept = {s.name: s for s in shared.get_finished_spans()}["chat claude-haiku"] + assert kept.parent is None + assert kept.attributes["langfuse.trace.name"] == "chat claude-haiku" + + def test_a_full_team_on_another_account_does_not_widen_the_operators_llm_only_exporter(self, monkeypatch): + self._additive(monkeypatch) + operator = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor( + _OverriddenBackendFilter( + SimpleSpanProcessor(operator), "langfuse_otel", "llm_only", TestRoutingMode.OPERATOR_SINK + ) + ) + provider.add_span_processor(TenantFanOutSpanProcessor(processor_factory=lambda _d: None)) + + self._run(provider, (LANGFUSE_DEST,)) + + kept = {s.name: s for s in operator.get_finished_spans()}["chat claude-haiku"] + assert names(operator) == LLM_SPANS + assert kept.parent is None + assert kept.attributes["langfuse.trace.name"] == "chat claude-haiku" + + def test_a_built_provider_knows_which_account_its_llm_only_exporter_writes_to(self, monkeypatch): + self._additive(monkeypatch) + shared = InMemorySpanExporter() + monkeypatch.setattr(otel_providers, "_exporter_from_spec", lambda _spec: shared) + config = OpenTelemetryV2Config( + langfuse_span_scope="llm_only", + exporters=[ + ExporterSpec( + kind="otlp_http", + endpoint=TestRoutingMode.OPERATOR_SINK[0], + headers="authorization=Basic op", + owner=ExporterOwner.LANGFUSE_OTEL, + ) + ], + ) + provider = build_tracer_provider(config, use_simple_processor=True) + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda _d: SimpleSpanProcessor(shared), + operator_sinks=operator_sink_scopes(config), + ) + ) + + self._run(provider, (self._same_account_destination("full"),)) + + whole = {s.name: s for s in shared.get_finished_spans()} + assert frozenset(whole) == REQUEST_TREE + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes + + def test_a_kept_generation_becomes_the_root_of_the_request_trace_with_its_trace_controls(self, monkeypatch): self._additive(monkeypatch) operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() self._run(self._operator_provider(operator, tenant), (LLM_ONLY_DEST,)) - root = next(s for s in operator.get_finished_spans() if s.name == "POST /v1/chat/completions") + full = {s.name: s for s in operator.get_finished_spans()} kept = {s.name: s for s in tenant.get_finished_spans()}["chat gpt-4"] - assert kept.context.trace_id == root.context.trace_id - assert kept.parent is not None and kept.parent.span_id == root.context.span_id, "no reparenting" - assert {k: kept.attributes[k] for k in TRACE_CONTROLS} == dict(TRACE_CONTROLS) + assert kept.context == full["chat gpt-4"].context, "same trace id and span id as the operator's copy" + assert kept.parent is None, "its parent is the request span the tenant never receives" + assert {k: kept.attributes[k] for k in TRACE_CONTROLS} == dict(TRACE_CONTROLS), "the caller's trace name wins" + assert full["chat gpt-4"].parent == full["POST /v1/chat/completions"].context, ( + "the operator's copy is untouched" + ) + + def test_a_kept_generation_with_no_trace_name_is_named_after_itself(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant, scope="llm_only"), (LLM_ONLY_DEST,)) + + for exporter in (operator, tenant): + kept = {s.name: s for s in exporter.get_finished_spans()}["chat claude-haiku"] + assert kept.parent is None + assert kept.attributes["langfuse.trace.name"] == "chat claude-haiku" + assert kept.attributes["gen_ai.request.model"] == "claude-haiku", "the rest of the attributes stay" + + def test_narrowing_one_exporter_leaves_the_other_exporters_view_of_the_span_alone(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant, scope="llm_only"), (LANGFUSE_DEST,)) + + whole = {s.name: s for s in tenant.get_finished_spans()} + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes + narrowed = {s.name: s for s in operator.get_finished_spans()}["chat claude-haiku"] + assert narrowed.parent is None + assert narrowed.attributes["langfuse.trace.name"] == "chat claude-haiku" + + def test_a_full_scope_exporter_gets_the_generation_under_its_request_span_and_unnamed(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant), (LANGFUSE_DEST,)) + + for exporter in (operator, tenant): + whole = {s.name: s for s in exporter.get_finished_spans()} + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes def test_a_non_langfuse_destination_of_the_same_request_keeps_the_full_tree(self, monkeypatch): self._additive(monkeypatch)