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