diff --git a/litellm/integrations/otel/plumbing/fan_out.py b/litellm/integrations/otel/plumbing/fan_out.py index 91973ff8b7c..eaf12bc0790 100644 --- a/litellm/integrations/otel/plumbing/fan_out.py +++ b/litellm/integrations/otel/plumbing/fan_out.py @@ -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 diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index d03d28faa87..39601f9f161 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -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 diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index 5c5a245e450..4af06d2f91a 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -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, ...]" diff --git a/litellm/proxy/credential_endpoints/endpoints.py b/litellm/proxy/credential_endpoints/endpoints.py index 48fb5ed65fc..5b359ba2eba 100644 --- a/litellm/proxy/credential_endpoints/endpoints.py +++ b/litellm/proxy/credential_endpoints/endpoints.py @@ -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"} diff --git a/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py b/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py index 3296e1e514a..fab16eba42a 100644 --- a/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py +++ b/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py @@ -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()