refactor(otel/v2): build destination collections immutably instead of raising the LIT002 ceiling
Some checks failed
Terraform Modules / fmt, validate, test (aws) (push) Has been cancelled
Terraform Provider / gofmt, vet, build, test (push) Has been cancelled
Terraform Provider / Provider endpoints vs proxy OpenAPI schema (push) Has been cancelled

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
shivam 2026-08-17 18:43:41 +00:00
parent 4a5dfddc23
commit 5396212902
17 changed files with 180 additions and 102 deletions

View file

@ -13,6 +13,7 @@ owning logger, and including it here would export the same call twice.
from collections.abc import Mapping
from datetime import datetime
from functools import lru_cache
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
from opentelemetry.trace import Span
@ -41,7 +42,7 @@ def _vocabulary_config(backend: str) -> OpenTelemetryV2Config:
config: Final = preset_fn(allow_missing_credentials=True)
except Exception: # noqa: BLE001 # an unbuildable preset still has a usable default vocabulary
return OpenTelemetryV2Config()
return config.model_copy(update={"exporters": ()})
return config.model_copy(update=MappingProxyType({"exporters": ()}))
class _DestinationOnlyOtel(OpenTelemetryV2):
@ -133,7 +134,8 @@ class AdminDestinationLogger(CustomLogger):
end_time: "datetime | float | None",
) -> None:
owned: Final = otel_v2_owned_backends()
for backend in sorted({d.callback_name for d in request_destinations() if d.callback_name} - owned):
requested: Final = frozenset(d.callback_name for d in request_destinations() if d.callback_name)
for backend in sorted(requested - owned):
try:
self._emitter_for(backend).export_to_destinations(kwargs, start_time, end_time)
except Exception as exc: # noqa: BLE001 # one destination's failure must not break the request or the others

View file

