fix(langfuse_otel): drop the caller identity first when a span hits its attribute limit

v1 wrote the identity attributes after the OpenInference ones, so a span near the
128-attribute limit (about 33+ tools) evicted openinference.span.kind, metadata
and llm.model_name instead. The identity now goes first on a fresh span, so the
SDK evicts it before anything else, and a caller's active span that already
carries attributes does not get it.

v2 sheds the Langfuse identity keys before any indexed message attribute when a
span is over budget, so a span keeps every attribute it carried without them.
This commit is contained in:
yucheng 2026-10-19 00:43:19 -07:00
parent c9410b2aeb
commit 3d3e76ce81
6 changed files with 313 additions and 20 deletions

View file

@ -51,7 +51,14 @@ class LangfuseOtelLogger(OpenTelemetry):
Sets OpenTelemetry span attributes for Langfuse observability.
Uses the same attribute setting logic as Arize Phoenix for consistency.
"""
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)
)
_utils.set_attributes(span, kwargs, response_obj, LangfuseLLMObsOTELAttributes)
span.set_attribute("langfuse.observation.type", "generation")
@ -278,7 +285,6 @@ class LangfuseOtelLogger(OpenTelemetry):
metadata: Final = LangfuseOtelLogger._extract_langfuse_metadata(kwargs)
LangfuseOtelLogger._set_metadata_attributes(span=span, metadata=metadata)
LangfuseOtelLogger._set_request_metadata_attributes(span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs))
messages: Final = kwargs.get("messages")
if messages:

View file

@ -13,6 +13,7 @@ from opentelemetry.trace.status import Status, StatusCode
from litellm.integrations.otel.mappers import resolve_mappers
from litellm.integrations.otel.mappers.base import AttributeMapper, AttrValue, SpanData
from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES
from litellm.integrations.otel.mappers.openinference import fit_indexed_messages
from litellm.integrations.otel.model.config import OpenTelemetryV2Config
from litellm.integrations.otel.model.payloads import (
@ -303,7 +304,9 @@ class SpanEmitter:
)
stamped_later: Final = error_attributes(error) if error else _NO_ATTRIBUTES
reserved: Final = len(stamped_later.keys() - mapped.keys())
for key, value in fit_indexed_messages(mapped, attribute_budget(span, reserved, frozenset(mapped))).items():
# The Langfuse request metadata sheds first, so a span at its attribute limit loses nothing else to it.
budget: Final = attribute_budget(span, reserved, frozenset(mapped))
for key, value in fit_indexed_messages(mapped, budget, LANGFUSE_REQUEST_METADATA_ATTRIBUTES).items():
span.set_attribute(key, value)
if error:
stamped: Final = stamp_error(span, error)

View file

@ -153,7 +153,9 @@ def _shed_order(groups: Mapping[tuple[str, int, int], tuple[str, ...]]) -> tuple
_METADATA_KEY: Final = "metadata"
def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) -> Mapping[str, AttrValue]:
def fit_indexed_messages(
attrs: Mapping[str, AttrValue], budget: int | None, shed_first: frozenset[str] = frozenset()
) -> Mapping[str, AttrValue]:
"""``attrs`` with indexed message attributes shed, least valuable first, until at most ``budget`` keys remain.
``None`` means the span has no attribute count limit. Every message still rides the ``input.value`` and
@ -161,12 +163,14 @@ def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) ->
after every indexed message attribute: message attributes are the indexed, queryable view (the blobs
carry no per-index keys), so a count-squeezed span keeps them and the single metadata key absorbs only
the residual shortfall. Shedding happens here, before ``span.set_attribute``, so the fit is exact and
the SDK's dropped-attributes counter never silently masks the choice.
the SDK's dropped-attributes counter never silently masks the choice. ``shed_first`` keys shed one at a
time before anything else, so they never cost the span any other attribute.
"""
if budget is None or len(attrs) <= budget:
return attrs
groups: Final = _message_key_groups(attrs)
sheddable: Final = (*(groups[group] for group in _shed_order(groups)),)
leading: Final = tuple((key,) for key in attrs if key in shed_first)
sheddable: Final = (*leading, *(groups[group] for group in _shed_order(groups)))
flex: Final = (_METADATA_KEY,) if _METADATA_KEY in attrs else ()
candidates: Final = (*sheddable, flex) if flex else sheddable
running: Final = tuple(accumulate(len(candidate) for candidate in candidates))

View file

@ -1396,3 +1396,157 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(
assert keyword not in json.dumps(langfuse_metadata, default=str), langfuse_metadata
if otel_v2:
assert keyword not in json.dumps(attrs, default=str), attrs
def _span_records(body: bytes) -> tuple[tuple[dict[str, object], int], ...]:
request: Final = ExportTraceServiceRequest.FromString(body)
return tuple(
(
{
attribute.key: getattr(attribute.value, attribute.value.WhichOneof("value"))
for attribute in span.attributes
},
span.dropped_attributes_count,
)
for resource in request.resource_spans
for scope in resource.scope_spans
for span in scope.spans
)
def _identity_attributes(attrs: Mapping[str, object]) -> frozenset[str]:
names: Final = {"langfuse.observation.metadata", *(f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)}
return frozenset(key for key in attrs if key in names)
def _limit_sweep(
rig: _Rig,
key: str,
marker: str,
bodies: Mapping[int, Mapping[str, object]],
) -> dict[int, tuple[dict[str, object], int]]:
"""The exported span and dropped-attribute count per size; each body's prompt carries ``<marker>:<size>:``."""
drained: list[
tuple[dict[str, object], int]
] = [] # mutable-ok: drain() consumes the wire queue, so observed spans must accumulate across eventually polls
for body in bodies.values():
response = rig.candidate.request("POST", "/v1/chat/completions", dict(body), key=key)
assert response.status_code == 200, response.text
def by_size() -> dict[int, tuple[dict[str, object], int] | None]:
drained.extend(record for batch in rig.collector.drain() for record in _span_records(batch.body))
return {
size: next(
(
(attrs, dropped)
for attrs, dropped in drained
if f"{marker}:{size}:" in str(attrs.get("langfuse.observation.input", ""))
),
None,
)
for size in bodies
}
observed: Final = eventually(by_size, lambda spans: None not in spans.values(), seconds=30)
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(
gateway: Gateway, tmp_path: Path
) -> None:
marker: Final = "lf-limit-" + uuid.uuid4().hex
with (
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
rig.candidate.scenario() as scenario,
):
key, identity = _scenario_identity(scenario, rig.provider)
def body(tool_count: int) -> dict[str, object]:
return {
"model": identity["model"],
"messages": [{"role": "user", "content": f"{marker}:{tool_count}:"}],
"user": identity["end_user"],
"tools": [
{
"type": "function",
"function": {
"name": f"tool_{n}",
"description": f"looks up thing {n}",
"parameters": {"type": "object", "properties": {"q": {"type": "string"}}},
},
}
for n in range(tool_count)
],
"cache": {"no-cache": True},
}
spans: Final = _limit_sweep(
rig, key, marker, {tool_count: body(tool_count) for tool_count in (0, *range(30, 41))}
)
reference, reference_dropped = spans[0]
identity_count: Final = len(_identity_attributes(reference))
assert reference_dropped == 0, reference_dropped
assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference)
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:
expected = {"openinference.span.kind", "metadata", "llm.model_name", "llm.request.type"}
assert expected <= set(attrs), (tool_count, dropped, 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()
}
def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: Gateway, tmp_path: Path) -> None:
marker: Final = "lf-v2-limit-" + uuid.uuid4().hex
with (
_langfuse_rig(
gateway,
tmp_path,
{},
marker,
overrides={
"LITELLM_OTEL_V2": "1",
"MAPPER_NAMES": "genai,openinference",
"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only",
},
) as rig,
rig.candidate.scenario() as scenario,
):
key, identity = _scenario_identity(scenario, rig.provider)
def body(message_count: int) -> dict[str, object]:
return {
"model": identity["model"],
"messages": [
{"role": ("user", "assistant")[n % 2], "content": f"{marker} turn {n}"}
for n in range(message_count - 1)
]
+ [{"role": "user", "content": f"{marker}:{message_count}: turn {message_count - 1}"}],
"user": identity["end_user"],
"cache": {"no-cache": True},
}
spans: Final = _limit_sweep(
rig, key, marker, {message_count: body(message_count) for message_count in (1, *range(20, 61))}
)
reference, _ = spans[1]
identity_count: Final = len(_identity_attributes(reference))
assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference)
shed_by_size: Final[dict[int, tuple[int, int]]] = {} # mutable-ok: filled once per swept span below
for message_count, (attrs, _) in spans.items():
kept = {n for n in range(message_count) if f"llm.input_messages.{n}.message.content" in attrs}
shed_messages = message_count - len(kept)
shed_identity = identity_count - len(_identity_attributes(attrs))
shed_by_size[message_count] = (shed_identity, shed_messages)
if shed_messages:
assert shed_identity == identity_count, (message_count, shed_identity, shed_messages)
assert [
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

View file

@ -795,6 +795,37 @@ def test_indexed_messages_follow_the_span_limits_of_a_per_request_tracer_overrid
assert span.attributes["llm.output_messages.0.message.content"] == "reply 0"
def _team_conversation_span(monkeypatch, redact, max_span_attributes):
monkeypatch.setattr("litellm.redact_user_api_key_info", redact)
return _conversation_span(
["genai", "openinference", "langfuse"],
_conversation_payload(60),
span_limits=SpanLimits(max_span_attributes=max_span_attributes),
)
def test_langfuse_identity_sheds_before_any_attribute_the_span_carries_without_it(monkeypatch):
"""At its attribute limit a span drops the Langfuse identity keys first, then exactly what it drops without them."""
plain = _team_conversation_span(monkeypatch, True, SpanLimits.UNSET).attributes
full = _team_conversation_span(monkeypatch, False, SpanLimits.UNSET).attributes
identity_keys = set(full) - set(plain)
assert identity_keys == {
"langfuse.observation.metadata",
"langfuse.trace.metadata.user_api_key_team_id",
"langfuse.trace.metadata.user_api_key_team_alias",
}
squeezed_limit = len(plain) - 10
squeezed_plain = _team_conversation_span(monkeypatch, True, squeezed_limit)
squeezed = _team_conversation_span(monkeypatch, False, squeezed_limit)
assert dict(squeezed.attributes) == dict(squeezed_plain.attributes)
assert squeezed.dropped_attributes == squeezed_plain.dropped_attributes == 0
roomy = _team_conversation_span(monkeypatch, False, len(plain) + 2).attributes
assert {key: roomy.get(key) for key in plain} == dict(plain)
assert len(set(roomy) & identity_keys) == 2
def test_span_attribute_limit_falls_back_to_the_environment_for_spans_outside_the_sdk(monkeypatch):
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "48")
assert span_attribute_limit(INVALID_SPAN) == 48

