diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index c871f78cf85..a663356ad7b 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -1,10 +1,12 @@ import base64 import json import os -from collections.abc import Sequence +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.arize import _utils from litellm.integrations.langfuse.langfuse_otel_attributes import ( @@ -25,19 +27,13 @@ else: Span = Any +_STANDARD_LOGGING_OBJECT_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) + 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" LANGFUSE_INGESTION_VERSION: Final = "4" -_TRACE_IDENTITY_FIELDS: Final = ( - "user_api_key_alias", - "user_api_key_user_id", - "user_api_key_end_user_id", - "user_api_key_team_id", - "user_api_key_team_alias", -) - class LangfuseOtelLogger(OpenTelemetry): def __init__(self, config=None, *args, **kwargs): @@ -135,26 +131,30 @@ class LangfuseOtelLogger(OpenTelemetry): safe_set_attribute(span, enum_attr.value, value) @staticmethod - def _set_request_metadata_attributes(span: Span, kwargs: dict[str, object]) -> None: + def _set_request_metadata_attributes( + span: Span, + kwargs: dict[str, object], # mutable-ok: required callback signature accepts kwargs as a dict + ) -> None: from litellm.integrations.arize._utils import safe_set_attribute - from litellm.integrations.langfuse.langfuse import log_requester_metadata - from litellm.litellm_core_utils.redact_messages import redact_user_api_key_info + from litellm.integrations.otel.mappers.langfuse import TRACE_IDENTITY_FIELDS + from litellm.integrations.otel.model.metadata import exported_request_metadata from litellm.litellm_core_utils.safe_json_dumps import safe_dumps - standard_logging_object: Final = kwargs.get("standard_logging_object") - request_metadata: Final = ( - standard_logging_object.get("metadata") if isinstance(standard_logging_object, dict) else None - ) - if not isinstance(request_metadata, dict): + raw_standard_logging_object: Final = kwargs.get("standard_logging_object") + if not isinstance(raw_standard_logging_object, dict): return - observation_metadata: Final = { # mutable-ok: stays a real dict for safe_dumps and .get reads below - key: value - for key, value in log_requester_metadata(redact_user_api_key_info(metadata=request_metadata)).items() - if value is not None - } - safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_METADATA.value, safe_dumps(observation_metadata)) + standard_logging_object: Final = _STANDARD_LOGGING_OBJECT_ADAPTER.validate_python(raw_standard_logging_object) + if not isinstance(standard_logging_object.get("metadata"), dict): + return + observation_metadata: Final = exported_request_metadata(standard_logging_object) + if observation_metadata: + safe_set_attribute( + span, + LangfuseSpanAttributes.OBSERVATION_METADATA.value, + safe_dumps(dict(observation_metadata)), + ) trace_prefix: Final = LangfuseSpanAttributes.TRACE_METADATA.value - for field in _TRACE_IDENTITY_FIELDS: + for field in TRACE_IDENTITY_FIELDS: if value := observation_metadata.get(field): safe_set_attribute(span, f"{trace_prefix}.{field}", value) diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index 68895d996ce..4f65111c4d8 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -11,10 +11,12 @@ the JSON-serialized payloads. ``trace_attributes`` maps the caller's trace contr """ import json -from collections.abc import Callable +from collections.abc import Callable, Mapping from types import MappingProxyType from typing import Final +from pydantic import TypeAdapter + from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData from litellm.integrations.otel.mappers.utils import ( collect, @@ -40,6 +42,7 @@ LANGFUSE_TRACE_SESSION_ID: Final = "session.id" LANGFUSE_TRACE_TAGS: Final = "langfuse.trace.tags" LANGFUSE_OBSERVATION_METADATA: Final = "langfuse.observation.metadata" LANGFUSE_TRACE_METADATA_PREFIX: Final = "langfuse.trace.metadata." +_REQUEST_METADATA_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) TRACE_IDENTITY_FIELDS: Final = ( "user_api_key_alias", "user_api_key_user_id", @@ -88,7 +91,7 @@ class LangfuseMapper: # JSON-payload attributes: each builder returns the serialized blob or None. _BLOB_ATTRS: dict[str, Callable[[LLMCallSpanData], AttrValue | None]] = { LANGFUSE_OBSERVATION_METADATA: lambda d: ( - safe_dumps(dict(d.request_metadata)) # mutable-ok: safe_dumps only serializes real dicts + safe_dumps({**_REQUEST_METADATA_ADAPTER.validate_python(d.request_metadata)}) if d.request_metadata else None ), diff --git a/litellm/integrations/otel/mappers/utils.py b/litellm/integrations/otel/mappers/utils.py index 80ef2dd553a..c7e99ae0535 100644 --- a/litellm/integrations/otel/mappers/utils.py +++ b/litellm/integrations/otel/mappers/utils.py @@ -135,9 +135,9 @@ def message_content(message: object) -> str | None: return None -def output_messages(data: LLMCallSpanData) -> list: +def output_messages(data: LLMCallSpanData) -> tuple[object, ...]: """The ``message`` payload of each response choice.""" - return [c.get("message") for c in data.choices_out if isinstance(c, dict)] + return tuple(c.get("message") for c in data.choices_out if isinstance(c, dict)) def message_tool_calls(message: object) -> tuple[MessageToolCall, ...]: diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index f48330498a4..5047cbfe979 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -49,11 +49,9 @@ from litellm.constants import ( LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, SESSION_ID_GENERATED_METADATA_KEY, ) -from litellm.integrations.langfuse.langfuse import log_requester_metadata from litellm.integrations.otel.model.semconv import resolve_operation from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls from litellm.integrations.otel.model.utils import as_str, as_str_mapping, to_seconds -from litellm.litellm_core_utils.redact_messages import redact_user_api_key_info from litellm.types.utils import InternalCallOrigin if TYPE_CHECKING: @@ -61,6 +59,7 @@ if TYPE_CHECKING: _INTERNAL_CALL_ORIGINS: Final[frozenset[str]] = frozenset(get_args(InternalCallOrigin)) _AUTH_ATTRIBUTE: Final = TypeAdapter(object) +_REQUEST_METADATA_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) REQUESTER_METADATA_KEY: Final = "requester_metadata" REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}." @@ -466,20 +465,25 @@ def _model_info_id(model_info: object) -> str | None: return None -def exported_request_metadata(payload: StandardLoggingPayload) -> Mapping[str, object]: +def exported_request_metadata(payload: Mapping[str, object]) -> Mapping[str, object]: """The request metadata as the logging callbacks export it: ``user_api_key_*`` dropped when ``litellm.redact_user_api_key_info`` is on, header keys nested under ``requester_metadata``, ``None`` values dropped.""" - raw_meta: Final = cast( # cast-ok: StandardLoggingPayload.get returns Any | None - Mapping[str, object], payload.get("metadata") or MappingProxyType({}) + from litellm.integrations.langfuse.langfuse import log_requester_metadata + from litellm.litellm_core_utils.redact_messages import ( # pyright: ignore[reportUnknownVariableType] # helper uses bare dict + redact_user_api_key_info as _redact_user_api_key_info, ) - redacted: Final = cast( # cast-ok: redact_user_api_key_info is untyped - Mapping[str, object], - redact_user_api_key_info(metadata=dict(raw_meta)), # mutable-ok: the redactor requires a real dict + + raw_metadata: Final = payload.get("metadata") + metadata: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python( + raw_metadata if isinstance(raw_metadata, Mapping) else MappingProxyType({}) ) - exported: Final = cast( # cast-ok: log_requester_metadata is untyped - Mapping[str, object], log_requester_metadata(redacted) + redact_metadata: Final = cast( # cast-ok: adapter validates the helper's output below + Callable[[dict[str, object]], object], + _redact_user_api_key_info, ) + redacted: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python(redact_metadata(dict(metadata))) + exported: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python(log_requester_metadata(redacted)) return MappingProxyType({key: value for key, value in exported.items() if value is not None}) diff --git a/tests/integration/observability/test_langfuse_otel_metadata.py b/tests/integration/observability/test_langfuse_otel_metadata.py index bf18bc967f7..68cbf7e5e89 100644 --- a/tests/integration/observability/test_langfuse_otel_metadata.py +++ b/tests/integration/observability/test_langfuse_otel_metadata.py @@ -11,7 +11,6 @@ from pathlib import Path from typing import Final import httpx -import psutil import yaml from anthropic import Anthropic, AsyncAnthropic from integration._support.client import Gateway, Scenario, eventually @@ -928,20 +927,16 @@ def test_langfuse_otel_worker_kill_mid_burst_keeps_serving_and_exports(gateway: return 0, repr(error) return response.status_code, response.json()["id"] if response.status_code == 200 else response.text - workers: Final = tuple( - child - for child in psutil.Process(rig.proxy.process.pid).children(recursive=True) - if "spawn_main" in " ".join(child.cmdline()) - ) - assert len(workers) >= 2, f"expected multiple uvicorn workers, got {workers!r}" - victim: Final = workers[0] + workers: Final = rig.proxy.worker_pids() + assert len(workers) >= 2, f"expected multiple uvicorn workers, got {sorted(workers)!r}" + victim: Final = min(workers) with ThreadPoolExecutor(max_workers=20) as pool: futures: Final = tuple(pool.submit(call, index) for index in range(20)) while sum(f.done() for f in futures) < 5: wait(futures, timeout=0.05) - victim.kill() + replacement: Final = rig.proxy.kill_worker(victim) outcomes: Final = tuple(f.result() for f in futures) - assert not victim.is_running() or victim.status() == psutil.STATUS_ZOMBIE + assert replacement != victim succeeded: Final = tuple(identifier for status, identifier in outcomes if status == 200) assert succeeded, outcomes post_kill: Final = tuple(call(20 + index) for index in range(5)) diff --git a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py index c5e3019cfda..e658f24b041 100644 --- a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py +++ b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py @@ -834,26 +834,22 @@ def test_langfuse_mapper_skips_empty_identity_values(): assert attrs["langfuse.trace.metadata.user_api_key_end_user_id"] == "end-1" -def test_langfuse_mapper_redacts_user_api_key_fields(): +def test_langfuse_mapper_redacts_user_api_key_fields(monkeypatch: pytest.MonkeyPatch) -> None: import litellm - saved: Final = litellm.redact_user_api_key_info - try: - litellm.redact_user_api_key_info = True - payload: Final[dict[str, object]] = { - "call_type": "acompletion", - "custom_llm_provider": "openai", - "model": "gpt-4o", - "metadata": { - "user_api_key_alias": "prod-key", - "user_api_key_team_id": "team-9", - "spend_logs_metadata": {"ticket": "LIT-8283"}, - }, - "response": {}, - } - attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload)) - finally: - litellm.redact_user_api_key_info = saved + monkeypatch.setattr(litellm, "redact_user_api_key_info", True) + payload: Final[dict[str, object]] = { + "call_type": "acompletion", + "custom_llm_provider": "openai", + "model": "gpt-4o", + "metadata": { + "user_api_key_alias": "prod-key", + "user_api_key_team_id": "team-9", + "spend_logs_metadata": {"ticket": "LIT-8283"}, + }, + "response": {}, + } + attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload)) exported: Final = json.loads(attrs["langfuse.observation.metadata"]) assert not [key for key in exported if key.startswith("user_api_key")] diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index 183f21bb806..ab0e7669fe5 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -500,7 +500,7 @@ class TestLangfuseOtelIntegration: "requester_ip_address": "10.0.0.1", "requester_metadata": {"ticket": "LIT-8283"}, } - kwargs: Final = { + kwargs: Final[dict[str, object]] = { "litellm_params": {"metadata": {}}, "standard_logging_object": {"metadata": request_metadata}, } @@ -519,9 +519,8 @@ class TestLangfuseOtelIntegration: "langfuse.trace.metadata.user_api_key_team_alias": "team-a", } - def test_request_metadata_redaction_matches_vanilla_langfuse(self) -> None: - previous_flag: Final = litellm.redact_user_api_key_info - litellm.redact_user_api_key_info = True + 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", @@ -533,25 +532,22 @@ class TestLangfuseOtelIntegration: "requester_ip_address": "10.0.0.1", "requester_metadata": {"ticket": "LIT-8283"}, } - kwargs: Final = { + kwargs: Final[dict[str, object]] = { "litellm_params": {"metadata": {}}, "standard_logging_object": {"metadata": request_metadata}, } - try: - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) + with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: + LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} + 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"]) == { - "spend_logs_metadata": {"env": "prod"}, - "requester_ip_address": "10.0.0.1", - "requester_metadata": {"ticket": "LIT-8283"}, - } - assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")] - finally: - litellm.redact_user_api_key_info = previous_flag + assert json.loads(actual["langfuse.observation.metadata"]) == { + "spend_logs_metadata": {"env": "prod"}, + "requester_ip_address": "10.0.0.1", + "requester_metadata": {"ticket": "LIT-8283"}, + } + assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")] def test_request_metadata_drops_null_fields_and_skips_empty_trace_identities(self) -> None: request_metadata: Final = { @@ -564,7 +560,7 @@ class TestLangfuseOtelIntegration: "team_alias": None, "spend_logs_metadata": {"env": "prod"}, } - kwargs: Final = { + kwargs: Final[dict[str, object]] = { "litellm_params": {"metadata": {}}, "standard_logging_object": {"metadata": request_metadata}, }