fix(langfuse_otel): keep other spans and dropped counts as main exports them
Some checks failed
LiteLLM Rust / rust-lint (push) Waiting to run
LiteLLM Rust / rust-test (push) Waiting to run
LiteLLM Rust / rust-wheel (push) Waiting to run
Terraform Modules / fmt, validate, test (aws) (push) Has been cancelled
Terraform Modules / fmt, validate, test (gcp) (push) Has been cancelled

- v1 writes the caller identity only into the room the span's attribute
  limit leaves next to the error attributes a failed call adds, so the
  exported dropped-attribute count matches main
- v2 strips the Langfuse request metadata only from LiteLLM's own spans;
  application, instrumentation and server spans keep every key they set
- a stripped v2 span keeps the dropped attribute, event and link counts
- a config that lists the Langfuse mapper without the langfuse_otel
  exporter also keeps the request metadata away from other exporters
This commit is contained in:
yucheng 2026-10-19 11:14:01 -07:00
parent 0df6eb1739
commit 4fa762baa9
6 changed files with 372 additions and 74 deletions

View file

@ -8,11 +8,12 @@ from typing import TYPE_CHECKING, Any, Final, Optional
from pydantic import TypeAdapter
from litellm._logging import verbose_logger
from litellm.integrations._types.open_inference import ErrorAttributes
from litellm.integrations.arize import _utils
from litellm.integrations.langfuse.langfuse_otel_attributes import (
LangfuseLLMObsOTELAttributes,
)
from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig
from litellm.integrations.opentelemetry import HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE, OpenTelemetry, OpenTelemetryConfig
from litellm.integrations.otel.mappers.langfuse import TRACE_IDENTITY_FIELDS
from litellm.integrations.otel.model.metadata import IDENTITY_METADATA_FIELDS, exported_request_metadata
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
@ -30,6 +31,17 @@ else:
_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[object, object]]] = TypeAdapter(Mapping[object, object])
# The attributes ``OpenTelemetry._record_exception_on_span`` adds to a failed call's span after these.
_ERROR_ATTRIBUTES: Final = frozenset(
{
ErrorAttributes.ERROR_CODE,
HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE,
ErrorAttributes.ERROR_TYPE,
ErrorAttributes.ERROR_MESSAGE,
ErrorAttributes.ERROR_LLM_PROVIDER,
ErrorAttributes.ERROR_STACK_TRACE,
}
)
LANGFUSE_CLOUD_EU_ENDPOINT: Final = "https://cloud.langfuse.com/api/public/otel"
LANGFUSE_CLOUD_US_ENDPOINT: Final = "https://us.cloud.langfuse.com/api/public/otel"
@ -53,12 +65,8 @@ class LangfuseOtelLogger(OpenTelemetry):
"""
from opentelemetry.sdk.trace import ReadableSpan
# The caller identity goes first, so a span that reaches its attribute limit evicts it before any
# other attribute. A span that already carries attributes (a caller's active span) does not get it.
if not (isinstance(span, ReadableSpan) and span.attributes):
LangfuseOtelLogger._set_request_metadata_attributes(
span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs)
)
# A span that already carries attributes (a caller's active span) does not get the caller identity.
fresh: Final = not (isinstance(span, ReadableSpan) and span.attributes)
_utils.set_attributes(span, kwargs, response_obj, LangfuseLLMObsOTELAttributes)
span.set_attribute("langfuse.observation.type", "generation")
@ -66,6 +74,10 @@ class LangfuseOtelLogger(OpenTelemetry):
# Set Langfuse specific attributes
#########################################################
LangfuseOtelLogger._set_langfuse_specific_attributes(span=span, kwargs=kwargs, response_obj=response_obj)
if fresh:
LangfuseOtelLogger._set_request_metadata_attributes(
span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs)
)
@staticmethod
def _extract_langfuse_metadata(kwargs: dict) -> dict:
@ -144,6 +156,10 @@ class LangfuseOtelLogger(OpenTelemetry):
span: Span,
kwargs: Mapping[object, object],
) -> None:
"""Set the caller identity into the room ``span``'s attribute limit leaves next to the error attributes
a failed call adds after it, so the identity never evicts or drops another attribute."""
from litellm.integrations.otel.emitter import carried_keys, span_attribute_limit
raw_standard_logging_object: Final = kwargs.get("standard_logging_object")
if not isinstance(raw_standard_logging_object, dict):
return
@ -154,16 +170,26 @@ class LangfuseOtelLogger(OpenTelemetry):
observation_metadata: Final = exported_request_metadata(
{field: metadata.get(field) for field in IDENTITY_METADATA_FIELDS}
)
if observation_metadata:
_utils.safe_set_attribute(
span,
LangfuseSpanAttributes.OBSERVATION_METADATA.value,
json.dumps(dict(observation_metadata)),
)
trace_prefix: Final = LangfuseSpanAttributes.TRACE_METADATA.value
for field in TRACE_IDENTITY_FIELDS:
if value := observation_metadata.get(field):
_utils.safe_set_attribute(span, f"{trace_prefix}.{field}", value)
identity: Final = (
*(
((LangfuseSpanAttributes.OBSERVATION_METADATA.value, json.dumps(dict(observation_metadata))),)
if observation_metadata
else ()
),
*(
(f"{trace_prefix}.{field}", value)
for field in TRACE_IDENTITY_FIELDS
if (value := observation_metadata.get(field))
),
)
limit: Final = span_attribute_limit(span)
carried: Final = carried_keys(span)
room: Final = (
len(identity) if limit is None else max(limit - len(carried) - len(_ERROR_ATTRIBUTES - carried), 0)
)
for key, value in identity[:room]:
_utils.safe_set_attribute(span, key, value)
@staticmethod
def _set_observation_output(span: Span, response_obj):

View file

@ -74,6 +74,7 @@ from litellm.integrations.otel.plumbing.metrics import (
create_genai_metrics,
)
from litellm.integrations.otel.plumbing.providers import (
LITELLM_TRACER_NAME,
attach_tenant_fan_out,
build_tracer_provider,
get_event_logger,
@ -97,7 +98,6 @@ if TYPE_CHECKING:
StandardLoggingPayload,
)
LITELLM_TRACER_NAME: Final = "litellm"
_published_v2_provider: ApiTracerProvider | None = None

View file

@ -374,6 +374,7 @@ class _DrainPool:
self._backlog -= 1
LITELLM_TRACER_NAME: Final = "litellm"
_NO_ATTRIBUTES: Final[Mapping[str, AttributeValue]] = MappingProxyType({})
_DB_SYSTEM_KEYS: Final = frozenset({DB.SYSTEM_NAME, DB.SYSTEM_LEGACY})
# Keys on a database span that describe the proxy's own datastore: its host, its
@ -460,15 +461,46 @@ def without_message_content(span: ReadableSpan) -> ReadableSpan:
return _SpanView(span, span.resource, kept, span.events, span.status, parent=span.parent)
class _CountedSpanView(_SpanView):
"""A ``_SpanView`` of ``inner`` with other attributes that reports the dropped counts ``inner`` reports."""
def __init__(self, inner: ReadableSpan, attributes: Attributes) -> None:
super().__init__(inner, inner.resource, attributes, inner.events, inner.status, parent=inner.parent)
self._dropped_counts: Final = (inner.dropped_attributes, inner.dropped_events, inner.dropped_links)
@property
def dropped_attributes(self) -> int:
return self._dropped_counts[0]
@property
def dropped_events(self) -> int:
return self._dropped_counts[1]
@property
def dropped_links(self) -> int:
return self._dropped_counts[2]
def _is_litellm_span(span: ReadableSpan) -> bool:
"""Whether LiteLLM's own tracer wrote ``span``, rather than application or instrumentation code."""
scope: Final = span.instrumentation_scope
return scope is not None and scope.name == LITELLM_TRACER_NAME
def without_langfuse_request_metadata(span: ReadableSpan) -> ReadableSpan:
"""``span`` without the request metadata the Langfuse mapper writes for a Langfuse destination."""
"""``span`` without the request metadata the Langfuse mapper writes for a Langfuse destination.
Only LiteLLM's own spans lose it: any other span keeps every key its own code set.
"""
attributes: Final = span.attributes or _NO_ATTRIBUTES
if not any(key in LANGFUSE_REQUEST_METADATA_ATTRIBUTES for key in attributes):
if not (_is_litellm_span(span) and any(key in LANGFUSE_REQUEST_METADATA_ATTRIBUTES for key in attributes)):
return span
kept: Final = MappingProxyType(
{key: value for key, value in attributes.items() if key not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES}
return _CountedSpanView(
span,
MappingProxyType(
{key: value for key, value in attributes.items() if key not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES}
),
)
return _SpanView(span, span.resource, kept, span.events, span.status, parent=span.parent)
def _is_database_span(attributes: Mapping[str, AttributeValue]) -> bool:
@ -627,34 +659,34 @@ def _for_destination(
and the query string of the request URL, which can hold the same key. A
destination whose effective setting disables span content loses the prompt and
response content too, and a destination for any backend but Langfuse loses the
request metadata the Langfuse mapper writes. The span itself stays, so the tenant
still gets the whole trace tree.
request metadata the Langfuse mapper writes on LiteLLM's own spans. The span itself
stays, so the tenant still gets the whole trace tree.
"""
extra: Final = destination.resource_attributes
defaults: Final = destination.resource_defaults
attributes: Final = span.attributes or _NO_ATTRIBUTES
source: Final = (
span if destination.callback_name == ExporterOwner.LANGFUSE_OTEL else without_langfuse_request_metadata(span)
)
attributes: Final = source.attributes or _NO_ATTRIBUTES
database: Final = _is_database_span(attributes)
owned: Final = _is_tenant_owned_span(attributes)
unreachable: Final = _guardrail_unreachable(attributes)
redacted: Final = not destination.captures_content(default_capture)
langfuse: Final = destination.callback_name == ExporterOwner.LANGFUSE_OTEL
kept: Final = MappingProxyType(
{
key: _without_query(key, value)
for key, value in attributes.items()
if _tenant_visible(key, database, owned, unreachable)
and not (redacted and _is_message_content(key))
and (langfuse or key not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES)
if _tenant_visible(key, database, owned, unreachable) and not (redacted and _is_message_content(key))
}
)
recorded: Final = span.events
recorded: Final = source.events
events: Final = tuple(_without_stack_trace(event) for event in recorded) if owned else ()
unchanged: Final = owned and _same_attributes(kept, attributes) and all(a is b for a, b in zip(events, recorded))
if not extra and not defaults and unchanged:
return span
resource: Final = _tenant_resource(span.resource, defaults, extra)
status: Final = span.status if owned else Status(span.status.status_code)
return _SpanView(span, resource, kept, events, status, parent=span.parent)
return source
resource: Final = _tenant_resource(source.resource, defaults, extra)
status: Final = source.status if owned else Status(source.status.status_code)
return _SpanView(source, resource, kept, events, status, parent=source.parent)
class TenantFanOutSpanProcessor(SpanProcessor):
@ -1123,7 +1155,7 @@ class _OverriddenBackendFilter(SpanProcessor):
class _LangfuseRequestMetadataFilter(SpanProcessor):
"""Strip the Langfuse request metadata from every span before ``inner`` exports it."""
"""Strip the Langfuse request metadata from LiteLLM's own spans before ``inner`` exports them."""
def __init__(self, inner: SpanProcessor) -> None:
self._inner: Final = inner
@ -1450,8 +1482,8 @@ def build_tracer_provider(
``config.langfuse_span_scope`` narrows the exporter owned by ``langfuse_otel``
alone; a collector or any other backend in the same config keeps the full tree.
In a config with a ``langfuse_otel`` exporter, every other exporter goes without
the request metadata the Langfuse mapper writes.
In a config that writes the Langfuse vocabulary, every exporter but ``langfuse_otel``'s
goes without the request metadata the Langfuse mapper writes on LiteLLM's own spans.
``strip_message_content`` removes prompt and response content before every
exporter, for a routed provider whose spans can carry content another
destination asked for.
@ -1467,7 +1499,9 @@ def build_tracer_provider(
# ``config._normalize`` guarantees at least one spec (it folds the top-level
# ``exporter``/``endpoint``/``headers`` fields in when ``exporters`` is empty).
langfuse_owned: Final = any(spec.owner is ExporterOwner.LANGFUSE_OTEL for spec in config.exporters)
writes_langfuse: Final = "langfuse" in config.mapper_names or any(
spec.owner is ExporterOwner.LANGFUSE_OTEL for spec in config.exporters
)
for spec in config.exporters:
if spec.requires_headers and not spec.headers:
continue
@ -1475,7 +1509,7 @@ def build_tracer_provider(
spec,
use_simple_processor,
strip_message_content,
strip_langfuse_request_metadata=langfuse_owned and spec.owner is not ExporterOwner.LANGFUSE_OTEL,
strip_langfuse_request_metadata=writes_langfuse and spec.owner is not ExporterOwner.LANGFUSE_OTEL,
)
owner = spec.owner.value if tenant_overrides and spec.owner is not None else None
scope = _operator_scope(config, spec)
@ -1624,7 +1658,7 @@ def _attached_processors(provider: trace.TracerProvider) -> "tuple[SpanProcessor
return tuple(getattr(multi, "_span_processors", ()))
def get_tracer(provider: TracerProvider, name: str = "litellm") -> Tracer:
def get_tracer(provider: TracerProvider, name: str = LITELLM_TRACER_NAME) -> Tracer:
# Stamp the instrumentation scope with the LiteLLM package version so every
# emitted span carries a deterministic ``scope.version`` (the standard OTel
# location for the emitting library's version) for downstream consumers.

View file

@ -1451,7 +1451,7 @@ def _limit_sweep(
return {size: record for size, record in observed.items() if record is not None}
def test_langfuse_otel_identity_is_evicted_before_any_other_attribute_of_a_full_span(
def test_langfuse_otel_identity_never_evicts_or_drops_another_attribute_of_a_full_span(
gateway: Gateway, tmp_path: Path
) -> None:
marker: Final = "lf-limit-" + uuid.uuid4().hex
@ -1488,17 +1488,20 @@ def test_langfuse_otel_identity_is_evicted_before_any_other_attribute_of_a_full_
identity_count: Final = len(_identity_attributes(reference))
assert reference_dropped == 0, reference_dropped
assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference)
kept_by_size: Final[dict[int, tuple[int, int]]] = {} # mutable-ok: filled once per swept span below
for tool_count, (attrs, dropped) in spans.items():
kept = _identity_attributes(attrs)
assert len(kept) == identity_count - min(dropped, identity_count), (tool_count, dropped, sorted(kept))
if dropped <= identity_count:
kept = len(_identity_attributes(attrs))
kept_by_size[tool_count] = (kept, dropped)
if kept:
assert dropped == 0, (tool_count, kept, dropped)
expected = {"openinference.span.kind", "metadata", "llm.model_name", "llm.request.type"}
assert expected <= set(attrs), (tool_count, dropped, sorted(expected - set(attrs)))
assert expected <= set(attrs), (tool_count, sorted(expected - set(attrs)))
tools = {f"llm.tools.{n}.name" for n in range(tool_count)}
assert tools <= set(attrs), (tool_count, dropped, sorted(tools - set(attrs)))
assert [tool_count for tool_count, (_, dropped) in spans.items() if 0 < dropped <= identity_count], {
tool_count: dropped for tool_count, (_, dropped) in spans.items()
}
assert tools <= set(attrs), (tool_count, sorted(tools - set(attrs)))
assert [size for size, (kept, dropped) in kept_by_size.items() if kept < identity_count and not dropped], (
kept_by_size
)
assert [size for size, (_, dropped) in kept_by_size.items() if dropped], kept_by_size
def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: Gateway, tmp_path: Path) -> None:
@ -1550,3 +1553,168 @@ def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: G
size for size, (shed_identity, shed_messages) in shed_by_size.items() if shed_identity and not shed_messages
], shed_by_size
assert [size for size, (_, shed_messages) in shed_by_size.items() if shed_messages], shed_by_size
_APP_SPAN_HOOK: Final = """
from opentelemetry import trace
from litellm.integrations.custom_logger import CustomLogger
class AppSpanHook(CustomLogger):
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
with trace.get_tracer("lit8283-app").start_as_current_span("app.audit") as span:
span.set_attributes(
{
"app.response_id": response_obj.id,
"langfuse.observation.metadata": '{"source": "app"}',
"langfuse.trace.metadata.user_api_key_alias": "app-set-alias",
}
)
instance = AppSpanHook()
"""
def test_otel_v2_span_another_tracer_wrote_keeps_its_langfuse_metadata_keys_at_the_collector(
gateway: Gateway, tmp_path: Path
) -> None:
marker: Final = "lf-v2-app-" + uuid.uuid4().hex
(tmp_path / "lit8283_app_span.py").write_text(_APP_SPAN_HOOK)
with (
wire_server(lambda _request: Reply()) as generic,
_langfuse_rig(
gateway,
tmp_path,
{"callbacks": ["langfuse_otel", "otel", "lit8283_app_span.instance"]},
marker,
overrides={"LITELLM_OTEL_V2": "1", "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url},
) as rig,
rig.candidate.scenario() as scenario,
):
key, identity = _scenario_identity(scenario, rig.provider)
response: Final = rig.candidate.request(
"POST",
"/v1/chat/completions",
{
"model": identity["model"],
"messages": [{"role": "user", "content": marker}],
"cache": {"no-cache": True},
},
key=key,
)
assert response.status_code == 200, response.text
response_id: Final = response.json()["id"]
def app(span: Mapping[str, object]) -> bool:
return span.get("app.response_id") == response_id
def llm(span: Mapping[str, object]) -> bool:
return _response_id(span.get("gen_ai.response.id")) == _response_id(response_id)
spans: Final = eventually(
_watched_spans(generic), lambda seen: any(map(app, seen)) and any(map(llm, seen)), seconds=25
)
app_span: Final = next(span for span in spans if app(span))
assert app_span["langfuse.observation.metadata"] == '{"source": "app"}', app_span
assert app_span["langfuse.trace.metadata.user_api_key_alias"] == "app-set-alias", app_span
collector_llm: Final = next(span for span in spans if llm(span))
assert not _identity_attributes(collector_llm), sorted(collector_llm)
def _chat_span_drops(body: bytes) -> tuple[tuple[int, int, int], ...]:
request: Final = ExportTraceServiceRequest.FromString(body)
return tuple(
(span.dropped_attributes_count, span.dropped_events_count, span.dropped_links_count)
for resource in request.resource_spans
for scope in resource.scope_spans
for span in scope.spans
if span.name.startswith("chat ")
)
def test_otel_v2_collector_keeps_the_dropped_counts_of_a_span_without_langfuse_request_metadata(
gateway: Gateway, tmp_path: Path
) -> None:
marker: Final = "lf-v2-drops-" + uuid.uuid4().hex
with (
wire_server(lambda _request: Reply()) as generic,
_langfuse_rig(
gateway,
tmp_path,
{"callbacks": ["langfuse_otel", "otel"]},
marker,
overrides={
"LITELLM_OTEL_V2": "1",
"OTEL_EXPORTER": "otlp_http",
"OTEL_ENDPOINT": generic.url,
"OTEL_SPAN_EVENT_COUNT_LIMIT": "0",
},
) as rig,
rig.candidate.scenario() as scenario,
):
key, identity = _scenario_identity(scenario, rig.provider)
response: Final = rig.candidate.request(
"POST",
"/v1/chat/completions",
{
"model": identity["model"],
"messages": [{"role": "user", "content": f"{marker} scripted -fail- upstream"}],
"cache": {"no-cache": True},
},
key=key,
)
assert response.status_code >= 500, response.text
drops: Final[dict[str, list[tuple[int, int, int]]]] = {"langfuse": [], "generic": []} # mutable-ok: drained
def observed() -> dict[str, list[tuple[int, int, int]]]:
for name, wire in (("langfuse", rig.collector), ("generic", generic)):
drops[name].extend(drop for batch in wire.drain() for drop in _chat_span_drops(batch.body))
return drops
settled: Final = eventually(
observed, lambda seen: bool(seen["langfuse"]) and len(seen["generic"]) == len(seen["langfuse"]), seconds=25
)
assert any(events for _, events, _ in settled["langfuse"]), settled
assert sorted(settled["generic"]) == sorted(settled["langfuse"]), settled
def test_otel_v2_langfuse_mapper_sends_no_langfuse_request_metadata_to_a_plain_collector(
gateway: Gateway, tmp_path: Path
) -> None:
marker: Final = "lf-v2-mapper-" + uuid.uuid4().hex
with (
_langfuse_rig(
gateway,
tmp_path,
{"callbacks": ["otel"]},
marker,
overrides={
"LITELLM_OTEL_V2": "1",
"MAPPER_NAMES": "genai,langfuse",
"OTEL_EXPORTER": "otlp_http",
"OTEL_ENDPOINT": "__collector__",
},
) as rig,
rig.candidate.scenario() as scenario,
):
key, identity = _scenario_identity(scenario, rig.provider)
response: Final = rig.candidate.request(
"POST",
"/v1/chat/completions",
{
"model": identity["model"],
"messages": [{"role": "user", "content": marker}],
"user": identity["end_user"],
"cache": {"no-cache": True},
},
key=key,
)
assert response.status_code == 200, response.text
collector: Final = _v2_llm_span(rig.collector, response.json()["id"])
assert collector["langfuse.trace.metadata.team_id"] == identity["team_id"], collector
assert not _identity_attributes(collector), sorted(collector)

View file

@ -18,7 +18,7 @@ from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import Status, StatusCode
from opentelemetry.trace import Link, SpanContext, Status, StatusCode
import litellm
from litellm.integrations.custom_logger import CustomLogger
@ -5075,11 +5075,12 @@ _LANGFUSE_REQUEST_METADATA_SPAN: Final = MappingProxyType(
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
"langfuse.trace.metadata.user_api_key_user_id": "user-1",
"langfuse.trace.metadata.team_id": "team-9",
"gen_ai.operation.name": "chat",
"gen_ai.request.model": "gpt-4o",
}
)
_WITHOUT_LANGFUSE_REQUEST_METADATA: Final = MappingProxyType(
{"langfuse.trace.metadata.team_id": "team-9", "gen_ai.request.model": "gpt-4o"}
{"langfuse.trace.metadata.team_id": "team-9", "gen_ai.operation.name": "chat", "gen_ai.request.model": "gpt-4o"}
)
@ -5087,10 +5088,19 @@ class TestLangfuseRequestMetadataStaysWithLangfuse:
"""The request metadata the Langfuse mapper writes reaches a Langfuse destination and no other."""
@staticmethod
def _emit(provider: TracerProvider, destinations: tuple[OtelDestination, ...] = ()) -> None:
def _emit(
provider: TracerProvider,
destinations: tuple[OtelDestination, ...] = (),
tracer_name: str = "litellm",
overflow: bool = False,
) -> None:
def run() -> None:
set_request_destinations(destinations)
with get_tracer(provider, "litellm").start_as_current_span("chat gpt-4o") as span:
links: Final = (Link(SpanContext(trace_id=1, span_id=1, is_remote=True)),) if overflow else ()
with get_tracer(provider, tracer_name).start_as_current_span("chat gpt-4o", links=links) as span:
if overflow:
span.set_attribute("app.evicted", "first")
span.add_event("app.event")
span.set_attributes(_LANGFUSE_REQUEST_METADATA_SPAN)
in_fresh_context(run)
@ -5112,6 +5122,19 @@ class TestLangfuseRequestMetadataStaysWithLangfuse:
)
return build_tracer_provider(config, use_simple_processor=True, tenant_overrides=True), exporters
@staticmethod
def _fan_out() -> tuple[TracerProvider, InMemorySpanExporter, InMemorySpanExporter]:
langfuse_exporter, arize_exporter = InMemorySpanExporter(), InMemorySpanExporter()
provider: Final = TracerProvider()
provider.add_span_processor(
TenantFanOutSpanProcessor(
processor_factory=lambda destination: SimpleSpanProcessor(
langfuse_exporter if destination.callback_name == "langfuse_otel" else arize_exporter
)
)
)
return provider, langfuse_exporter, arize_exporter
def test_a_collector_beside_the_langfuse_exporter_gets_no_langfuse_request_metadata(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
@ -5124,27 +5147,49 @@ class TestLangfuseRequestMetadataStaysWithLangfuse:
)
assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)
def test_a_collector_with_no_langfuse_exporter_keeps_what_its_mappers_write(
def test_a_collector_with_the_langfuse_mapper_and_no_langfuse_exporter_gets_no_langfuse_request_metadata(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
provider, exporters = self._provider(monkeypatch, None)
self._emit(provider)
assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN)
assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)
def test_the_fan_out_forwards_langfuse_request_metadata_to_langfuse_destinations_only(self) -> None:
langfuse_exporter, arize_exporter = InMemorySpanExporter(), InMemorySpanExporter()
provider: Final = TracerProvider()
provider.add_span_processor(
TenantFanOutSpanProcessor(
processor_factory=lambda destination: SimpleSpanProcessor(
langfuse_exporter if destination.callback_name == "langfuse_otel" else arize_exporter
)
)
)
provider, langfuse_exporter, arize_exporter = self._fan_out()
self._emit(provider, (LANGFUSE_DEST, INHERITING_DEST))
assert dict(by_name(langfuse_exporter)["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN)
assert dict(by_name(arize_exporter)["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)
def test_a_span_another_tracer_wrote_keeps_the_langfuse_metadata_keys_it_set(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL)
fanned, _langfuse_exporter, arize_exporter = self._fan_out()
self._emit(provider, tracer_name="app")
self._emit(fanned, (LANGFUSE_DEST, INHERITING_DEST), tracer_name="app")
assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN)
assert dict(by_name(arize_exporter)["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN)
def test_a_span_without_the_langfuse_request_metadata_keeps_its_dropped_counts(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", str(len(_LANGFUSE_REQUEST_METADATA_SPAN)))
monkeypatch.setenv("OTEL_SPAN_EVENT_COUNT_LIMIT", "0")
monkeypatch.setenv("OTEL_SPAN_LINK_COUNT_LIMIT", "0")
provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL)
fanned, _langfuse_exporter, arize_exporter = self._fan_out()
self._emit(provider, overflow=True)
self._emit(fanned, (LANGFUSE_DEST, INHERITING_DEST), overflow=True)
stripped: Final = (by_name(exporters[None])["chat gpt-4o"], by_name(arize_exporter)["chat gpt-4o"])
assert [dict(span.attributes or {}) for span in stripped] == [dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)] * 2
assert [(span.dropped_attributes, span.dropped_events, span.dropped_links) for span in stripped] == [
(1, 1, 1)
] * 2

View file

@ -1,7 +1,9 @@
import base64
import json
import os
from collections.abc import Mapping
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Final
from unittest.mock import MagicMock, patch
@ -700,9 +702,24 @@ class TestLangfuseOtelIntegration:
assert json.loads(actual["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"}
assert json.loads(actual["langfuse.observation.output"]) == {"role": "assistant", "content": "hello"}
# What ``OpenTelemetry._record_exception_on_span`` adds to a failed call's span after its attributes.
_FAILED_CALL_ATTRIBUTES: Final = MappingProxyType(
{
"error.code": "429",
"http.response.status_code": 429,
"error.type": "RateLimitError",
"error.message": "rate limited",
"error.llm_provider": "openai",
"error.stack_trace": "Traceback",
}
)
@staticmethod
def _exported_span(
kwargs: dict[str, object], max_span_attributes: int | None, carried: dict[str, str] | None = None
kwargs: dict[str, object],
max_span_attributes: int | None,
carried: dict[str, str] | None = None,
added_after: Mapping[str, str | int] = MappingProxyType({}),
) -> ReadableSpan:
exporter: Final = InMemorySpanExporter()
provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=max_span_attributes))
@ -710,6 +727,7 @@ class TestLangfuseOtelIntegration:
span: Final = provider.get_tracer("langfuse-otel-test").start_span("litellm_request", attributes=carried)
response_obj: Final = {"id": "chatcmpl-1", "choices": [{"message": {"role": "assistant", "content": "hello"}}]}
LangfuseOtelLogger.set_langfuse_otel_attributes(span, kwargs, response_obj)
span.set_attributes(added_after)
span.end()
(finished,) = exporter.get_finished_spans()
return finished
@ -739,14 +757,16 @@ class TestLangfuseOtelIntegration:
},
}
def test_request_metadata_never_evicts_another_attribute_at_the_span_limit(
def test_request_metadata_never_evicts_or_drops_another_attribute_at_the_span_limit(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
kwargs: Final = self._tool_call_kwargs()
failed: Final = self._FAILED_CALL_ATTRIBUTES
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
plain: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET).attributes or {})
plain: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET, added_after=failed).attributes or {})
succeeded: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET).attributes or {})
monkeypatch.setattr(litellm, "redact_user_api_key_info", False)
full: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET).attributes or {})
full: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET, added_after=failed).attributes or {})
identity_keys: Final = set(full) - set(plain)
assert identity_keys == {
"langfuse.observation.metadata",
@ -754,12 +774,17 @@ class TestLangfuseOtelIntegration:
"langfuse.trace.metadata.user_api_key_team_id",
}
at_limit: Final = self._exported_span(kwargs, len(plain))
assert dict(at_limit.attributes or {}) == plain
one_spare: Final = dict(self._exported_span(kwargs, len(plain) + 1).attributes or {})
assert {key: one_spare.get(key) for key in plain} == plain
assert len(set(one_spare) & identity_keys) == 1
kept_identity: list[int] = [] # mutable-ok: one count per limit
for limit in range(len(plain), len(full) + 1):
failed_span = self._exported_span(kwargs, limit, added_after=failed)
succeeded_span = self._exported_span(kwargs, limit - len(failed))
for span, expected in ((failed_span, plain), (succeeded_span, succeeded)):
attributes = dict(span.attributes or {})
assert span.dropped_attributes == 0
assert {key: attributes.get(key) for key in expected} == expected
assert set(attributes) - set(expected) <= identity_keys
kept_identity.append(len(set(failed_span.attributes or {}) & identity_keys))
assert kept_identity == list(range(len(identity_keys) + 1))
def test_request_metadata_skips_a_span_that_already_carries_attributes(self) -> None:
attributes: Final = dict(