diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 342b8ef5148..69d03fdcdd5 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -1,6 +1,7 @@ import base64 import json import os +from collections.abc import Mapping from datetime import datetime from typing import TYPE_CHECKING, Any, Final, Optional @@ -42,7 +43,7 @@ class LangfuseOtelLogger(OpenTelemetry): super().__init__(config=config, *args, **kwargs) @staticmethod - def set_langfuse_otel_attributes(span: Span, kwargs: dict[str, object], response_obj) -> None: + def set_langfuse_otel_attributes(span: Span, kwargs, response_obj) -> None: """ Sets OpenTelemetry span attributes for Langfuse observability. Uses the same attribute setting logic as Arize Phoenix for consistency. @@ -55,7 +56,13 @@ class LangfuseOtelLogger(OpenTelemetry): # Set Langfuse specific attributes ######################################################### LangfuseOtelLogger._set_langfuse_specific_attributes(span=span, kwargs=kwargs, response_obj=response_obj) - LangfuseOtelLogger._set_trace_user_attribute(span=span, kwargs=kwargs) + LangfuseOtelLogger._set_trace_user_attribute( + span=span, + kwargs=kwargs, + metadata=as_str_mapping( + LangfuseOtelLogger._extract_langfuse_metadata(kwargs) # pyright: ignore[reportUnknownArgumentType,reportUnknownMemberType] # helper returns a loosely typed dict + ), + ) @staticmethod def _extract_langfuse_metadata(kwargs: dict) -> dict: @@ -260,14 +267,13 @@ class LangfuseOtelLogger(OpenTelemetry): LangfuseOtelLogger._set_observation_output(span=span, response_obj=response_obj) @staticmethod - def _set_trace_user_attribute(span: Span, kwargs: dict[str, object]) -> None: + def _set_trace_user_attribute( + span: Span, kwargs: Mapping[str, object], metadata: Mapping[str, object] | None + ) -> None: slp: Final = as_str_mapping(kwargs.get("standard_logging_object")) slp_metadata: Final = as_str_mapping(slp.get("metadata")) if slp is not None else None if slp is None or slp_metadata is None: return - metadata: Final = as_str_mapping( - LangfuseOtelLogger._extract_langfuse_metadata(kwargs) # pyright: ignore[reportUnknownArgumentType,reportUnknownMemberType] # helper returns a loosely typed dict - ) litellm_params: Final = as_str_mapping(kwargs.get("litellm_params")) or {} bodies: Final = metadata_bodies(litellm_params) caller: Final = (as_str(metadata.get("trace_user_id")) if metadata is not None else None) or next( @@ -279,7 +285,7 @@ class LangfuseOtelLogger(OpenTelemetry): end_user: Final = ( slp_metadata.get("user_api_key_end_user_id") or slp.get("end_user") - or next((value for body in bodies if (value := as_str(body.get("user_api_key_end_user_id")))), None) + or next((found for body in bodies if (found := as_str(body.get("user_api_key_end_user_id")))), None) ) if end_user: safe_set_attribute(span, LangfuseSpanAttributes.TRACE_USER_ID.value, str(end_user)) diff --git a/tests/integration/observability/test_langfuse_otel_identity_chaos.py b/tests/integration/observability/test_langfuse_otel_identity_chaos.py index 4a627ea7e73..24125cd0410 100644 --- a/tests/integration/observability/test_langfuse_otel_identity_chaos.py +++ b/tests/integration/observability/test_langfuse_otel_identity_chaos.py @@ -1,6 +1,4 @@ import concurrent.futures -import os -import signal import threading import time import uuid @@ -19,7 +17,7 @@ from _langfuse_otel import ( _upstream_reply_for, ) from integration._support.client import Gateway, eventually -from integration._support.process import group_members, owned_proxy_process +from integration._support.process import owned_proxy_process from integration._support.wire import Reply, Request, Wire, wire_server @@ -208,11 +206,9 @@ def test_langfuse_otel_worker_kill_mid_burst_keeps_every_marker(gateway: Gateway for future in concurrent.futures.as_completed(futures): results[futures[future]] = future.result() if len(results) >= 10 and not killed["done"]: - children: Final = tuple( - process for process in group_members(owned.process.pid) if process.pid != owned.process.pid - ) - assert children, "no uvicorn worker child found" - os.kill(children[0].pid, signal.SIGKILL) + serving: Final = owned.worker_pids() + assert serving, "no uvicorn worker child found" + owned.kill_worker(min(serving)) killed["done"] = True assert killed["done"], "worker kill never fired" accepted: Final = {m for m, r in results.items() if r["status"] == 200}