From 1c46a698893e50aa853ca1a4972ef8b89b38ddb2 Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 27 Oct 2026 22:46:28 -0700 Subject: [PATCH 01/13] fix(otel)!: keep a credential-less langfuse_otel on OTel v2 so key destinations get the Langfuse mapper With LITELLM_OTEL_V2 on, langfuse_otel in callbacks and no operator LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY, the preset raised at startup and the callback fell back to the legacy logger. Key and team destinations were then served by a generic v2 logger without the Langfuse mapper, so their traces lost tags, user and cost, and the legacy logger added an orphaned raw_gen_ai_request span. The preset now always degrades to the credential-gated exporter when the operator has no keys, the same way arize_preset does, so the startup logger stays on v2, exports nowhere for traffic without a destination, and carries the Langfuse mapper to every key and team destination. --- litellm/integrations/otel/presets/base.py | 9 +- litellm/integrations/otel/presets/langfuse.py | 4 +- .../otel/test_otel_v2_destinations.py | 111 +++++++++++++++--- 3 files changed, 100 insertions(+), 24 deletions(-) diff --git a/litellm/integrations/otel/presets/base.py b/litellm/integrations/otel/presets/base.py index 3a768a08a4f..795b24bc525 100644 --- a/litellm/integrations/otel/presets/base.py +++ b/litellm/integrations/otel/presets/base.py @@ -19,12 +19,13 @@ class Preset(Protocol): ``config_overrides`` lets one preset layer onto another's config (or onto test-supplied defaults); the factory calls presets with no arguments. - ``allow_missing_credentials`` lets a credential-mandatory backend (langfuse and - weave) degrade to an exporter-less, mapper-only config instead of raising when the - operator set no env credentials of their own. That is a real + ``allow_missing_credentials`` lets a credential-mandatory backend (weave) degrade + to an exporter-less, mapper-only config instead of raising when the operator set + no env credentials of their own. That is a real deployment: every team brings its own account and the operator keeps none, and without it the whole V2 path silently falls back to the legacy integration, so - no team destination is ever reached. Credential-optional backends ignore it. + no team destination is ever reached. Backends that always degrade without + operator credentials (arize, langfuse) and credential-optional ones ignore it. """ def __call__( diff --git a/litellm/integrations/otel/presets/langfuse.py b/litellm/integrations/otel/presets/langfuse.py index 3ff3b521d29..1c4a6189dd4 100644 --- a/litellm/integrations/otel/presets/langfuse.py +++ b/litellm/integrations/otel/presets/langfuse.py @@ -26,9 +26,7 @@ def langfuse_preset( mappers: Final = ensure_mappers(base.mapper_names, "langfuse") try: cfg: Final = _V1Langfuse.get_langfuse_otel_config() - except Exception: - if not allow_missing_credentials: - raise + except ValueError: return base.model_copy( update={ "exporters": credential_gated_exporters(base.exporters, ExporterOwner.LANGFUSE_OTEL), diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 78dcb7bc8f1..fb169141d09 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -2945,8 +2945,20 @@ def credential_less_proxy(monkeypatch) -> None: """An operator with no Langfuse account and no generic OTLP collector.""" for name in ("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", *_OTEL_SHORTHAND_ENV): monkeypatch.delenv(name, raising=False) - with pytest.raises(ValueError, match="LANGFUSE_PUBLIC_KEY"): - langfuse_preset() + assert all(spec.requires_headers and not spec.headers for spec in langfuse_preset().exporters) + + +def credential_less_weave(monkeypatch) -> None: + """An operator with no Weave account and no generic OTLP collector.""" + for name in ("WANDB_API_KEY", "WANDB_PROJECT_ID", *_OTEL_SHORTHAND_ENV): + monkeypatch.delenv(name, raising=False) + + +WEAVE_DEST = OtelDestination( + endpoint="https://trace.wandb.ai/otel/v1/traces", + headers={"Authorization": "Basic dGVuYW50"}, + callback_name="weave_otel", +) def _closed_chat_call_kwargs() -> dict[str, object]: @@ -2986,12 +2998,14 @@ class TestPresetDegradation: assert '"name": "chat gpt-4"' not in capfd.readouterr().out assert "langfuse" in config.mapper_names - def test_langfuse_still_raises_for_a_global_callback_with_no_credentials(self, monkeypatch): + def test_a_global_langfuse_callback_with_no_credentials_degrades_instead_of_raising(self, monkeypatch): monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) - with pytest.raises(ValueError, match="LANGFUSE_PUBLIC_KEY"): - langfuse_preset() + config = langfuse_preset() + + assert any(spec.owner == ExporterOwner.LANGFUSE_OTEL and spec.requires_headers for spec in config.exporters) + assert "langfuse" in config.mapper_names def test_a_credential_less_proxy_builds_the_gated_logger_beside_a_v2_carrier(self, monkeypatch): from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 @@ -3011,10 +3025,9 @@ class TestPresetDegradation: assert logger is not None assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) - def test_a_credential_less_proxy_with_no_destinations_falls_back_to_the_legacy_path(self, monkeypatch): - """Nothing can use a credential-less langfuse here, so the operator has to get - the same story as before v2: the legacy integration, not a global provider - that exports nowhere.""" + def test_a_credential_less_proxy_with_no_destinations_stays_on_v2_and_exports_nowhere(self, monkeypatch): + """The proxy builds the operator's callback at startup, before any key or team + destination has resolved, so this is the logger those destinations ride on.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 credential_less_proxy(monkeypatch) @@ -3024,7 +3037,9 @@ class TestPresetDegradation: logger = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", []) is_otel_v2_enabled.cache_clear() - assert logger is None + assert isinstance(logger, OpenTelemetryV2), "None hands 'langfuse_otel' to the legacy logger" + assert "langfuse" in logger.config.mapper_names + assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) def test_a_valid_newrelic_base_exporter_survives_without_a_license_key(self, monkeypatch): from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 @@ -3099,17 +3114,17 @@ class TestPresetDegradation: without one is logged, long after the degrade was justified.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - credential_less_proxy(monkeypatch) + credential_less_weave(monkeypatch) monkeypatch.setenv("LITELLM_OTEL_V2", "true") loggers = [build_otel_v2_logger(OpenTelemetryV2Config(exporter="in_memory"))] def with_destination(): - set_request_destinations((LANGFUSE_DEST,)) - return _maybe_construct_otel_v2("langfuse_otel", loggers) + set_request_destinations((WEAVE_DEST,)) + return _maybe_construct_otel_v2("weave_otel", loggers) is_otel_v2_enabled.cache_clear() degraded = in_fresh_context(with_destination) - plain = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) + plain = in_fresh_context(_maybe_construct_otel_v2, "weave_otel", loggers) is_otel_v2_enabled.cache_clear() assert degraded is not None @@ -3166,12 +3181,12 @@ class TestPresetDegradation: callback at all, and the operator keeps the pre-V2 story.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - credential_less_proxy(monkeypatch) + credential_less_weave(monkeypatch) monkeypatch.setenv("LITELLM_OTEL_V2", "true") def run(): - set_request_destinations((LANGFUSE_DEST,)) - return _maybe_construct_otel_v2("langfuse_otel", list(registered)) + set_request_destinations((WEAVE_DEST,)) + return _maybe_construct_otel_v2("weave_otel", list(registered)) is_otel_v2_enabled.cache_clear() logger = in_fresh_context(run) @@ -3435,6 +3450,68 @@ class TestArizeTenantOnly: assert loggers == [at_startup] +class TestLangfuseTenantOnly: + """An operator whose keys and teams each bring their own Langfuse project keeps no + Langfuse credentials of their own; the startup logger then exports nowhere for + traffic without a destination and carries the Langfuse mapper to the destinations.""" + + @staticmethod + def _startup_logger(monkeypatch, loggers: list) -> OpenTelemetryV2: + from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 + + credential_less_proxy(monkeypatch) + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + is_otel_v2_enabled.cache_clear() + logger = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) + is_otel_v2_enabled.cache_clear() + assert isinstance(logger, OpenTelemetryV2) + return logger + + def test_the_published_provider_is_the_langfuse_logger(self, monkeypatch): + """The published provider carries the key/team fan-out, so its mappers are the + attributes every destination receives.""" + loggers = [] + logger = self._startup_logger(monkeypatch, loggers) + + published = publish_global_otel_v2_provider(loggers, lambda _p: None) + + assert published is logger + assert "langfuse" in published.config.mapper_names + + def test_the_startup_logger_delivers_a_key_destination_and_nothing_else(self, monkeypatch): + from litellm.integrations.otel.plumbing.providers import attach_tenant_fan_out + + capture = _ExporterCapture() + for kind in ("otlp_grpc", "otlp_http", "console"): + monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) + logger = self._startup_logger(monkeypatch, []) + provider = logger.tracer_provider + attach_tenant_fan_out(provider, logger.config) + + def with_destination(): + set_request_destinations(deliverable_destinations((LANGFUSE_DEST,), provider)) + emit(provider) + + in_fresh_context(with_destination) + in_fresh_context(emit, provider, "chat without a destination") + provider.force_flush() + + assert capture.received() == {LANGFUSE_DEST.header_string(): ("chat gpt-4",)} + + def test_the_startup_logger_is_reused_by_later_requests_without_a_destination(self, monkeypatch): + from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 + + loggers = [] + at_startup = self._startup_logger(monkeypatch, loggers) + + is_otel_v2_enabled.cache_clear() + later = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) + is_otel_v2_enabled.cache_clear() + + assert later is at_startup + assert loggers == [at_startup] + + class TestContextIsolation: def test_destinations_do_not_leak_between_requests(self): def first(): From 3272253d05bcbab3b80f6019125ad100d3c385f1 Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 00:09:45 -0700 Subject: [PATCH 02/13] test(otel): type the langfuse tenant-only test helpers --- .../otel/test_otel_v2_destinations.py | 52 +++++++++++-------- 1 file changed, 30 insertions(+), 22 deletions(-) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index fb169141d09..92394e6ac4a 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -2941,20 +2941,20 @@ _OTEL_SHORTHAND_ENV = ( ) -def credential_less_proxy(monkeypatch) -> None: +def credential_less_proxy(monkeypatch: pytest.MonkeyPatch) -> None: """An operator with no Langfuse account and no generic OTLP collector.""" for name in ("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", *_OTEL_SHORTHAND_ENV): monkeypatch.delenv(name, raising=False) assert all(spec.requires_headers and not spec.headers for spec in langfuse_preset().exporters) -def credential_less_weave(monkeypatch) -> None: +def credential_less_weave(monkeypatch: pytest.MonkeyPatch) -> None: """An operator with no Weave account and no generic OTLP collector.""" for name in ("WANDB_API_KEY", "WANDB_PROJECT_ID", *_OTEL_SHORTHAND_ENV): monkeypatch.delenv(name, raising=False) -WEAVE_DEST = OtelDestination( +WEAVE_DEST: Final = OtelDestination( endpoint="https://trace.wandb.ai/otel/v1/traces", headers={"Authorization": "Basic dGVuYW50"}, callback_name="weave_otel", @@ -2998,11 +2998,13 @@ class TestPresetDegradation: assert '"name": "chat gpt-4"' not in capfd.readouterr().out assert "langfuse" in config.mapper_names - def test_a_global_langfuse_callback_with_no_credentials_degrades_instead_of_raising(self, monkeypatch): + def test_a_global_langfuse_callback_with_no_credentials_degrades_instead_of_raising( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) - config = langfuse_preset() + config: Final = langfuse_preset() assert any(spec.owner == ExporterOwner.LANGFUSE_OTEL and spec.requires_headers for spec in config.exporters) assert "langfuse" in config.mapper_names @@ -3025,7 +3027,9 @@ class TestPresetDegradation: assert logger is not None assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) - def test_a_credential_less_proxy_with_no_destinations_stays_on_v2_and_exports_nowhere(self, monkeypatch): + def test_a_credential_less_proxy_with_no_destinations_stays_on_v2_and_exports_nowhere( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: """The proxy builds the operator's callback at startup, before any key or team destination has resolved, so this is the logger those destinations ride on.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 @@ -3034,7 +3038,7 @@ class TestPresetDegradation: monkeypatch.setenv("LITELLM_OTEL_V2", "true") is_otel_v2_enabled.cache_clear() - logger = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", []) + logger: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", []) is_otel_v2_enabled.cache_clear() assert isinstance(logger, OpenTelemetryV2), "None hands 'langfuse_otel' to the legacy logger" @@ -3456,39 +3460,41 @@ class TestLangfuseTenantOnly: traffic without a destination and carries the Langfuse mapper to the destinations.""" @staticmethod - def _startup_logger(monkeypatch, loggers: list) -> OpenTelemetryV2: + def _startup_logger(monkeypatch: pytest.MonkeyPatch, loggers: list[CustomLogger]) -> OpenTelemetryV2: from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 credential_less_proxy(monkeypatch) monkeypatch.setenv("LITELLM_OTEL_V2", "true") is_otel_v2_enabled.cache_clear() - logger = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) + logger: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) is_otel_v2_enabled.cache_clear() assert isinstance(logger, OpenTelemetryV2) return logger - def test_the_published_provider_is_the_langfuse_logger(self, monkeypatch): + def test_the_published_provider_is_the_langfuse_logger(self, monkeypatch: pytest.MonkeyPatch) -> None: """The published provider carries the key/team fan-out, so its mappers are the attributes every destination receives.""" - loggers = [] - logger = self._startup_logger(monkeypatch, loggers) + loggers: Final[list[CustomLogger]] = [] + logger: Final = self._startup_logger(monkeypatch, loggers) - published = publish_global_otel_v2_provider(loggers, lambda _p: None) + published: Final = publish_global_otel_v2_provider(loggers, lambda _p: None) assert published is logger assert "langfuse" in published.config.mapper_names - def test_the_startup_logger_delivers_a_key_destination_and_nothing_else(self, monkeypatch): + def test_the_startup_logger_delivers_a_key_destination_and_nothing_else( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: from litellm.integrations.otel.plumbing.providers import attach_tenant_fan_out - capture = _ExporterCapture() + capture: Final = _ExporterCapture() for kind in ("otlp_grpc", "otlp_http", "console"): monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) - logger = self._startup_logger(monkeypatch, []) - provider = logger.tracer_provider + logger: Final = self._startup_logger(monkeypatch, []) + provider: Final = logger.tracer_provider attach_tenant_fan_out(provider, logger.config) - def with_destination(): + def with_destination() -> None: set_request_destinations(deliverable_destinations((LANGFUSE_DEST,), provider)) emit(provider) @@ -3498,14 +3504,16 @@ class TestLangfuseTenantOnly: assert capture.received() == {LANGFUSE_DEST.header_string(): ("chat gpt-4",)} - def test_the_startup_logger_is_reused_by_later_requests_without_a_destination(self, monkeypatch): + def test_the_startup_logger_is_reused_by_later_requests_without_a_destination( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - loggers = [] - at_startup = self._startup_logger(monkeypatch, loggers) + loggers: Final[list[CustomLogger]] = [] + at_startup: Final = self._startup_logger(monkeypatch, loggers) is_otel_v2_enabled.cache_clear() - later = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) + later: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) is_otel_v2_enabled.cache_clear() assert later is at_startup From 44d055e4d5b5fe6a007016322c43ca14162181dd Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 00:47:19 -0700 Subject: [PATCH 03/13] fix(otel): give key/team Langfuse destinations the Langfuse fields when langfuse_otel is not a callback With langfuse_otel only in a key's or team's logging, the startup logger is the generic one and its spans reached the key's Langfuse project without user, session or tags. The emitter now adds the Langfuse mapper for a request with a Langfuse destination, and every non-Langfuse exporter strips the Langfuse request metadata, since any logger can now write it. --- litellm/integrations/otel/emitter.py | 22 ++- .../integrations/otel/plumbing/providers.py | 10 +- .../test_langfuse_otel_tenant_only.py | 131 ++++++++++++++++++ .../otel/test_otel_v2_destinations.py | 63 ++++++++- 4 files changed, 213 insertions(+), 13 deletions(-) create mode 100644 tests/integration/observability/test_langfuse_otel_tenant_only.py diff --git a/litellm/integrations/otel/emitter.py b/litellm/integrations/otel/emitter.py index c02ea6032a0..3784efcc2f2 100644 --- a/litellm/integrations/otel/emitter.py +++ b/litellm/integrations/otel/emitter.py @@ -13,9 +13,9 @@ 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.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LangfuseMapper from litellm.integrations.otel.mappers.openinference import fit_indexed_messages -from litellm.integrations.otel.model.config import OpenTelemetryV2Config +from litellm.integrations.otel.model.config import ExporterOwner, OpenTelemetryV2Config from litellm.integrations.otel.model.payloads import ( GuardrailSpanData, LLMCallSpanData, @@ -34,6 +34,7 @@ from litellm.integrations.otel.model.spans import ( mcp_tool_call_span_name, service_span_name, ) +from litellm.integrations.otel.plumbing.context import destination_backends from litellm.integrations.otel.plumbing.events import GenAIEventRecorder from litellm.integrations.otel.plumbing.providers import to_otel_span_kind @@ -168,6 +169,10 @@ class SpanEmitter: self._mappers: list[AttributeMapper] = ( list(mappers) if mappers is not None else resolve_mappers(config.mapper_names) ) + # A key's or team's Langfuse destination reads the Langfuse vocabulary, which this logger may not write. + self._langfuse: Final = ( + None if any(isinstance(mapper, LangfuseMapper) for mapper in self._mappers) else LangfuseMapper() + ) # Bounded LRU (ordered by insertion / most-recent touch). Storing keys # only — the value is unused — so it behaves like a capped set. self._emitted: OrderedDict[tuple[str, SpanRole], None] = OrderedDict() @@ -295,17 +300,22 @@ class SpanEmitter: ) else None ) - mapped: Final = MappingProxyType( - {key: value for mapper in self._mappers for key, value in mapper.map(data).items()} + configured: Final = {key: value for mapper in self._mappers for key, value in mapper.map(data).items()} + for_langfuse: Final = ( + self._langfuse.map(data) + if self._langfuse is not None and ExporterOwner.LANGFUSE_OTEL in destination_backends() + else _NO_ATTRIBUTES ) + mapped: Final = MappingProxyType({**configured, **for_langfuse}) stamped_later: Final = error_attributes(error) if error else _NO_ATTRIBUTES reserved: Final = len(stamped_later.keys() - mapped.keys()) overwrites: Final = carried_keys(span) & mapped.keys() - # The Langfuse request metadata sheds first, so a span at its attribute limit loses nothing else to it. + # The Langfuse request metadata, and the Langfuse vocabulary only a destination asked for, shed first, + # so a span at its attribute limit loses nothing else to them. fitted: Final = fit_indexed_messages( MappingProxyType({key: value for key, value in mapped.items() if key not in overwrites}), attribute_budget(span, reserved), - LANGFUSE_REQUEST_METADATA_ATTRIBUTES, + LANGFUSE_REQUEST_METADATA_ATTRIBUTES.union(for_langfuse), ) for key, value in mapped.items(): if key in overwrites or key in fitted: diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 9736192a816..a65a75d2128 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -1524,8 +1524,9 @@ 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 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. + Every exporter but ``langfuse_otel``'s goes without the request metadata the Langfuse + mapper writes on LiteLLM's own spans, which any logger writes for a request with a + Langfuse destination. ``strip_message_content`` removes prompt and response content before every exporter, for a routed provider whose spans can carry content another destination asked for. @@ -1541,9 +1542,6 @@ 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). - 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 @@ -1551,7 +1549,7 @@ def build_tracer_provider( spec, use_simple_processor, strip_message_content, - strip_langfuse_request_metadata=writes_langfuse and spec.owner is not ExporterOwner.LANGFUSE_OTEL, + strip_langfuse_request_metadata=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) diff --git a/tests/integration/observability/test_langfuse_otel_tenant_only.py b/tests/integration/observability/test_langfuse_otel_tenant_only.py new file mode 100644 index 00000000000..29ae7f4c990 --- /dev/null +++ b/tests/integration/observability/test_langfuse_otel_tenant_only.py @@ -0,0 +1,131 @@ +"""A key's or team's own Langfuse project gets the Langfuse trace fields when the operator has no Langfuse keys, +whether or not ``langfuse_otel`` is listed in ``litellm_settings.callbacks``.""" + +from __future__ import annotations + +import json +import uuid +from collections.abc import Iterator +from typing import Final, Literal + +import pytest +from integration._support.client import Gateway, eventually, gateway_from_environment +from integration._support.otlp_sink import ( + AuditConfigWriter, + Span, + SpanSinks, + langfuse_callback_variables, + recorded_spans, + spans_for_trace, +) +from integration._support.process import owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue + +OPERATION: Final = "gen_ai.operation.name" +TAGS: Final = ["agent:orders", "run:r1"] + + +def _chat(request: Request) -> Reply: + marker: Final = json.loads(request.body)["messages"][-1]["content"] + return Reply( + body=json.dumps( + { + "id": f"chatcmpl-{uuid.uuid4().hex}", + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "message": {"role": "assistant", "content": marker}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}, + } + ).encode() + ) + + +@pytest.fixture(scope="module") +def provider() -> Iterator[Wire]: + with wire_server(_chat) as wire: + yield wire + + +@pytest.fixture(scope="module", params=[["langfuse_otel"], []], ids=["listed", "not-listed"]) +def proxy( + request: pytest.FixtureRequest, + audit_sinks: SpanSinks, + otel_audit_config: AuditConfigWriter, + tmp_path_factory: pytest.TempPathFactory, +) -> Iterator[Gateway]: + directory: Final = tmp_path_factory.mktemp("langfuse-tenant-only") + with ( + gateway_from_environment() as gateway, + owned_proxy_process( + gateway, + directory, + { + "LITELLM_OTEL_V2": "1", + "OTEL_BSP_SCHEDULE_DELAY": "300", + "OTEL_EXPORTER": "otlp_http", + "OTEL_ENDPOINT": audit_sinks.operator, + }, + config=otel_audit_config(directory, {"callbacks": request.param}), + remove_environment=("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"), + workers=2, + ) as owned, + ): + yield owned.gateway + + +def _model_spans(url: str, since: int, trace_id: str) -> tuple[Span, ...]: + return tuple( + span for span in spans_for_trace(recorded_spans(url, since)[1], trace_id) if OPERATION in span["attributes"] + ) + + +@pytest.mark.parametrize("level", ["key", "team"]) +def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_keys( + level: Literal["key", "team"], proxy: Gateway, provider: Wire, audit_sinks: SpanSinks +) -> None: + logging: Final[list[JsonValue]] = [ + { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": langfuse_callback_variables(audit_sinks), + } + ] + with proxy.scenario() as scenario: + model: Final = scenario.model(api_base=provider.url + "/v1") + alias: Final = f"lf-tenant-only-{uuid.uuid4().hex}" + key: Final = ( + scenario.key(key_alias=alias, metadata={"logging": logging}) + if level == "key" + else scenario.key(key_alias=alias, team_id=scenario.team(metadata={"logging": logging})) + ) + since: Final = (recorded_spans(audit_sinks.tenant)[0], recorded_spans(audit_sinks.operator)[0]) + trace_id: Final = uuid.uuid4().hex + session: Final = f"sess-{trace_id}" + response: Final = proxy.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": f"lf-tenant-only-{trace_id}"}], + "metadata": {"tags": TAGS, "session_id": session, "trace_user_id": "user-42"}, + }, + key=key, + headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + ) + assert response.status_code == 200, response.text + + tenant: Final = eventually( + lambda: _model_spans(audit_sinks.tenant, since[0], trace_id), lambda spans: bool(spans), seconds=40 + ) + operator: Final = eventually( + lambda: _model_spans(audit_sinks.operator, since[1], trace_id), lambda spans: bool(spans), seconds=40 + ) + + assert [ + {name: span["attributes"].get(name) for name in ("user.id", "session.id", "langfuse.trace.tags")} + for span in tenant + ] == [{"user.id": "user-42", "session.id": session, "langfuse.trace.tags": TAGS}], [s["name"] for s in tenant] + assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"]) + assert "langfuse.observation.metadata" not in operator[0]["attributes"], sorted(operator[0]["attributes"]) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 92394e6ac4a..03c545334df 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -17,7 +17,7 @@ import pytest from opentelemetry import trace as trace_api from opentelemetry.context import Context from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import ReadableSpan, 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 from opentelemetry.sdk.trace.id_generator import RandomIdGenerator @@ -27,6 +27,7 @@ import litellm from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.otel import GenAIOperation from litellm.integrations.otel import logger as otel_logger +from litellm.integrations.otel.emitter import SpanEmitter from litellm.integrations.otel.logger import ( OpenTelemetryV2, build_otel_v2_logger, @@ -52,6 +53,8 @@ from litellm.integrations.otel.model.payloads import ( ServerInfo, ToolDefinition, ) +from litellm.integrations.otel.model.spans import SpanRole +from litellm.integrations.otel.model.trace_controls import TraceControls from litellm.integrations.otel.plumbing import providers as otel_providers from litellm.integrations.otel.plumbing.context import ( destination_backends, @@ -3519,6 +3522,64 @@ class TestLangfuseTenantOnly: assert later is at_startup assert loggers == [at_startup] + def test_a_key_destination_gets_the_langfuse_fields_from_the_generic_startup_logger( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """With ``langfuse_otel`` only in a key's or team's logging, the startup logger is the + generic one; the key's project still gets the user, session and tags, and the + operator's stdout placeholder still gets no Langfuse request metadata.""" + credential_less_proxy(monkeypatch) + capture: Final = _ExporterCapture() + for kind in ("otlp_grpc", "otlp_http", "console"): + monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) + config: Final = OpenTelemetryV2Config() + provider: Final = build_tracer_provider(config, tenant_overrides=True) + attach_tenant_fan_out(provider, config) + emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config) + call: Final = replace( + _MODEL_CALL_WITHOUT_CONTENT, + trace=TraceControls(user_id="user-42", session_id="sess-1", tags=("agent:orders",)), + request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}), + ) + + def with_destination() -> None: + set_request_destinations(deliverable_destinations((LANGFUSE_DEST,), provider)) + emitter.emit(SpanRole.LLM_CALL, call) + + in_fresh_context(with_destination) + provider.force_flush() + + received: Final = { + headers: exporter.get_finished_spans()[0].attributes or {} for headers, exporter in capture.built + } + to_langfuse: Final = received[LANGFUSE_DEST.header_string()] + assert {key: to_langfuse.get(key) for key in ("user.id", "session.id", "langfuse.trace.tags")} == { + "user.id": "user-42", + "session.id": "sess-1", + "langfuse.trace.tags": ("agent:orders",), + } + assert "langfuse.observation.metadata" in to_langfuse + assert "langfuse.observation.metadata" not in received[None] + + def test_a_span_at_its_attribute_limit_sheds_the_destinations_langfuse_fields_first(self) -> None: + def attributes(destinations: tuple[OtelDestination, ...], limit: int | None) -> dict[str, object]: + exporter: Final = InMemorySpanExporter() + provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=limit)) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), OpenTelemetryV2Config()) + + def run() -> None: + set_request_destinations(destinations) + emitter.emit(SpanRole.LLM_CALL, _MODEL_CALL_WITH_CONTENT) + + in_fresh_context(run) + return dict(exporter.get_finished_spans()[0].attributes or {}) + + generic: Final = attributes((), None) + + assert "user.id" not in generic + assert attributes((LANGFUSE_DEST,), len(generic)) == generic, "the generic attributes all survive" + class TestContextIsolation: def test_destinations_do_not_leak_between_requests(self): From 1963a42bac3d88778dc8c931d8e77bf6d2870f85 Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 01:32:37 -0700 Subject: [PATCH 04/13] fix(otel): stop a key's Langfuse logger from re-sending spans to the operator's collector With langfuse_otel only in key or team logging, the proxy publishes the generic V2 logger at startup and it already exports to the operator's OTLP collector. The Langfuse logger a key's request builds kept the same base exporter, so the collector got every model span twice. A logger built after the publish now keeps only its preset's own exporters, as it already did beside a configured V2 logger, and a later call that would rebuild one exporting nowhere reuses it. --- litellm/integrations/otel/logger.py | 5 ++ litellm/litellm_core_utils/litellm_logging.py | 21 +++++---- .../test_langfuse_otel_tenant_only.py | 1 + .../otel/test_otel_v2_destinations.py | 46 +++++++++++++++++++ 4 files changed, 64 insertions(+), 9 deletions(-) diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index e09226ac9cd..21414e80c00 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -1038,6 +1038,11 @@ def seed_request_identity(user_api_key_dict: object, model: str | None = None) - logger.seed_request_identity(user_api_key_dict, model=model) +def v2_provider_published() -> bool: + """Whether the proxy published a v2 provider at startup, which delivers to the operator's own backends.""" + return _published_v2_provider is not None + + def fan_out_provider() -> ApiTracerProvider: """The provider :func:`publish_global_otel_v2_provider` gave the tenant fan-out. diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 48e8a44b9a4..69e9d338ba7 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -5971,11 +5971,12 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom Returns ``None`` when V2 is off OR when there's no preset registered for ``callback_name`` — callers should then fall through to the legacy path. - A logger built while another V2 logger is already registered keeps only the - exporters its own preset contributed, whether or not the operator holds - credentials for that backend and whether or not a destination is anchored: the - registered logger owns operator delivery, so a copy of the operator's base OTLP - exporters here would emit every LLM call a second time into the operator's sink. + A logger built while another V2 logger is already registered, or after the proxy + published one at startup, keeps only the exporters its own preset contributed, + whether or not the operator holds credentials for that backend and whether or not + a destination is anchored: that logger owns operator delivery, so a copy of the + operator's base OTLP exporters here would emit every LLM call a second time into + the operator's sink. A preset that contributes no exporter of its own (a mapper over the operator's collector) keeps the base exporters, since it has nothing else to deliver through. A preset that needs operator credentials it cannot find is allowed to build only @@ -5986,7 +5987,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom until a key/team destination appears: the proxy builds the operator's callback at startup, before any request has resolved a destination, and the legacy path there would post every non-team request to the backend keyless. That logger is reused by - later calls as long as the preset would again build exporting nowhere; one that + later calls as long as a rebuild would again export nowhere; one that was degraded for a destination while the preset raises without one is not, since the degrade was justified by that destination alone. """ @@ -5994,7 +5995,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom if not is_otel_v2_enabled(): return None - from litellm.integrations.otel.logger import OpenTelemetryV2, build_otel_v2_logger + from litellm.integrations.otel.logger import OpenTelemetryV2, build_otel_v2_logger, v2_provider_published from litellm.integrations.otel.plumbing.context import destination_backends from litellm.integrations.otel.presets import PRESET_BY_CALLBACK @@ -6022,9 +6023,11 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom return None if _is_credential_gated(built) and not carried and _only_the_placeholder_would_export(built): return None - if existing is not None and _exports_nowhere(built): + config: Final = ( + _only_the_presets_own_exporters(built, callback_name) if has_v2_logger or v2_provider_published() else built + ) + if existing is not None and _exports_nowhere(config): return existing - config: Final = _only_the_presets_own_exporters(built, callback_name) if has_v2_logger else built if _exports_nowhere(config): verbose_logger.warning( "OTel V2: no operator credentials for '%s'; only key/team destinations will receive its traces", diff --git a/tests/integration/observability/test_langfuse_otel_tenant_only.py b/tests/integration/observability/test_langfuse_otel_tenant_only.py index 29ae7f4c990..9ea50faf439 100644 --- a/tests/integration/observability/test_langfuse_otel_tenant_only.py +++ b/tests/integration/observability/test_langfuse_otel_tenant_only.py @@ -128,4 +128,5 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k for span in tenant ] == [{"user.id": "user-42", "session.id": session, "langfuse.trace.tags": TAGS}], [s["name"] for s in tenant] assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"]) + assert len(operator) == 1, [span["name"] for span in operator] assert "langfuse.observation.metadata" not in operator[0]["attributes"], sorted(operator[0]["attributes"]) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 03c545334df..cd2ae99e93c 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -3522,6 +3522,52 @@ class TestLangfuseTenantOnly: assert later is at_startup assert loggers == [at_startup] + @staticmethod + def _key_logger_beside_the_generic_startup_logger( + monkeypatch: pytest.MonkeyPatch, loggers: list[CustomLogger] + ) -> OpenTelemetryV2: + """``langfuse_otel`` only in a key's logging: the proxy published the generic logger, + which exports to the operator's collector, and a key's request builds the Langfuse one.""" + from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 + + credential_less_proxy(monkeypatch) + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.local:4318") + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + publish_global_otel_v2_provider(loggers, lambda _p: None) + + def with_destination() -> OpenTelemetryV2 | None: + set_request_destinations((LANGFUSE_DEST,)) + return _maybe_construct_otel_v2("langfuse_otel", loggers) + + is_otel_v2_enabled.cache_clear() + logger: Final = in_fresh_context(with_destination) + is_otel_v2_enabled.cache_clear() + assert isinstance(logger, OpenTelemetryV2) + return logger + + def test_a_key_destinations_logger_leaves_the_operators_collector_to_the_published_logger( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The published logger already sends every span to the collector; a copy here doubles each one.""" + logger: Final = self._key_logger_beside_the_generic_startup_logger(monkeypatch, []) + + assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) + + def test_a_key_destinations_logger_is_reused_by_later_requests_without_a_destination( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 + + loggers: Final[list[CustomLogger]] = [] + for_the_key: Final = self._key_logger_beside_the_generic_startup_logger(monkeypatch, loggers) + + is_otel_v2_enabled.cache_clear() + later: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) + is_otel_v2_enabled.cache_clear() + + assert later is for_the_key + assert loggers == [for_the_key] + def test_a_key_destination_gets_the_langfuse_fields_from_the_generic_startup_logger( self, monkeypatch: pytest.MonkeyPatch ) -> None: From 0dca18e19a981685ab7d6e7f1e745387e042ae3a Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 01:54:26 -0700 Subject: [PATCH 05/13] test(otel): assert the collector gets one copy instead of reading the key logger's exporters --- .../otel/test_otel_v2_destinations.py | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index cd2ae99e93c..3fed1801b21 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -3525,7 +3525,7 @@ class TestLangfuseTenantOnly: @staticmethod def _key_logger_beside_the_generic_startup_logger( monkeypatch: pytest.MonkeyPatch, loggers: list[CustomLogger] - ) -> OpenTelemetryV2: + ) -> tuple[OpenTelemetryV2, OpenTelemetryV2]: """``langfuse_otel`` only in a key's logging: the proxy published the generic logger, which exports to the operator's collector, and a key's request builds the Langfuse one.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 @@ -3533,7 +3533,7 @@ class TestLangfuseTenantOnly: credential_less_proxy(monkeypatch) monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.local:4318") monkeypatch.setenv("LITELLM_OTEL_V2", "true") - publish_global_otel_v2_provider(loggers, lambda _p: None) + published: Final = publish_global_otel_v2_provider(loggers, lambda _p: None) def with_destination() -> OpenTelemetryV2 | None: set_request_destinations((LANGFUSE_DEST,)) @@ -3543,15 +3543,20 @@ class TestLangfuseTenantOnly: logger: Final = in_fresh_context(with_destination) is_otel_v2_enabled.cache_clear() assert isinstance(logger, OpenTelemetryV2) - return logger + return published, logger - def test_a_key_destinations_logger_leaves_the_operators_collector_to_the_published_logger( + def test_the_operators_collector_gets_one_copy_of_a_call_both_loggers_emit( self, monkeypatch: pytest.MonkeyPatch ) -> None: - """The published logger already sends every span to the collector; a copy here doubles each one.""" - logger: Final = self._key_logger_beside_the_generic_startup_logger(monkeypatch, []) + capture: Final = _ExporterCapture() + for kind in ("otlp_grpc", "otlp_http", "console"): + monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) - assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) + for logger in self._key_logger_beside_the_generic_startup_logger(monkeypatch, []): + emit(logger.tracer_provider) + logger.tracer_provider.force_flush() + + assert [span.name for _, exporter in capture.built for span in exporter.get_finished_spans()] == ["chat gpt-4"] def test_a_key_destinations_logger_is_reused_by_later_requests_without_a_destination( self, monkeypatch: pytest.MonkeyPatch @@ -3559,7 +3564,7 @@ class TestLangfuseTenantOnly: from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 loggers: Final[list[CustomLogger]] = [] - for_the_key: Final = self._key_logger_beside_the_generic_startup_logger(monkeypatch, loggers) + for_the_key: Final = self._key_logger_beside_the_generic_startup_logger(monkeypatch, loggers)[1] is_otel_v2_enabled.cache_clear() later: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) From dfa95e4f050edddcbb4e7fbcf758f227146c0738 Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 02:12:46 -0700 Subject: [PATCH 06/13] test(otel): count the collector's spans with one comprehension clause --- tests/unit/integrations/otel/test_otel_v2_destinations.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 3fed1801b21..79a345c5c08 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -3556,7 +3556,7 @@ class TestLangfuseTenantOnly: emit(logger.tracer_provider) logger.tracer_provider.force_flush() - assert [span.name for _, exporter in capture.built for span in exporter.get_finished_spans()] == ["chat gpt-4"] + assert sum(len(exporter.get_finished_spans()) for _, exporter in capture.built) == 1 def test_a_key_destinations_logger_is_reused_by_later_requests_without_a_destination( self, monkeypatch: pytest.MonkeyPatch From ce1b34dbc25142640803ee2dfb052649d5d2104e Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 12:08:24 -0700 Subject: [PATCH 07/13] fix(otel): keep the legacy langfuse_otel logger and stand it down for requests a Langfuse destination carries Reverts the preset degrade and the published-provider exporter trim, so every request without a key or team Langfuse destination keeps main's legacy logger. For a request with one, the V2 fan-out already delivers the whole trace with the Langfuse fields, so the legacy logger emits nothing for it. --- .../integrations/langfuse/langfuse_otel.py | 32 ++++ litellm/integrations/otel/logger.py | 5 - litellm/integrations/otel/presets/base.py | 9 +- litellm/integrations/otel/presets/langfuse.py | 4 +- litellm/litellm_core_utils/litellm_logging.py | 21 +-- .../otel/test_otel_v2_destinations.py | 174 +++--------------- tests/unit/integrations/test_langfuse_otel.py | 58 ++++++ 7 files changed, 127 insertions(+), 176 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 34d2c9fce0d..dcccd1155c5 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -35,6 +35,8 @@ from litellm.types.utils import StandardCallbackDynamicParams if TYPE_CHECKING: from opentelemetry.trace import Span as _Span + from litellm.proxy._types import UserAPIKeyAuth + Span = _Span else: Span = Any @@ -126,6 +128,15 @@ def _starts_its_own_trace(span: object, kwargs: Mapping[str, object]) -> bool: ) +def _a_destination_carries_the_trace() -> bool: + """Whether this request resolved a key's or team's ``langfuse_otel`` destination, whose fan-out sends it the whole + trace.""" + from litellm.integrations.otel.model.config import ExporterOwner + from litellm.integrations.otel.plumbing.context import destination_backends + + return ExporterOwner.LANGFUSE_OTEL in destination_backends() + + class LangfuseOtelLogger(OpenTelemetry): def __init__(self, config=None, *args, **kwargs): # Prevent LangfuseOtelLogger from modifying global environment variables by constructing config manually @@ -541,6 +552,27 @@ class LangfuseOtelLogger(OpenTelemetry): """ return None + def _emit_once( + self, + kwargs: dict[str, object], # mutable-ok: the v1 dedupe marker is written into the request's kwargs + *scope: object, + ) -> bool: + """Whether to emit this span: once per request, and never for a request a Langfuse destination carries.""" + return not _a_destination_carries_the_trace() and super()._emit_once(kwargs, *scope) # pyright: ignore[reportUnknownMemberType] # the v1 hook takes a bare dict + + async def async_post_call_failure_hook( + self, + request_data: dict[str, object], # mutable-ok: the v1 hook contract + original_exception: Exception, + user_api_key_dict: "UserAPIKeyAuth", + traceback_str: str | None = None, + ) -> None: + """Record the proxy failure of a request no Langfuse destination carries.""" + if not _a_destination_carries_the_trace(): + await super().async_post_call_failure_hook( # pyright: ignore[reportUnknownMemberType] # the v1 hook takes a bare dict + request_data, original_exception, user_api_key_dict, traceback_str + ) + async def async_service_success_hook(self, *args, **kwargs): """ Langfuse should not receive service success logs. diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 21414e80c00..e09226ac9cd 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -1038,11 +1038,6 @@ def seed_request_identity(user_api_key_dict: object, model: str | None = None) - logger.seed_request_identity(user_api_key_dict, model=model) -def v2_provider_published() -> bool: - """Whether the proxy published a v2 provider at startup, which delivers to the operator's own backends.""" - return _published_v2_provider is not None - - def fan_out_provider() -> ApiTracerProvider: """The provider :func:`publish_global_otel_v2_provider` gave the tenant fan-out. diff --git a/litellm/integrations/otel/presets/base.py b/litellm/integrations/otel/presets/base.py index 795b24bc525..3a768a08a4f 100644 --- a/litellm/integrations/otel/presets/base.py +++ b/litellm/integrations/otel/presets/base.py @@ -19,13 +19,12 @@ class Preset(Protocol): ``config_overrides`` lets one preset layer onto another's config (or onto test-supplied defaults); the factory calls presets with no arguments. - ``allow_missing_credentials`` lets a credential-mandatory backend (weave) degrade - to an exporter-less, mapper-only config instead of raising when the operator set - no env credentials of their own. That is a real + ``allow_missing_credentials`` lets a credential-mandatory backend (langfuse and + weave) degrade to an exporter-less, mapper-only config instead of raising when the + operator set no env credentials of their own. That is a real deployment: every team brings its own account and the operator keeps none, and without it the whole V2 path silently falls back to the legacy integration, so - no team destination is ever reached. Backends that always degrade without - operator credentials (arize, langfuse) and credential-optional ones ignore it. + no team destination is ever reached. Credential-optional backends ignore it. """ def __call__( diff --git a/litellm/integrations/otel/presets/langfuse.py b/litellm/integrations/otel/presets/langfuse.py index 1c4a6189dd4..3ff3b521d29 100644 --- a/litellm/integrations/otel/presets/langfuse.py +++ b/litellm/integrations/otel/presets/langfuse.py @@ -26,7 +26,9 @@ def langfuse_preset( mappers: Final = ensure_mappers(base.mapper_names, "langfuse") try: cfg: Final = _V1Langfuse.get_langfuse_otel_config() - except ValueError: + except Exception: + if not allow_missing_credentials: + raise return base.model_copy( update={ "exporters": credential_gated_exporters(base.exporters, ExporterOwner.LANGFUSE_OTEL), diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 69e9d338ba7..48e8a44b9a4 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -5971,12 +5971,11 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom Returns ``None`` when V2 is off OR when there's no preset registered for ``callback_name`` — callers should then fall through to the legacy path. - A logger built while another V2 logger is already registered, or after the proxy - published one at startup, keeps only the exporters its own preset contributed, - whether or not the operator holds credentials for that backend and whether or not - a destination is anchored: that logger owns operator delivery, so a copy of the - operator's base OTLP exporters here would emit every LLM call a second time into - the operator's sink. + A logger built while another V2 logger is already registered keeps only the + exporters its own preset contributed, whether or not the operator holds + credentials for that backend and whether or not a destination is anchored: the + registered logger owns operator delivery, so a copy of the operator's base OTLP + exporters here would emit every LLM call a second time into the operator's sink. A preset that contributes no exporter of its own (a mapper over the operator's collector) keeps the base exporters, since it has nothing else to deliver through. A preset that needs operator credentials it cannot find is allowed to build only @@ -5987,7 +5986,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom until a key/team destination appears: the proxy builds the operator's callback at startup, before any request has resolved a destination, and the legacy path there would post every non-team request to the backend keyless. That logger is reused by - later calls as long as a rebuild would again export nowhere; one that + later calls as long as the preset would again build exporting nowhere; one that was degraded for a destination while the preset raises without one is not, since the degrade was justified by that destination alone. """ @@ -5995,7 +5994,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom if not is_otel_v2_enabled(): return None - from litellm.integrations.otel.logger import OpenTelemetryV2, build_otel_v2_logger, v2_provider_published + from litellm.integrations.otel.logger import OpenTelemetryV2, build_otel_v2_logger from litellm.integrations.otel.plumbing.context import destination_backends from litellm.integrations.otel.presets import PRESET_BY_CALLBACK @@ -6023,11 +6022,9 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom return None if _is_credential_gated(built) and not carried and _only_the_placeholder_would_export(built): return None - config: Final = ( - _only_the_presets_own_exporters(built, callback_name) if has_v2_logger or v2_provider_published() else built - ) - if existing is not None and _exports_nowhere(config): + if existing is not None and _exports_nowhere(built): return existing + config: Final = _only_the_presets_own_exporters(built, callback_name) if has_v2_logger else built if _exports_nowhere(config): verbose_logger.warning( "OTel V2: no operator credentials for '%s'; only key/team destinations will receive its traces", diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 79a345c5c08..c0b81c49a4f 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -2944,24 +2944,12 @@ _OTEL_SHORTHAND_ENV = ( ) -def credential_less_proxy(monkeypatch: pytest.MonkeyPatch) -> None: +def credential_less_proxy(monkeypatch) -> None: """An operator with no Langfuse account and no generic OTLP collector.""" for name in ("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", *_OTEL_SHORTHAND_ENV): monkeypatch.delenv(name, raising=False) - assert all(spec.requires_headers and not spec.headers for spec in langfuse_preset().exporters) - - -def credential_less_weave(monkeypatch: pytest.MonkeyPatch) -> None: - """An operator with no Weave account and no generic OTLP collector.""" - for name in ("WANDB_API_KEY", "WANDB_PROJECT_ID", *_OTEL_SHORTHAND_ENV): - monkeypatch.delenv(name, raising=False) - - -WEAVE_DEST: Final = OtelDestination( - endpoint="https://trace.wandb.ai/otel/v1/traces", - headers={"Authorization": "Basic dGVuYW50"}, - callback_name="weave_otel", -) + with pytest.raises(ValueError, match="LANGFUSE_PUBLIC_KEY"): + langfuse_preset() def _closed_chat_call_kwargs() -> dict[str, object]: @@ -3001,16 +2989,12 @@ class TestPresetDegradation: assert '"name": "chat gpt-4"' not in capfd.readouterr().out assert "langfuse" in config.mapper_names - def test_a_global_langfuse_callback_with_no_credentials_degrades_instead_of_raising( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: + def test_langfuse_still_raises_for_a_global_callback_with_no_credentials(self, monkeypatch): monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) - config: Final = langfuse_preset() - - assert any(spec.owner == ExporterOwner.LANGFUSE_OTEL and spec.requires_headers for spec in config.exporters) - assert "langfuse" in config.mapper_names + with pytest.raises(ValueError, match="LANGFUSE_PUBLIC_KEY"): + langfuse_preset() def test_a_credential_less_proxy_builds_the_gated_logger_beside_a_v2_carrier(self, monkeypatch): from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 @@ -3030,23 +3014,20 @@ class TestPresetDegradation: assert logger is not None assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) - def test_a_credential_less_proxy_with_no_destinations_stays_on_v2_and_exports_nowhere( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - """The proxy builds the operator's callback at startup, before any key or team - destination has resolved, so this is the logger those destinations ride on.""" + def test_a_credential_less_proxy_with_no_destinations_falls_back_to_the_legacy_path(self, monkeypatch): + """Nothing can use a credential-less langfuse here, so the operator has to get + the same story as before v2: the legacy integration, not a global provider + that exports nowhere.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 credential_less_proxy(monkeypatch) monkeypatch.setenv("LITELLM_OTEL_V2", "true") is_otel_v2_enabled.cache_clear() - logger: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", []) + logger = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", []) is_otel_v2_enabled.cache_clear() - assert isinstance(logger, OpenTelemetryV2), "None hands 'langfuse_otel' to the legacy logger" - assert "langfuse" in logger.config.mapper_names - assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) + assert logger is None def test_a_valid_newrelic_base_exporter_survives_without_a_license_key(self, monkeypatch): from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 @@ -3121,17 +3102,17 @@ class TestPresetDegradation: without one is logged, long after the degrade was justified.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - credential_less_weave(monkeypatch) + credential_less_proxy(monkeypatch) monkeypatch.setenv("LITELLM_OTEL_V2", "true") loggers = [build_otel_v2_logger(OpenTelemetryV2Config(exporter="in_memory"))] def with_destination(): - set_request_destinations((WEAVE_DEST,)) - return _maybe_construct_otel_v2("weave_otel", loggers) + set_request_destinations((LANGFUSE_DEST,)) + return _maybe_construct_otel_v2("langfuse_otel", loggers) is_otel_v2_enabled.cache_clear() degraded = in_fresh_context(with_destination) - plain = in_fresh_context(_maybe_construct_otel_v2, "weave_otel", loggers) + plain = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) is_otel_v2_enabled.cache_clear() assert degraded is not None @@ -3188,12 +3169,12 @@ class TestPresetDegradation: callback at all, and the operator keeps the pre-V2 story.""" from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - credential_less_weave(monkeypatch) + credential_less_proxy(monkeypatch) monkeypatch.setenv("LITELLM_OTEL_V2", "true") def run(): - set_request_destinations((WEAVE_DEST,)) - return _maybe_construct_otel_v2("weave_otel", list(registered)) + set_request_destinations((LANGFUSE_DEST,)) + return _maybe_construct_otel_v2("langfuse_otel", list(registered)) is_otel_v2_enabled.cache_clear() logger = in_fresh_context(run) @@ -3457,121 +3438,8 @@ class TestArizeTenantOnly: assert loggers == [at_startup] -class TestLangfuseTenantOnly: - """An operator whose keys and teams each bring their own Langfuse project keeps no - Langfuse credentials of their own; the startup logger then exports nowhere for - traffic without a destination and carries the Langfuse mapper to the destinations.""" - - @staticmethod - def _startup_logger(monkeypatch: pytest.MonkeyPatch, loggers: list[CustomLogger]) -> OpenTelemetryV2: - from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - - credential_less_proxy(monkeypatch) - monkeypatch.setenv("LITELLM_OTEL_V2", "true") - is_otel_v2_enabled.cache_clear() - logger: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) - is_otel_v2_enabled.cache_clear() - assert isinstance(logger, OpenTelemetryV2) - return logger - - def test_the_published_provider_is_the_langfuse_logger(self, monkeypatch: pytest.MonkeyPatch) -> None: - """The published provider carries the key/team fan-out, so its mappers are the - attributes every destination receives.""" - loggers: Final[list[CustomLogger]] = [] - logger: Final = self._startup_logger(monkeypatch, loggers) - - published: Final = publish_global_otel_v2_provider(loggers, lambda _p: None) - - assert published is logger - assert "langfuse" in published.config.mapper_names - - def test_the_startup_logger_delivers_a_key_destination_and_nothing_else( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - from litellm.integrations.otel.plumbing.providers import attach_tenant_fan_out - - capture: Final = _ExporterCapture() - for kind in ("otlp_grpc", "otlp_http", "console"): - monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) - logger: Final = self._startup_logger(monkeypatch, []) - provider: Final = logger.tracer_provider - attach_tenant_fan_out(provider, logger.config) - - def with_destination() -> None: - set_request_destinations(deliverable_destinations((LANGFUSE_DEST,), provider)) - emit(provider) - - in_fresh_context(with_destination) - in_fresh_context(emit, provider, "chat without a destination") - provider.force_flush() - - assert capture.received() == {LANGFUSE_DEST.header_string(): ("chat gpt-4",)} - - def test_the_startup_logger_is_reused_by_later_requests_without_a_destination( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - - loggers: Final[list[CustomLogger]] = [] - at_startup: Final = self._startup_logger(monkeypatch, loggers) - - is_otel_v2_enabled.cache_clear() - later: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) - is_otel_v2_enabled.cache_clear() - - assert later is at_startup - assert loggers == [at_startup] - - @staticmethod - def _key_logger_beside_the_generic_startup_logger( - monkeypatch: pytest.MonkeyPatch, loggers: list[CustomLogger] - ) -> tuple[OpenTelemetryV2, OpenTelemetryV2]: - """``langfuse_otel`` only in a key's logging: the proxy published the generic logger, - which exports to the operator's collector, and a key's request builds the Langfuse one.""" - from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - - credential_less_proxy(monkeypatch) - monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.local:4318") - monkeypatch.setenv("LITELLM_OTEL_V2", "true") - published: Final = publish_global_otel_v2_provider(loggers, lambda _p: None) - - def with_destination() -> OpenTelemetryV2 | None: - set_request_destinations((LANGFUSE_DEST,)) - return _maybe_construct_otel_v2("langfuse_otel", loggers) - - is_otel_v2_enabled.cache_clear() - logger: Final = in_fresh_context(with_destination) - is_otel_v2_enabled.cache_clear() - assert isinstance(logger, OpenTelemetryV2) - return published, logger - - def test_the_operators_collector_gets_one_copy_of_a_call_both_loggers_emit( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - capture: Final = _ExporterCapture() - for kind in ("otlp_grpc", "otlp_http", "console"): - monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) - - for logger in self._key_logger_beside_the_generic_startup_logger(monkeypatch, []): - emit(logger.tracer_provider) - logger.tracer_provider.force_flush() - - assert sum(len(exporter.get_finished_spans()) for _, exporter in capture.built) == 1 - - def test_a_key_destinations_logger_is_reused_by_later_requests_without_a_destination( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 - - loggers: Final[list[CustomLogger]] = [] - for_the_key: Final = self._key_logger_beside_the_generic_startup_logger(monkeypatch, loggers)[1] - - is_otel_v2_enabled.cache_clear() - later: Final = in_fresh_context(_maybe_construct_otel_v2, "langfuse_otel", loggers) - is_otel_v2_enabled.cache_clear() - - assert later is for_the_key - assert loggers == [for_the_key] +class TestLangfuseDestinationFields: + """A key's or team's Langfuse destination gets the Langfuse fields from whichever logger the proxy published.""" def test_a_key_destination_gets_the_langfuse_fields_from_the_generic_startup_logger( self, monkeypatch: pytest.MonkeyPatch diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index 5d37a145de6..a8ead8e3953 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -1,9 +1,12 @@ +import asyncio import base64 +import contextvars import json import os from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import datetime, timezone +from types import SimpleNamespace from typing import Final from unittest.mock import MagicMock, patch @@ -12,12 +15,16 @@ from openai.types.responses import ResponseFunctionToolCall from opentelemetry import trace from opentelemetry.context import Context from opentelemetry.sdk.trace import SpanLimits, TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from opentelemetry.trace import Span import litellm from litellm.constants import REDACTED_BY_LITELLM from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME, OpenTelemetry, OpenTelemetryConfig +from litellm.integrations.otel.model.destination import OtelDestination +from litellm.integrations.otel.plumbing.context import set_request_destinations from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.utils import Choices, ModelResponse, StandardCallbackDynamicParams @@ -1507,3 +1514,54 @@ def test_langfuse_otel_does_not_start_proxy_request_span() -> None: finally: langfuse_provider.shutdown() generic_provider.shutdown() + + +def _spans_the_legacy_logger_exports(destinations: tuple[OtelDestination, ...]) -> tuple[str, ...]: + exporter: Final = InMemorySpanExporter() + provider: Final = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True): + logger: Final = LangfuseOtelLogger(tracer_provider=provider) + at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc) + kwargs: Final = {**CHAT_KWARGS, "litellm_params": {"metadata": {}}} + server_span: Final = provider.get_tracer("proxy").start_span("POST /v1/chat/completions") + + async def log() -> None: + set_request_destinations(destinations) + logger.log_success_event(dict(kwargs), _chat_response("ok"), at, at) + await logger.async_log_failure_event({**kwargs, "exception": ValueError("boom")}, None, at, at) + await logger.async_post_call_failure_hook( + {}, ValueError("boom"), SimpleNamespace(parent_otel_span=server_span, team_id=None, team_alias=None) + ) + + try: + contextvars.copy_context().run(asyncio.run, log()) + return tuple(span.name for span in exporter.get_finished_spans()) + finally: + provider.shutdown() + + +_LANGFUSE_DESTINATION: Final = OtelDestination( + endpoint="http://tenant.local/api/public/otel", + headers={"Authorization": "Basic dGVuYW50"}, + callback_name="langfuse_otel", +) + + +def test_a_request_with_a_langfuse_destination_exports_nothing_through_the_legacy_logger() -> None: + assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,)) == () + + +def test_a_request_without_a_langfuse_destination_still_exports_through_the_legacy_logger() -> None: + logged: Final = ( + "litellm_request", + "raw_gen_ai_request", + "litellm_request", + "Failed Proxy Server Request", + "POST /v1/chat/completions", + ) + + assert _spans_the_legacy_logger_exports(()) == logged + assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION.model_copy(update={"callback_name": "arize"}),)) == ( + logged + ) From f372899f1238a4f41bfcef6591ef821a767a8de3 Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 13:46:52 -0700 Subject: [PATCH 08/13] test(otel): pin the legacy langfuse_otel paths a tenant's project still gets without a destination --- .../test_langfuse_otel_legacy_destinations.py | 171 ++++++++++++++++++ 1 file changed, 171 insertions(+) create mode 100644 tests/integration/observability/test_langfuse_otel_legacy_destinations.py diff --git a/tests/integration/observability/test_langfuse_otel_legacy_destinations.py b/tests/integration/observability/test_langfuse_otel_legacy_destinations.py new file mode 100644 index 00000000000..e136b9fda90 --- /dev/null +++ b/tests/integration/observability/test_langfuse_otel_legacy_destinations.py @@ -0,0 +1,171 @@ +"""Without operator Langfuse keys on OTel v2, Langfuse logging that resolves no key or team destination still reaches +the tenant's own Langfuse project: a key's failure-only entry on a failed call, and a config ``default_team_settings`` +entry. A ``success_and_failure`` key entry sends that project exactly one copy of a failed call.""" + +from __future__ import annotations + +import json +import uuid +from collections.abc import Iterator +from typing import Final, Literal + +import pytest +from integration._support.client import Gateway, eventually, gateway_from_environment +from integration._support.otlp_sink import ( + AuditConfigWriter, + Span, + SpanSinks, + langfuse_callback_variables, + recorded_spans, + spans_for_trace, +) +from integration._support.process import owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue + +OPERATION: Final = "gen_ai.operation.name" +TAGS: Final = ["agent:orders", "run:r1"] +CONFIG_TEAM: Final = f"lf-config-team-{uuid.uuid4().hex[:8]}" +FAIL: Final = "please-fail" + + +def _chat(request: Request) -> Reply: + if FAIL.encode() in request.body: + return Reply( + status=500, body=json.dumps({"error": {"message": "upstream boom", "type": "server_error"}}).encode() + ) + return Reply( + body=json.dumps( + { + "id": f"chatcmpl-{uuid.uuid4().hex}", + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}, + } + ).encode() + ) + + +@pytest.fixture(scope="module") +def provider() -> Iterator[Wire]: + with wire_server(_chat) as wire: + yield wire + + +@pytest.fixture(scope="module") +def proxy( + audit_sinks: SpanSinks, + otel_audit_config: AuditConfigWriter, + tmp_path_factory: pytest.TempPathFactory, +) -> Iterator[Gateway]: + directory: Final = tmp_path_factory.mktemp("langfuse-legacy-destinations") + settings: Final[dict[str, JsonValue]] = { + "callbacks": [], + "default_team_settings": [ + { + "team_id": CONFIG_TEAM, + "success_callback": ["langfuse_otel"], + "callback_vars": dict(langfuse_callback_variables(audit_sinks)), + } + ], + } + with ( + gateway_from_environment() as gateway, + owned_proxy_process( + gateway, + directory, + { + "LITELLM_OTEL_V2": "1", + "OTEL_BSP_SCHEDULE_DELAY": "300", + "OTEL_EXPORTER": "otlp_http", + "OTEL_ENDPOINT": audit_sinks.operator, + }, + config=otel_audit_config(directory, settings), + remove_environment=("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"), + workers=2, + ) as owned, + ): + yield owned.gateway + + +def _tenant_spans(url: str, since: int, trace_id: str) -> tuple[Span, ...]: + return spans_for_trace(recorded_spans(url, since)[1], trace_id) + + +def _tags(value: object) -> object: + """The tags a span carries: the legacy logger writes them as a JSON string, OTel v2 as a list.""" + return json.loads(value) if isinstance(value, str) else value + + +def _settled(spans: tuple[Span, ...]) -> bool: + """The model span and the request's root span have both arrived.""" + return any(OPERATION in span["attributes"] for span in spans) and any( + str(span["name"]).startswith("POST ") for span in spans + ) + + +def _send( + proxy: Gateway, provider: Wire, audit_sinks: SpanSinks, setup: Literal["failure", "config_team", "both"] +) -> tuple[int, tuple[Span, ...]]: + logging_entry: Final[list[JsonValue]] = [ + { + "callback_name": "langfuse_otel", + "callback_type": "failure" if setup == "failure" else "success_and_failure", + "callback_vars": langfuse_callback_variables(audit_sinks), + } + ] + with proxy.scenario() as scenario: + model: Final = scenario.model(api_base=provider.url + "/v1", num_retries=0) + alias: Final = f"lf-legacy-{setup}-{uuid.uuid4().hex}" + key: Final = ( + scenario.key(key_alias=alias, team_id=scenario.team(team_id=CONFIG_TEAM)) + if setup == "config_team" + else scenario.key(key_alias=alias, metadata={"logging": logging_entry}) + ) + since: Final = recorded_spans(audit_sinks.tenant)[0] + trace_id: Final = uuid.uuid4().hex + content: Final = f"lf-legacy-{setup}-{trace_id}" + ("" if setup == "config_team" else f" {FAIL}") + body: Final[dict[str, JsonValue]] = { + "model": model, + "messages": [{"role": "user", "content": content}], + "metadata": {"tags": [*TAGS], "session_id": f"sess-{trace_id}", "trace_user_id": "user-42"}, + } + response: Final = proxy.request( + "POST", + "/v1/chat/completions", + body, + key=key, + headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + ) + tenant: Final = eventually( + lambda: _tenant_spans(audit_sinks.tenant, since, trace_id), + _settled if setup == "both" else lambda spans: any(span["attributes"].get("user.id") for span in spans), + seconds=40, + ) + return response.status_code, tenant + + +@pytest.mark.parametrize("setup", ["failure", "config_team"]) +def test_a_tenant_langfuse_project_without_a_destination_still_gets_the_trace( + setup: Literal["failure", "config_team"], proxy: Gateway, provider: Wire, audit_sinks: SpanSinks +) -> None: + status, tenant = _send(proxy, provider, audit_sinks, setup) + + assert status == (500 if setup == "failure" else 200) + assert [ + (span["attributes"]["user.id"], _tags(span["attributes"].get("langfuse.trace.tags"))) + for span in tenant + if "user.id" in span["attributes"] + ] == [("user-42", TAGS)], [span["name"] for span in tenant] + + +def test_a_failed_call_reaches_a_success_and_failure_destination_once( + proxy: Gateway, provider: Wire, audit_sinks: SpanSinks +) -> None: + status, tenant = _send(proxy, provider, audit_sinks, "both") + + assert status == 500 + assert len([span for span in tenant if OPERATION in span["attributes"]]) == 1, [span["name"] for span in tenant] + assert "litellm_request" not in {span["name"] for span in tenant}, [span["name"] for span in tenant] From cea1923222372b289c794ab0cd26fb01c9a74dd9 Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 14:40:04 -0700 Subject: [PATCH 09/13] fix(otel): shed a destination's Langfuse request metadata before its other Langfuse fields At its attribute limit a span sheds the Langfuse fields only a destination asked for first, in the mapper's order, so the generation marker went before the request metadata blob. Put the request metadata first, as a Langfuse-configured logger already does. --- litellm/integrations/otel/emitter.py | 8 ++- .../otel/test_otel_v2_destinations.py | 52 +++++++++++++------ 2 files changed, 42 insertions(+), 18 deletions(-) diff --git a/litellm/integrations/otel/emitter.py b/litellm/integrations/otel/emitter.py index 3784efcc2f2..87a432ee7ce 100644 --- a/litellm/integrations/otel/emitter.py +++ b/litellm/integrations/otel/emitter.py @@ -169,7 +169,6 @@ class SpanEmitter: self._mappers: list[AttributeMapper] = ( list(mappers) if mappers is not None else resolve_mappers(config.mapper_names) ) - # A key's or team's Langfuse destination reads the Langfuse vocabulary, which this logger may not write. self._langfuse: Final = ( None if any(isinstance(mapper, LangfuseMapper) for mapper in self._mappers) else LangfuseMapper() ) @@ -302,7 +301,12 @@ class SpanEmitter: ) configured: Final = {key: value for mapper in self._mappers for key, value in mapper.map(data).items()} for_langfuse: Final = ( - self._langfuse.map(data) + dict( + sorted( + self._langfuse.map(data).items(), + key=lambda item: item[0] not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES, + ) + ) if self._langfuse is not None and ExporterOwner.LANGFUSE_OTEL in destination_backends() else _NO_ATTRIBUTES ) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index c0b81c49a4f..572960726e2 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -35,6 +35,7 @@ from litellm.integrations.otel.logger import ( publish_global_otel_v2_provider, ) from litellm.integrations.otel.mappers import resolve_mappers +from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES from litellm.integrations.otel.model.config import ( CaptureMessageContent, ExporterOwner, @@ -3450,7 +3451,7 @@ class TestLangfuseDestinationFields: credential_less_proxy(monkeypatch) capture: Final = _ExporterCapture() for kind in ("otlp_grpc", "otlp_http", "console"): - monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) + monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) # pyright: ignore[reportPrivateUsage] # the factory table is the seam for capturing each exporter config: Final = OpenTelemetryV2Config() provider: Final = build_tracer_provider(config, tenant_overrides=True) attach_tenant_fan_out(provider, config) @@ -3480,24 +3481,43 @@ class TestLangfuseDestinationFields: assert "langfuse.observation.metadata" in to_langfuse assert "langfuse.observation.metadata" not in received[None] + @staticmethod + def _attributes( + destinations: tuple[OtelDestination, ...], limit: int | None, call: LLMCallSpanData + ) -> dict[str, object]: + exporter: Final = InMemorySpanExporter() + provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=limit)) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), OpenTelemetryV2Config()) + + def run() -> None: + set_request_destinations(destinations) + emitter.emit(SpanRole.LLM_CALL, call) + + in_fresh_context(run) + return dict(exporter.get_finished_spans()[0].attributes or {}) + def test_a_span_at_its_attribute_limit_sheds_the_destinations_langfuse_fields_first(self) -> None: - def attributes(destinations: tuple[OtelDestination, ...], limit: int | None) -> dict[str, object]: - exporter: Final = InMemorySpanExporter() - provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=limit)) - provider.add_span_processor(SimpleSpanProcessor(exporter)) - emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), OpenTelemetryV2Config()) - - def run() -> None: - set_request_destinations(destinations) - emitter.emit(SpanRole.LLM_CALL, _MODEL_CALL_WITH_CONTENT) - - in_fresh_context(run) - return dict(exporter.get_finished_spans()[0].attributes or {}) - - generic: Final = attributes((), None) + generic: Final = self._attributes((), None, _MODEL_CALL_WITH_CONTENT) assert "user.id" not in generic - assert attributes((LANGFUSE_DEST,), len(generic)) == generic, "the generic attributes all survive" + assert self._attributes((LANGFUSE_DEST,), len(generic), _MODEL_CALL_WITH_CONTENT) == generic, ( + "the generic attributes all survive" + ) + + def test_a_span_at_its_attribute_limit_sheds_the_destinations_request_metadata_before_its_other_fields( + self, + ) -> None: + call: Final = replace( + _MODEL_CALL_WITH_CONTENT, request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}) + ) + whole: Final = self._attributes((LANGFUSE_DEST,), None, call) + + squeezed: Final = self._attributes((LANGFUSE_DEST,), len(whole) - 1, call) + + assert len(squeezed) == len(whole) - 1 + assert whole.keys() - squeezed.keys() <= LANGFUSE_REQUEST_METADATA_ATTRIBUTES + assert squeezed["langfuse.observation.type"] == "generation" class TestContextIsolation: From 0f1cfe4365fad1c2ddb8e8a341c172b25b7bdf2c Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 14:40:04 -0700 Subject: [PATCH 10/13] fix(otel): keep the legacy langfuse_otel span for a destination's call rejected before any provider The OTel v2 fan-out records no call for a request the proxy rejects before a provider (rate limit, pre-call hook), so standing the legacy logger down there lost the user, session and tags main sent the tenant. Stand it down only for a call the fan-out records, using the same check the v2 logger uses. The tenant and collector integration tests now hold their exact counts for a few seconds, so a late duplicate copy fails them. --- .../integrations/langfuse/langfuse_otel.py | 15 ++- litellm/integrations/otel/logger.py | 4 +- litellm/integrations/otel/model/metadata.py | 5 + .../test_langfuse_otel_legacy_destinations.py | 125 +++++++++++++----- .../test_langfuse_otel_tenant_only.py | 49 ++++--- tests/unit/integrations/test_langfuse_otel.py | 44 ++++-- 6 files changed, 169 insertions(+), 73 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index dcccd1155c5..ee75e6501e8 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -128,6 +128,17 @@ def _starts_its_own_trace(span: object, kwargs: Mapping[str, object]) -> bool: ) +def _a_destination_records_the_call(kwargs: Mapping[str, object]) -> bool: + """Whether a key's or team's ``langfuse_otel`` destination gets this call from the OTel v2 fan-out, which records + every call but one whose logging payload shows it never reached a provider.""" + from litellm.integrations.otel.model.metadata import LLMCallEvent + + if not _a_destination_carries_the_trace(): + return False + call: Final = LLMCallEvent.from_dict(kwargs) + return call.payload is None or call.reached_a_provider + + def _a_destination_carries_the_trace() -> bool: """Whether this request resolved a key's or team's ``langfuse_otel`` destination, whose fan-out sends it the whole trace.""" @@ -557,8 +568,8 @@ class LangfuseOtelLogger(OpenTelemetry): kwargs: dict[str, object], # mutable-ok: the v1 dedupe marker is written into the request's kwargs *scope: object, ) -> bool: - """Whether to emit this span: once per request, and never for a request a Langfuse destination carries.""" - return not _a_destination_carries_the_trace() and super()._emit_once(kwargs, *scope) # pyright: ignore[reportUnknownMemberType] # the v1 hook takes a bare dict + """Whether to emit this span: once per request, and never for a call a Langfuse destination records.""" + return not _a_destination_records_the_call(kwargs) and super()._emit_once(kwargs, *scope) # pyright: ignore[reportUnknownMemberType] # the v1 hook takes a bare dict async def async_post_call_failure_hook( self, diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index e09226ac9cd..cb5a6bd82ea 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -541,9 +541,7 @@ class OpenTelemetryV2(CustomLogger): # gate rejection carries ``is_no_upstream_call`` and gets no span. A cache # hit never hands off to a provider, so ``cache_hit`` substitutes for the # upstream stamp, since the payload proves the call completed. - if carrier is None and ( - call.is_no_upstream_call or call.payload is None or (not call.upstream_started and not call.cache_hit) - ): + if carrier is None and not call.reached_a_provider: return None try: return self._finish_carrier(carrier, call, start_time, end_time) diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index b96ed55872d..b92f478c7a2 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -271,6 +271,11 @@ class LLMCallEvent: trace: TraceControls session_id: str | None + @property + def reached_a_provider(self) -> bool: + """Whether the call was handed to a provider or answered from the response cache.""" + return not self.is_no_upstream_call and self.payload is not None and (self.upstream_started or self.cache_hit) + @classmethod def from_dict(cls, kwargs: Mapping[str, object]) -> LLMCallEvent: raw_payload: Final = kwargs.get("standard_logging_object") diff --git a/tests/integration/observability/test_langfuse_otel_legacy_destinations.py b/tests/integration/observability/test_langfuse_otel_legacy_destinations.py index e136b9fda90..aaff856fda8 100644 --- a/tests/integration/observability/test_langfuse_otel_legacy_destinations.py +++ b/tests/integration/observability/test_langfuse_otel_legacy_destinations.py @@ -1,16 +1,17 @@ -"""Without operator Langfuse keys on OTel v2, Langfuse logging that resolves no key or team destination still reaches -the tenant's own Langfuse project: a key's failure-only entry on a failed call, and a config ``default_team_settings`` -entry. A ``success_and_failure`` key entry sends that project exactly one copy of a failed call.""" +"""Without operator Langfuse keys on OTel v2, Langfuse logging the OTel v2 fan-out does not carry still reaches the +tenant's own Langfuse project: a key's failure-only entry on a failed call, a config ``default_team_settings`` entry, +and a key destination's call the proxy rejected before any provider. A ``success_and_failure`` key entry sends that +project exactly one copy of a failed call.""" from __future__ import annotations import json import uuid -from collections.abc import Iterator +from collections.abc import Callable, Iterator from typing import Final, Literal import pytest -from integration._support.client import Gateway, eventually, gateway_from_environment +from integration._support.client import Gateway, gateway_from_environment, held from integration._support.otlp_sink import ( AuditConfigWriter, Span, @@ -106,16 +107,59 @@ def _settled(spans: tuple[Span, ...]) -> bool: ) -def _send( - proxy: Gateway, provider: Wire, audit_sinks: SpanSinks, setup: Literal["failure", "config_team", "both"] -) -> tuple[int, tuple[Span, ...]]: - logging_entry: Final[list[JsonValue]] = [ +def _logging_entry(audit_sinks: SpanSinks, callback_type: str) -> list[JsonValue]: + return [ { "callback_name": "langfuse_otel", - "callback_type": "failure" if setup == "failure" else "success_and_failure", + "callback_type": callback_type, "callback_vars": langfuse_callback_variables(audit_sinks), } ] + + +def _chat_status(proxy: Gateway, model: str, key: str, trace_id: str, content: str) -> int: + body: Final[dict[str, JsonValue]] = { + "model": model, + "messages": [{"role": "user", "content": content}], + "metadata": {"tags": [*TAGS], "session_id": f"sess-{trace_id}", "trace_user_id": "user-42"}, + } + return proxy.request( + "POST", + "/v1/chat/completions", + body, + key=key, + headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + ).status_code + + +def _trace_fields(spans: tuple[Span, ...]) -> list[tuple[object, object]]: + return [ + (span["attributes"]["user.id"], _tags(span["attributes"].get("langfuse.trace.tags"))) + for span in spans + if "user.id" in span["attributes"] + ] + + +def _carries_the_trace_fields_once(spans: tuple[Span, ...]) -> bool: + return _trace_fields(spans) == [("user-42", TAGS)] + + +def _one_failed_call(spans: tuple[Span, ...]) -> bool: + return ( + _settled(spans) + and len([span for span in spans if OPERATION in span["attributes"]]) == 1 + and "litellm_request" not in {span["name"] for span in spans} + ) + + +def _send( + proxy: Gateway, + provider: Wire, + audit_sinks: SpanSinks, + setup: Literal["failure", "config_team", "both"], + expected: Callable[[tuple[Span, ...]], bool], +) -> tuple[int, tuple[Span, ...]]: + logging_entry: Final = _logging_entry(audit_sinks, "failure" if setup == "failure" else "success_and_failure") with proxy.scenario() as scenario: model: Final = scenario.model(api_base=provider.url + "/v1", num_retries=0) alias: Final = f"lf-legacy-{setup}-{uuid.uuid4().hex}" @@ -127,45 +171,54 @@ def _send( since: Final = recorded_spans(audit_sinks.tenant)[0] trace_id: Final = uuid.uuid4().hex content: Final = f"lf-legacy-{setup}-{trace_id}" + ("" if setup == "config_team" else f" {FAIL}") - body: Final[dict[str, JsonValue]] = { - "model": model, - "messages": [{"role": "user", "content": content}], - "metadata": {"tags": [*TAGS], "session_id": f"sess-{trace_id}", "trace_user_id": "user-42"}, - } - response: Final = proxy.request( - "POST", - "/v1/chat/completions", - body, - key=key, - headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + status: Final = _chat_status(proxy, model, key, trace_id, content) + tenant: Final = held( + lambda: _tenant_spans(audit_sinks.tenant, since, trace_id), expected, holding=3, seconds=40 ) - tenant: Final = eventually( - lambda: _tenant_spans(audit_sinks.tenant, since, trace_id), - _settled if setup == "both" else lambda spans: any(span["attributes"].get("user.id") for span in spans), - seconds=40, - ) - return response.status_code, tenant + return status, tenant @pytest.mark.parametrize("setup", ["failure", "config_team"]) def test_a_tenant_langfuse_project_without_a_destination_still_gets_the_trace( setup: Literal["failure", "config_team"], proxy: Gateway, provider: Wire, audit_sinks: SpanSinks ) -> None: - status, tenant = _send(proxy, provider, audit_sinks, setup) + status, tenant = _send(proxy, provider, audit_sinks, setup, _carries_the_trace_fields_once) assert status == (500 if setup == "failure" else 200) - assert [ - (span["attributes"]["user.id"], _tags(span["attributes"].get("langfuse.trace.tags"))) - for span in tenant - if "user.id" in span["attributes"] - ] == [("user-42", TAGS)], [span["name"] for span in tenant] + assert _trace_fields(tenant) == [("user-42", TAGS)], [span["name"] for span in tenant] + + +def test_a_key_destinations_call_rejected_before_any_provider_still_gets_the_trace_fields( + proxy: Gateway, provider: Wire, audit_sinks: SpanSinks +) -> None: + with proxy.scenario() as scenario: + model: Final = scenario.model(api_base=provider.url + "/v1", num_retries=0) + key: Final = scenario.key( + key_alias=f"lf-legacy-rejected-{uuid.uuid4().hex}", + metadata={"logging": _logging_entry(audit_sinks, "success_and_failure")}, + rpm_limit=1, + ) + since: Final = recorded_spans(audit_sinks.tenant)[0] + statuses: Final = { + trace_id: _chat_status(proxy, model, key, trace_id, f"lf-legacy-rejected-{trace_id}") + for trace_id in [uuid.uuid4().hex for _ in range(4)] + } + rejected: Final = [trace_id for trace_id, status in statuses.items() if status == 429] + assert rejected, statuses + tenant: Final = held( + lambda: _tenant_spans(audit_sinks.tenant, since, rejected[0]), + _carries_the_trace_fields_once, + holding=3, + seconds=40, + ) + + assert _trace_fields(tenant) == [("user-42", TAGS)], [span["name"] for span in tenant] def test_a_failed_call_reaches_a_success_and_failure_destination_once( proxy: Gateway, provider: Wire, audit_sinks: SpanSinks ) -> None: - status, tenant = _send(proxy, provider, audit_sinks, "both") + status, tenant = _send(proxy, provider, audit_sinks, "both", _one_failed_call) assert status == 500 - assert len([span for span in tenant if OPERATION in span["attributes"]]) == 1, [span["name"] for span in tenant] - assert "litellm_request" not in {span["name"] for span in tenant}, [span["name"] for span in tenant] + assert _one_failed_call(tenant), [span["name"] for span in tenant] diff --git a/tests/integration/observability/test_langfuse_otel_tenant_only.py b/tests/integration/observability/test_langfuse_otel_tenant_only.py index 9ea50faf439..1163a4a52b2 100644 --- a/tests/integration/observability/test_langfuse_otel_tenant_only.py +++ b/tests/integration/observability/test_langfuse_otel_tenant_only.py @@ -1,15 +1,15 @@ -"""A key's or team's own Langfuse project gets the Langfuse trace fields when the operator has no Langfuse keys, -whether or not ``langfuse_otel`` is listed in ``litellm_settings.callbacks``.""" +"""A key's or team's own Langfuse project gets one copy of the call with the Langfuse trace fields when the operator +has no Langfuse keys, whether or not ``langfuse_otel`` is listed in ``litellm_settings.callbacks``.""" from __future__ import annotations import json import uuid from collections.abc import Iterator -from typing import Final, Literal +from typing import Final, Literal, cast import pytest -from integration._support.client import Gateway, eventually, gateway_from_environment +from integration._support.client import Gateway, gateway_from_environment, held from integration._support.otlp_sink import ( AuditConfigWriter, Span, @@ -24,10 +24,10 @@ from pydantic import JsonValue OPERATION: Final = "gen_ai.operation.name" TAGS: Final = ["agent:orders", "run:r1"] +LEGACY_SPANS: Final = frozenset({"litellm_request", "raw_gen_ai_request"}) -def _chat(request: Request) -> Reply: - marker: Final = json.loads(request.body)["messages"][-1]["content"] +def _chat(_request: Request) -> Reply: return Reply( body=json.dumps( { @@ -35,7 +35,7 @@ def _chat(request: Request) -> Reply: "object": "chat.completion", "created": 1, "model": "gpt-4o-mini", - "choices": [{"index": 0, "message": {"role": "assistant", "content": marker}, "finish_reason": "stop"}], + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}, } ).encode() @@ -67,7 +67,7 @@ def proxy( "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": audit_sinks.operator, }, - config=otel_audit_config(directory, {"callbacks": request.param}), + config=otel_audit_config(directory, {"callbacks": cast("list[JsonValue]", request.param)}), remove_environment=("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"), workers=2, ) as owned, @@ -75,10 +75,16 @@ def proxy( yield owned.gateway -def _model_spans(url: str, since: int, trace_id: str) -> tuple[Span, ...]: - return tuple( - span for span in spans_for_trace(recorded_spans(url, since)[1], trace_id) if OPERATION in span["attributes"] - ) +def _trace_spans(url: str, since: int, trace_id: str) -> tuple[Span, ...]: + return spans_for_trace(recorded_spans(url, since)[1], trace_id) + + +def _model_spans(spans: tuple[Span, ...]) -> tuple[Span, ...]: + return tuple(span for span in spans if OPERATION in span["attributes"]) + + +def _one_model_span(spans: tuple[Span, ...]) -> bool: + return len(_model_spans(spans)) == 1 and not LEGACY_SPANS & {str(span["name"]) for span in spans} @pytest.mark.parametrize("level", ["key", "team"]) @@ -103,24 +109,25 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k since: Final = (recorded_spans(audit_sinks.tenant)[0], recorded_spans(audit_sinks.operator)[0]) trace_id: Final = uuid.uuid4().hex session: Final = f"sess-{trace_id}" + body: Final[dict[str, JsonValue]] = { + "model": model, + "messages": [{"role": "user", "content": f"lf-tenant-only-{trace_id}"}], + "metadata": {"tags": [*TAGS], "session_id": session, "trace_user_id": "user-42"}, + } response: Final = proxy.request( "POST", "/v1/chat/completions", - { - "model": model, - "messages": [{"role": "user", "content": f"lf-tenant-only-{trace_id}"}], - "metadata": {"tags": TAGS, "session_id": session, "trace_user_id": "user-42"}, - }, + body, key=key, headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, ) assert response.status_code == 200, response.text - tenant: Final = eventually( - lambda: _model_spans(audit_sinks.tenant, since[0], trace_id), lambda spans: bool(spans), seconds=40 + tenant: Final = _model_spans( + held(lambda: _trace_spans(audit_sinks.tenant, since[0], trace_id), _one_model_span, holding=3, seconds=40) ) - operator: Final = eventually( - lambda: _model_spans(audit_sinks.operator, since[1], trace_id), lambda spans: bool(spans), seconds=40 + operator: Final = _model_spans( + held(lambda: _trace_spans(audit_sinks.operator, since[1], trace_id), _one_model_span, holding=3, seconds=40) ) assert [ diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index a8ead8e3953..a2803719c75 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -6,7 +6,7 @@ import os from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import datetime, timezone -from types import SimpleNamespace +from types import MappingProxyType from typing import Final from unittest.mock import MagicMock, patch @@ -20,11 +20,12 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanE from opentelemetry.trace import Span import litellm -from litellm.constants import REDACTED_BY_LITELLM +from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, REDACTED_BY_LITELLM from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME, OpenTelemetry, OpenTelemetryConfig from litellm.integrations.otel.model.destination import OtelDestination from litellm.integrations.otel.plumbing.context import set_request_destinations +from litellm.proxy._types import UserAPIKeyAuth from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.utils import Choices, ModelResponse, StandardCallbackDynamicParams @@ -1516,23 +1517,26 @@ def test_langfuse_otel_does_not_start_proxy_request_span() -> None: generic_provider.shutdown() -def _spans_the_legacy_logger_exports(destinations: tuple[OtelDestination, ...]) -> tuple[str, ...]: +_HOOK_CALL: Final[Mapping[str, object]] = MappingProxyType({}) + + +def _spans_the_legacy_logger_exports( + destinations: tuple[OtelDestination, ...], call: Mapping[str, object] = _HOOK_CALL +) -> tuple[str, ...]: exporter: Final = InMemorySpanExporter() provider: Final = TracerProvider() provider.add_span_processor(SimpleSpanProcessor(exporter)) with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True): logger: Final = LangfuseOtelLogger(tracer_provider=provider) at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc) - kwargs: Final = {**CHAT_KWARGS, "litellm_params": {"metadata": {}}} + kwargs: Final[dict[str, object]] = {**CHAT_KWARGS, **call, "litellm_params": {"metadata": {}}} server_span: Final = provider.get_tracer("proxy").start_span("POST /v1/chat/completions") async def log() -> None: set_request_destinations(destinations) - logger.log_success_event(dict(kwargs), _chat_response("ok"), at, at) - await logger.async_log_failure_event({**kwargs, "exception": ValueError("boom")}, None, at, at) - await logger.async_post_call_failure_hook( - {}, ValueError("boom"), SimpleNamespace(parent_otel_span=server_span, team_id=None, team_alias=None) - ) + logger.log_success_event(dict(kwargs), _chat_response("ok"), at, at) # pyright: ignore[reportUnknownMemberType] # the v1 logger hooks are untyped + await logger.async_log_failure_event({**kwargs, "exception": ValueError("boom")}, None, at, at) # pyright: ignore[reportUnknownMemberType] # the v1 logger hooks are untyped + await logger.async_post_call_failure_hook({}, ValueError("boom"), UserAPIKeyAuth(parent_otel_span=server_span)) try: contextvars.copy_context().run(asyncio.run, log()) @@ -1548,8 +1552,26 @@ _LANGFUSE_DESTINATION: Final = OtelDestination( ) -def test_a_request_with_a_langfuse_destination_exports_nothing_through_the_legacy_logger() -> None: - assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,)) == () +_PROVIDER_CALL: Final[Mapping[str, object]] = MappingProxyType( + {"standard_logging_object": {"id": "c1"}, "api_call_start_time": datetime(2026, 1, 1, tzinfo=timezone.utc)} +) + + +@pytest.mark.parametrize("call", [_HOOK_CALL, _PROVIDER_CALL], ids=["hook", "provider-call"]) +def test_a_request_with_a_langfuse_destination_exports_nothing_through_the_legacy_logger( + call: Mapping[str, object], +) -> None: + assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,), call) == () + + +def test_a_call_rejected_before_any_provider_still_exports_through_the_legacy_logger() -> None: + rejected: Final = {"standard_logging_object": {"id": "c1"}, LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL: True} + + assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,), rejected) == ( + "litellm_request", + "raw_gen_ai_request", + "litellm_request", + ) def test_a_request_without_a_langfuse_destination_still_exports_through_the_legacy_logger() -> None: From 732de4340ed5580e94fe5ad66646f6ea7ec23a34 Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 15:48:51 -0700 Subject: [PATCH 11/13] fix(otel): give a key's Langfuse destination the langfuse_environment the legacy logger wrote The destination's resource carries langfuse.environment from the key's or team's langfuse_environment, else LANGFUSE_TRACING_ENVIRONMENT, so a key's project files its traces under the environment it chose instead of default. Only that destination's export carries it. --- .../integrations/otel/presets/destinations.py | 26 +++++--- .../test_langfuse_otel_tenant_only.py | 37 ++++++++++++ .../otel/test_otel_v2_destinations.py | 59 +++++++++++++++++++ 3 files changed, 114 insertions(+), 8 deletions(-) diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py index a705fc16bbc..82828574b3e 100644 --- a/litellm/integrations/otel/presets/destinations.py +++ b/litellm/integrations/otel/presets/destinations.py @@ -111,7 +111,7 @@ _REQUIRED_HEADERS_BY_CALLBACK: Final[Mapping[str, frozenset[str]]] = MappingProx _NO_ATTRS: Final[Mapping[str, str]] = MappingProxyType({}) -def _arize_project() -> Mapping[str, str]: +def _arize_project(_params: StandardCallbackDynamicParams) -> Mapping[str, str]: from litellm.integrations.arize.arize import ArizeLogger project: Final = ArizeLogger.get_arize_config().project_name @@ -124,10 +124,18 @@ def _arize_default_project() -> Mapping[str, str]: return MappingProxyType({"model_id": ARIZE_DEFAULT_PROJECT}) -#: Resource attributes the operator chose for a backend, applied to every export the -#: team receives over the span's own resource: ``ARIZE_PROJECT_NAME`` for Arize. -_RESOURCE_ATTRIBUTES_BY_CALLBACK: Final[Mapping[str, Callable[[], Mapping[str, str]]]] = MappingProxyType( - {"arize": _arize_project} +def _langfuse_environment(params: StandardCallbackDynamicParams) -> Mapping[str, str]: + """The key's or team's ``langfuse_environment``, else ``LANGFUSE_TRACING_ENVIRONMENT``.""" + from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes + + environment: Final = params.get("langfuse_environment") or os.environ.get("LANGFUSE_TRACING_ENVIRONMENT") + return MappingProxyType({LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value: environment} if environment else {}) + + +#: Resource attributes a backend gets on every export the team receives, over the +#: span's own resource: ``ARIZE_PROJECT_NAME`` for Arize, the Langfuse environment. +_RESOURCE_ATTRIBUTES_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], Mapping[str, str]]]] = ( + MappingProxyType({"arize": _arize_project, "langfuse_otel": _langfuse_environment}) ) #: Resource attributes a backend needs on every export to accept it, filled only where @@ -137,12 +145,14 @@ _RESOURCE_DEFAULTS_BY_CALLBACK: Final[Mapping[str, Callable[[], Mapping[str, str ) -def _resource_attributes(callback_name: str, service_name: str | None) -> Mapping[str, str]: +def _resource_attributes( + callback_name: str, service_name: str | None, params: StandardCallbackDynamicParams +) -> Mapping[str, str]: backend: Final = _RESOURCE_ATTRIBUTES_BY_CALLBACK.get(callback_name) return MappingProxyType( { **({"service.name": service_name} if service_name else {}), - **(backend() if backend is not None else {}), + **(backend(params) if backend is not None else {}), } ) @@ -214,7 +224,7 @@ def destination_for( return OtelDestination( endpoint=endpoint, headers=MappingProxyType(dict(headers)), - resource_attributes=_resource_attributes(callback_name, service_name), + resource_attributes=_resource_attributes(callback_name, service_name, params), resource_defaults=_resource_defaults(callback_name), callback_name=callback_name, protocol=protocol, diff --git a/tests/integration/observability/test_langfuse_otel_tenant_only.py b/tests/integration/observability/test_langfuse_otel_tenant_only.py index 1163a4a52b2..88398e367b2 100644 --- a/tests/integration/observability/test_langfuse_otel_tenant_only.py +++ b/tests/integration/observability/test_langfuse_otel_tenant_only.py @@ -137,3 +137,40 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"]) assert len(operator) == 1, [span["name"] for span in operator] assert "langfuse.observation.metadata" not in operator[0]["attributes"], sorted(operator[0]["attributes"]) + + +def test_a_tenant_langfuse_project_gets_the_keys_langfuse_environment( + proxy: Gateway, provider: Wire, audit_sinks: SpanSinks +) -> None: + logging: Final[list[JsonValue]] = [ + { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": {**langfuse_callback_variables(audit_sinks), "langfuse_environment": "prod"}, + } + ] + with proxy.scenario() as scenario: + model: Final = scenario.model(api_base=provider.url + "/v1") + key: Final = scenario.key(key_alias=f"lf-tenant-env-{uuid.uuid4().hex}", metadata={"logging": logging}) + since: Final = (recorded_spans(audit_sinks.tenant)[0], recorded_spans(audit_sinks.operator)[0]) + trace_id: Final = uuid.uuid4().hex + response: Final = proxy.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": f"lf-tenant-env-{trace_id}"}]}, + key=key, + headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + ) + assert response.status_code == 200, response.text + + tenant: Final = held( + lambda: _trace_spans(audit_sinks.tenant, since[0], trace_id), _one_model_span, holding=3, seconds=40 + ) + operator: Final = held( + lambda: _trace_spans(audit_sinks.operator, since[1], trace_id), _one_model_span, holding=3, seconds=40 + ) + + assert [span["resource"].get("langfuse.environment") for span in tenant] == ["prod"] * len(tenant), [ + span["name"] for span in tenant + ] + assert not any("langfuse.environment" in span["resource"] for span in operator), [span["name"] for span in operator] diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 572960726e2..d26d238b0b7 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -3481,6 +3481,65 @@ class TestLangfuseDestinationFields: assert "langfuse.observation.metadata" in to_langfuse assert "langfuse.observation.metadata" not in received[None] + @pytest.mark.parametrize( + ("key_environment", "tracing_environment", "expected"), + [ + ("prod", None, {"langfuse.environment": "prod"}), + ("prod", "staging", {"langfuse.environment": "prod"}), + (None, "staging", {"langfuse.environment": "staging"}), + (None, None, {}), + ], + ids=["key", "key-over-env", "env", "neither"], + ) + def test_a_langfuse_destination_names_the_environment_the_legacy_logger_wrote( + self, + monkeypatch: pytest.MonkeyPatch, + key_environment: str | None, + tracing_environment: str | None, + expected: Mapping[str, str], + ) -> None: + if tracing_environment is None: + monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False) + else: + monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", tracing_environment) + + destination: Final = destination_for( + "langfuse_otel", + StandardCallbackDynamicParams( + langfuse_public_key="pk", langfuse_secret_key="sk", langfuse_environment=key_environment + ), + ) + + assert destination is not None + assert dict(destination.resource_attributes) == expected + + def test_only_the_langfuse_destination_gets_the_keys_environment(self, monkeypatch: pytest.MonkeyPatch) -> None: + credential_less_proxy(monkeypatch) + capture: Final = _ExporterCapture() + for kind in ("otlp_grpc", "otlp_http", "console"): + monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) # pyright: ignore[reportPrivateUsage] # the factory table is the seam for capturing each exporter + config: Final = OpenTelemetryV2Config() + provider: Final = build_tracer_provider(config, tenant_overrides=True) + attach_tenant_fan_out(provider, config) + emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config) + destination: Final = destination_for( + "langfuse_otel", {"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_environment": "prod"} + ) + assert destination is not None + + def with_destination() -> None: + set_request_destinations(deliverable_destinations((destination,), provider)) + emitter.emit(SpanRole.LLM_CALL, _MODEL_CALL_WITHOUT_CONTENT) + + in_fresh_context(with_destination) + provider.force_flush() + + environments: Final = { + headers: exporter.get_finished_spans()[0].resource.attributes.get("langfuse.environment") + for headers, exporter in capture.built + } + assert environments == {destination.header_string(): "prod", None: None} + @staticmethod def _attributes( destinations: tuple[OtelDestination, ...], limit: int | None, call: LLMCallSpanData From f74c14ecc9a53be341c603f7f3983e7c2f145cfc Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 16:29:40 -0700 Subject: [PATCH 12/13] fix(otel): keep the Langfuse fields written for a key's Langfuse destination off every other exporter --- litellm/integrations/otel/mappers/langfuse.py | 12 +++ .../integrations/otel/plumbing/providers.py | 78 +++++++++++++------ .../test_langfuse_otel_tenant_only.py | 4 +- .../otel/test_otel_v2_destinations.py | 58 +++++++++++++- 4 files changed, 126 insertions(+), 26 deletions(-) diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index b55d9746f63..5aed173254c 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -103,6 +103,18 @@ class LangfuseMapper: ), } + #: Every attribute this mapper can write on a model call. + LLM_CALL_ATTRIBUTES: Final[frozenset[str]] = frozenset( + { + *_LLM_CALL_ATTRS, + *_BLOB_ATTRS, + LANGFUSE_TRACE_NAME, + LANGFUSE_TRACE_USER_ID, + LANGFUSE_TRACE_SESSION_ID, + LANGFUSE_TRACE_TAGS, + } + ) + def map(self, data: SpanData) -> AttributeMap: match data: case LLMCallSpanData(): diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index a65a75d2128..bcbeb5bd2f3 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -46,6 +46,7 @@ from litellm.integrations.otel.mappers.langfuse import ( LANGFUSE_OBSERVATION_OUTPUT, LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LANGFUSE_TRACE_NAME, + LangfuseMapper, ) from litellm.integrations.otel.model.config import ( CaptureMessageContent, @@ -488,19 +489,25 @@ def _is_litellm_span(span: ReadableSpan) -> bool: 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. +def without_destination_langfuse_attributes(span: ReadableSpan, langfuse_mapped: bool) -> ReadableSpan: + """``span`` without the Langfuse attributes only a Langfuse destination asked for. - Only LiteLLM's own spans lose it: any other span keeps every key its own code set. + That is the request metadata the Langfuse mapper writes and, on a model call written by + a logger whose own mappers leave the Langfuse vocabulary out (``langfuse_mapped`` off), + every Langfuse attribute, since that logger wrote them for a key's or team's Langfuse + destination alone. Only LiteLLM's own spans lose them: any other span keeps every key + its own code set. """ attributes: Final = span.attributes or _NO_ATTRIBUTES - if not (_is_litellm_span(span) and any(key in LANGFUSE_REQUEST_METADATA_ATTRIBUTES for key in attributes)): + dropped: Final = ( + LANGFUSE_REQUEST_METADATA_ATTRIBUTES + if langfuse_mapped or not is_llm_call_span(span) + else LangfuseMapper.LLM_CALL_ATTRIBUTES + ) + if not (_is_litellm_span(span) and any(key in dropped for key in attributes)): return span return _CountedSpanView( - span, - MappingProxyType( - {key: value for key, value in attributes.items() if key not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES} - ), + span, MappingProxyType({key: value for key, value in attributes.items() if key not in dropped}) ) @@ -656,6 +663,7 @@ def _for_destination( span: ReadableSpan, destination: "OtelDestination", default_capture: str, + langfuse_mapped: bool, ) -> ReadableSpan: """The view of ``span`` a tenant destination receives. @@ -670,13 +678,16 @@ 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 on LiteLLM's own spans. The span itself - stays, so the tenant still gets the whole trace tree. + Langfuse attributes only a Langfuse destination asked for (see + :func:`without_destination_langfuse_attributes`). The span itself stays, so the + tenant still gets the whole trace tree. """ extra: Final = destination.resource_attributes defaults: Final = destination.resource_defaults source: Final = ( - span if destination.callback_name == ExporterOwner.LANGFUSE_OTEL else without_langfuse_request_metadata(span) + span + if destination.callback_name == ExporterOwner.LANGFUSE_OTEL + else without_destination_langfuse_attributes(span, langfuse_mapped) ) attributes: Final = source.attributes or _NO_ATTRIBUTES database: Final = _is_database_span(attributes) @@ -726,8 +737,10 @@ class TenantFanOutSpanProcessor(SpanProcessor): default_capture_by_backend: Mapping[str, str] = MappingProxyType({}), pending_drains: int = _MAX_PENDING_DRAINS, drain_pool: _DrainPool | None = None, + langfuse_mapped: bool = True, ) -> None: self._operator_sinks: Final = operator_sinks + self._langfuse_mapped: Final = langfuse_mapped self._excluded_db_systems: Final = excluded_db_systems self._default_capture: Final = default_capture self._default_capture_by_backend: Final = default_capture_by_backend @@ -820,7 +833,8 @@ class TenantFanOutSpanProcessor(SpanProcessor): try: processor.on_end( _scoped( - _for_destination(span, destination, self._default_capture_for(destination)), destination.span_scope + _for_destination(span, destination, self._default_capture_for(destination), self._langfuse_mapped), + destination.span_scope, ) ) except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span @@ -1196,17 +1210,19 @@ class _OverriddenBackendFilter(SpanProcessor): return self._inner.force_flush(timeout_millis) -class _LangfuseRequestMetadataFilter(SpanProcessor): - """Strip the Langfuse request metadata from LiteLLM's own spans before ``inner`` exports them.""" +class _DestinationLangfuseAttributeFilter(SpanProcessor): + """Strip the Langfuse attributes only a Langfuse destination asked for from LiteLLM's own spans before ``inner`` + exports them.""" - def __init__(self, inner: SpanProcessor) -> None: + def __init__(self, inner: SpanProcessor, langfuse_mapped: bool) -> None: self._inner: Final = inner + self._langfuse_mapped: Final = langfuse_mapped def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: self._inner.on_start(span, parent_context) def on_end(self, span: ReadableSpan) -> None: - self._inner.on_end(without_langfuse_request_metadata(span)) + self._inner.on_end(without_destination_langfuse_attributes(span, self._langfuse_mapped)) def shutdown(self) -> None: self._inner.shutdown() @@ -1489,13 +1505,18 @@ def _spec_processor( spec: ExporterSpec, use_simple_processor: bool | None, strip_message_content: bool, - strip_langfuse_request_metadata: bool, + strip_destination_langfuse_attributes: bool, + langfuse_mapped: bool, ) -> SpanProcessor: exporting: Final = _processor_for( _exporter_from_spec(spec), (spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor), ) - scoped: Final = _LangfuseRequestMetadataFilter(exporting) if strip_langfuse_request_metadata else exporting + scoped: Final = ( + _DestinationLangfuseAttributeFilter(exporting, langfuse_mapped) + if strip_destination_langfuse_attributes + else exporting + ) return _MessageContentFilter(scoped) if strip_message_content else scoped @@ -1524,9 +1545,9 @@ 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. - Every exporter but ``langfuse_otel``'s goes without the request metadata the Langfuse - mapper writes on LiteLLM's own spans, which any logger writes for a request with a - Langfuse destination. + Every exporter but ``langfuse_otel``'s goes without the Langfuse attributes only a + Langfuse destination asked for (see :func:`without_destination_langfuse_attributes`), + which any logger writes for a request with a Langfuse destination. ``strip_message_content`` removes prompt and response content before every exporter, for a routed provider whose spans can carry content another destination asked for. @@ -1540,6 +1561,7 @@ def build_tracer_provider( provider.add_span_processor(_processor_for(exporter, use_simple_processor)) return provider + langfuse_mapped: Final = _maps_langfuse(config) # ``config._normalize`` guarantees at least one spec (it folds the top-level # ``exporter``/``endpoint``/``headers`` fields in when ``exporters`` is empty). for spec in config.exporters: @@ -1549,7 +1571,8 @@ def build_tracer_provider( spec, use_simple_processor, strip_message_content, - strip_langfuse_request_metadata=spec.owner is not ExporterOwner.LANGFUSE_OTEL, + strip_destination_langfuse_attributes=spec.owner is not ExporterOwner.LANGFUSE_OTEL, + langfuse_mapped=langfuse_mapped, ) owner = spec.owner.value if tenant_overrides and spec.owner is not None else None scope = _operator_scope(config, spec) @@ -1586,8 +1609,9 @@ def attach_tenant_fan_out( so exactly one fan-out lands. ``configs`` name the operator's own exporters, one config per v2 logger since each keeps its own provider and still writes its account, so an additive destination pointing at any of them is delivered once - rather than twice. ``excluded_db_systems`` only filters what the fan-out - delivers, never the operator's own exporters. + rather than twice. The first is the config of the logger that owns ``provider``, + whose model calls the fan-out carries. ``excluded_db_systems`` only filters what + the fan-out delivers, never the operator's own exporters. """ with _FAN_OUT_ATTACH_LOCK: if any(isinstance(processor, TenantFanOutSpanProcessor) for processor in _attached_processors(provider)): @@ -1598,10 +1622,16 @@ def attach_tenant_fan_out( excluded_db_systems=excluded_db_systems, default_capture=_operator_capture_mode(configs), default_capture_by_backend=MappingProxyType(dict(_owned_capture_modes(configs))), + langfuse_mapped=any(_maps_langfuse(config) for config in configs[:1]), ) ) +def _maps_langfuse(config: OpenTelemetryV2Config) -> bool: + """Whether a logger built from ``config`` writes the Langfuse vocabulary on every model call.""" + return "langfuse" in config.mapper_names + + def _operator_capture_mode(configs: Sequence[OpenTelemetryV2Config]) -> str: """The mode of the operator's generic ``otel`` callback, so ``callback_settings.otel`` counts as the global default""" return next( diff --git a/tests/integration/observability/test_langfuse_otel_tenant_only.py b/tests/integration/observability/test_langfuse_otel_tenant_only.py index 88398e367b2..3ab1b30562d 100644 --- a/tests/integration/observability/test_langfuse_otel_tenant_only.py +++ b/tests/integration/observability/test_langfuse_otel_tenant_only.py @@ -136,7 +136,9 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k ] == [{"user.id": "user-42", "session.id": session, "langfuse.trace.tags": TAGS}], [s["name"] for s in tenant] assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"]) assert len(operator) == 1, [span["name"] for span in operator] - assert "langfuse.observation.metadata" not in operator[0]["attributes"], sorted(operator[0]["attributes"]) + assert [ + name for name in operator[0]["attributes"] if name in ("user.id", "session.id") or name.startswith("langfuse.") + ] == [], sorted(operator[0]["attributes"]) def test_a_tenant_langfuse_project_gets_the_keys_langfuse_environment( diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index d26d238b0b7..a35580e0e8f 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -35,7 +35,7 @@ from litellm.integrations.otel.logger import ( publish_global_otel_v2_provider, ) from litellm.integrations.otel.mappers import resolve_mappers -from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES +from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LangfuseMapper from litellm.integrations.otel.model.config import ( CaptureMessageContent, ExporterOwner, @@ -3481,6 +3481,62 @@ class TestLangfuseDestinationFields: assert "langfuse.observation.metadata" in to_langfuse assert "langfuse.observation.metadata" not in received[None] + @staticmethod + def _received_by_account( + monkeypatch: pytest.MonkeyPatch, config: OpenTelemetryV2Config, destinations: tuple[OtelDestination, ...] + ) -> Mapping[str | None, Mapping[str, object]]: + """The attributes of the model call each account received, keyed by the headers it was sent with.""" + credential_less_proxy(monkeypatch) + capture: Final = _ExporterCapture() + for kind in ("otlp_grpc", "otlp_http", "console"): + monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) # pyright: ignore[reportPrivateUsage] # the factory table is the seam for capturing each exporter + provider: Final = build_tracer_provider(config, tenant_overrides=True) + attach_tenant_fan_out(provider, config) + emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config) + call: Final = replace( + _MODEL_CALL_WITHOUT_CONTENT, + trace=TraceControls(user_id="user-42", session_id="sess-1", tags=("agent:orders",)), + request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}), + ) + + def with_destinations() -> None: + set_request_destinations(deliverable_destinations(destinations, provider)) + emitter.emit(SpanRole.LLM_CALL, call) + + in_fresh_context(with_destinations) + provider.force_flush() + return MappingProxyType( + {headers: dict(exporter.get_finished_spans()[0].attributes or {}) for headers, exporter in capture.built} + ) + + def test_only_the_langfuse_destination_gets_the_langfuse_fields_written_for_it( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The operator's own exporter and the key's Arize destination get the model call as they would with no + Langfuse destination on the request.""" + received: Final = self._received_by_account(monkeypatch, OpenTelemetryV2Config(), (LANGFUSE_DEST, ARIZE_DEST)) + + assert received[LANGFUSE_DEST.header_string()]["user.id"] == "user-42" + assert { + headers: sorted(LangfuseMapper.LLM_CALL_ATTRIBUTES & attributes.keys()) + for headers, attributes in received.items() + if headers != LANGFUSE_DEST.header_string() + } == {None: [], ARIZE_DEST.header_string(): []} + + def test_an_exporter_whose_config_maps_langfuse_keeps_the_langfuse_fields_but_the_request_metadata( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + received: Final = self._received_by_account( + monkeypatch, OpenTelemetryV2Config(mapper_names=["genai", "langfuse"]), (LANGFUSE_DEST,) + ) + + assert {key: received[None].get(key) for key in ("user.id", "session.id", "langfuse.trace.tags")} == { + "user.id": "user-42", + "session.id": "sess-1", + "langfuse.trace.tags": ("agent:orders",), + } + assert not LANGFUSE_REQUEST_METADATA_ATTRIBUTES & received[None].keys() + @pytest.mark.parametrize( ("key_environment", "tracing_environment", "expected"), [ From 4737d491d683f46c1a504d8b837bb348471b3cb5 Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 28 Oct 2026 16:43:00 -0700 Subject: [PATCH 13/13] fix(otel): give a key's Langfuse destination the prompt and response the legacy logger sent, only where it ran --- .../integrations/otel/presets/destinations.py | 23 +++++- .../test_langfuse_otel_tenant_only.py | 4 + .../otel/test_otel_v2_destinations.py | 81 ++++++++++++++++++- 3 files changed, 103 insertions(+), 5 deletions(-) diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py index 82828574b3e..673337854ce 100644 --- a/litellm/integrations/otel/presets/destinations.py +++ b/litellm/integrations/otel/presets/destinations.py @@ -15,7 +15,7 @@ import litellm from litellm._logging import verbose_logger from litellm.integrations.otel.model.destination import OtelDestination from litellm.litellm_core_utils.url_utils import is_url_destination_allowed_by_host -from litellm.types.utils import OtelSpanScope, StandardCallbackDynamicParams +from litellm.types.utils import CaptureMessageContent, OtelSpanScope, StandardCallbackDynamicParams #: An endpoint plus the OTLP transport to reach it with, or ``None`` when the backend #: names no destination. The transport is ``None`` where the backend has only one. @@ -124,10 +124,19 @@ def _arize_default_project() -> Mapping[str, str]: return MappingProxyType({"model_id": ARIZE_DEFAULT_PROJECT}) +def _stands_in_for_the_legacy_langfuse_logger() -> bool: + """Whether a Langfuse destination replaces the older Langfuse logger's copy, which a proxy without Langfuse keys + of its own sends for a key's or team's ``langfuse_otel``.""" + return not (os.environ.get("LANGFUSE_PUBLIC_KEY") and os.environ.get("LANGFUSE_SECRET_KEY")) + + def _langfuse_environment(params: StandardCallbackDynamicParams) -> Mapping[str, str]: - """The key's or team's ``langfuse_environment``, else ``LANGFUSE_TRACING_ENVIRONMENT``.""" + """The key's or team's ``langfuse_environment``, else ``LANGFUSE_TRACING_ENVIRONMENT``, where the older Langfuse + logger wrote it.""" from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes + if not _stands_in_for_the_legacy_langfuse_logger(): + return _NO_ATTRS environment: Final = params.get("langfuse_environment") or os.environ.get("LANGFUSE_TRACING_ENVIRONMENT") return MappingProxyType({LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value: environment} if environment else {}) @@ -162,6 +171,14 @@ def _resource_defaults(callback_name: str) -> Mapping[str, str]: return backend() if backend is not None else _NO_ATTRS +def _capture_message_content(callback_name: str, configured: str | None) -> str | None: + """The key's or team's own setting, else the prompt and response where the older Langfuse logger sent them + whatever the global setting.""" + if configured is not None or callback_name != "langfuse_otel" or not _stands_in_for_the_legacy_langfuse_logger(): + return configured + return CaptureMessageContent.SPAN_ONLY + + def _span_scope(callback_name: str, params: StandardCallbackDynamicParams) -> OtelSpanScope: if callback_name != "langfuse_otel": return "full" @@ -229,7 +246,7 @@ def destination_for( callback_name=callback_name, protocol=protocol, span_scope=_span_scope(callback_name, params), - capture_message_content=capture_message_content, + capture_message_content=_capture_message_content(callback_name, capture_message_content), excluded_db_systems=excluded_db_systems, success_sampling_rate=success_sampling_rate, error_sampling_rate=error_sampling_rate, diff --git a/tests/integration/observability/test_langfuse_otel_tenant_only.py b/tests/integration/observability/test_langfuse_otel_tenant_only.py index 3ab1b30562d..4204d286d82 100644 --- a/tests/integration/observability/test_langfuse_otel_tenant_only.py +++ b/tests/integration/observability/test_langfuse_otel_tenant_only.py @@ -135,7 +135,11 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k for span in tenant ] == [{"user.id": "user-42", "session.id": session, "langfuse.trace.tags": TAGS}], [s["name"] for s in tenant] assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"]) + assert f"lf-tenant-only-{trace_id}" in str(tenant[0]["attributes"].get("langfuse.observation.input")), sorted( + tenant[0]["attributes"] + ) assert len(operator) == 1, [span["name"] for span in operator] + assert f"lf-tenant-only-{trace_id}" not in json.dumps(operator[0]["attributes"]), sorted(operator[0]["attributes"]) assert [ name for name in operator[0]["attributes"] if name in ("user.id", "session.id") or name.startswith("langfuse.") ] == [], sorted(operator[0]["attributes"]) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index a35580e0e8f..eb422508789 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -3483,7 +3483,10 @@ class TestLangfuseDestinationFields: @staticmethod def _received_by_account( - monkeypatch: pytest.MonkeyPatch, config: OpenTelemetryV2Config, destinations: tuple[OtelDestination, ...] + monkeypatch: pytest.MonkeyPatch, + config: OpenTelemetryV2Config, + destinations: tuple[OtelDestination, ...], + with_content: bool = False, ) -> Mapping[str | None, Mapping[str, object]]: """The attributes of the model call each account received, keyed by the headers it was sent with.""" credential_less_proxy(monkeypatch) @@ -3494,7 +3497,7 @@ class TestLangfuseDestinationFields: attach_tenant_fan_out(provider, config) emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config) call: Final = replace( - _MODEL_CALL_WITHOUT_CONTENT, + _MODEL_CALL_WITH_CONTENT if with_content else _MODEL_CALL_WITHOUT_CONTENT, trace=TraceControls(user_id="user-42", session_id="sess-1", tags=("agent:orders",)), request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}), ) @@ -3554,6 +3557,7 @@ class TestLangfuseDestinationFields: tracing_environment: str | None, expected: Mapping[str, str], ) -> None: + credential_less_proxy(monkeypatch) if tracing_environment is None: monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False) else: @@ -3569,6 +3573,76 @@ class TestLangfuseDestinationFields: assert destination is not None assert dict(destination.resource_attributes) == expected + def test_a_langfuse_destination_on_a_proxy_with_langfuse_keys_names_no_environment( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """With operator keys the older Langfuse logger never runs, so there is no environment of its to keep.""" + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator") + + destination: Final = destination_for( + "langfuse_otel", + StandardCallbackDynamicParams( + langfuse_public_key="pk", langfuse_secret_key="sk", langfuse_environment="prod" + ), + ) + + assert destination is not None + assert dict(destination.resource_attributes) == {} + + @pytest.mark.parametrize( + ("configured", "operator_keys", "expected"), + [ + (None, False, "span_only"), + ("no_content", False, "no_content"), + (None, True, None), + ("span_only", True, "span_only"), + ], + ids=["unset", "key-no-content", "operator-keys", "operator-keys-key-span-only"], + ) + def test_a_langfuse_destination_captures_the_prompt_where_the_legacy_logger_sent_it( + self, + monkeypatch: pytest.MonkeyPatch, + configured: str | None, + operator_keys: bool, + expected: str | None, + ) -> None: + credential_less_proxy(monkeypatch) + if operator_keys: + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator") + + destination: Final = destination_for( + "langfuse_otel", + StandardCallbackDynamicParams(langfuse_public_key="pk", langfuse_secret_key="sk"), + capture_message_content=configured, + ) + + assert destination is not None + assert destination.capture_message_content == expected + + def test_only_the_langfuse_destination_gets_the_prompt_it_captures_by_default( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + credential_less_proxy(monkeypatch) + destination: Final = destination_for( + "langfuse_otel", StandardCallbackDynamicParams(langfuse_public_key="pk", langfuse_secret_key="sk") + ) + arize: Final = destination_for("arize", StandardCallbackDynamicParams(arize_space_id="s", arize_api_key="k")) + assert destination is not None + assert arize is not None + + received: Final = self._received_by_account( + monkeypatch, OpenTelemetryV2Config(), (destination, arize), with_content=True + ) + + assert {headers: "langfuse.observation.input" in attributes for headers, attributes in received.items()} == { + destination.header_string(): True, + arize.header_string(): False, + None: False, + } + assert not [key for key in received[None] if key.startswith("gen_ai.input")] + def test_only_the_langfuse_destination_gets_the_keys_environment(self, monkeypatch: pytest.MonkeyPatch) -> None: credential_less_proxy(monkeypatch) capture: Final = _ExporterCapture() @@ -5303,7 +5377,10 @@ class TestCaptureMessageContent: def test_the_setting_rides_the_teams_destination_and_omission_stays_omitted( self, monkeypatch: pytest.MonkeyPatch, setting: str | None ) -> None: + """On a proxy with Langfuse keys of its own; without them see ``TestLangfuseDestinationFields``.""" monkeypatch.setenv("LITELLM_OTEL_V2", "true") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator") is_otel_v2_enabled.cache_clear() callback_vars: Final = { "langfuse_public_key": "pk-team",