View file

@ -6,7 +6,9 @@ from typing import Final
from unittest.mock import MagicMock, patch
import pytest
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
import litellm
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
@ -506,8 +508,11 @@ class TestLangfuseOtelIntegration:
"standard_logging_object": {"metadata": request_metadata},
}
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
with (
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
patch("litellm.integrations.arize._utils.set_attributes"),
):
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
@ -545,8 +550,11 @@ class TestLangfuseOtelIntegration:
"standard_logging_object": {"metadata": request_metadata},
}
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
with (
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
patch("litellm.integrations.arize._utils.set_attributes"),
):
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
@ -574,8 +582,11 @@ class TestLangfuseOtelIntegration:
"standard_logging_object": {"metadata": request_metadata},
}
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
with (
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
patch("litellm.integrations.arize._utils.set_attributes"),
):
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
langfuse_metadata: Final = {
@ -605,8 +616,11 @@ class TestLangfuseOtelIntegration:
"standard_logging_object": {"metadata": request_metadata},
}
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
with (
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
patch("litellm.integrations.arize._utils.set_attributes"),
):
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
@ -627,8 +641,11 @@ class TestLangfuseOtelIntegration:
},
}
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
with (
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
patch("litellm.integrations.arize._utils.set_attributes"),
):
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
@ -638,8 +655,11 @@ class TestLangfuseOtelIntegration:
def test_request_metadata_keys_are_absent_without_standard_logging_object(self) -> None:
kwargs: Final = {"litellm_params": {"metadata": {"trace_metadata": {"k": "v"}}}}
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
with (
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
patch("litellm.integrations.arize._utils.set_attributes"),
):
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
@ -669,14 +689,89 @@ class TestLangfuseOtelIntegration:
}
response_obj: Final = {"choices": [{"message": {"role": "assistant", "content": "hello"}}]}
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, response_obj)
with (
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
patch("litellm.integrations.arize._utils.set_attributes"),
):
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, response_obj)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
assert json.loads(actual["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"}
assert json.loads(actual["langfuse.observation.output"]) == {"role": "assistant", "content": "hello"}
@staticmethod
def _exported_span(
kwargs: dict[str, object], max_span_attributes: int | None, carried: dict[str, str] | None = None
) -> ReadableSpan:
exporter: Final = InMemorySpanExporter()
provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=max_span_attributes))
provider.add_span_processor(SimpleSpanProcessor(exporter))
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.end()
(finished,) = exporter.get_finished_spans()
return finished
@staticmethod
def _tool_call_kwargs() -> dict[str, object]:
tools: Final = [
{
"type": "function",
"function": {"name": f"tool_{n}", "description": f"does thing {n}", "parameters": {"type": "object"}},
}
for n in range(3)
]
messages: Final = [{"role": "user", "content": "hello"}]
return {
"model": "gpt-4o",
"messages": messages,
"optional_params": {"tools": tools},
"litellm_params": {"metadata": {}},
"standard_logging_object": {
"model": "gpt-4o",
"custom_llm_provider": "openai",
"call_type": "completion",
"messages": messages,
"model_parameters": {"tools": tools},
"metadata": {"user_api_key_alias": "prod-key", "user_api_key_team_id": "team-1"},
},
}
def test_request_metadata_never_evicts_another_attribute_at_the_span_limit(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
kwargs: Final = self._tool_call_kwargs()
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
plain: 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 {})
identity_keys: Final = set(full) - set(plain)
assert identity_keys == {
"langfuse.observation.metadata",
"langfuse.trace.metadata.user_api_key_alias",
"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
def test_request_metadata_skips_a_span_that_already_carries_attributes(self) -> None:
attributes: Final = dict(
self._exported_span(self._tool_call_kwargs(), SpanLimits.UNSET, {"caller.attribute": "kept"}).attributes
or {}
)
assert attributes["caller.attribute"] == "kept"
assert attributes["langfuse.observation.type"] == "generation"
assert "langfuse.observation.metadata" not in attributes
assert not [key for key in attributes if key.startswith("langfuse.trace.metadata.")]
class TestLangfuseOtelKeyDynamicConfig:
"""Key/team-scoped Langfuse credentials must define the full export target