mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-26 06:03:19 +00:00
fix(otel v2): keep the end user off user.id when the metadata allowlist drops it
The end-user fallback only reaches user.id, on the model-call span and the Langfuse root, when baggage_metadata_keys includes user_api_key_end_user_id. A caller's own trace_user_id is always sent. Drops a comment the repo policy does not allow.
This commit is contained in:
parent
d7f270ff03
commit
abc7fe5a67
8 changed files with 131 additions and 24 deletions
|
|
@ -31,7 +31,11 @@ class LangfuseOpenTelemetryV2(OpenTelemetryV2):
|
|||
def log_pre_api_call(self, model: str, messages: object, kwargs: Mapping[str, object]) -> None:
|
||||
root: Final = request_root_span()
|
||||
if root is not None and root.is_recording():
|
||||
controls: Final = request_trace_controls(kwargs) if self._started(root) else caller_trace_controls(kwargs)
|
||||
controls: Final = (
|
||||
request_trace_controls(kwargs).released(self.config.baggage_metadata_keys)
|
||||
if self._started(root)
|
||||
else caller_trace_controls(kwargs)
|
||||
)
|
||||
root.set_attributes(LangfuseMapper.trace_attributes(controls))
|
||||
super().log_pre_api_call(model, messages, kwargs)
|
||||
|
||||
|
|
|
|||
|
|
@ -47,7 +47,6 @@ class GenAIMapper:
|
|||
GenAI.PROVIDER_NAME: lambda d: d.provider or None,
|
||||
GenAI.OUTPUT_TYPE: lambda d: d.output_type.value if d.output_type else None,
|
||||
GenAI.CONVERSATION_ID: lambda d: d.session_id,
|
||||
# The caller's ``trace_user_id``, else the proxy end user, for every backend that reads ``user.id``.
|
||||
User.ID: lambda d: d.trace.user_id,
|
||||
GenAI.REQUEST_MODEL: lambda d: d.request_model or None,
|
||||
GenAI.REQUEST_TEMPERATURE: lambda d: d.request_params.temperature,
|
||||
|
|
|
|||
|
|
@ -503,7 +503,7 @@ class LLMCallSpanData:
|
|||
call_type=call_type or None,
|
||||
request_route=request_route or context.identity.request_route,
|
||||
request_purpose=request_purpose,
|
||||
trace=trace or TraceControls(),
|
||||
trace=(trace or TraceControls()).released(metadata_keys),
|
||||
session_id=session_id or None,
|
||||
embedding_output=embedding_output if capture_content else None,
|
||||
routing_attributes=routing_decision_attributes(decision),
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import Collection, Mapping
|
||||
from dataclasses import dataclass, replace
|
||||
from typing import Final
|
||||
|
||||
|
|
@ -11,6 +11,7 @@ from pydantic import TypeAdapter, ValidationError
|
|||
from litellm.integrations.otel.model.utils import as_str, as_str_mapping
|
||||
|
||||
LANGFUSE_HEADER_PREFIX: Final = "langfuse_"
|
||||
END_USER_METADATA_KEY: Final = "user_api_key_end_user_id"
|
||||
_ITEMS: Final = TypeAdapter(tuple[object, ...])
|
||||
|
||||
|
||||
|
|
@ -26,6 +27,14 @@ class TraceControls:
|
|||
user_id: str | None = None
|
||||
session_id: str | None = None
|
||||
tags: tuple[str, ...] = ()
|
||||
user_id_is_end_user: bool = False
|
||||
|
||||
def released(self, metadata_keys: Collection[str]) -> TraceControls:
|
||||
"""These controls minus an end-user ``user_id`` when the operator's ``baggage_metadata_keys`` keeps the
|
||||
end-user id in the proxy. A caller's own ``trace_user_id`` is always kept."""
|
||||
if self.user_id_is_end_user and END_USER_METADATA_KEY not in metadata_keys:
|
||||
return replace(self, user_id=None, user_id_is_end_user=False)
|
||||
return self
|
||||
|
||||
|
||||
def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls:
|
||||
|
|
@ -58,10 +67,10 @@ def request_trace_controls(kwargs: Mapping[str, object]) -> TraceControls:
|
|||
if request is None:
|
||||
return controls
|
||||
end_user: Final = next(
|
||||
(value for body in _metadata_bodies(request) if (value := as_str(body.get("user_api_key_end_user_id")))),
|
||||
(value for body in _metadata_bodies(request) if (value := as_str(body.get(END_USER_METADATA_KEY)))),
|
||||
None,
|
||||
)
|
||||
return replace(controls, user_id=end_user)
|
||||
return replace(controls, user_id=end_user, user_id_is_end_user=end_user is not None)
|
||||
|
||||
|
||||
def _metadata_bodies(request: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from pathlib import Path
|
|||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from _openinference_support import _chat_plain_response, _rig, _spans
|
||||
from _openinference_support import Rig, _chat_plain_response, _rig, _spans
|
||||
from integration._support.client import Gateway, eventually
|
||||
from integration._support.wire import Reply, Request, Wire
|
||||
from pydantic import JsonValue
|
||||
|
|
@ -14,6 +14,7 @@ from pydantic import JsonValue
|
|||
_OPENINFERENCE_CALLBACKS: Final = ("arize", "arize_phoenix", "weave_otel")
|
||||
_GENAI_CALLBACKS: Final = ("otel", "langfuse_otel", "langtrace", "signoz", "levo", "newrelic", "agentops")
|
||||
_UNUSED_ENDPOINT: Final[dict[str, JsonValue]] = {"otel": {"exporter": "http/protobuf", "endpoint": "unused"}}
|
||||
_METADATA_ALLOWLIST: Final = "LITELLM_OTEL_BAGGAGE_METADATA_KEYS"
|
||||
|
||||
|
||||
def _settings(callback: str) -> Mapping[str, JsonValue] | None:
|
||||
|
|
@ -64,6 +65,17 @@ def _model_call_spans(destination: Wire, markers: tuple[str, ...]) -> Mapping[st
|
|||
return eventually(by_marker, lambda found: all(found.values()), seconds=30)
|
||||
|
||||
|
||||
def _chat(rig: Rig, marker: str, key: str, headers: Mapping[str, str], extra: Mapping[str, JsonValue]) -> None:
|
||||
response: Final = rig.proxy.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{"model": rig.model, "messages": [{"role": "user", "content": marker}], "cache": {"no-cache": True}, **extra},
|
||||
key=key,
|
||||
headers=headers,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("callback", (*_OPENINFERENCE_CALLBACKS, *_GENAI_CALLBACKS))
|
||||
def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gateway: Gateway, tmp_path: Path) -> None:
|
||||
run: Final = uuid.uuid4().hex
|
||||
|
|
@ -78,7 +90,8 @@ def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gatew
|
|||
callbacks=(callback,),
|
||||
callback_settings=_settings(callback),
|
||||
environment=_environment(callback),
|
||||
remove_environment=_removed_environment(callback),
|
||||
remove_environment=(*_removed_environment(callback), _METADATA_ALLOWLIST),
|
||||
disabled_environment=(_METADATA_ALLOWLIST,),
|
||||
) as rig,
|
||||
rig.proxy.scenario() as scenario,
|
||||
):
|
||||
|
|
@ -90,21 +103,41 @@ def test_otel_v2_callback_sends_the_request_user_as_user_id(callback: str, gatew
|
|||
(markers[2], owned_key, {}, {}),
|
||||
)
|
||||
for marker, key, headers, extra in sent:
|
||||
response = rig.proxy.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": rig.model,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"cache": {"no-cache": True},
|
||||
**extra,
|
||||
},
|
||||
key=key,
|
||||
headers=headers,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
_chat(rig, marker, key, headers, extra)
|
||||
spans: Final = _model_call_spans(rig.destination, markers)
|
||||
users: Final = {marker: {attributes.get("user.id") for attributes in spans[marker]} for marker in markers}
|
||||
assert users == {markers[0]: {end_user}, markers[1]: {caller}, markers[2]: {None}}, spans
|
||||
sessions: Final = {attributes.get("session.id") for attributes in itertools.chain.from_iterable(spans.values())}
|
||||
assert end_user not in sessions, spans
|
||||
|
||||
|
||||
@pytest.mark.parametrize("callback", ("arize", "otel", "langfuse_otel"))
|
||||
def test_otel_v2_callback_withholds_an_end_user_the_metadata_allowlist_drops(
|
||||
callback: str, gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
run: Final = uuid.uuid4().hex
|
||||
end_user: Final = f"end-user-{run}"
|
||||
caller: Final = f"caller-{run}"
|
||||
markers: Final = (f"end-{run}", f"caller-{run}-marker")
|
||||
with (
|
||||
_rig(
|
||||
gateway,
|
||||
tmp_path,
|
||||
_upstream,
|
||||
callbacks=(callback,),
|
||||
callback_settings=_settings(callback),
|
||||
environment={**_environment(callback), _METADATA_ALLOWLIST: "user_api_key_org_id"},
|
||||
) as rig,
|
||||
rig.proxy.scenario() as scenario,
|
||||
):
|
||||
service_key: Final = scenario.key()
|
||||
_chat(rig, markers[0], service_key, {"x-litellm-end-user-id": end_user}, {})
|
||||
_chat(
|
||||
rig, markers[1], service_key, {"x-litellm-end-user-id": end_user}, {"metadata": {"trace_user_id": caller}}
|
||||
)
|
||||
spans: Final = _model_call_spans(rig.destination, markers)
|
||||
users: Final = {marker: {attributes.get("user.id") for attributes in spans[marker]} for marker in markers}
|
||||
assert users == {markers[0]: {None}, markers[1]: {caller}}, spans
|
||||
model_call_spans: Final = tuple(itertools.chain.from_iterable(spans.values()))
|
||||
values: Final = tuple(itertools.chain.from_iterable(span.values() for span in model_call_spans))
|
||||
assert not [value for value in values if end_user in str(value)], spans
|
||||
|
|
|
|||
|
|
@ -466,6 +466,31 @@ def test_a_callers_trace_user_id_still_wins_over_the_proxy_end_user():
|
|||
assert generation_attrs["user.id"] == "caller-1"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("metadata", "expected_user"),
|
||||
[
|
||||
({"user_api_key_end_user_id": "end-1"}, None),
|
||||
({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "caller-1"),
|
||||
],
|
||||
ids=["end-user-withheld", "caller-trace-user-still-sent"],
|
||||
)
|
||||
def test_an_end_user_the_metadata_allowlist_drops_stays_off_the_root_and_the_generation(metadata, expected_user):
|
||||
cfg: Final = OpenTelemetryV2Config(
|
||||
exporter="in_memory", mapper_names=["genai", "langfuse"], baggage_metadata_keys=["user_api_key_org_id"]
|
||||
)
|
||||
exporter: Final = InMemorySpanExporter()
|
||||
logger: Final = build_otel_v2_logger(
|
||||
config=cfg, tracer_provider=providers.build_tracer_provider(cfg, exporter=exporter)
|
||||
)
|
||||
|
||||
root_attrs, generation_attrs = _run_named_request(
|
||||
logger, exporter, {"metadata": metadata, "proxy_server_request": {"headers": {}}}
|
||||
)
|
||||
|
||||
assert root_attrs.get("user.id") == expected_user
|
||||
assert generation_attrs.get("user.id") == expected_user
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("metadata", "expected_root_user"),
|
||||
[
|
||||
|
|
|
|||
|
|
@ -318,6 +318,24 @@ def test_llm_call_span_carries_the_request_user_as_user_id(metadata, expected):
|
|||
assert span.attributes.get(User.ID) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("metadata", "expected"),
|
||||
[
|
||||
({"user_api_key_end_user_id": "end-1"}, None),
|
||||
({"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}, "caller-1"),
|
||||
],
|
||||
ids=["end-user-withheld", "caller-trace-user-still-sent"],
|
||||
)
|
||||
def test_llm_call_span_withholds_an_end_user_the_metadata_allowlist_drops(metadata, expected):
|
||||
cfg: Final = OpenTelemetryV2Config(exporter="in_memory", baggage_metadata_keys=["user_api_key_org_id"])
|
||||
exporter: Final = InMemorySpanExporter()
|
||||
logger: Final = OpenTelemetryV2(config=cfg, tracer_provider=providers.build_tracer_provider(cfg, exporter=exporter))
|
||||
_emit_llm(logger, {**_kwargs(), "litellm_params": {"metadata": metadata}})
|
||||
(span,) = exporter.get_finished_spans()
|
||||
assert span.attributes.get(User.ID) == expected
|
||||
assert "end-1" not in {str(value) for value in span.attributes.values()}
|
||||
|
||||
|
||||
def test_streaming_span_carries_time_to_first_chunk():
|
||||
logger, exporter = _logger()
|
||||
kwargs = {
|
||||
|
|
|
|||
|
|
@ -1358,8 +1358,11 @@ def test_caller_trace_controls_carry_user_session_and_tags(request_data, expecte
|
|||
@pytest.mark.parametrize(
|
||||
("request_data", "expected"),
|
||||
[
|
||||
({"metadata": {"user_api_key_end_user_id": "end-1"}}, TraceControls(user_id="end-1")),
|
||||
({"litellm_metadata": {"user_api_key_end_user_id": "end-2"}}, TraceControls(user_id="end-2")),
|
||||
({"metadata": {"user_api_key_end_user_id": "end-1"}}, TraceControls(user_id="end-1", user_id_is_end_user=True)),
|
||||
(
|
||||
{"litellm_metadata": {"user_api_key_end_user_id": "end-2"}},
|
||||
TraceControls(user_id="end-2", user_id_is_end_user=True),
|
||||
),
|
||||
(
|
||||
{"metadata": {"trace_user_id": "caller-1", "user_api_key_end_user_id": "end-1"}},
|
||||
TraceControls(user_id="caller-1"),
|
||||
|
|
@ -1373,7 +1376,7 @@ def test_caller_trace_controls_carry_user_session_and_tags(request_data, expecte
|
|||
),
|
||||
(
|
||||
{"metadata": {"user_api_key_end_user_id": "end-1", "session_id": "s-1"}},
|
||||
TraceControls(user_id="end-1", session_id="s-1"),
|
||||
TraceControls(user_id="end-1", session_id="s-1", user_id_is_end_user=True),
|
||||
),
|
||||
({"metadata": {"user_api_key_user_id": "internal-1"}}, TraceControls()),
|
||||
({}, TraceControls()),
|
||||
|
|
@ -1553,6 +1556,22 @@ def test_llm_span_stamps_user_id_only_when_the_request_names_a_user():
|
|||
assert User.ID not in GenAIMapper().map(without)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("trace", "metadata_keys", "expected"),
|
||||
[
|
||||
(TraceControls(user_id="end-1", user_id_is_end_user=True), ("user_api_key_end_user_id",), "end-1"),
|
||||
(TraceControls(user_id="end-1", user_id_is_end_user=True), ("user_api_key_org_id",), None),
|
||||
(TraceControls(user_id="caller-1"), ("user_api_key_org_id",), "caller-1"),
|
||||
],
|
||||
ids=["end-user-allowlisted", "end-user-withheld", "caller-trace-user-always-sent"],
|
||||
)
|
||||
def test_llm_span_sends_the_end_user_only_when_the_metadata_allowlist_releases_it(trace, metadata_keys, expected):
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_sample_payload(), trace=trace, metadata_keys=metadata_keys
|
||||
)
|
||||
assert GenAIMapper().map(data).get(User.ID) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("callback_name", ["otel", *sorted(PRESET_BY_CALLBACK)])
|
||||
def test_every_v2_callback_sends_the_request_user_as_user_id(callback_name, monkeypatch):
|
||||
for name in ("LEVOAI_API_KEY", "LEVOAI_ORG_ID", "LEVOAI_WORKSPACE_ID"):
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue