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: