chore(release): backport #42388 and #41462 to stable/1.102.x

litellm/integrations/opentelemetry.py

@@ -5,6 +5,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from datetime import datetime
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, TypedDict, cast
 
 import litellm
@@ -20,7 +21,9 @@
     OTELSemconvCategory,
     parse_semconv_opt_in,
 )
+from litellm.integrations.otel.model.baggage import promoted_metadata
 from litellm.integrations.otel.model.db_endpoint import db_span_attributes
+from litellm.integrations.otel.model.metadata import flatten_metadata
 from litellm.integrations.otel.model.semconv import Metric
 from litellm.litellm_core_utils.internal_call_metadata import is_unbilled_non_inference_call_from_params
 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
@@ -288,6 +291,7 @@ class OpenTelemetryConfig:
     # under ``litellm.team.metadata``. Empty by default so none of a team's
     # metadata leaves the process until explicitly allowlisted.
     baggage_team_metadata_keys: list[str] = field(default_factory=list)
+    baggage_metadata_keys: list[str] = field(default_factory=list)
     # Prometheus-style include/exclude control over which attributes are stamped
     # on emitted metrics, to cap metric cardinality.
     attributes: OTELMetricAttributeFilter | None = None
@@ -314,6 +318,9 @@ def __post_init__(self) -> None:
         self.baggage_team_metadata_keys = _normalize_team_metadata_keys(
             self.baggage_team_metadata_keys
         ) or _normalize_team_metadata_keys(os.getenv("LITELLM_OTEL_BAGGAGE_TEAM_METADATA_KEYS"))
+        self.baggage_metadata_keys = _normalize_team_metadata_keys(
+            self.baggage_metadata_keys
+        ) or _normalize_team_metadata_keys(os.getenv("LITELLM_OTEL_BAGGAGE_METADATA_KEYS"))
 
     @classmethod
     def from_env(cls):
@@ -366,11 +373,14 @@ def __init__(
         **kwargs,
     ):
         team_metadata_keys_override: Final = kwargs.pop("baggage_team_metadata_keys", None)
+        metadata_keys_override: Final = kwargs.pop("baggage_metadata_keys", None)
         metric_attributes_override: Final = kwargs.pop("attributes", None)
         if config is None:
             config = OpenTelemetryConfig.from_env()
         if team_metadata_keys_override is not None:
             config.baggage_team_metadata_keys = _normalize_team_metadata_keys(team_metadata_keys_override)
+        if metadata_keys_override is not None:
+            config.baggage_metadata_keys = _normalize_team_metadata_keys(metadata_keys_override)
         if metric_attributes_override is not None:
             config.attributes = _build_metric_attribute_filter(metric_attributes_override)
 
