scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 84

fix(otel/v2): address Greptile review (admin gate, LRU eviction, transport kind)

Browse source 
Three findings from the Greptile review on PR #30873:

1. Admin gate on credential update/delete consulted only the in-process
   litellm.credential_list. A logging credential resident only in the DB (created
   on another horizontally-scaled instance, or before a restart) bypassed the
   proxy-admin check, so a credential_values-only patch could redirect a logging
   destination's endpoint. The gate now falls back to the DB via
   _credential_for_admin_gate so it sees the authoritative credential_info.
   Adds two regression tests (update + delete of a DB-only logging credential by
   a non-admin -> 403), mutation-verified.

2. The fan-out LRU evicted a cached BatchSpanProcessor and called shutdown() on
   it synchronously, which can drop or raise on spans still queued for that
   processor's exporter thread (custom rule a961cf42). The evicted entry is now
   dropped without shutdown; the BatchSpanProcessor worker keeps draining on its
   own schedule and is reclaimed at process exit, and the cache stays bounded.

3. fan_out hardcoded _GRPC_BACKENDS={'arize'} separately from
   TenantTracerCache._owned_otlp_kind. Centralized the backend->transport default
   in providers.default_otlp_kind_for_backend, used by both, so the per-tenant
   destination transport and the owned-exporter fallback can no longer diverge.
This commit is contained in:
yucheng-berriai 2026-06-23 17:34:26 -07:00
parent 742fde2330
commit 9af9714ed3
5 changed files with 133 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
litellm/integrations/otel/plumbing/fan_out.py
View file

@ -122,11 +122,12 @@ class TenantFanOutSpanProcessor(SpanProcessor):
from litellm.integrations.otel.plumbing.providers import (
_exporter_from_spec,
_processor_for,
default_otlp_kind_for_backend,
)
try:
spec = ExporterSpec(
kind=_resolve_kind(destination),
kind=default_otlp_kind_for_backend(destination.callback_name),
endpoint=destination.endpoint,
headers=destination.header_string(),
owner=None,
@ -142,24 +143,17 @@ class TenantFanOutSpanProcessor(SpanProcessor):
return None
self._processors[key] = processor
if len(self._processors) > _MAX_CACHED_PROCESSORS:
_, evicted = self._processors.popitem(last=False)
try:
evicted.shutdown()
except Exception:
pass
# Evict the LRU entry but do NOT shut it down here: a
# ``BatchSpanProcessor`` may still hold spans queued on its exporter
# thread, and calling ``shutdown`` synchronously can drop or raise on
# those in-flight spans. Dropping the reference lets the worker drain
# naturally and be reclaimed at process exit. The cache is bounded at
# ``_MAX_CACHED_PROCESSORS``, so the un-shut-down working set stays
# bounded.
self._processors.popitem(last=False)
return processor
# Per-backend transport: Arize speaks OTLP/gRPC, every other current preset
# speaks OTLP/HTTP. Kept here so the fan-out picks the same transport the
# preset's own exporter uses, mirroring ``TenantTracerCache._owned_otlp_kind``.
_GRPC_BACKENDS = frozenset({"arize"})
def _resolve_kind(destination: "OtelDestination") -> str:
return "otlp_grpc" if destination.callback_name in _GRPC_BACKENDS else "otlp_http"
# Attribute set on every gen-AI LLM-call span by the v2 emitter. Used as the
# unambiguous skip signal: only the LLM-call span carries this, and the
# per-backend v2 logger already routes it to per-tenant destinations through

40
litellm/integrations/otel/plumbing/providers.py
View file

@ -53,9 +53,7 @@ def to_otel_span_kind(kind: LiteLLMSpanKind) -> SpanKind:
_EXPORTER_FACTORIES: dict[str, Callable[[ExporterSpec], SpanExporter]] = {}
def register_exporter_factory(
kind: str, factory: Callable[[ExporterSpec], SpanExporter]
) -> None:
def register_exporter_factory(kind: str, factory: Callable[[ExporterSpec], SpanExporter]) -> None:
"""Register a custom exporter ``factory`` for the exporter ``kind``."""
_EXPORTER_FACTORIES[kind.lower()] = factory
@ -72,9 +70,7 @@ class LiteLLMBaggageSpanProcessor(SpanProcessor):
self._allowed_prefixes = tuple(allowed_prefixes)
def _is_allowed(self, key: str) -> bool:
return key in self._allowed_keys or any(
key.startswith(prefix) for prefix in self._allowed_prefixes
)
return key in self._allowed_keys or any(key.startswith(prefix) for prefix in self._allowed_prefixes)
def on_start(self, span: Span, parent_context: Context | None = None) -> None:
for key, value in baggage.get_all(parent_context).items():
@ -112,6 +108,18 @@ def _otlp_traces_endpoint(endpoint: str | None) -> str | None:
return endpoint + "/v1/traces"
# Backends whose OTLP transport is gRPC. Arize's OTLP endpoint
# (``otlp.arize.com``) speaks gRPC; every other current preset speaks OTLP/HTTP.
# Single source of truth shared by the per-tenant fan-out processor and the
# ``TenantTracerCache`` so the two never disagree on a destination's transport.
_GRPC_BACKENDS = frozenset({"arize"})
def default_otlp_kind_for_backend(callback_name: "str | None") -> str:
"""The intrinsic OTLP transport for a backend's own OTLP endpoint."""
return "otlp_grpc" if callback_name in _GRPC_BACKENDS else "otlp_http"
def _exporter_from_spec(spec: ExporterSpec) -> SpanExporter:
kind = (spec.kind or "console").lower()
factory = _EXPORTER_FACTORIES.get(kind)
@ -156,11 +164,7 @@ def build_span_exporter(config: OpenTelemetryV2Config) -> SpanExporter:
``exporter`` / ``endpoint`` / ``headers`` fields. To configure multiple
exporters, populate ``config.exporters`` directly.
"""
return _exporter_from_spec(
ExporterSpec(
kind=config.exporter, endpoint=config.endpoint, headers=config.headers
)
)
return _exporter_from_spec(ExporterSpec(kind=config.exporter, endpoint=config.endpoint, headers=config.headers))
def _otlp_metrics_endpoint(endpoint: str | None) -> str | None:
@ -309,9 +313,7 @@ def build_tracer_provider(
"""
provider = TracerProvider(resource=build_resource(config))
if baggage_processor is None:
baggage_processor = LiteLLMBaggageSpanProcessor(
allowed_keys=config.baggage_promoted_keys
)
baggage_processor = LiteLLMBaggageSpanProcessor(allowed_keys=config.baggage_promoted_keys)
provider.add_span_processor(baggage_processor)
if tenant_fan_out_owner is not None:
@ -319,9 +321,7 @@ def build_tracer_provider(
TenantFanOutSpanProcessor,
)
provider.add_span_processor(
TenantFanOutSpanProcessor(owner_callback_name=tenant_fan_out_owner)
)
provider.add_span_processor(TenantFanOutSpanProcessor(owner_callback_name=tenant_fan_out_owner))
if exporter is not None:
provider.add_span_processor(_processor_for(exporter, use_simple_processor))
@ -334,11 +334,7 @@ def build_tracer_provider(
provider.add_span_processor(
_processor_for(
exp,
(
spec.use_simple_processor
if spec.use_simple_processor is not None
else use_simple_processor
),
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
)
)
return provider

14
litellm/integrations/otel/plumbing/routing.py
View file

@ -95,14 +95,24 @@ class TenantTracerCache:
def _owned_otlp_kind(self) -> str:
"""The OTLP transport of this integration's own exporter (langfuse -> http,
arize -> grpc), used for the destinations appended below."""
arize -> grpc), used for the destinations appended below.
Prefer the admin's configured exporter kind for this backend; fall back to
the backend's intrinsic default (shared with the fan-out processor via
``default_otlp_kind_for_backend``) so a lazily-activated backend with no
owned spec still picks the right transport (e.g. arize -> grpc, not http).
"""
from litellm.integrations.otel.plumbing.providers import (
default_otlp_kind_for_backend,
)
for spec in self._config.exporters:
if (
spec.owner == self._callback_name
and spec.kind.lower() not in _NON_OTLP_KINDS
):
return spec.kind
return "otlp_http"
return default_otlp_kind_for_backend(self._callback_name)
def _config_with_destinations(
self, destinations: "tuple[OtelDestination, ...]"

80
litellm/proxy/credential_endpoints/endpoints.py
View file

@ -34,15 +34,34 @@ def _require_proxy_admin(user_api_key_dict: UserAPIKeyAuth) -> None:
def _credential_in_memory(credential_name: str) -> Optional[CredentialItem]:
return next(
(
cred
for cred in litellm.credential_list
if cred.credential_name == credential_name
),
(cred for cred in litellm.credential_list if cred.credential_name == credential_name),
None,
)
async def _credential_for_admin_gate(credential_name: str, prisma_client: object) -> Optional[CredentialItem]:
"""Authoritative credential lookup for the admin gate on update/delete.
The in-process ``litellm.credential_list`` can be stale: a credential created
via the API on another horizontally-scaled instance, or before a restart,
exists only in the DB. Gating on the in-memory copy alone would let a logging
credential that isn't resident be updated/deleted without the proxy-admin
check. Prefer the in-memory copy, fall back to the DB so the gate sees the
real ``credential_info``.
"""
existing = _credential_in_memory(credential_name)
if existing is not None:
return existing
if prisma_client is None:
return None
try:
return await CredentialsRepository(
prisma_client # type: ignore[arg-type]
).find_by_name(credential_name)
except Exception:
return None
class CredentialHelperUtils:
@staticmethod
def encrypt_credential_values(
@ -51,9 +70,7 @@ class CredentialHelperUtils:
"""Encrypt values in credential.credential_values and add to DB"""
encrypted_credential_values = {}
for key, value in (credential.credential_values or {}).items():
encrypted_credential_values[key] = encrypt_value_helper(
value, new_encryption_key
)
encrypted_credential_values[key] = encrypt_value_helper(value, new_encryption_key)
# Return a new object to avoid mutating the caller's credential, which
# is kept in memory and should remain unencrypted.
@ -102,9 +119,7 @@ async def create_credential(
model = llm_router.get_deployment(credential.model_id)
if model is None:
raise HTTPException(status_code=404, detail="Model not found")
credential_values = llm_router.get_deployment_credentials(
credential.model_id
)
credential_values = llm_router.get_deployment_credentials(credential.model_id)
if credential_values is None:
raise HTTPException(status_code=404, detail="Model not found")
credential.credential_values = credential_values
@ -119,9 +134,7 @@ async def create_credential(
credential_values=credential.credential_values,
credential_info=credential.credential_info,
)
encrypted_credential = CredentialHelperUtils.encrypt_credential_values(
processed_credential
)
encrypted_credential = CredentialHelperUtils.encrypt_credential_values(processed_credential)
credentials_dict = encrypted_credential.model_dump()
credentials_dict_jsonified = jsonify_object(credentials_dict)
await CredentialsRepository(prisma_client).create(
@ -177,9 +190,7 @@ async def get_credentials(
async def get_credential_by_name(
request: Request,
fastapi_response: Response,
credential_name: str = Path(
..., description="The credential name, percent-decoded; may contain slashes"
),
credential_name: str = Path(..., description="The credential name, percent-decoded; may contain slashes"),
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
@ -257,9 +268,7 @@ async def get_credential_by_model(
async def delete_credential(
request: Request,
fastapi_response: Response,
credential_name: str = Path(
..., description="The credential name, percent-decoded; may contain slashes"
),
credential_name: str = Path(..., description="The credential name, percent-decoded; may contain slashes"),
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
@ -267,10 +276,8 @@ async def delete_credential(
"""
from litellm.proxy.proxy_server import prisma_client
existing = _credential_in_memory(credential_name)
if existing is not None and is_admin_gated_credential_info(
existing.credential_info
):
existing = await _credential_for_admin_gate(credential_name, prisma_client)
if existing is not None and is_admin_gated_credential_info(existing.credential_info):
_require_proxy_admin(user_api_key_dict)
try:
@ -282,11 +289,7 @@ async def delete_credential(
await CredentialsRepository(prisma_client).delete_by_name(credential_name)
## DELETE FROM LITELLM ##
litellm.credential_list = [
cred
for cred in litellm.credential_list
if cred.credential_name != credential_name
]
litellm.credential_list = [cred for cred in litellm.credential_list if cred.credential_name != credential_name]
return {"success": True, "message": "Credential deleted successfully"}
except Exception as e:
return handle_exception_on_proxy(e)
@ -317,9 +320,7 @@ def update_db_credential(
# update litellm params
if encrypted_credential.credential_values:
# Encrypt any sensitive values
encrypted_params = {
k: v for k, v in encrypted_credential.credential_values.items()
}
encrypted_params = {k: v for k, v in encrypted_credential.credential_values.items()}
merged_credential.credential_values.update(encrypted_params)
@ -344,9 +345,7 @@ async def update_credential(
request: Request,
fastapi_response: Response,
credential: CredentialItem,
credential_name: str = Path(
..., description="The credential name, percent-decoded; may contain slashes"
),
credential_name: str = Path(..., description="The credential name, percent-decoded; may contain slashes"),
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
@ -354,10 +353,9 @@ async def update_credential(
"""
from litellm.proxy.proxy_server import prisma_client
existing = _credential_in_memory(credential_name)
existing = await _credential_for_admin_gate(credential_name, prisma_client)
if is_admin_gated_credential_info(credential.credential_info) or (
existing is not None
and is_admin_gated_credential_info(existing.credential_info)
existing is not None and is_admin_gated_credential_info(existing.credential_info)
):
_require_proxy_admin(user_api_key_dict)
validate_credential_access(credential.credential_info)
@ -404,11 +402,7 @@ async def update_credential(
)
# Remove old entry if renamed, then use upsert_credentials to handle duplicates
if new_name != credential_name:
litellm.credential_list = [
c
for c in litellm.credential_list
if c.credential_name != credential_name
]
litellm.credential_list = [c for c in litellm.credential_list if c.credential_name != credential_name]
CredentialAccessor.upsert_credentials([updated_in_memory])
return {"success": True, "message": "Credential updated successfully"}

65
tests/test_litellm/proxy/credential_endpoints/test_endpoints.py
View file

@ -44,9 +44,7 @@ def _connected_db(monkeypatch):
repo.create = AsyncMock()
repo.delete_by_name = AsyncMock()
monkeypatch.setattr(endpoints, "CredentialsRepository", lambda _client: repo)
monkeypatch.setattr(
endpoints.CredentialAccessor, "upsert_credentials", lambda creds: None
)
monkeypatch.setattr(endpoints.CredentialAccessor, "upsert_credentials", lambda creds: None)
return repo
@ -117,9 +115,7 @@ async def test_update_logging_credential_forbidden_for_non_admin(_connected_db):
@pytest.mark.asyncio
async def test_update_existing_logging_credential_forbidden_even_without_logging_patch(
_connected_db, monkeypatch
):
async def test_update_existing_logging_credential_forbidden_even_without_logging_patch(_connected_db, monkeypatch):
"""A non-admin cannot edit a stored logging credential's values, even with a patch
that omits credential_info (the gate consults the in-memory credential too)."""
monkeypatch.setattr(
@ -180,9 +176,7 @@ def test_update_db_credential_preserves_existing_info_on_partial_patch():
@pytest.mark.asyncio
async def test_delete_logging_credential_forbidden_for_non_admin(
_connected_db, monkeypatch
):
async def test_delete_logging_credential_forbidden_for_non_admin(_connected_db, monkeypatch):
monkeypatch.setattr(
litellm,
"credential_list",
@ -203,3 +197,56 @@ async def test_delete_logging_credential_forbidden_for_non_admin(
)
assert exc.value.status_code == 403
_connected_db.delete_by_name.assert_not_awaited()
@pytest.mark.asyncio
async def test_update_db_only_logging_credential_forbidden_for_non_admin(_connected_db, monkeypatch):
"""A logging credential that exists ONLY in the DB (not resident in the
in-memory ``credential_list`` -- e.g. created on another scaled instance or
before a restart) must still gate a non-admin update. The gate falls back to
the DB so a credential_values-only patch can't redirect a logging
destination's endpoint without the proxy-admin check."""
monkeypatch.setattr(litellm, "credential_list", []) # nothing in memory
_connected_db.find_by_name = AsyncMock(
return_value=CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "h"},
credential_info=_LOGGING_INFO,
)
)
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "evil"},
credential_info={},
),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
@pytest.mark.asyncio
async def test_delete_db_only_logging_credential_forbidden_for_non_admin(_connected_db, monkeypatch):
"""Same DB-only fallback for delete: a non-admin can't delete a logging
credential that is resident only in the DB."""
monkeypatch.setattr(litellm, "credential_list", [])
_connected_db.find_by_name = AsyncMock(
return_value=CredentialItem(
credential_name="dest",
credential_values={},
credential_info=_LOGGING_INFO,
)
)
with pytest.raises(HTTPException) as exc:
await endpoints.delete_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.delete_by_name.assert_not_awaited()