@ -1,6 +1,7 @@
"""Provider / exporter factory + the Baggage span processor."""
from collections.abc import Callable, Iterable, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
from opentelemetry import _logs, baggage, metrics
@ -133,7 +134,7 @@ def destination_resource_attrs(destination: "OtelDestination") -> Mapping[str, s
``model_id`` / ``arize.project.name``; empty for header-routed backends), read
by both export paths so the gen-AI span and its parents share one Resource.
"""
return dict(destination.resource_attributes)
return MappingProxyType(dict(destination.resource_attributes))
def parse_headers(raw: str | None) -> dict[str, str]:

View file

@ -6,6 +6,8 @@
import threading
from collections import OrderedDict
from itertools import chain
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
from opentelemetry.context import Context
@ -178,11 +180,15 @@ class TenantTracerCache:
)
if not owns_exporter:
return self._config.model_copy(
update={"exporters": [*self._config.exporters, *self._synthesized_exporter(header_str, dynamic_params)]}
update=MappingProxyType(
{
"exporters": [*self._config.exporters, *self._synthesized_exporter(header_str, dynamic_params)]
} # mutable-ok: model_copy bypasses validation, so the field's declared list/dict type must be built as-is
)
)
exporters: Final = [
(
spec.model_copy(update={"headers": header_str})
spec.model_copy(update=MappingProxyType({"headers": header_str}))
if spec.owner == self._callback_name and spec.kind.lower() not in _NON_OTLP_KINDS
else spec
)
@ -228,13 +234,14 @@ class TenantTracerCache:
destination_resource_attrs,
)
groups: OrderedDict[tuple[tuple[str, str], ...], list[OtelDestination]] = (
OrderedDict()
) # mutable-ok: insertion-order grouping accumulator, frozen before return
for destination in destinations:
key = tuple(sorted(destination_resource_attrs(destination).items()))
groups.setdefault(key, []).append(destination)
return tuple((key, tuple(group)) for key, group in sorted(groups.items()))
keyed: Final = tuple(
(tuple(sorted(destination_resource_attrs(destination).items())), destination)
for destination in destinations
)
return tuple(
(key, tuple(destination for other_key, destination in keyed if other_key == key))
for key in sorted(frozenset(key for key, _ in keyed))
)
def _tracer_for_group(
self,
@ -300,15 +307,24 @@ class TenantTracerCache:
for d in destinations
)
base_exporters: Final = (*self._config.exporters,) if include_base_exporters else ()
merged_resource_attrs: Final = {
**self._config.resource_attributes,
**{key: value for d in destinations for key, value in destination_resource_attrs(d).items()},
}
merged_resource_attrs: Final = (
dict( # mutable-ok: model_copy bypasses validation, so the declared dict field must be built as-is
chain(
self._config.resource_attributes.items(),
((key, value) for d in destinations for key, value in destination_resource_attrs(d).items()),
)
)
)
return self._config.model_copy(
update={
"exporters": [*base_exporters, *appended],
"resource_attributes": merged_resource_attrs,
}
update=MappingProxyType(
{
"exporters": [
*base_exporters,
*appended,
], # mutable-ok: model_copy bypasses validation, so the field's declared list/dict type must be built as-is
"resource_attributes": merged_resource_attrs,
}
)
)
@ -322,8 +338,7 @@ def _processor_key(destination: OtelDestination) -> "tuple[str, tuple[tuple[str,
def _is_genai_span(span: ReadableSpan) -> bool:
attributes: Final = span.attributes or {}
return _GENAI_SPAN_ATTR in attributes
return span.attributes is not None and _GENAI_SPAN_ATTR in span.attributes
def _with_destination_resource(span: ReadableSpan, destination: OtelDestination) -> ReadableSpan:
@ -336,7 +351,7 @@ def _with_destination_resource(span: ReadableSpan, destination: OtelDestination)
extra: Final = destination_resource_attrs(destination)
if not extra:
return span
merged: Final = Resource.create({**dict(span.resource.attributes), **extra})
merged: Final = Resource.create(MappingProxyType(dict(chain(span.resource.attributes.items(), extra.items()))))
return _ResourceWrappedReadableSpan(span, merged)

View file

@ -15,7 +15,8 @@ tracer for the integrations that support it; ``dynamic_otlp_headers`` below buil
those per-request headers.
"""
from collections.abc import Callable
from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
from litellm.integrations.otel.presets.agentops import agentops_preset
@ -39,17 +40,21 @@ if TYPE_CHECKING:
#: routing). Only integrations that support dynamic credentials appear here —
#: Arize-Phoenix/Langtrace/Levo/AgentOps/generic don't, so they use the logger's
#: default tracer.
DYNAMIC_HEADERS_BY_CALLBACK: Final[dict[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]]] = {
"arize": arize_dynamic_headers,
"langfuse_otel": langfuse_dynamic_headers,
"weave_otel": weave_dynamic_headers,
}
DYNAMIC_HEADERS_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], Mapping[str, str]]]] = (
MappingProxyType(
{
"arize": arize_dynamic_headers,
"langfuse_otel": langfuse_dynamic_headers,
"weave_otel": weave_dynamic_headers,
}
)
)
def dynamic_otlp_headers(
callback_name: str | None,
dynamic_params: "StandardCallbackDynamicParams | None",
) -> dict[str, str] | None:
) -> Mapping[str, str] | None:
"""Per-request OTLP headers for ``callback_name``, or ``None`` if N/A.
``None`` means "no per-request routing" the caller uses its default tracer.
@ -76,22 +81,26 @@ def dynamic_otlp_destination(
if callback_name not in DYNAMIC_HEADERS_BY_CALLBACK or not dynamic_params:
return None
values: Final = {str(key): str(value) for key, value in dynamic_params.items() if isinstance(value, str)}
values: Final = MappingProxyType(
{str(key): str(value) for key, value in dynamic_params.items() if isinstance(value, str)}
)
return build_destination(callback_name or "", values)
#: Callback name → preset. The ``Preset`` annotation makes mypy verify every
#: registered value matches the preset interface.
PRESET_BY_CALLBACK: Final[dict[str, Preset]] = {
"agentops": agentops_preset,
"arize": arize_preset,
"arize_phoenix": phoenix_preset,
"generic": generic_preset,
"langfuse_otel": langfuse_preset,
"langtrace": langtrace_preset,
"levo": levo_preset,
"weave_otel": weave_preset,
}
PRESET_BY_CALLBACK: Final[Mapping[str, Preset]] = MappingProxyType(
{
"agentops": agentops_preset,
"arize": arize_preset,
"arize_phoenix": phoenix_preset,
"generic": generic_preset,
"langfuse_otel": langfuse_preset,
"langtrace": langtrace_preset,
"levo": levo_preset,
"weave_otel": weave_preset,
}
)
__all__ = [

View file

@ -57,14 +57,19 @@ def agentops_preset(
ExporterSpec(
kind=_AGENTOPS_EXPORTER_KIND,
endpoint=_AGENTOPS_ENDPOINT,
options=({"api_key": settings.api_key} if settings.api_key else None),
options=(
{"api_key": settings.api_key} if settings.api_key else None
), # mutable-ok: ExporterSpec.options is a mutable dict field
owner=ExporterOwner.AGENTOPS,
),
)
)
return base.model_copy(
update={
"exporters": [*base.exporters, *global_exporter],
"exporters": [
*base.exporters,
*global_exporter,
], # mutable-ok: model_copy bypasses validation, so the field's declared list/dict type must be built as-is
"resource_attributes": {
**base.resource_attributes,
"service.name": settings.service_name,

View file

@ -51,7 +51,10 @@ def arize_preset(
)
return base.model_copy(
update={
"exporters": [*base.exporters, *global_exporter],
"exporters": [
*base.exporters,
*global_exporter,
], # mutable-ok: model_copy bypasses validation, so the field's declared list/dict type must be built as-is
"mapper_names": ensure_mappers(base.mapper_names, "openinference"),
"resource_attributes": {
**base.resource_attributes,

View file

@ -12,6 +12,7 @@ credential values only.
import os
from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import Final
from litellm.integrations.langfuse.langfuse_otel import (
@ -26,7 +27,7 @@ from litellm.integrations.weave.weave_otel import (
def _parse_header_string(raw: str) -> Mapping[str, str]:
pairs: Final = (item.split("=", 1) for item in raw.split(",") if "=" in item)
return {key.strip(): value.strip() for key, value in pairs}
return MappingProxyType({key.strip(): value.strip() for key, value in pairs})
def _langfuse_endpoint(host: str) -> str:
@ -44,7 +45,7 @@ def _langfuse_destination(values: Mapping[str, str]) -> OtelDestination | None:
auth: Final = LangfuseOtelLogger._get_langfuse_authorization_header( # pyright: ignore[reportPrivateUsage] # shared v1 header builder
public_key=public_key, secret_key=secret_key
)
return OtelDestination(endpoint=endpoint, headers={"Authorization": auth})
return OtelDestination(endpoint=endpoint, headers=MappingProxyType({"Authorization": auth}))
def _arize_destination(values: Mapping[str, str]) -> OtelDestination | None:
@ -60,10 +61,12 @@ def _arize_destination(values: Mapping[str, str]) -> OtelDestination | None:
project: Final = (
values.get("arize_project_name") or values.get("project_name") or os.environ.get("ARIZE_PROJECT_NAME")
)
resource_attributes: Final = {"model_id": project, "arize.project.name": project} if project else {}
resource_attributes: Final = MappingProxyType(
{"model_id": project, "arize.project.name": project} if project else {}
)
return OtelDestination(
endpoint=endpoint,
headers={"space_id": space, "api_key": api_key},
headers=MappingProxyType({"space_id": space, "api_key": api_key}),
resource_attributes=resource_attributes,
protocol="otlp_http" if http_endpoint and not values.get("arize_endpoint") else None,
)
@ -80,10 +83,15 @@ def _weave_destination(values: Mapping[str, str]) -> OtelDestination | None:
base: Final = (values.get("weave_endpoint") or WEAVE_BASE_URL).rstrip("/")
endpoint: Final = base if base.endswith("/v1/traces") else base.removesuffix("/otel") + WEAVE_OTEL_ENDPOINT
headers: Final = {"Authorization": _get_weave_authorization_header(api_key=api_key)}
project_id: Final = values.get("weave_project_id")
if project_id:
headers["project_id"] = project_id
headers: Final = MappingProxyType(
dict(
(
("Authorization", _get_weave_authorization_header(api_key=api_key)),
*((("project_id", project_id),) if project_id else ()),
)
)
)
return OtelDestination(endpoint=endpoint, headers=headers)
@ -108,11 +116,13 @@ def _generic_destination(values: Mapping[str, str]) -> OtelDestination | None:
)
_ADAPTERS: Final[Mapping[str, Callable[[Mapping[str, str]], OtelDestination | None]]] = {
"langfuse_otel": _langfuse_destination,
"arize": _arize_destination,
"weave_otel": _weave_destination,
}
_ADAPTERS: Final[Mapping[str, Callable[[Mapping[str, str]], OtelDestination | None]]] = MappingProxyType(
{
"langfuse_otel": _langfuse_destination,
"arize": _arize_destination,
"weave_otel": _weave_destination,
}
)
OTEL_V2_DESTINATION_CALLBACKS: Final = frozenset(_ADAPTERS)
@ -125,7 +135,7 @@ def build_destination(callback_name: str, values: Mapping[str, str]) -> OtelDest
host (an easy slip in the create form) yields a malformed OTLP URL the
exporter rejects with a 404, so whitespace is never significant here.
"""
trimmed: Final = {key: value.strip() for key, value in values.items()}
trimmed: Final = MappingProxyType({key: value.strip() for key, value in values.items()})
adapter: Final = _ADAPTERS.get(callback_name)
if adapter is not None:
destination: Final = adapter(trimmed)

View file

@ -1,5 +1,7 @@
"""Langfuse-OTEL preset."""
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
from litellm.integrations.langfuse.langfuse_otel import (
@ -14,7 +16,7 @@ from litellm.integrations.otel.presets.utils import ensure_mappers
from litellm.types.utils import StandardCallbackDynamicParams
def langfuse_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]:
def langfuse_dynamic_headers(params: StandardCallbackDynamicParams) -> Mapping[str, str]:
"""Per-request Langfuse OTLP headers from team/key dynamic params."""
public_key: Final = params.get("langfuse_public_key")
secret_key: Final = params.get("langfuse_secret_key")
@ -24,7 +26,7 @@ def langfuse_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str,
public_key=public_key, secret_key=secret_key
)
)
return {}
return MappingProxyType({})
def langfuse_preset(
@ -39,7 +41,7 @@ def langfuse_preset(
except Exception:
if not allow_missing_credentials:
raise
return base.model_copy(update={"mapper_names": mappers})
return base.model_copy(update=MappingProxyType({"mapper_names": mappers}))
kind: Final = cfg.exporter if isinstance(cfg.exporter, str) else "otlp_http"
return base.model_copy(
update={

View file

@ -44,7 +44,10 @@ def phoenix_preset(
)
return base.model_copy(
update={
"exporters": [*base.exporters, *global_exporter],
"exporters": [
*base.exporters,
*global_exporter,
], # mutable-ok: model_copy bypasses validation, so the field's declared list/dict type must be built as-is
"mapper_names": ensure_mappers(base.mapper_names, "openinference"),
"resource_attributes": {
**base.resource_attributes,

View file

@ -1,5 +1,7 @@
"""Weave (W&B) preset."""
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
from litellm.integrations.otel.model.config import (
@ -15,16 +17,18 @@ from litellm.integrations.weave.weave_otel import (
from litellm.types.utils import StandardCallbackDynamicParams
def weave_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]:
def weave_dynamic_headers(params: StandardCallbackDynamicParams) -> Mapping[str, str]:
"""Per-request Weave OTLP headers from team/key dynamic params."""
headers: Final[dict[str, str]] = {}
api_key: Final = params.get("wandb_api_key")
if api_key:
headers["Authorization"] = _get_weave_authorization_header(api_key=api_key)
project_id: Final = params.get("weave_project_id")
if project_id:
headers["project_id"] = project_id
return headers
return MappingProxyType(
dict(
(
*((("Authorization", _get_weave_authorization_header(api_key=api_key)),) if api_key else ()),
*((("project_id", project_id),) if project_id else ()),
)
)
)
def weave_preset(
@ -39,7 +43,7 @@ def weave_preset(
except Exception:
if not allow_missing_credentials:
raise
return base.model_copy(update={"mapper_names": mappers})
return base.model_copy(update=MappingProxyType({"mapper_names": mappers}))
return base.model_copy(
update={
"exporters": [

View file

@ -4036,7 +4036,12 @@ def _init_custom_logger_compatible_class(
config: Final = (
initial_config
if initial_config.exporters
else OpenTelemetryV2Config(**{**settings, "exporter": initial_config.exporter})
else OpenTelemetryV2Config(
**{
**settings,
"exporter": initial_config.exporter,
} # mutable-ok: keyword expansion needs a dict
)
)
otel_logger_v2: Final = OpenTelemetryV2(config=config)
_in_memory_loggers.append(otel_logger_v2)

View file

@ -13,6 +13,7 @@ import re
import secrets
from contextlib import suppress
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Any, Final, NamedTuple, Protocol, Union, cast
import fastapi
@ -1039,8 +1040,8 @@ async def _hoist_request_destinations(request: Request, user_api_key_dict: UserA
OtelDestination(
callback_name=item.get("callback_name"),
endpoint=item.get("endpoint", ""),
headers=item.get("headers") or {},
resource_attributes=item.get("resource_attributes") or {},
headers=item.get("headers") or MappingProxyType({}),
resource_attributes=item.get("resource_attributes") or MappingProxyType({}),
protocol=item.get("protocol"),
)
for item in destinations_raw

View file

@ -2,6 +2,7 @@
CRUD endpoints for storing reusable credentials.
"""
from types import MappingProxyType
from typing import Final
from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
@ -138,9 +139,9 @@ async def get_credentials(
"credential_values": _get_masked_values(credential.credential_values),
"credential_info": credential.credential_info,
**(
{"resolves_to_destination": destination_for_credential(credential) is not None}
MappingProxyType({"resolves_to_destination": destination_for_credential(credential) is not None})
if is_logging_credential(credential.credential_info)
else {}
else MappingProxyType({})
),
}
for credential in litellm.credential_list

View file

@ -743,6 +743,17 @@ async def _effective_org_id(user_api_key_dict: UserAPIKeyAuth) -> str | None:
return getattr(team_obj, "organization_id", None)
def _destination_params(backend: str, destination: "OtelDestination") -> "OtelDestinationParams":
params: Final[OtelDestinationParams] = {
"callback_name": backend,
"endpoint": destination.endpoint,
"headers": destination.headers,
"resource_attributes": destination.resource_attributes,
"protocol": destination.protocol,
}
return params
async def _resolve_logging_exporters(
user_api_key_dict: UserAPIKeyAuth,
) -> "tuple[tuple[OtelDestinationParams, ...], tuple[str, ...]]":
@ -795,26 +806,21 @@ async def _resolve_logging_exporters(
if access_grants(info.access, team_ids, org_ids)
if (result := destination_for_credential(credential)) is not None
)
deduped: Final = {
(
destination.endpoint,
tuple(sorted(destination.headers.items())),
tuple(sorted(destination.resource_attributes.items())),
): (
backend,
destination,
)
for backend, destination in built
}
destinations: Final[tuple[OtelDestinationParams, ...]] = tuple(
deduped: Final = MappingProxyType(
{
"callback_name": backend,
"endpoint": destination.endpoint,
"headers": destination.headers,
"resource_attributes": destination.resource_attributes,
"protocol": destination.protocol,
(
destination.endpoint,
tuple(sorted(destination.headers.items())),
tuple(sorted(destination.resource_attributes.items())),
): (
backend,
destination,
)
for backend, destination in built
}
for backend, destination in deduped.values()
)
destinations: Final[tuple[OtelDestinationParams, ...]] = tuple(
_destination_params(backend, destination) for backend, destination in deduped.values()
)
backends: Final = tuple(dict.fromkeys(backend for backend, _ in deduped.values()))
return destinations, backends

View file

@ -10,6 +10,7 @@ scope is the given set of team ids and org ids. The resolver passes a
one-element scope built with ``identity_scope``.
"""
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
from pydantic import BaseModel, ConfigDict, ValidationError
@ -127,15 +128,19 @@ def destination_for_credential(credential: CredentialItem) -> 'tuple[str, "OtelD
from litellm.integrations.otel.presets import PRESET_BY_CALLBACK
from litellm.integrations.otel.presets.destinations import build_destination
backend: Final = (credential.credential_info or {}).get("description")
backend: Final = (credential.credential_info or MappingProxyType({})).get("description")
if not backend:
return None
# Drop unset (``None``) values rather than stringifying them: ``str(None)`` is the
# literal ``"None"``, which would land in the exporter endpoint/headers and break
# the export (e.g. an empty ``otel_endpoint`` becoming the URL ``"None"``).
values: Final = {
str(key): str(value) for key, value in (credential.credential_values or {}).items() if value is not None
}
values: Final = MappingProxyType(
{
str(key): str(value)
for key, value in (credential.credential_values or MappingProxyType({})).items()
if value is not None
}
)
destination: Final = build_destination(backend, values)
if destination is None:
return None

View file

@ -24,23 +24,29 @@ def validate_credential_access(credential_info: Mapping[str, object] | None) ->
if not isinstance(access, dict):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": "credential_info.access must be an object"},
detail={
"error": "credential_info.access must be an object"
}, # mutable-ok: FastAPI serializes the detail dict
)
if "global" in access and not isinstance(access["global"], bool):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": "access.global must be a boolean"},
detail={"error": "access.global must be a boolean"}, # mutable-ok: FastAPI serializes the detail dict
)
for field in ("teams", "orgs"):
bucket = access.get(field)
if bucket is not None and not (isinstance(bucket, list) and all(isinstance(item, str) for item in bucket)):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": f"access.{field} must be a list of strings"},
detail={
"error": f"access.{field} must be a list of strings"
}, # mutable-ok: FastAPI serializes the detail dict
)
unknown: Final = set(access) - {"global", "teams", "orgs"}
unknown: Final = frozenset(access) - frozenset(("global", "teams", "orgs"))
if unknown:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": f"access contains unknown field(s): {sorted(unknown)}"},
detail={
"error": f"access contains unknown field(s): {sorted(unknown)}"
}, # mutable-ok: FastAPI serializes the detail dict
)

View file

@ -3,7 +3,7 @@
"limit": 22909
},
"LIT002": {
"limit": 26927
"limit": 26898
},
"LIT003": {
"limit": 269