mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
chore(langfuse_otel): meet the type discipline and test quality gates tightened on main
Read the trace-user inputs as Mapping views, keep the walrus targets distinct, and kill the chaos worker through the harness kill_worker API instead of a raw SIGKILL
This commit is contained in:
parent
1d3c99f445
commit
5ac2bd29a9
2 changed files with 17 additions and 15 deletions
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue