mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(langfuse otel): keep the identity keys on the OTEL v2 Langfuse mapper only
The v1 langfuse_otel callback goes back to main: it sends no langfuse.observation.metadata or langfuse.trace.metadata.user_api_key_* keys. The OTEL v2 Langfuse mapper keeps them. The integration cells now run every endpoint, client, redaction, failure and chaos case on v2.
This commit is contained in:
yucheng
parent
b474b2eabc
commit
fe120be28d
5 changed files with 47 additions and 819 deletions
|
|
@ -5,18 +5,13 @@ from collections.abc import Mapping, Sequence
|
|||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING, Any, Final, Optional
|
||||
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.integrations._types.open_inference import ErrorAttributes
|
||||
from litellm.integrations.arize import _utils
|
||||
from litellm.integrations.langfuse.langfuse_otel_attributes import (
|
||||
LangfuseLLMObsOTELAttributes,
|
||||
serialize_langfuse_datetime,
|
||||
)
|
||||
from litellm.integrations.opentelemetry import HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE, OpenTelemetry, OpenTelemetryConfig
|
||||
from litellm.integrations.otel.mappers.langfuse import TRACE_IDENTITY_FIELDS
|
||||
from litellm.integrations.otel.model.metadata import IDENTITY_METADATA_FIELDS, exported_request_metadata
|
||||
from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig
|
||||
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
|
||||
from litellm.types.integrations.langfuse_otel import (
|
||||
LangfuseSpanAttributes,
|
||||
|
|
@ -31,19 +26,6 @@ else:
|
|||
Span = Any
|
||||
|
||||
|
||||
_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[object, object]]] = TypeAdapter(Mapping[object, object])
|
||||
# The attributes ``OpenTelemetry._record_exception_on_span`` adds to a failed call's span after these.
|
||||
_ERROR_ATTRIBUTES: Final = frozenset(
|
||||
{
|
||||
ErrorAttributes.ERROR_CODE,
|
||||
HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE,
|
||||
ErrorAttributes.ERROR_TYPE,
|
||||
ErrorAttributes.ERROR_MESSAGE,
|
||||
ErrorAttributes.ERROR_LLM_PROVIDER,
|
||||
ErrorAttributes.ERROR_STACK_TRACE,
|
||||
}
|
||||
)
|
||||
|
||||
LANGFUSE_CLOUD_EU_ENDPOINT: Final = "https://cloud.langfuse.com/api/public/otel"
|
||||
LANGFUSE_CLOUD_US_ENDPOINT: Final = "https://us.cloud.langfuse.com/api/public/otel"
|
||||
LANGFUSE_INGESTION_VERSION_HEADER: Final = "x-langfuse-ingestion-version"
|
||||
|
|
@ -64,10 +46,7 @@ class LangfuseOtelLogger(OpenTelemetry):
|
|||
Sets OpenTelemetry span attributes for Langfuse observability.
|
||||
Uses the same attribute setting logic as Arize Phoenix for consistency.
|
||||
"""
|
||||
from opentelemetry import trace
|
||||
|
||||
# The caller's active span, which LiteLLM writes onto instead of starting its own, gets no caller identity.
|
||||
own_span: Final = span is not trace.get_current_span()
|
||||
_utils.set_attributes(span, kwargs, response_obj, LangfuseLLMObsOTELAttributes)
|
||||
span.set_attribute("langfuse.observation.type", "generation")
|
||||
|
||||
|
|
@ -75,10 +54,6 @@ class LangfuseOtelLogger(OpenTelemetry):
|
|||
# Set Langfuse specific attributes
|
||||
#########################################################
|
||||
LangfuseOtelLogger._set_langfuse_specific_attributes(span=span, kwargs=kwargs, response_obj=response_obj)
|
||||
if own_span:
|
||||
LangfuseOtelLogger._set_request_metadata_attributes(
|
||||
span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _extract_langfuse_metadata(kwargs: Mapping[str, Any]) -> dict:
|
||||
|
|
@ -152,46 +127,6 @@ class LangfuseOtelLogger(OpenTelemetry):
|
|||
value = str(value)
|
||||
safe_set_attribute(span, enum_attr.value, value)
|
||||
|
||||
@staticmethod
|
||||
def _set_request_metadata_attributes(
|
||||
span: Span,
|
||||
kwargs: Mapping[object, object],
|
||||
) -> None:
|
||||
"""Set the caller identity into the room ``span``'s attribute limit leaves next to the error attributes
|
||||
a failed call adds after it, so the identity never evicts or drops another attribute."""
|
||||
from litellm.integrations.otel.emitter import carried_keys, span_attribute_limit
|
||||
|
||||
raw_standard_logging_object: Final = kwargs.get("standard_logging_object")
|
||||
if not isinstance(raw_standard_logging_object, dict):
|
||||
return
|
||||
raw_metadata: Final = _MAPPING_ADAPTER.validate_python(raw_standard_logging_object).get("metadata")
|
||||
if not isinstance(raw_metadata, dict):
|
||||
return
|
||||
metadata: Final = _MAPPING_ADAPTER.validate_python(raw_metadata)
|
||||
observation_metadata: Final = exported_request_metadata(
|
||||
{field: metadata.get(field) for field in IDENTITY_METADATA_FIELDS}
|
||||
)
|
||||
trace_prefix: Final = LangfuseSpanAttributes.TRACE_METADATA.value
|
||||
identity: Final = (
|
||||
*(
|
||||
((LangfuseSpanAttributes.OBSERVATION_METADATA.value, json.dumps(dict(observation_metadata))),)
|
||||
if observation_metadata
|
||||
else ()
|
||||
),
|
||||
*(
|
||||
(f"{trace_prefix}.{field}", value)
|
||||
for field in TRACE_IDENTITY_FIELDS
|
||||
if (value := observation_metadata.get(field))
|
||||
),
|
||||
)
|
||||
limit: Final = span_attribute_limit(span)
|
||||
carried: Final = carried_keys(span)
|
||||
room: Final = (
|
||||
len(identity) if limit is None else max(limit - len(carried) - len(_ERROR_ATTRIBUTES - carried), 0)
|
||||
)
|
||||
for key, value in identity[:room]:
|
||||
_utils.safe_set_attribute(span, key, value)
|
||||
|
||||
@staticmethod
|
||||
def _set_observation_output(span: Span, response_obj):
|
||||
"""Helper to set observation output attributes."""
|
||||
|
|
|
|||
|
|
@ -30,7 +30,6 @@ class LangfuseSpanAttributes(str, Enum):
|
|||
OBSERVATION_INPUT = "langfuse.observation.input"
|
||||
OBSERVATION_OUTPUT = "langfuse.observation.output"
|
||||
OBSERVATION_COMPLETION_START_TIME = "langfuse.observation.completion_start_time"
|
||||
OBSERVATION_METADATA = "langfuse.observation.metadata"
|
||||
|
||||
# ---- Trace-level metadata ----
|
||||
TRACE_USER_ID = "user.id"
|
||||
|
|
|
|||
|
|
@ -256,6 +256,7 @@ def _langfuse_rig(
|
|||
"LANGFUSE_PUBLIC_KEY": "pk-lf-integration",
|
||||
"LANGFUSE_SECRET_KEY": "sk-lf-integration",
|
||||
"LANGFUSE_HOST": collector.url,
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
**{
|
||||
name: (collector.url if value == "__collector__" else value)
|
||||
for name, value in dict(overrides or {}).items()
|
||||
|
|
@ -282,7 +283,7 @@ def _response_id(value: object) -> str | None:
|
|||
|
||||
|
||||
def _span_response_id(span: Mapping[str, object]) -> str | None:
|
||||
return _response_id(span.get("llm.response.id"))
|
||||
return _response_id(span.get("gen_ai.response.id"))
|
||||
|
||||
|
||||
def _watched_spans(collector: Wire) -> Callable[[], tuple[dict[str, object], ...]]:
|
||||
|
|
@ -307,32 +308,6 @@ def _v2_llm_span(collector: Wire, response_id: str) -> dict[str, object]:
|
|||
return next(span for span in observed if matches(span))
|
||||
|
||||
|
||||
def _observation_span(collector: Wire, response_id: str) -> dict[str, object]:
|
||||
def spans() -> tuple[dict[str, object], ...]:
|
||||
return tuple(
|
||||
attributes
|
||||
for batch in collector.drain()
|
||||
for attributes in _span_attributes(batch.body)
|
||||
if _span_response_id(attributes) == _response_id(response_id)
|
||||
)
|
||||
|
||||
observed: Final = eventually(spans, lambda values: len(values) == 1, seconds=25)
|
||||
return observed[0]
|
||||
|
||||
|
||||
def _marked_span(collector: Wire, marker: str, needle: str) -> dict[str, object]:
|
||||
def spans() -> tuple[dict[str, object], ...]:
|
||||
return tuple(
|
||||
attributes
|
||||
for batch in collector.drain()
|
||||
for attributes in _span_attributes(batch.body)
|
||||
if needle in json.dumps(attributes, default=str) and marker in json.dumps(attributes, default=str)
|
||||
)
|
||||
|
||||
observed: Final = eventually(spans, lambda values: len(values) == 1, seconds=25)
|
||||
return observed[0]
|
||||
|
||||
|
||||
def _scenario_identity(
|
||||
scenario: Scenario,
|
||||
provider: Wire,
|
||||
|
|
@ -360,7 +335,6 @@ def _scenario_identity(
|
|||
|
||||
|
||||
def _assert_identity(attrs: Mapping[str, object], expected: Mapping[str, object]) -> None:
|
||||
assert "metadata" in attrs, sorted(attrs)
|
||||
assert "langfuse.observation.metadata" in attrs, sorted(attrs)
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert {key: observation.get(key) for key in expected} == dict(expected), observation
|
||||
|
|
@ -383,63 +357,6 @@ def _chat_identity_fields(identity: Mapping[str, str | None]) -> dict[str, objec
|
|||
}
|
||||
|
||||
|
||||
def test_langfuse_otel_emits_request_metadata_under_langfuse_observation_and_trace_keys(
|
||||
gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
marker: Final = "lf-meta-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
response: Final = rig.candidate.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": identity["model"],
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"user": identity["end_user"],
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
key=key,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json()["choices"][0]["message"]["content"] == "langfuse-echo"
|
||||
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_metadata_redacts_user_api_key_fields_like_vanilla_langfuse(
|
||||
gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
marker: Final = "lf-redact-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {"redact_user_api_key_info": True}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key: Final = scenario.key(
|
||||
key_alias="lf-redact-" + uuid.uuid4().hex,
|
||||
metadata={"spend_logs_metadata": {"ticket": "LIT-8283"}},
|
||||
)
|
||||
model: Final = scenario.model(api_base=rig.provider.url + "/v1")
|
||||
response: Final = rig.candidate.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
key=key,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert "langfuse.observation.metadata" not in attrs, attrs["langfuse.observation.metadata"]
|
||||
assert not [name for name in attrs if name.startswith("langfuse.trace.metadata.")], sorted(attrs)
|
||||
|
||||
|
||||
def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_stream(gateway: Gateway, tmp_path: Path) -> None:
|
||||
marker: Final = "lf-sdkstream-" + uuid.uuid4().hex
|
||||
with (
|
||||
|
|
@ -463,7 +380,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_stream(gateway:
|
|||
)
|
||||
assert chunks, "stream produced no chunks"
|
||||
response_id: Final = chunks[-1].id
|
||||
attrs: Final = _observation_span(rig.collector, response_id)
|
||||
attrs: Final = _v2_llm_span(rig.collector, response_id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
|
|
@ -485,7 +402,7 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_async(gate
|
|||
user=identity["end_user"],
|
||||
extra_body={"cache": {"no-cache": True}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, response.id)
|
||||
attrs: Final = _v2_llm_span(rig.collector, response.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
|
|
@ -508,7 +425,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_messages(gate
|
|||
messages=[{"role": "user", "content": marker}],
|
||||
extra_body={"litellm_metadata": {"tags": ["lf8283"]}, "metadata": {"user_id": identity["end_user"]}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, message.id)
|
||||
attrs: Final = _v2_llm_span(rig.collector, message.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
|
|
@ -536,7 +453,7 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_stream(
|
|||
) as stream,
|
||||
):
|
||||
final: Final = await stream.get_final_message()
|
||||
attrs: Final = _observation_span(rig.collector, final.id)
|
||||
attrs: Final = _v2_llm_span(rig.collector, final.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
|
|
@ -558,7 +475,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_responses_api(gateway: Gate
|
|||
user=identity["end_user"],
|
||||
extra_body={"cache": {"no-cache": True}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, response.id)
|
||||
attrs: Final = _v2_llm_span(rig.collector, response.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
|
|
@ -589,32 +506,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_responses_sse_stream(gatewa
|
|||
response_id: Final = next(
|
||||
event["response_id"] for event in events if event.get("type") == "response.output_text.delta"
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, response_id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_caller_trace_metadata_survives_flattened_identity(gateway: Gateway, tmp_path: Path) -> None:
|
||||
marker: Final = "lf-deploy-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
response: Final = rig.candidate.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": identity["model"],
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"user": identity["end_user"],
|
||||
"metadata": {"trace_metadata": {"deploy": "blue"}, "tags": ["lf8283"]},
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
key=key,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert json.loads(str(attrs["langfuse.trace.metadata"])) == {"deploy": "blue"}, attrs
|
||||
attrs: Final = _v2_llm_span(rig.collector, response_id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
|
|
@ -637,65 +529,19 @@ def test_langfuse_otel_failure_span_carries_request_metadata(gateway: Gateway, t
|
|||
key=key,
|
||||
)
|
||||
assert response.status_code >= 500, response.text
|
||||
attrs: Final = _marked_span(rig.collector, marker, "langfuse.observation.")
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
def failed_calls(spans: Sequence[Mapping[str, object]]) -> tuple[Mapping[str, object], ...]:
|
||||
return tuple(
|
||||
span
|
||||
for span in spans
|
||||
if span.get("langfuse.trace.metadata.team_id") == identity["team_id"] and "error.type" in span
|
||||
)
|
||||
|
||||
def test_langfuse_otel_teamless_key_emits_no_team_identity(gateway: Gateway, tmp_path: Path) -> None:
|
||||
marker: Final = "lf-noteam-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider, team=False)
|
||||
response: Final = rig.candidate.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": identity["model"],
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
key=key,
|
||||
failed: Final = failed_calls(
|
||||
eventually(_watched_spans(rig.collector), lambda spans: bool(failed_calls(spans)), seconds=25)
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert "langfuse.observation.metadata" in attrs, sorted(attrs)
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
for field in ("user_api_key_team_id", "user_api_key_team_alias", "user_api_key_end_user_id"):
|
||||
assert field not in observation, observation
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
assert {
|
||||
attribute: attrs.get(attribute)
|
||||
for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)
|
||||
if attribute in attrs
|
||||
} == {"langfuse.trace.metadata.user_api_key_alias": identity["key_alias"]}, attrs
|
||||
|
||||
|
||||
def test_langfuse_otel_teamless_key_keeps_caller_trace_metadata(gateway: Gateway, tmp_path: Path) -> None:
|
||||
marker: Final = "lf-noteam-meta-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider, team=False)
|
||||
response: Final = rig.candidate.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": identity["model"],
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"metadata": {"trace_metadata": {"deploy": "blue"}},
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
key=key,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert json.loads(str(attrs["langfuse.trace.metadata"])) == {"deploy": "blue"}, attrs
|
||||
assert "langfuse.trace.metadata.user_api_key_team_id" not in attrs, attrs
|
||||
assert "langfuse.trace.metadata.user_api_key_team_alias" not in attrs, attrs
|
||||
assert attrs.get("langfuse.trace.metadata.user_api_key_alias") == identity["key_alias"], attrs
|
||||
for attrs in failed:
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_oversized_caller_metadata_is_not_copied_into_the_langfuse_keys(
|
||||
|
|
@ -727,11 +573,18 @@ def test_langfuse_otel_oversized_caller_metadata_is_not_copied_into_the_langfuse
|
|||
key=key,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
attrs: Final = _v2_llm_span(rig.collector, response.json()["id"])
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
assert len(str(attrs["langfuse.observation.metadata"])) < 1000, len(str(attrs["langfuse.observation.metadata"]))
|
||||
assert caller_note in str(attrs["metadata"]), "the bare metadata attribute keeps the caller metadata"
|
||||
assert spend_logs_metadata["blob"] in str(attrs["metadata"]), "the bare metadata attribute keeps it"
|
||||
langfuse_metadata: Final = json.dumps(
|
||||
{
|
||||
name: value
|
||||
for name, value in attrs.items()
|
||||
if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.")
|
||||
}
|
||||
)
|
||||
assert caller_note not in langfuse_metadata, len(langfuse_metadata)
|
||||
assert spend_logs_metadata["blob"] not in langfuse_metadata, len(langfuse_metadata)
|
||||
|
||||
|
||||
def test_langfuse_otel_unauthenticated_request_leaks_no_identity(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -779,20 +632,13 @@ def test_langfuse_otel_unauthenticated_request_leaks_no_identity(gateway: Gatewa
|
|||
settled: Final = eventually(
|
||||
watch, lambda spans: sum(_span_response_id(s) == second_id for s in spans) == 1, seconds=25
|
||||
)
|
||||
denied_spans: Final = tuple(span for span in settled if f"{marker}-denied" in json.dumps(span, default=str))
|
||||
assert len(denied_spans) == 1, [s.get("llm.response.id") for s in settled]
|
||||
denied_span: Final = denied_spans[0]
|
||||
assert not [name for name in denied_span if name.startswith("langfuse.trace.metadata.user_api_key")], (
|
||||
denied_span
|
||||
carriers: Final = tuple(
|
||||
_span_response_id(span)
|
||||
for span in settled
|
||||
if "langfuse.observation.metadata" in span
|
||||
or any(name.startswith("langfuse.trace.metadata.user_api_key_") for name in span)
|
||||
)
|
||||
leaked: Final = json.loads(str(denied_span.get("langfuse.observation.metadata", "{}")))
|
||||
for field in (
|
||||
"user_api_key_alias",
|
||||
"user_api_key_team_id",
|
||||
"user_api_key_team_alias",
|
||||
"user_api_key_end_user_id",
|
||||
):
|
||||
assert field not in leaked, leaked
|
||||
assert sorted(carriers, key=str) == sorted((first_id, second_id)), carriers
|
||||
|
||||
|
||||
def test_langfuse_otel_repeated_identical_requests_emit_one_span_each(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -903,10 +749,12 @@ def test_langfuse_otel_sink_outage_mid_burst_lands_every_response_id_once(gatewa
|
|||
watch,
|
||||
lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in response_ids),
|
||||
seconds=80,
|
||||
return_last_on_timeout=True,
|
||||
)
|
||||
assert sorted(_span_response_id(span) for span in settled if _span_response_id(span) in response_ids) == sorted(
|
||||
response_ids
|
||||
landed: Final = sorted(
|
||||
str(_span_response_id(span)) for span in settled if _span_response_id(span) in response_ids
|
||||
)
|
||||
assert landed == sorted(response_ids), sorted(set(response_ids) - set(landed))
|
||||
assert len(rejected) >= 1, "sink outage never served a 503"
|
||||
|
||||
|
||||
|
|
@ -1006,7 +854,6 @@ def test_langfuse_otel_v2_mapper_emits_request_metadata_and_identity(gateway: Ga
|
|||
tmp_path,
|
||||
{},
|
||||
marker,
|
||||
overrides={"LITELLM_OTEL_V2": "1"},
|
||||
) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
|
|
@ -1053,7 +900,6 @@ def test_langfuse_otel_v2_mapper_redacts_user_api_key_fields(gateway: Gateway, t
|
|||
tmp_path,
|
||||
{"redact_user_api_key_info": True},
|
||||
marker,
|
||||
overrides={"LITELLM_OTEL_V2": "1"},
|
||||
) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
|
|
@ -1092,7 +938,6 @@ def test_langfuse_otel_v2_mapper_teamless_key_emits_no_team_identity(gateway: Ga
|
|||
tmp_path,
|
||||
{},
|
||||
marker,
|
||||
overrides={"LITELLM_OTEL_V2": "1"},
|
||||
) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
|
|
@ -1154,7 +999,6 @@ def test_otel_v2_scalar_requester_metadata_keeps_the_llm_span(
|
|||
{"callbacks": [callback]},
|
||||
marker,
|
||||
overrides={
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
**({"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": "__collector__"} if callback == "otel" else {}),
|
||||
},
|
||||
) as rig,
|
||||
|
|
@ -1179,26 +1023,6 @@ def test_otel_v2_scalar_requester_metadata_keeps_the_llm_span(
|
|||
assert observation["user_api_key_alias"] == identity["key_alias"], observation
|
||||
|
||||
|
||||
def test_langfuse_otel_scalar_requester_metadata_keeps_the_observation_input(gateway: Gateway, tmp_path: Path) -> None:
|
||||
marker: Final = "lf-scalar-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
model: Final = scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url)
|
||||
response: Final = rig.candidate.request(
|
||||
"POST", "/v1/messages", _scalar_requester_metadata_request("/v1/messages", model, marker), key=key
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert marker in str(attrs.get("langfuse.observation.input")), sorted(attrs)
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert "requester_metadata" not in observation, observation
|
||||
assert observation["user_api_key_alias"] == identity["key_alias"], observation
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"callbacks",
|
||||
[["langfuse_otel", "otel"], ["langfuse_otel"]],
|
||||
|
|
@ -1216,7 +1040,7 @@ def test_otel_v2_collector_beside_langfuse_gets_no_langfuse_request_metadata(
|
|||
tmp_path,
|
||||
{"callbacks": callbacks},
|
||||
marker,
|
||||
overrides={"LITELLM_OTEL_V2": "1", "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url},
|
||||
overrides={"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url},
|
||||
) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
|
|
@ -1258,7 +1082,7 @@ def test_otel_v2_langfuse_gets_no_request_headers_or_caller_metadata(
|
|||
marker: Final = "lf-v2-probe-" + uuid.uuid4().hex
|
||||
probe: Final = "lf-probe-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker, overrides={"LITELLM_OTEL_V2": "1"}) as rig,
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
|
|
@ -1346,10 +1170,7 @@ def _keyword_router(router: str, keyword: str) -> Callable[[str], Sequence[Mappi
|
|||
return models
|
||||
|
||||
|
||||
@pytest.mark.parametrize("otel_v2", [False, True], ids=["v1", "v2"])
|
||||
def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(
|
||||
otel_v2: bool, gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(gateway: Gateway, tmp_path: Path) -> None:
|
||||
marker: Final = "lf-route-" + uuid.uuid4().hex
|
||||
router: Final = "lf-router-" + uuid.uuid4().hex[:8]
|
||||
keyword: Final = "acquire-globex-" + uuid.uuid4().hex[:8]
|
||||
|
|
@ -1359,11 +1180,7 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(
|
|||
tmp_path,
|
||||
{},
|
||||
marker,
|
||||
overrides=(
|
||||
{"LITELLM_OTEL_V2": "1", "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "no_content"}
|
||||
if otel_v2
|
||||
else None
|
||||
),
|
||||
overrides={"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "no_content"},
|
||||
model_list=_keyword_router(router, keyword),
|
||||
) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
|
|
@ -1381,11 +1198,7 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(
|
|||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
attrs: Final = (
|
||||
_v2_llm_span(rig.collector, response.json()["id"])
|
||||
if otel_v2
|
||||
else _observation_span(rig.collector, response.json()["id"])
|
||||
)
|
||||
attrs: Final = _v2_llm_span(rig.collector, response.json()["id"])
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
langfuse_metadata: Final = {
|
||||
|
|
@ -1394,8 +1207,7 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(
|
|||
if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.")
|
||||
}
|
||||
assert keyword not in json.dumps(langfuse_metadata, default=str), langfuse_metadata
|
||||
if otel_v2:
|
||||
assert keyword not in json.dumps(attrs, default=str), attrs
|
||||
assert keyword not in json.dumps(attrs, default=str), attrs
|
||||
|
||||
|
||||
def _span_records(body: bytes) -> tuple[tuple[dict[str, object], int], ...]:
|
||||
|
|
@ -1451,59 +1263,6 @@ def _limit_sweep(
|
|||
return {size: record for size, record in observed.items() if record is not None}
|
||||
|
||||
|
||||
def test_langfuse_otel_identity_never_evicts_or_drops_another_attribute_of_a_full_span(
|
||||
gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
marker: Final = "lf-limit-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
|
||||
def body(tool_count: int) -> dict[str, object]:
|
||||
return {
|
||||
"model": identity["model"],
|
||||
"messages": [{"role": "user", "content": f"{marker}:{tool_count}:"}],
|
||||
"user": identity["end_user"],
|
||||
"tools": [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": f"tool_{n}",
|
||||
"description": f"looks up thing {n}",
|
||||
"parameters": {"type": "object", "properties": {"q": {"type": "string"}}},
|
||||
},
|
||||
}
|
||||
for n in range(tool_count)
|
||||
],
|
||||
"cache": {"no-cache": True},
|
||||
}
|
||||
|
||||
spans: Final = _limit_sweep(
|
||||
rig, key, marker, {tool_count: body(tool_count) for tool_count in (0, *range(30, 41))}
|
||||
)
|
||||
|
||||
reference, reference_dropped = spans[0]
|
||||
identity_count: Final = len(_identity_attributes(reference))
|
||||
assert reference_dropped == 0, reference_dropped
|
||||
assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference)
|
||||
kept_by_size: Final[dict[int, tuple[int, int]]] = {} # mutable-ok: filled once per swept span below
|
||||
for tool_count, (attrs, dropped) in spans.items():
|
||||
kept = len(_identity_attributes(attrs))
|
||||
kept_by_size[tool_count] = (kept, dropped)
|
||||
if kept:
|
||||
assert dropped == 0, (tool_count, kept, dropped)
|
||||
expected = {"openinference.span.kind", "metadata", "llm.model_name", "llm.request.type"}
|
||||
assert expected <= set(attrs), (tool_count, sorted(expected - set(attrs)))
|
||||
tools = {f"llm.tools.{n}.name" for n in range(tool_count)}
|
||||
assert tools <= set(attrs), (tool_count, sorted(tools - set(attrs)))
|
||||
assert [size for size, (kept, dropped) in kept_by_size.items() if kept < identity_count and not dropped], (
|
||||
kept_by_size
|
||||
)
|
||||
assert [size for size, (_, dropped) in kept_by_size.items() if dropped], kept_by_size
|
||||
|
||||
|
||||
def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: Gateway, tmp_path: Path) -> None:
|
||||
marker: Final = "lf-v2-limit-" + uuid.uuid4().hex
|
||||
with (
|
||||
|
|
@ -1513,7 +1272,6 @@ def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: G
|
|||
{},
|
||||
marker,
|
||||
overrides={
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
"MAPPER_NAMES": "genai,openinference",
|
||||
"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only",
|
||||
},
|
||||
|
|
@ -1589,7 +1347,7 @@ def test_otel_v2_span_another_tracer_wrote_keeps_its_langfuse_metadata_keys_at_t
|
|||
tmp_path,
|
||||
{"callbacks": ["langfuse_otel", "otel", "lit8283_app_span.instance"]},
|
||||
marker,
|
||||
overrides={"LITELLM_OTEL_V2": "1", "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url},
|
||||
overrides={"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url},
|
||||
) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
|
|
@ -1647,7 +1405,6 @@ def test_otel_v2_collector_keeps_the_dropped_counts_of_a_span_without_langfuse_r
|
|||
{"callbacks": ["langfuse_otel", "otel"]},
|
||||
marker,
|
||||
overrides={
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
"OTEL_EXPORTER": "otlp_http",
|
||||
"OTEL_ENDPOINT": generic.url,
|
||||
"OTEL_SPAN_EVENT_COUNT_LIMIT": "0",
|
||||
|
|
@ -1693,7 +1450,6 @@ def test_otel_v2_langfuse_mapper_sends_no_langfuse_request_metadata_to_a_plain_c
|
|||
{"callbacks": ["otel"]},
|
||||
marker,
|
||||
overrides={
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
"MAPPER_NAMES": "genai,langfuse",
|
||||
"OTEL_EXPORTER": "otlp_http",
|
||||
"OTEL_ENDPOINT": "__collector__",
|
||||
|
|
|
|||
|
|
@ -1,113 +0,0 @@
|
|||
import asyncio
|
||||
import json
|
||||
import time
|
||||
from collections.abc import Mapping
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
|
||||
|
||||
_IDENTITY: Final = {
|
||||
"user_api_key_alias": "caller-span-alias",
|
||||
"user_api_key_user_id": "caller-span-user",
|
||||
"user_api_key_team_id": "caller-span-team",
|
||||
}
|
||||
_CHAT_REPLY: Final = Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": "chatcmpl-caller-span",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o-mini",
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
|
||||
|
||||
def _is_identity(key: str) -> bool:
|
||||
return key == "langfuse.observation.metadata" or key.startswith("langfuse.trace.metadata.user_api_key_")
|
||||
|
||||
|
||||
def _exported(langfuse: Wire, seen: list[tuple[str, dict[str, object]]]) -> list[tuple[str, dict[str, object]]]:
|
||||
for batch in langfuse.drain():
|
||||
request: ExportTraceServiceRequest = ExportTraceServiceRequest.FromString(batch.body)
|
||||
seen.extend(
|
||||
(
|
||||
span.name,
|
||||
{
|
||||
attribute.key: getattr(attribute.value, attribute.value.WhichOneof("value"))
|
||||
for attribute in span.attributes
|
||||
},
|
||||
)
|
||||
for resource in request.resource_spans
|
||||
for scope in resource.scope_spans
|
||||
for span in scope.spans
|
||||
)
|
||||
return seen
|
||||
|
||||
|
||||
async def _exported_until(
|
||||
langfuse: Wire, seen: list[tuple[str, dict[str, object]]], name: str, prompt: str
|
||||
) -> list[tuple[str, dict[str, object]]]:
|
||||
"""The spans the Langfuse sink got, once a ``name`` span for ``prompt`` is among them; polls without blocking the loop."""
|
||||
deadline: Final = time.monotonic() + 25
|
||||
while not any(span == name and prompt in json.dumps(attributes) for span, attributes in _exported(langfuse, seen)):
|
||||
assert time.monotonic() < deadline, [span for span, _ in seen]
|
||||
await asyncio.sleep(0.2)
|
||||
return seen
|
||||
|
||||
|
||||
async def _complete(upstream: Wire, prompt: str) -> None:
|
||||
await litellm.acompletion(
|
||||
model="openai/gpt-4o-mini",
|
||||
api_base=upstream.url + "/v1",
|
||||
api_key="synthetic-caller-span-key",
|
||||
messages=[{"role": "user", "content": prompt}],
|
||||
metadata=dict(_IDENTITY),
|
||||
)
|
||||
await GLOBAL_LOGGING_WORKER.flush()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_langfuse_otel_keeps_the_caller_identity_off_the_callers_active_span(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
app_exporter: Final = InMemorySpanExporter()
|
||||
app_provider: Final = TracerProvider()
|
||||
app_provider.add_span_processor(SimpleSpanProcessor(app_exporter))
|
||||
|
||||
def chat(_request: Request) -> Reply:
|
||||
return _CHAT_REPLY
|
||||
|
||||
with wire_server(chat) as upstream, wire_server(lambda _request: Reply()) as langfuse:
|
||||
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-caller-span")
|
||||
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-caller-span")
|
||||
monkeypatch.setenv("LANGFUSE_OTEL_HOST", langfuse.url)
|
||||
monkeypatch.setenv("OTEL_BSP_SCHEDULE_DELAY", "100")
|
||||
monkeypatch.setattr(litellm, "callbacks", ["langfuse_otel"])
|
||||
|
||||
seen: Final[list[tuple[str, dict[str, object]]]] = [] # mutable-ok: drain() consumes the sink queue
|
||||
with app_provider.get_tracer("app").start_as_current_span("app.operation"):
|
||||
await _complete(upstream, "inside the caller span")
|
||||
await _exported_until(langfuse, seen, "raw_gen_ai_request", "inside the caller span")
|
||||
await _complete(upstream, "no caller span")
|
||||
exported: Final = await _exported_until(langfuse, seen, "litellm_request", "no caller span")
|
||||
|
||||
(app_span,) = app_exporter.get_finished_spans()
|
||||
app_attributes: Final[Mapping[str, object]] = app_span.attributes or {}
|
||||
assert app_attributes["langfuse.observation.type"] == "generation", sorted(app_attributes)
|
||||
assert not [key for key in app_attributes if _is_identity(key)], sorted(app_attributes)
|
||||
(own_span,) = [attributes for name, attributes in exported if name == "litellm_request"]
|
||||
assert json.loads(str(own_span["langfuse.observation.metadata"])) == _IDENTITY, own_span
|
||||
assert {key: value for key, value in own_span.items() if key.startswith("langfuse.trace.metadata.")} == {
|
||||
f"langfuse.trace.metadata.{field}": value for field, value in _IDENTITY.items()
|
||||
}, sorted(own_span)
|
||||
|
|
@ -1,20 +1,13 @@
|
|||
import base64
|
||||
import json
|
||||
import os
|
||||
from collections.abc import Mapping
|
||||
from contextlib import nullcontext
|
||||
from datetime import datetime, timezone
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from opentelemetry import trace
|
||||
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 import TracerProvider
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
|
||||
from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig
|
||||
from litellm.types.llms.openai import ResponsesAPIResponse
|
||||
|
|
@ -519,313 +512,6 @@ class TestLangfuseOtelIntegration:
|
|||
assert isinstance(config, OpenTelemetryConfig)
|
||||
# Endpoint assertion removed as side effect is gone
|
||||
|
||||
def test_request_metadata_is_emitted_under_langfuse_observation_and_trace_metadata_keys(self) -> None:
|
||||
request_metadata: Final = {
|
||||
"user_api_key_hash": "hash123",
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": "end-user-1",
|
||||
"user_api_key_team_id": "team-1",
|
||||
"user_api_key_team_alias": "team-a",
|
||||
"user_api_key_org_id": "org-1",
|
||||
"spend_logs_metadata": {"env": "prod"},
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"requester_metadata": {"ticket": "LIT-8283"},
|
||||
}
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {"metadata": request_metadata},
|
||||
}
|
||||
|
||||
with (
|
||||
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
|
||||
patch("litellm.integrations.arize._utils.set_attributes"),
|
||||
):
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert json.loads(actual["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": "end-user-1",
|
||||
"user_api_key_team_id": "team-1",
|
||||
"user_api_key_team_alias": "team-a",
|
||||
"user_api_key_org_id": "org-1",
|
||||
}
|
||||
assert {key: value for key, value in actual.items() if key.startswith("langfuse.trace.metadata.")} == {
|
||||
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
|
||||
"langfuse.trace.metadata.user_api_key_user_id": "user-1",
|
||||
"langfuse.trace.metadata.user_api_key_end_user_id": "end-user-1",
|
||||
"langfuse.trace.metadata.user_api_key_team_id": "team-1",
|
||||
"langfuse.trace.metadata.user_api_key_team_alias": "team-a",
|
||||
}
|
||||
|
||||
def test_request_metadata_redaction_matches_vanilla_langfuse(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
|
||||
request_metadata: Final = {
|
||||
"user_api_key_hash": "hash123",
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": "end-user-1",
|
||||
"user_api_key_team_id": "team-1",
|
||||
"user_api_key_team_alias": "team-a",
|
||||
"spend_logs_metadata": {"env": "prod"},
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"requester_metadata": {"ticket": "LIT-8283"},
|
||||
}
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {"metadata": request_metadata},
|
||||
}
|
||||
|
||||
with (
|
||||
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
|
||||
patch("litellm.integrations.arize._utils.set_attributes"),
|
||||
):
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert "langfuse.observation.metadata" not in actual
|
||||
assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")]
|
||||
|
||||
def test_request_metadata_beyond_the_identity_fields_stays_off_the_langfuse_metadata_keys(self) -> None:
|
||||
request_metadata: Final = {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"user_api_key_user_email": "caller-email-secret",
|
||||
"user_api_key_auth_metadata": {"note": "auth-metadata-secret"},
|
||||
"user_agent": "user-agent-secret",
|
||||
"headers": {"x-tenant-secret": "header-secret"},
|
||||
"requester_custom_headers": {"x-tenant-secret": "custom-header-secret"},
|
||||
"requester_metadata": {"note": "requester-metadata-secret"},
|
||||
"spend_logs_metadata": {"note": "spend-logs-secret"},
|
||||
"prompt_management_metadata": {"prompt_variables": {"input": "prompt-variables-secret"}},
|
||||
"mcp_tool_call_metadata": {"arguments": {"location": "mcp-arguments-secret"}},
|
||||
"vector_store_request_metadata": [{"query": "vector-query-secret"}],
|
||||
"routing_decision": {"matched_keyword": "routing-prompt-secret"},
|
||||
}
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {"metadata": request_metadata},
|
||||
}
|
||||
|
||||
with (
|
||||
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
|
||||
patch("litellm.integrations.arize._utils.set_attributes"),
|
||||
):
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
langfuse_metadata: Final = {
|
||||
key: value
|
||||
for key, value in actual.items()
|
||||
if key == "langfuse.observation.metadata" or key.startswith("langfuse.trace.metadata.")
|
||||
}
|
||||
|
||||
assert langfuse_metadata == {
|
||||
"langfuse.observation.metadata": json.dumps({"user_api_key_alias": "prod-key"}),
|
||||
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
|
||||
}
|
||||
|
||||
def test_request_metadata_drops_empty_fields_and_skips_empty_trace_identities(self) -> None:
|
||||
request_metadata: Final = {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": None,
|
||||
"user_api_key_team_id": "",
|
||||
"user_api_key_team_alias": None,
|
||||
"team_id": None,
|
||||
"team_alias": None,
|
||||
"spend_logs_metadata": {"env": "prod"},
|
||||
}
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {"metadata": request_metadata},
|
||||
}
|
||||
|
||||
with (
|
||||
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
|
||||
patch("litellm.integrations.arize._utils.set_attributes"),
|
||||
):
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert json.loads(actual["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
}
|
||||
assert {key: value for key, value in actual.items() if key.startswith("langfuse.trace.metadata.")} == {
|
||||
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
|
||||
"langfuse.trace.metadata.user_api_key_user_id": "user-1",
|
||||
}
|
||||
|
||||
def test_request_metadata_keys_are_absent_without_identity_fields(self) -> None:
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {
|
||||
"metadata": {"spend_logs_metadata": {"env": "prod"}, "requester_ip_address": "10.0.0.1"}
|
||||
},
|
||||
}
|
||||
|
||||
with (
|
||||
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
|
||||
patch("litellm.integrations.arize._utils.set_attributes"),
|
||||
):
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert "langfuse.observation.metadata" not in actual
|
||||
assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")]
|
||||
|
||||
def test_request_metadata_keys_are_absent_without_standard_logging_object(self) -> None:
|
||||
kwargs: Final = {"litellm_params": {"metadata": {"trace_metadata": {"k": "v"}}}}
|
||||
|
||||
with (
|
||||
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
|
||||
patch("litellm.integrations.arize._utils.set_attributes"),
|
||||
):
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert [key for key in actual if key.startswith("langfuse.trace.metadata")] == ["langfuse.trace.metadata"]
|
||||
assert "langfuse.observation.metadata" not in actual
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"unexpected_shape",
|
||||
[
|
||||
{"prompt_management_metadata": {1: "one", "prompt_id": "prompt-1"}},
|
||||
{"mcp_tool_call_metadata": {1: "one", "name": "lookup_weather"}},
|
||||
{"vector_store_request_metadata": [{1: "one", "vector_store_id": "vs-1"}]},
|
||||
{"requester_metadata": 1},
|
||||
{"requester_metadata": True},
|
||||
{"requester_metadata": 1.5},
|
||||
{1: "one"},
|
||||
{"user_api_key_user_id": 5},
|
||||
{"user_api_key_team_id": {"id": "team-1"}},
|
||||
],
|
||||
)
|
||||
def test_request_metadata_of_unexpected_shape_keeps_the_observation_output(
|
||||
self, unexpected_shape: dict[object, object]
|
||||
) -> None:
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {"metadata": {**unexpected_shape, "user_api_key_alias": "prod-key"}},
|
||||
}
|
||||
response_obj: Final = {"choices": [{"message": {"role": "assistant", "content": "hello"}}]}
|
||||
|
||||
with (
|
||||
patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute,
|
||||
patch("litellm.integrations.arize._utils.set_attributes"),
|
||||
):
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, response_obj)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert json.loads(actual["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"}
|
||||
assert json.loads(actual["langfuse.observation.output"]) == {"role": "assistant", "content": "hello"}
|
||||
|
||||
# What ``OpenTelemetry._record_exception_on_span`` adds to a failed call's span after its attributes.
|
||||
_FAILED_CALL_ATTRIBUTES: Final = MappingProxyType(
|
||||
{
|
||||
"error.code": "429",
|
||||
"http.response.status_code": 429,
|
||||
"error.type": "RateLimitError",
|
||||
"error.message": "rate limited",
|
||||
"error.llm_provider": "openai",
|
||||
"error.stack_trace": "Traceback",
|
||||
}
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _exported_span(
|
||||
kwargs: dict[str, object],
|
||||
max_span_attributes: int | None,
|
||||
carried: dict[str, str] | None = None,
|
||||
added_after: Mapping[str, str | int] = MappingProxyType({}),
|
||||
active: bool = False,
|
||||
) -> ReadableSpan:
|
||||
exporter: Final = InMemorySpanExporter()
|
||||
provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=max_span_attributes))
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
span: Final = provider.get_tracer("langfuse-otel-test").start_span("litellm_request", attributes=carried)
|
||||
response_obj: Final = {"id": "chatcmpl-1", "choices": [{"message": {"role": "assistant", "content": "hello"}}]}
|
||||
with trace.use_span(span, end_on_exit=False) if active else nullcontext():
|
||||
LangfuseOtelLogger.set_langfuse_otel_attributes(span, kwargs, response_obj)
|
||||
span.set_attributes(added_after)
|
||||
span.end()
|
||||
(finished,) = exporter.get_finished_spans()
|
||||
return finished
|
||||
|
||||
@staticmethod
|
||||
def _tool_call_kwargs() -> dict[str, object]:
|
||||
tools: Final = [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {"name": f"tool_{n}", "description": f"does thing {n}", "parameters": {"type": "object"}},
|
||||
}
|
||||
for n in range(3)
|
||||
]
|
||||
messages: Final = [{"role": "user", "content": "hello"}]
|
||||
return {
|
||||
"model": "gpt-4o",
|
||||
"messages": messages,
|
||||
"optional_params": {"tools": tools},
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {
|
||||
"model": "gpt-4o",
|
||||
"custom_llm_provider": "openai",
|
||||
"call_type": "completion",
|
||||
"messages": messages,
|
||||
"model_parameters": {"tools": tools},
|
||||
"metadata": {"user_api_key_alias": "prod-key", "user_api_key_team_id": "team-1"},
|
||||
},
|
||||
}
|
||||
|
||||
def test_request_metadata_never_evicts_or_drops_another_attribute_at_the_span_limit(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
kwargs: Final = self._tool_call_kwargs()
|
||||
failed: Final = self._FAILED_CALL_ATTRIBUTES
|
||||
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
|
||||
plain: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET, added_after=failed).attributes or {})
|
||||
succeeded: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET).attributes or {})
|
||||
monkeypatch.setattr(litellm, "redact_user_api_key_info", False)
|
||||
full: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET, added_after=failed).attributes or {})
|
||||
identity_keys: Final = set(full) - set(plain)
|
||||
assert identity_keys == {
|
||||
"langfuse.observation.metadata",
|
||||
"langfuse.trace.metadata.user_api_key_alias",
|
||||
"langfuse.trace.metadata.user_api_key_team_id",
|
||||
}
|
||||
|
||||
kept_identity: list[int] = [] # mutable-ok: one count per limit
|
||||
for limit in range(len(plain), len(full) + 1):
|
||||
failed_span = self._exported_span(kwargs, limit, added_after=failed)
|
||||
succeeded_span = self._exported_span(kwargs, limit - len(failed))
|
||||
for span, expected in ((failed_span, plain), (succeeded_span, succeeded)):
|
||||
attributes = dict(span.attributes or {})
|
||||
assert span.dropped_attributes == 0
|
||||
assert {key: attributes.get(key) for key in expected} == expected
|
||||
assert set(attributes) - set(expected) <= identity_keys
|
||||
kept_identity.append(len(set(failed_span.attributes or {}) & identity_keys))
|
||||
assert kept_identity == list(range(len(identity_keys) + 1))
|
||||
|
||||
@pytest.mark.parametrize("carried", [None, {"caller.attribute": "kept"}], ids=["bare", "with_attributes"])
|
||||
def test_request_metadata_skips_the_callers_active_span(self, carried: dict[str, str] | None) -> None:
|
||||
attributes: Final = dict(
|
||||
self._exported_span(self._tool_call_kwargs(), SpanLimits.UNSET, carried, active=True).attributes or {}
|
||||
)
|
||||
|
||||
assert {key: attributes[key] for key in carried or {}} == (carried or {})
|
||||
assert attributes["langfuse.observation.type"] == "generation"
|
||||
assert "langfuse.observation.metadata" not in attributes
|
||||
assert not [key for key in attributes if key.startswith("langfuse.trace.metadata.")]
|
||||
|
||||
|
||||
class TestLangfuseOtelKeyDynamicConfig:
|
||||
"""Key/team-scoped Langfuse credentials must define the full export target
|
||||
|
|
@ -1405,38 +1091,3 @@ def test_langfuse_otel_does_not_start_proxy_request_span() -> None:
|
|||
finally:
|
||||
langfuse_provider.shutdown()
|
||||
generic_provider.shutdown()
|
||||
|
||||
|
||||
def test_langfuse_otel_success_writes_identity_onto_its_own_span_and_never_onto_the_callers() -> None:
|
||||
langfuse_exporter: Final = InMemorySpanExporter()
|
||||
langfuse_provider: Final = TracerProvider()
|
||||
langfuse_provider.add_span_processor(SimpleSpanProcessor(langfuse_exporter))
|
||||
app_exporter: Final = InMemorySpanExporter()
|
||||
app_provider: Final = TracerProvider()
|
||||
app_provider.add_span_processor(SimpleSpanProcessor(app_exporter))
|
||||
with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True):
|
||||
logger: Final = LangfuseOtelLogger(callback_name="langfuse_otel", tracer_provider=langfuse_provider)
|
||||
response_obj: Final = {"id": "chatcmpl-1", "choices": [{"message": {"role": "assistant", "content": "hello"}}]}
|
||||
started_at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
try:
|
||||
with app_provider.get_tracer("app").start_as_current_span("app.operation"):
|
||||
logger._handle_success(
|
||||
TestLangfuseOtelIntegration._tool_call_kwargs(), response_obj, started_at, started_at
|
||||
)
|
||||
logger._handle_success(TestLangfuseOtelIntegration._tool_call_kwargs(), response_obj, started_at, started_at)
|
||||
|
||||
(app_span,) = app_exporter.get_finished_spans()
|
||||
(own_span,) = [span for span in langfuse_exporter.get_finished_spans() if span.name == "litellm_request"]
|
||||
app_attributes: Final = dict(app_span.attributes or {})
|
||||
own_attributes: Final = dict(own_span.attributes or {})
|
||||
assert app_attributes["langfuse.observation.type"] == "generation"
|
||||
identity_prefixes: Final = ("langfuse.observation.metadata", "langfuse.trace.metadata.")
|
||||
assert not [key for key in app_attributes if key.startswith(identity_prefixes)]
|
||||
assert json.loads(str(own_attributes["langfuse.observation.metadata"])) == {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_team_id": "team-1",
|
||||
}
|
||||
assert own_attributes["langfuse.trace.metadata.user_api_key_team_id"] == "team-1"
|
||||
finally:
|
||||
langfuse_provider.shutdown()
|
||||
app_provider.shutdown()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue