scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-10 08:28:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

fix(otel): keep the legacy langfuse_otel span for a destination's call rejected before any provider

Browse source 
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.
This commit is contained in:
yucheng 2026-10-28 14:40:04 -07:00
parent cea1923222
commit 0f1cfe4365
6 changed files with 169 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
litellm/integrations/langfuse/langfuse_otel.py
View file

@ -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,

4
litellm/integrations/otel/logger.py
View file

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

5
litellm/integrations/otel/model/metadata.py
View file

@ -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")

125
tests/integration/observability/test_langfuse_otel_legacy_destinations.py
View file

@ -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]

49
tests/integration/observability/test_langfuse_otel_tenant_only.py
View file

@ -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 [

44
tests/unit/integrations/test_langfuse_otel.py
View file

@ -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: