scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-24 00:52:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 91

feat(otel v2): name and re-root the kept generation under llm_only, widening to the account's widest scope

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-19 00:33:46 +00:00
parent c5cf32b49d
commit 95c1d5b0a6
2 changed files with 170 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

62
litellm/integrations/otel/plumbing/providers.py
View file

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

124
tests/test_litellm/integrations/otel/test_otel_v2_destinations.py
View file

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