@@ -1542,6 +1552,11 @@ def _set_inference_identity_attributes(
         if team_metadata:
             self.safe_set_attribute(span=span, key=TEAM_METADATA_ATTRIBUTE, value=team_metadata)
 
+        if self.config.baggage_metadata_keys:
+            flat_metadata: Final = MappingProxyType(dict(flatten_metadata(metadata)))
+            for key, value in promoted_metadata(flat_metadata, tuple(self.config.baggage_metadata_keys)).items():
+                self.safe_set_attribute(span=span, key=key, value=value)
+
         model_group: Final = standard_logging_payload.get("model_group")
         if model_group:
             self.safe_set_attribute(span=span, key=MODEL_GROUP_ATTRIBUTE, value=model_group)

litellm/integrations/otel/logger.py

@@ -33,6 +33,7 @@
     LLMCallEvent,
     RequestIdentity,
     auth_metadata,
+    metadata_from_request_data,
     model_from_request_data,
 )
 from litellm.integrations.otel.model.payloads import (
@@ -679,7 +680,12 @@ def _emit_service(
     #  / errors are the FastAPI instrumentor's job, so we don't touch it here.
     # ====================================================================== #
 
-    def seed_request_identity(self, user_api_key_dict: object, model: str | None = None) -> None:
+    def seed_request_identity(
+        self,
+        user_api_key_dict: object,
+        model: str | None = None,
+        request_metadata: Mapping[str, object] | None = None,
+    ) -> None:
         """Attach request-identity Baggage to the current context + server span.
 
         Seeding identity into Baggage makes **every** span emitted afterwards for
@@ -691,7 +697,7 @@ def seed_request_identity(self, user_api_key_dict: object, model: str | None = N
         isn't determined yet, which is correct.
         """
         try:
-            identity: Final = RequestIdentity.from_user_api_key_auth(user_api_key_dict)
+            identity: Final = RequestIdentity.from_user_api_key_auth(user_api_key_dict, request_metadata)
             bag: Final = promoted_baggage(
                 identity,
                 model,
@@ -743,6 +749,7 @@ async def async_pre_call_hook(
         self.seed_request_identity(
             user_api_key_dict,
             model=model_from_request_data(data),
+            request_metadata=metadata_from_request_data(data),
         )
         return data
 

litellm/integrations/otel/model/baggage.py

@@ -15,9 +15,10 @@
 
 import json
 from collections.abc import Callable, Mapping
+from types import MappingProxyType
 from typing import Final
 
-from litellm.integrations.otel.model.metadata import RequestIdentity
+from litellm.integrations.otel.model.metadata import REQUESTER_METADATA_PATH, RequestIdentity
 from litellm.integrations.otel.model.semconv import GenAI, LiteLLM
 
 # Attribute key -> value extractor over (identity, request_model,
@@ -79,17 +80,23 @@ def promoted_baggage(
     ``team_metadata_keys`` selects sub-keys of the team's metadata to promote
     under ``litellm.team.metadata``. Empty values are dropped.
     """
-    out: Final[dict[str, str]] = {}
-    for key, extract in _PROMOTABLE.items():
-        if key in promoted_keys:
-            value = extract(identity, request_model, team_metadata_keys)
-            if value:
-                out[key] = value
-    for meta_key in metadata_keys:
-        value = identity.metadata.get(meta_key)
-        if value:
-            out[f"{LiteLLM.METADATA_PREFIX}{meta_key}"] = value
-    return out
+    identity_values: Final = {
+        key: value
+        for key, extract in _PROMOTABLE.items()
+        if key in promoted_keys and (value := extract(identity, request_model, team_metadata_keys))
+    }
+    return {**identity_values, **promoted_metadata(identity.metadata, metadata_keys)}
+
+
+def promoted_metadata(metadata: Mapping[str, str], metadata_keys: tuple[str, ...]) -> Mapping[str, str]:
+    """Allowlisted entries of a flattened metadata mapping under ``litellm.metadata.*``."""
+    return MappingProxyType(
+        {
+            f"{LiteLLM.METADATA_PREFIX}{meta_key.removeprefix(REQUESTER_METADATA_PATH)}": value
+            for meta_key in metadata_keys
+            if (value := metadata.get(meta_key))
+        }
+    )
 
 
 def _filtered_team_metadata_json(

litellm/integrations/otel/model/config.py

@@ -210,7 +210,10 @@ class OpenTelemetryV2Config(BaseSettings):
         validation_alias=AliasChoices("baggage_metadata_keys", "LITELLM_OTEL_BAGGAGE_METADATA_KEYS"),
         description=(
             "Metadata sub-keys promoted under the ``litellm.metadata.*`` "
-            "namespace. Configure via the ``LITELLM_OTEL_BAGGAGE_METADATA_KEYS`` "
+            "namespace. A dotted path such as ``requester_metadata.trace_id`` "
+            "reads the caller's nested ``metadata.trace_id`` and is promoted as "
+            "``litellm.metadata.trace_id``; other dotted keys keep their full path. "
+            "Configure via the ``LITELLM_OTEL_BAGGAGE_METADATA_KEYS`` "
             "env var (comma-separated) or "
             "``callback_settings.otel.baggage_metadata_keys`` in config.yaml."
         ),

litellm/integrations/otel/model/metadata.py

@@ -49,6 +49,8 @@
     from litellm.types.utils import StandardLoggingPayload
 
 LANGFUSE_TRACE_NAME_HEADER: Final = "langfuse_trace_name"
+REQUESTER_METADATA_KEY: Final = "requester_metadata"
+REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
 
 
 @dataclass(frozen=True)
@@ -78,7 +80,7 @@ def from_payload(cls, payload: StandardLoggingPayload) -> RequestIdentity:
         model, not just the user-facing one.
         """
         raw_meta: Final = cast(Mapping[str, object], payload.get("metadata") or {})
-        metadata = {key: str(value) for key, value in raw_meta.items() if isinstance(value, (str, bool, int, float))}
+        metadata: Final = MappingProxyType(dict(flatten_metadata(raw_meta)))
         return cls(
             call_id=as_str(payload.get("litellm_call_id")) or as_str(payload.get("id")),
             # StandardLoggingMetadata's canonical key is ``user_api_key_team_id``;
@@ -95,27 +97,33 @@ def from_payload(cls, payload: StandardLoggingPayload) -> RequestIdentity:
         )
 
     @classmethod
-    def from_user_api_key_auth(cls, auth: object) -> RequestIdentity:
+    def from_user_api_key_auth(
+        cls, auth: object, request_metadata: Mapping[str, object] | None = None
+    ) -> RequestIdentity:
         """Identity from a ``UserAPIKeyAuth`` (duck-typed to keep this module
         free of a proxy import).
 
         Used in the pre-call hook to seed Baggage early — before any LLM,
         guardrail, or service span is created — so the whole request's spans
         inherit identity, not just the LLM-call span. Metadata sub-keys use the
         ``user_api_key_*`` names that ``baggage.DEFAULT_BAGGAGE_METADATA_KEYS``
-        promotes.
+        promotes; ``request_metadata`` (the caller's ``requester_metadata``
+        snapshot) is flattened to dotted keys so ``requester_metadata.<key>``
+        resolves too.
         """
         get: Final = lambda name: getattr(auth, name, None)  # noqa: E731
-        metadata: Final = {
-            meta_key: str(value)
+        auth_meta: Final = tuple(
+            (meta_key, str(value))
             for meta_key, attr in (
                 ("user_api_key_user_id", "user_id"),
                 ("user_api_key_org_id", "org_id"),
                 ("user_api_key_alias", "key_alias"),
                 ("user_api_key_end_user_id", "end_user_id"),
             )
             if (value := get(attr))
-        }
+        )
+        request_meta: Final = flatten_metadata(request_metadata) if request_metadata is not None else ()
+        metadata: Final = MappingProxyType(dict((*request_meta, *auth_meta)))
         return cls(
             team_id=as_str(get("team_id")),
             team_alias=as_str(get("team_alias")),
@@ -351,6 +359,35 @@ def model_from_request_data(data: object) -> str | None:
     return None
 
 
+def metadata_from_request_data(data: object) -> Mapping[str, object] | None:
+    """The caller's ``requester_metadata`` snapshot from a pre-call ``data`` dict, keyed under its wrapper.
+
+    The proxy stores it under ``metadata`` or ``litellm_metadata`` depending on the route;
+    the proxy-owned siblings (``user_api_key_*``, ``requester_ip_address``) are not read.
+    """
+    top: Final = _as_str_mapping(data)
+    if top is None:
+        return None
+    snapshots: Final = tuple(
+        snapshot
+        for name in ("metadata", "litellm_metadata")
+        if (nested := _as_str_mapping(top.get(name))) is not None
+        and (snapshot := _as_str_mapping(nested.get(REQUESTER_METADATA_KEY))) is not None
+    )
+    return MappingProxyType({REQUESTER_METADATA_KEY: snapshots[0]}) if snapshots else None
+
+
+def flatten_metadata(raw: Mapping[str, object]) -> Iterator[tuple[str, str]]:
+    """Scalar leaves of a nested metadata mapping, keyed by their dotted path."""
+    stack: Final = list(tuple(raw.items())[::-1])  # mutable-ok: iterative worklist keeps the walk off the call stack
+    while stack:
+        key, value = stack.pop()
+        if (nested := _as_str_mapping(value)) is not None:
+            stack.extend(tuple((f"{key}.{sub_key}", sub_value) for sub_key, sub_value in nested.items())[::-1])
+        elif isinstance(value, (str, bool, int, float)):
+            yield key, str(value)
+
+
 def resolve_provider_model(payload: StandardLoggingPayload) -> str | None:
     """The model litellm dispatched to the provider, from the payload.
 

litellm/litellm_core_utils/realtime_errors.py

@@ -9,13 +9,20 @@
 """
 
 import json
-from typing import Final
+from types import MappingProxyType
+from typing import Final, Protocol
 
 from litellm.types.realtime import RealtimeErrorDetail, RealtimeErrorEvent
 
 WEBSOCKET_CLOSE_REASON_MAX_BYTES: Final = 123
 
 
+class _ClientWebSocket(Protocol):
+    async def send_text(self, data: str) -> None: ...
+
+    async def close(self, code: int = ..., reason: str | None = ...) -> None: ...
+
+
 def realtime_error_event(message: str, error_type: str) -> str:
     detail: Final[RealtimeErrorDetail] = {"type": error_type, "message": message}
     event: Final[RealtimeErrorEvent] = {"type": "error", "error": detail}
@@ -37,3 +44,28 @@ def client_close_code(upstream_code: int) -> int:
     if upstream_code in EXTERNAL_CLOSE_CODES or 3000 <= upstream_code < 5000:
         return upstream_code
     return int(CloseCode.INTERNAL_ERROR)
+
+
+def upstream_handshake_close_code(status_code: int) -> int:
+    from websockets.frames import CloseCode
+
+    refusal_codes: Final = MappingProxyType(
+        {
+            401: int(CloseCode.POLICY_VIOLATION),
+            403: int(CloseCode.POLICY_VIOLATION),
+            429: int(CloseCode.TRY_AGAIN_LATER),
+        }
+    )
+    return refusal_codes.get(status_code, int(CloseCode.INTERNAL_ERROR))
+
+
+async def close_after_upstream_handshake_refusal(websocket: _ClientWebSocket, status_code: int) -> None:
+    message: Final = f"Upstream realtime handshake rejected with HTTP {status_code}"
+    try:
+        await websocket.send_text(realtime_error_event(message, error_type="server_error"))
+    except Exception:  # noqa: BLE001  # best-effort notice: a dead client socket must not skip the close below
+        pass
+    await websocket.close(
+        code=upstream_handshake_close_code(status_code),
+        reason=websocket_close_reason(message, fallback="Upstream handshake rejected"),
+    )

litellm/llms/azure/realtime/handler.py

@@ -8,11 +8,15 @@
 from types import MappingProxyType
 from typing import Any, Final, Protocol, cast
 
-from litellm._logging import _redact_string, verbose_proxy_logger
+from litellm._logging import verbose_proxy_logger
 from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
 from litellm.types.realtime import RealtimeQueryParams
 
 from ....litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
+from ....litellm_core_utils.realtime_errors import (
+    close_after_upstream_handshake_refusal,
+    realtime_error_event,
+)
 from ....litellm_core_utils.realtime_streaming import (
     RealTimeStreaming,
     ScopedWebSocket,
@@ -49,7 +53,9 @@ def azure_realtime_protocol_for_client(
 
 
 class _ProxyClientWebSocket(Protocol):
-    """Client-facing websocket handle: this path only closes it after a failed handshake."""
+    """Client-facing websocket handle: this path only writes to it after a failed handshake."""
+
+    async def send_text(self, data: str) -> None: ...
 
     async def close(self, code: int = ..., reason: str | None = ...) -> None: ...
 
@@ -181,7 +187,16 @@ async def async_realtime(
                 )
                 await realtime_streaming.bidirectional_forward()
 
-        except websockets.exceptions.InvalidStatusCode as e:
-            await websocket.close(code=e.status_code, reason=_redact_string(str(e)))
+        except websockets.exceptions.InvalidStatus as e:
+            verbose_proxy_logger.exception("Error in AzureOpenAIRealtime.async_realtime")
+            await close_after_upstream_handshake_refusal(websocket, e.response.status_code)
         except Exception:
             verbose_proxy_logger.exception("Error in AzureOpenAIRealtime.async_realtime")
+            try:
+                await websocket.send_text(realtime_error_event("Internal server error", error_type="server_error"))
+            except Exception:  # noqa: BLE001  # best-effort notice: a dead client socket must not skip the close below
+                pass
+            try:
+                await websocket.close(code=1011, reason="Internal server error")
+            except Exception:  # noqa: BLE001  # the lower layer may have closed the socket already; closing twice is not an error
+                pass

litellm/llms/custom_httpx/llm_http_handler.py

@@ -30,7 +30,11 @@
 )
 from litellm.litellm_core_utils.get_litellm_params import AWS_CREDENTIAL_KWARGS_KEYS
 from litellm.litellm_core_utils.llm_request_utils import serialize_multipart_form_fields
-from litellm.litellm_core_utils.realtime_errors import realtime_error_event, websocket_close_reason
+from litellm.litellm_core_utils.realtime_errors import (
+    close_after_upstream_handshake_refusal,
+    realtime_error_event,
+    websocket_close_reason,
+)
 from litellm.litellm_core_utils.realtime_streaming import RealTimeStreaming
 from litellm.litellm_core_utils.url_utils import encode_url_path_segment
 from litellm.llms.base_llm.anthropic_messages.transformation import (
@@ -6228,9 +6232,9 @@ async def async_realtime(
 
                 await realtime_streaming.bidirectional_forward()
 
-        except websockets.exceptions.InvalidStatusCode as e:
+        except websockets.exceptions.InvalidStatus as e:
             verbose_logger.exception("Error connecting to backend: %s", e)
-            await websocket.close(code=e.status_code, reason=_redact_string(str(e)))
+            await close_after_upstream_handshake_refusal(websocket, e.response.status_code)
         except Exception as e:
             verbose_logger.exception("Error connecting to backend: %s", e)
             redacted_error: Final = _redact_string(str(e))
@@ -6640,9 +6644,9 @@ async def _backend_connection():
                 )
                 await streaming.bidirectional_forward()
 
-        except websockets.exceptions.InvalidStatusCode as e:
+        except websockets.exceptions.InvalidStatus as e:
             verbose_logger.exception("Error connecting to responses WS backend: %s", e)
-            await websocket.close(code=e.status_code, reason=_redact_string(str(e)))
+            await close_after_upstream_handshake_refusal(websocket, e.response.status_code)
         except Exception as e:
             verbose_logger.exception("Error in responses WS: %s", e)
             try: