diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 3efa474b6db..6db26c09c32 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -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): diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 73448221acb..697b524f8f0 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -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 diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 0f298d572b0..db56da033a5 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -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. diff --git a/tests/integration/observability/test_langfuse_otel_metadata.py b/tests/integration/observability/test_langfuse_otel_metadata.py index d43ff4779bb..ddd9bca229b 100644 --- a/tests/integration/observability/test_langfuse_otel_metadata.py +++ b/tests/integration/observability/test_langfuse_otel_metadata.py @@ -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) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 7b79dc034df..fdd1e0fc451 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -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 diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index eb7630781e4..fec14efa477 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -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(