From 231721f20affb1b48824b0197cd5b6b15c8ded84 Mon Sep 17 00:00:00 2001 From: mrinal-berri Date: Mon, 12 Oct 2026 19:25:52 -0400 Subject: [PATCH] feat(otel): let a team or key excluded_services override the global setting for its OTel v2 destination (#45880) * feat(otel): let a team or key excluded_services override the global setting for its OTel v2 destination Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * style: ruff format otel excluded_services changes Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test: use raw regex for excluded_services match Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): reject non-string destination callback vars Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): harden destination callback var validation Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): compare parsed values in callback var conflict check Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/integrations/otel/README.md | 11 +- litellm/integrations/otel/model/config.py | 23 ++- .../integrations/otel/model/destination.py | 11 +- .../integrations/otel/plumbing/providers.py | 7 +- .../integrations/otel/presets/destinations.py | 2 + .../initialize_dynamic_callback_params.py | 12 ++ litellm/proxy/_types.py | 12 +- .../callback_config_validation.py | 128 +++++++++--- litellm/proxy/litellm_pre_call_utils.py | 8 +- .../team_callback_endpoints.py | 28 ++- litellm/types/utils.py | 1 + .../test_otel_excluded_services.py | 113 ++++++++++- .../otel/test_otel_v2_destinations.py | 191 ++++++++++++++++++ .../test_callback_config_validation.py | 129 +++++++++++- .../test_team_callback_endpoints.py | 129 +++++++++++- tests/unit/types/test_utils.py | 7 + ui/litellm-dashboard/src/lib/http/schema.d.ts | 1 + 17 files changed, 755 insertions(+), 58 deletions(-) diff --git a/litellm/integrations/otel/README.md b/litellm/integrations/otel/README.md index 4b963ca3415..340bd71c983 100644 --- a/litellm/integrations/otel/README.md +++ b/litellm/integrations/otel/README.md @@ -311,10 +311,15 @@ nothing here imports outside it: `LITELLM_OTEL_EXCLUDED_SERVICES` (comma-separated) or `excluded_services` (a YAML list) under `callback_settings.otel`, naming the datastore services to withhold (`redis`, `postgres`, `batch_write_to_db`, `redis_*`, or their - `db.system.name` spellings `redis` / `postgresql`). Unknown names are logged - as an error and ignored. A span is withheld when its `db.system.name` / + `db.system.name` spellings `redis` / `postgresql`). Unknown global names are logged + as an error and ignored. Team and key callback values are validated when saved, + and invalid values are rejected. A span is withheld when its `db.system.name` / `db.system` attribute is in the set, so request root, auth, guardrail and - model spans can never be excluded. + model spans can never be excluded. Team and key destinations can replace the + global set with `excluded_services` in their own `callback_vars`; `"none"` + means no datastore spans are excluded, while an omitted value follows the + global set. These destination overrides require OTel v2 and are accepted only + on `langfuse_otel`, `arize`, `weave_otel` and `newrelic` callbacks. - [`baggage.py`](./model/baggage.py) — the single definition of which request-identity values are promoted into Baggage (so child spans inherit them) and under which attribute keys. diff --git a/litellm/integrations/otel/model/config.py b/litellm/integrations/otel/model/config.py index 55ef28bf470..cca12b885c1 100644 --- a/litellm/integrations/otel/model/config.py +++ b/litellm/integrations/otel/model/config.py @@ -408,6 +408,23 @@ def excluded_db_systems_from(value: object) -> frozenset[str]: return _normalize_excluded_services(excluded_service_names(value)) +def invalid_excluded_service_names(value: str) -> tuple[str, ...]: + names: Final = tuple(service.strip().lower() for service in value.split(",")) + return tuple( + name + for name in names + if name != "none" and (not name or _resolved_db_system_for_excluded_service(name) is None) + ) + + +def parse_excluded_services(value: object) -> frozenset[str] | None: + if not isinstance(value, str): + return None + if value.strip().lower() == "none": + return frozenset() + return excluded_db_systems_from(value) + + def parse_capture_message_content(value: object) -> str | None: return value if isinstance(value, str) and value in CAPTURE_MESSAGE_CONTENT_VALUES else None @@ -448,9 +465,13 @@ def _normalize_excluded_services(services: frozenset[str]) -> frozenset[str]: def _db_system_for_excluded_service(service: str) -> str | None: - resolved: Final = db_system(service) if service != POSTGRESQL else POSTGRESQL + resolved: Final = _resolved_db_system_for_excluded_service(service) if resolved is None: verbose_logger.error( "excluded_services: %r is not a datastore service; ignored. Allowed: postgres, redis", service ) return resolved + + +def _resolved_db_system_for_excluded_service(service: str) -> str | None: + return db_system(service) if service != POSTGRESQL else POSTGRESQL diff --git a/litellm/integrations/otel/model/destination.py b/litellm/integrations/otel/model/destination.py index b0c22dea292..4226b4361dd 100644 --- a/litellm/integrations/otel/model/destination.py +++ b/litellm/integrations/otel/model/destination.py @@ -57,6 +57,10 @@ class OtelDestination(LiteLLMBaseModel): "Omitted follows the global setting." ), ) + excluded_db_systems: frozenset[str] | None = Field( + default=None, + description="An explicit set replaces the global database-span exclusions; ``None`` follows the global set.", + ) def captures_content(self, default: str) -> bool: setting: Final = self.capture_message_content @@ -75,9 +79,10 @@ class OtelDestination(LiteLLMBaseModel): def cache_key(self) -> tuple[str, tuple[tuple[str, str], ...], tuple[tuple[str, str], ...], str | None]: """Identity for processor reuse, so one destination means one exporter. - ``span_scope``, the sampling rates and ``capture_message_content`` are left out on - purpose: they decide which spans reach the processor and what they carry, not how - the processor exports them, so two views of the same account share one exporter. + ``span_scope``, the sampling rates, ``capture_message_content`` and + ``excluded_db_systems`` are left out on purpose: they decide which spans reach + the processor and what they carry, not how the processor exports them, so two + views of the same account share one exporter. """ return ( self.endpoint, diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 4e7d4f231ed..898eab5715b 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -705,7 +705,12 @@ class TenantFanOutSpanProcessor(SpanProcessor): if ( self._operator_already_writes(span, destination, suppressed) or not _in_scope(span, destination.span_scope) - or _is_excluded_database_span(attributes, self._excluded_db_systems) + or _is_excluded_database_span( + attributes, + self._excluded_db_systems + if destination.excluded_db_systems is None + else destination.excluded_db_systems, + ) ): continue if not _is_sampled(destination): diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py index c4273fbb337..a705fc16bbc 100644 --- a/litellm/integrations/otel/presets/destinations.py +++ b/litellm/integrations/otel/presets/destinations.py @@ -188,6 +188,7 @@ def destination_for( params: StandardCallbackDynamicParams, service_name: str | None = None, capture_message_content: str | None = None, + excluded_db_systems: frozenset[str] | None = None, ) -> OtelDestination | None: """The destination ``params`` names for ``callback_name``, or ``None``. @@ -219,6 +220,7 @@ def destination_for( protocol=protocol, span_scope=_span_scope(callback_name, params), capture_message_content=capture_message_content, + excluded_db_systems=excluded_db_systems, success_sampling_rate=success_sampling_rate, error_sampling_rate=error_sampling_rate, ) diff --git a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py index 2efeead6bb2..5a533054f0b 100644 --- a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py +++ b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py @@ -99,6 +99,18 @@ def validate_capture_message_content_value(value: str) -> None: ) +def validate_excluded_services_value(value: str) -> None: + from litellm.integrations.otel.model.config import invalid_excluded_service_names + + names: Final = tuple(name.strip().lower() for name in value.split(",")) + invalid_names: Final = invalid_excluded_service_names(value) + allowed: Final = "redis, postgres, postgresql, batch_write_to_db, redis_-prefixed service names, or none" + if invalid_names or ("none" in names and len(names) != 1): + raise ValueError( + f"Invalid excluded_services {value!r}: use {allowed}; 'none' must be used alone and names cannot be empty" + ) + + # Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict _supported_callback_params: Final[tuple[str, ...]] = ( "langfuse_public_key", diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index b6c81f17761..2c75a608d0d 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -32,6 +32,7 @@ from litellm.constants import DEFAULT_STAGGER_WINDOW_SECONDS, MCP_STDIO_ALLOWED_ from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( validate_arize_otlp_protocol_value, validate_capture_message_content_value, + validate_excluded_services_value, validate_langfuse_environment_value, validate_langfuse_span_scope_value, validate_no_callback_env_reference, @@ -90,6 +91,7 @@ from litellm.types.router_weights import validate_router_settings_dict from litellm.types.secret_managers.main import KeyManagementSystem from litellm.types.utils import ( CAPTURE_MESSAGE_CONTENT_VAR, + EXCLUDED_SERVICES_VAR, AzureSpillover, CallTypes, CostBreakdown, @@ -2046,10 +2048,16 @@ class AddTeamCallback(LiteLLMPydanticObjectBase): @classmethod def validate_callback_vars(cls, values): callback_vars: Final = values.get("callback_vars", {}) - valid_keys: Final = {*StandardCallbackDynamicParams.__annotations__, CAPTURE_MESSAGE_CONTENT_VAR} + valid_keys: Final = { + *StandardCallbackDynamicParams.__annotations__, + CAPTURE_MESSAGE_CONTENT_VAR, + EXCLUDED_SERVICES_VAR, + } for key, value in callback_vars.items(): if key not in valid_keys: raise ValueError(f"Invalid callback variable: {key}. Must be one of {valid_keys}") + if key in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) and not isinstance(value, str): + raise ValueError(f"{key} must be a string, got {type(value).__name__}") callback_vars[key] = str(value) validate_no_callback_env_reference(key, callback_vars[key], source="key/team callback metadata") if key == "langfuse_environment": @@ -2060,6 +2068,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase): validate_arize_otlp_protocol_value(callback_vars[key]) if key == CAPTURE_MESSAGE_CONTENT_VAR: validate_capture_message_content_value(callback_vars[key]) + if key == EXCLUDED_SERVICES_VAR: + validate_excluded_services_value(callback_vars[key]) return values diff --git a/litellm/proxy/common_utils/callback_config_validation.py b/litellm/proxy/common_utils/callback_config_validation.py index 4956da853c6..302978d17a4 100644 --- a/litellm/proxy/common_utils/callback_config_validation.py +++ b/litellm/proxy/common_utils/callback_config_validation.py @@ -6,11 +6,11 @@ per-integration checks here. """ import math -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from types import MappingProxyType -from typing import Final +from typing import Final, cast # noqa: TID251 # preserve arbitrary keys from validated callback mappings -from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VAR +from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR _NEWRELIC_CALLBACK: Final = "newrelic" _NEWRELIC_VAR_PREFIX: Final = "newrelic_" @@ -38,9 +38,29 @@ def callback_config_error( ) if langfuse_error is not None: return langfuse_error - capture_error: Final = _capture_message_content_error(callback_name, callback_vars, callback_type) - if capture_error is not None: - return capture_error + from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( + validate_capture_message_content_value, + validate_excluded_services_value, + ) + + destination_setting_error: Final = next( + ( + error + for variable_name, validator in ( + (CAPTURE_MESSAGE_CONTENT_VAR, validate_capture_message_content_value), + (EXCLUDED_SERVICES_VAR, validate_excluded_services_value), + ) + if ( + error := _destination_setting_error( + variable_name, callback_name, callback_vars, callback_type, validator + ) + ) + is not None + ), + None, + ) + if destination_setting_error is not None: + return destination_setting_error if callback_name != _NEWRELIC_CALLBACK: return None return _newrelic_config_error(callback_vars) @@ -85,29 +105,30 @@ def _langfuse_span_scope_error(callback_name: str | None, callback_vars: Mapping return None -def _capture_message_content_error( - callback_name: str | None, callback_vars: Mapping[str, str], callback_type: str | None +def _destination_setting_error( + variable_name: str, + callback_name: str | None, + callback_vars: Mapping[str, str], + callback_type: str | None, + validator: Callable[[str], None], ) -> str | None: - value: Final = callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR) + value: Final = callback_vars.get(variable_name) if value is None: return None from litellm.integrations.otel.model.config import is_otel_v2_enabled from litellm.integrations.otel.presets.destinations import destination_capable_backends - from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( - validate_capture_message_content_value, - ) supported: Final = sorted(destination_capable_backends()) if callback_name not in supported: - return f"{CAPTURE_MESSAGE_CONTENT_VAR} applies to the OTel v2 callbacks {supported} only, not {callback_name!r}" + return f"{variable_name} applies to the OTel v2 callbacks {supported} only, not {callback_name!r}" if not is_otel_v2_enabled(): - return f"Per-destination {CAPTURE_MESSAGE_CONTENT_VAR} requires the proxy to run with LITELLM_OTEL_V2=true." + return f"Per-destination {variable_name} requires the proxy to run with LITELLM_OTEL_V2=true." try: - validate_capture_message_content_value(value) + validator(value) except ValueError as e: return str(e) if callback_type == "failure": - return f"{CAPTURE_MESSAGE_CONTENT_VAR} needs callback_type 'success' or 'success_and_failure'" + return f"{variable_name} needs callback_type 'success' or 'success_and_failure'" return None @@ -221,25 +242,36 @@ def conflicting_span_scope_error( ) -def conflicting_capture_error( +def conflicting_callback_var_error( callback_name: str, + variable_name: str, callback_vars: Mapping[str, str] | None, stored_entries: Sequence[tuple[str | None, Mapping[str, str]]], ) -> str | None: - incoming: Final = None if callback_vars is None else callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR) + incoming: Final = None if callback_vars is None else callback_vars.get(variable_name) if incoming is None: return None return next( ( - f"{CAPTURE_MESSAGE_CONTENT_VAR} is already set to {stored!r} by another {callback_name} entry. " + f"{variable_name} is already set to {stored!r} by another {callback_name} entry. " f"Every {callback_name} entry shares one value: remove that entry or send the same value." for name, entry_vars in stored_entries - if name == callback_name and (stored := entry_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)) not in (None, incoming) + if name == callback_name + and (stored := entry_vars.get(variable_name)) is not None + and not _same_callback_var_value(variable_name, stored, incoming) ), None, ) +def _same_callback_var_value(variable_name: str, first: str, second: str) -> bool: + if variable_name != EXCLUDED_SERVICES_VAR: + return first == second + from litellm.integrations.otel.model.config import parse_excluded_services + + return parse_excluded_services(first) == parse_excluded_services(second) + + def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None: """Validate every ``logging`` entry of a team/key metadata payload.""" if not metadata: @@ -247,15 +279,20 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str entries: Final = _object_sequence(metadata.get("logging")) if entries is None: return None + entry_error: Final = next( + (error for entry in entries if (error := _logging_entry_error(entry)) is not None), + None, + ) + if entry_error is not None: + return entry_error entry_vars: Final = tuple(_entry_callback_vars(entry) for entry in entries) named_vars: Final = tuple(zip((_entry_callback_name(entry) for entry in entries), entry_vars)) return next( ( error for error in ( - *(_logging_entry_error(entry) for entry in entries), *(conflicting_span_scope_error(entry_vars[i], entry_vars[:i]) for i in range(len(entry_vars))), - *(_conflicting_entry_capture_error(named_vars[i], named_vars[:i]) for i in range(len(named_vars))), + *(_conflicting_entry_setting_error(named_vars[i], named_vars[:i]) for i in range(len(named_vars))), ) if error is not None ), @@ -263,8 +300,8 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str ) -def stored_capture_entries(logging_entries: object) -> tuple[tuple[str | None, Mapping[str, str]], ...]: - """The callback name and vars of every stored ``logging`` entry, for ``conflicting_capture_error``.""" +def stored_callback_entries(logging_entries: object) -> tuple[tuple[str | None, Mapping[str, str]], ...]: + """The callback name and vars of every stored ``logging`` entry.""" entries: Final = _object_sequence(logging_entries) or () return tuple((_entry_callback_name(entry), _entry_callback_vars(entry)) for entry in entries) @@ -281,11 +318,21 @@ def _entry_callback_name(entry: object) -> str | None: return None -def _conflicting_entry_capture_error( +def _conflicting_entry_setting_error( entry: tuple[str | None, Mapping[str, str]], earlier: Sequence[tuple[str | None, Mapping[str, str]]] ) -> str | None: callback_name, callback_vars = entry - return None if callback_name is None else conflicting_capture_error(callback_name, callback_vars, earlier) + if callback_name is None: + return None + return next( + ( + error + for variable_name in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) + if (error := conflicting_callback_var_error(callback_name, variable_name, callback_vars, earlier)) + is not None + ), + None, + ) def _entry_callback_vars(entry: object) -> Mapping[str, str]: @@ -298,12 +345,33 @@ def _entry_callback_vars(entry: object) -> Mapping[str, str]: def _logging_entry_error(entry: object) -> str | None: if not isinstance(entry, Mapping): return None - callback_name: Final = entry.get("callback_name") - if not isinstance(callback_name, str) or not isinstance(entry.get("callback_vars"), Mapping): + typed_entry: Final[Mapping[object, object]] = ( + cast( # cast-ok: entry is Mapping-checked but may have non-string keys + Mapping[object, object], entry + ) + ) + callback_name: Final = typed_entry.get("callback_name") + callback_vars_raw: Final = typed_entry.get("callback_vars") + if not isinstance(callback_name, str) or not isinstance(callback_vars_raw, Mapping): return None - callback_type_raw: Final = entry.get("callback_type") + callback_vars: Final[Mapping[object, object]] = cast( # cast-ok: Mapping check preserves arbitrary key types + Mapping[object, object], callback_vars_raw + ) + destination_var_error: Final = next( + ( + f"{variable_name} must be a string, got {type(callback_vars[variable_name]).__name__}" + for variable_name in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) + if variable_name in callback_vars and not isinstance(callback_vars[variable_name], str) + ), + None, + ) + if destination_var_error is not None: + return destination_var_error + callback_type_raw: Final = typed_entry.get("callback_type") return callback_config_error( - callback_name, _entry_callback_vars(entry), callback_type_raw if isinstance(callback_type_raw, str) else None + callback_name, + _entry_callback_vars(typed_entry), + callback_type_raw if isinstance(callback_type_raw, str) else None, ) diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index fdc2e6d94b8..ba319be01cb 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -228,6 +228,7 @@ from litellm.types.llms.anthropic import ANTHROPIC_API_HEADERS from litellm.types.services import ServiceTypes from litellm.types.utils import ( CAPTURE_MESSAGE_CONTENT_VAR, + EXCLUDED_SERVICES_VAR, CustomPricingLiteLLMParams, LlmProviders, ProviderSpecificHeader, @@ -236,6 +237,8 @@ from litellm.types.utils import ( SupportedCacheControls, ) +_DESTINATION_ONLY_CALLBACK_VARS: Final[frozenset[str]] = frozenset({CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR}) + _CALLBACK_CREDENTIAL_KEYS: Final = frozenset(StandardCallbackDynamicParams.__annotations__) | frozenset( {TRUSTED_CALLBACK_VARS_FIELD} ) @@ -976,7 +979,7 @@ def convert_key_logging_metadata_to_callback( team_callback_settings_obj.callbacks.append(data.callback_name) for var, value in data.callback_vars.items(): - if var == CAPTURE_MESSAGE_CONTENT_VAR: + if var in _DESTINATION_ONLY_CALLBACK_VARS: continue # New Relic routing reads these from the trusted-vars overlay with no # callback-name check, so scope them to the newrelic entry: a team that @@ -1195,7 +1198,7 @@ def resolve_tenant_otel_destinations( def _tenant_destination( name: str, callbacks: Sequence[AddTeamCallback], service_name: str | None ) -> "OtelDestination | None": - from litellm.integrations.otel.model.config import parse_capture_message_content + from litellm.integrations.otel.model.config import parse_capture_message_content, parse_excluded_services from litellm.integrations.otel.presets.destinations import destination_for callback_vars: Final = MappingProxyType( @@ -1210,6 +1213,7 @@ def _tenant_destination( _tenant_otel_params(callback_vars), service_name, parse_capture_message_content(callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)), + parse_excluded_services(callback_vars.get(EXCLUDED_SERVICES_VAR)), ) diff --git a/litellm/proxy/management_endpoints/team_callback_endpoints.py b/litellm/proxy/management_endpoints/team_callback_endpoints.py index 3908ab3c927..c2d7033c06c 100644 --- a/litellm/proxy/management_endpoints/team_callback_endpoints.py +++ b/litellm/proxy/management_endpoints/team_callback_endpoints.py @@ -31,10 +31,10 @@ from litellm.proxy._types import ( from litellm.proxy.auth.user_api_key_auth import user_api_key_auth from litellm.proxy.common_utils.callback_config_validation import ( callback_config_error, - conflicting_capture_error, + conflicting_callback_var_error, conflicting_span_scope_error, cross_entry_family_error, - stored_capture_entries, + stored_callback_entries, ) from litellm.proxy.common_utils.callback_utils import ( # noqa: F401 # legacy module exports _CALLBACK_VAR_ENCRYPTED_PREFIX, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export @@ -56,6 +56,7 @@ from litellm.proxy.management_endpoints.team_endpoints import ( # noqa: F401 # ) from litellm.proxy.management_helpers.utils import management_endpoint_wrapper from litellm.repositories.team_repository import TeamRepository +from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR router: Final = APIRouter() @@ -293,6 +294,7 @@ async def add_team_callbacks( - capture_message_content: For OTel v2 callbacks, "no_content" or "span_only". An explicit value overrides the global setting for this destination; omitted follows the proxy's own callback for that backend, else the global setting. "span_only" puts prompt and response content on the destination's spans, "no_content" leaves it out + - excluded_services: For OTel v2 callbacks, a comma-separated list of datastore names such as "redis,postgres", or "none" to exclude nothing. An explicit value replaces the global exclusions; omitted follows them - gcs_bucket_name: The name of the GCS bucket - gcs_path_service_account: The path to the GCS service account - langsmith_api_key: The API key for the Langsmith callback @@ -356,19 +358,27 @@ async def add_team_callbacks( # Decrypted, because the checks compare the incoming values against # the stored ones and the credentials are encrypted at rest. decrypted_logging: Final = decrypt_callback_vars(team_metadata).get("logging") - stored_capture: Final = stored_capture_entries(decrypted_logging) + stored_callbacks: Final = stored_callback_entries(decrypted_logging) stored_entries: Final = decrypted_logging if isinstance(decrypted_logging, list) else () stored_entry_vars: Final = [entry.get("callback_vars") or {} for entry in stored_entries] scope_error: Final = conflicting_span_scope_error(data.callback_vars, stored_entry_vars) if scope_error is not None: raise _callback_config_error(scope_error) - capture_error: Final = conflicting_capture_error( - data.callback_name, - data.callback_vars, - stored_capture, + destination_setting_error: Final = next( + ( + error + for variable_name in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) + if ( + error := conflicting_callback_var_error( + data.callback_name, variable_name, data.callback_vars, stored_callbacks + ) + ) + is not None + ), + None, ) - if capture_error is not None: - raise _callback_config_error(capture_error) + if destination_setting_error is not None: + raise _callback_config_error(destination_setting_error) # One entry has to own a credential family end to end. The entries are # flattened into one dict before a request reads them, so an entry # naming only a destination would pair with a key written on another diff --git a/litellm/types/utils.py b/litellm/types/utils.py index e6103270ee3..06af506076a 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3737,6 +3737,7 @@ ARIZE_OTLP_PROTOCOLS: Final[frozenset[str]] = frozenset(get_args(ArizeOtlpProtoc CAPTURE_MESSAGE_CONTENT_VAR: Final = "capture_message_content" +EXCLUDED_SERVICES_VAR: Final = "excluded_services" class CaptureMessageContent(str): diff --git a/tests/integration/observability/test_otel_excluded_services.py b/tests/integration/observability/test_otel_excluded_services.py index d62537e2317..a90d1836d24 100644 --- a/tests/integration/observability/test_otel_excluded_services.py +++ b/tests/integration/observability/test_otel_excluded_services.py @@ -19,10 +19,12 @@ from integration._support.client import ( ) from integration._support.otlp_sink import ( AuditConfigWriter, + ConnectSink, Span, SpanSinks, audit_configuration_writer, langfuse_callback_variables, + owned_connect_sink, owned_sinks, recorded_spans, spans_for_trace, @@ -88,19 +90,43 @@ def _operator_langfuse(audit_sinks: SpanSinks) -> dict[str, str]: } -def _add_callback(gateway: Gateway, team_id: str, callback_vars: Mapping[str, JsonValue]) -> httpx.Response: +def _newrelic_environment(audit_sinks: SpanSinks, newrelic_sink: ConnectSink) -> dict[str, str]: + return { + "LITELLM_OTEL_V2": "1", + **_operator_langfuse(audit_sinks), + "HTTPS_PROXY": newrelic_sink.proxy_url, + "NO_PROXY": "127.0.0.1,localhost", + "REQUESTS_CA_BUNDLE": newrelic_sink.ca_pem, + } + + +def _newrelic_vars(excluded_services: str | None = None) -> dict[str, JsonValue]: + variables: Final[dict[str, JsonValue]] = {"newrelic_api_key": "nr-team-excluded-services"} + return variables if excluded_services is None else {**variables, "excluded_services": excluded_services} + + +def _add_callback( + gateway: Gateway, + team_id: str, + callback_vars: Mapping[str, JsonValue], + callback_name: str = "langfuse_otel", +) -> httpx.Response: return gateway.request( "POST", f"/team/{team_id}/callback", - {"callback_name": "langfuse_otel", "callback_vars": dict(callback_vars)}, + {"callback_name": callback_name, "callback_vars": dict(callback_vars)}, ) -def _drive(candidate: Gateway, langfuse_vars: Mapping[str, JsonValue]) -> httpx.Response: +def _drive( + candidate: Gateway, + callback_vars: Mapping[str, JsonValue], + callback_name: str = "langfuse_otel", +) -> httpx.Response: with candidate.scenario() as scenario: model: Final = scenario.model(model="openai/audit-chat", api_base=f"{candidate.upstream_url}/v1") team_id: Final = scenario.team() - callback: Final = _add_callback(candidate, team_id, langfuse_vars) + callback: Final = _add_callback(candidate, team_id, callback_vars, callback_name) assert callback.status_code == 200, callback.text key: Final = scenario.key(team_id=team_id) response: Final = candidate.request( @@ -299,6 +325,85 @@ def test_without_excluded_services_the_tenant_still_gets_redis_and_postgres_span assert {"redis", "postgresql"} <= systems, f"datastore spans missing at tenant: {systems}" +@pytest.mark.timeout(180) +def test_team_excluded_services_replaces_global_and_keeps_operator_db_spans( + gateway: Gateway, + audit_sinks: SpanSinks, + otel_audit_config: AuditConfigWriter, + tmp_path: Path, +) -> None: + config: Final = _config_with(tmp_path, otel_audit_config, extra=_guardrail_block) + with owned_connect_sink(tmp_path / "newrelic") as newrelic_sink: + overrides: Final = _newrelic_environment(audit_sinks, newrelic_sink) + with owned_proxy(gateway, tmp_path, overrides, config=config, workers=2) as candidate: + operator_start, _ = recorded_spans(audit_sinks.operator) + tenant_start, _ = recorded_spans(newrelic_sink.control_url) + traffic: Final = _drive(candidate, _newrelic_vars("redis,postgres"), callback_name="newrelic") + _await_db_span( + audit_sinks.operator, + None, + "postgres.update LiteLLM_VerificationToken", + seconds=60, + since=operator_start, + ) + operator_trace: Final = _trace_id(audit_sinks.operator, traffic) + tenant_trace: Final = _trace_id(newrelic_sink.control_url, traffic) + assert operator_trace == tenant_trace + operator_trace_spans: Final = _trace_spans(audit_sinks.operator, operator_trace, seconds=15) + assert "redis" in _db_systems(operator_trace_spans), _db_systems(operator_trace_spans) + _, operator_spans = recorded_spans(audit_sinks.operator, operator_start) + assert "postgresql" in _db_systems(operator_spans), _db_systems(operator_spans) + tenant_trace_spans: Final = _trace_spans(newrelic_sink.control_url, tenant_trace, seconds=15) + _, tenant_spans = recorded_spans(newrelic_sink.control_url, tenant_start) + assert _db_systems(tenant_trace_spans) == set(), _db_systems(tenant_trace_spans) + assert _db_systems(tenant_spans) == set(), _db_systems(tenant_spans) + + +@pytest.mark.timeout(180) +def test_team_excluded_services_none_overrides_global( + gateway: Gateway, + audit_sinks: SpanSinks, + otel_audit_config: AuditConfigWriter, + tmp_path: Path, +) -> None: + config: Final = _config_with( + tmp_path, otel_audit_config, otel={"excluded_services": ["redis", "postgres"]}, extra=_guardrail_block + ) + with owned_connect_sink(tmp_path / "newrelic") as newrelic_sink: + overrides: Final = _newrelic_environment(audit_sinks, newrelic_sink) + with owned_proxy(gateway, tmp_path, overrides, config=config, workers=2) as candidate: + tenant_start, _ = recorded_spans(newrelic_sink.control_url) + traffic: Final = _drive(candidate, _newrelic_vars("none"), callback_name="newrelic") + tenant_trace: Final = _trace_id(newrelic_sink.control_url, traffic) + _await_db_span(newrelic_sink.control_url, tenant_trace, "redis", seconds=60) + _await_db_span(newrelic_sink.control_url, None, "postgresql", seconds=60, since=tenant_start) + tenant_trace_spans: Final = _trace_spans(newrelic_sink.control_url, tenant_trace, seconds=15) + _, tenant_spans = recorded_spans(newrelic_sink.control_url, tenant_start) + assert {"redis", "postgresql"} <= _db_systems(tenant_spans), _db_systems(tenant_spans) + assert {"redis", "postgresql"} <= _db_systems(tenant_trace_spans), _db_systems(tenant_trace_spans) + + +@pytest.mark.timeout(180) +def test_team_excluded_services_unset_inherits_global( + gateway: Gateway, + audit_sinks: SpanSinks, + otel_audit_config: AuditConfigWriter, + tmp_path: Path, +) -> None: + config: Final = _config_with( + tmp_path, otel_audit_config, otel={"excluded_services": ["redis", "postgres"]}, extra=_guardrail_block + ) + with owned_connect_sink(tmp_path / "newrelic") as newrelic_sink: + overrides: Final = _newrelic_environment(audit_sinks, newrelic_sink) + with owned_proxy(gateway, tmp_path, overrides, config=config, workers=2) as candidate: + tenant_start, _ = recorded_spans(newrelic_sink.control_url) + traffic: Final = _drive(candidate, _newrelic_vars(), callback_name="newrelic") + tenant_trace: Final = _trace_id(newrelic_sink.control_url, traffic) + _trace_spans(newrelic_sink.control_url, tenant_trace, seconds=15) + _, tenant_spans = recorded_spans(newrelic_sink.control_url, tenant_start) + assert _db_systems(tenant_spans) == set(), _db_systems(tenant_spans) + + @pytest.mark.parametrize("otel", [None, True, "on", "", []], ids=["null", "true", "on", "empty_string", "empty_list"]) @pytest.mark.timeout(180) def test_a_non_mapping_otel_block_still_publishes_the_tenant_fan_out( diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 5c9cab88784..068b2aa2e2e 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -36,6 +36,7 @@ from litellm.integrations.otel.model.config import ( ExporterSpec, OpenTelemetryV2Config, is_otel_v2_enabled, + parse_excluded_services, ) from litellm.integrations.otel.model.destination import OtelDestination from litellm.integrations.otel.model.payloads import ( @@ -580,6 +581,101 @@ class TestFanOut: "chat gpt-4", } + def test_team_exclusions_replace_empty_global_and_operator_keeps_all_database_spans(self): + team, inherited, operator = InMemorySpanExporter(), InMemorySpanExporter(), InMemorySpanExporter() + exporters: Final = {"http://team.local": team, "http://inherited.local": inherited} + provider: Final = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(operator)) + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda destination: SimpleSpanProcessor(exporters[destination.endpoint]) + ) + ) + tracer: Final = get_tracer(provider, "litellm") + + def run(): + set_request_destinations( + ( + OtelDestination( + endpoint="http://team.local", excluded_db_systems=frozenset({"redis", "postgresql"}) + ), + OtelDestination(endpoint="http://inherited.local"), + ) + ) + with tracer.start_as_current_span("redis read") as span: + span.set_attribute("db.system.name", "redis") + with tracer.start_as_current_span("postgres write") as span: + span.set_attribute("db.system", "postgresql") + with tracer.start_as_current_span("chat gpt-4"): + pass + + in_fresh_context(run) + + assert {span.name for span in team.get_finished_spans()} == {"chat gpt-4"} + assert {span.name for span in inherited.get_finished_spans()} == { + "redis read", + "postgres write", + "chat gpt-4", + } + assert {span.name for span in operator.get_finished_spans()} == { + "redis read", + "postgres write", + "chat gpt-4", + } + + def test_none_and_partial_team_exclusions_replace_the_global_set(self): + none, redis_only, inherited, operator = (InMemorySpanExporter() for _ in range(4)) + exporters: Final = { + "http://none.local": none, + "http://redis.local": redis_only, + "http://inherited.local": inherited, + } + provider: Final = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(operator)) + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda destination: SimpleSpanProcessor(exporters[destination.endpoint]), + excluded_db_systems=frozenset({"redis", "postgresql"}), + ) + ) + tracer: Final = get_tracer(provider, "litellm") + + def run(): + set_request_destinations( + ( + OtelDestination(endpoint="http://none.local", excluded_db_systems=frozenset()), + OtelDestination(endpoint="http://redis.local", excluded_db_systems=frozenset({"redis"})), + OtelDestination(endpoint="http://inherited.local"), + ) + ) + with tracer.start_as_current_span("redis read") as span: + span.set_attribute("db.system.name", "redis") + with tracer.start_as_current_span("postgres write") as span: + span.set_attribute("db.system", "postgresql") + with tracer.start_as_current_span("chat gpt-4"): + pass + + in_fresh_context(run) + + assert {span.name for span in none.get_finished_spans()} == { + "redis read", + "postgres write", + "chat gpt-4", + } + assert {span.name for span in redis_only.get_finished_spans()} == {"postgres write", "chat gpt-4"} + assert {span.name for span in inherited.get_finished_spans()} == {"chat gpt-4"} + assert {span.name for span in operator.get_finished_spans()} == { + "redis read", + "postgres write", + "chat gpt-4", + } + + def test_excluded_system_policy_does_not_change_exporter_cache_identity(self): + inherited: Final = OtelDestination(endpoint="http://same.local") + overridden: Final = OtelDestination(endpoint="http://same.local", excluded_db_systems=frozenset()) + + assert inherited.cache_key() == overridden.cache_key() + def test_a_team_naming_two_backends_gets_the_trace_at_both(self): """The fan-out rides one provider, so it cannot skip a destination on the grounds that some other backend owns it: nothing else would deliver it.""" @@ -4727,6 +4823,83 @@ class TestCaptureMessageContent: assert destinations["langfuse_otel"].capture_message_content == setting assert destinations["arize"].capture_message_content is None, "the setting stays on its own callback" + @pytest.mark.usefixtures("allow_test_hosts") + @pytest.mark.parametrize( + ("setting", "expected"), + [ + ("redis,postgres", frozenset({"redis", "postgresql"})), + ("none", frozenset()), + (None, None), + ], + ) + def test_excluded_services_stays_on_its_destination_and_normalizes_aliases( + self, monkeypatch: pytest.MonkeyPatch, setting: str | None, expected: frozenset[str] | None + ) -> None: + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + is_otel_v2_enabled.cache_clear() + callback_vars: Final = { + "langfuse_public_key": "pk-team", + "langfuse_secret_key": "sk-team", + "langfuse_host": "http://team.local", + **({"excluded_services": setting} if setting is not None else {}), + } + auth: Final = UserAPIKeyAuth( + team_metadata={ + "logging": [ + {"callback_name": "langfuse_otel", "callback_type": "success", "callback_vars": callback_vars}, + { + "callback_name": "arize", + "callback_type": "success", + "callback_vars": {"arize_api_key": "k", "arize_space_id": "s"}, + }, + ] + } + ) + + destinations: Final = {d.callback_name: d for d in resolve_tenant_otel_destinations(auth)} + + assert destinations["langfuse_otel"].excluded_db_systems == expected + assert destinations["arize"].excluded_db_systems is None + + @pytest.mark.usefixtures("allow_test_hosts") + def test_key_excluded_services_is_parsed_onto_its_destination(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + is_otel_v2_enabled.cache_clear() + auth: Final = UserAPIKeyAuth( + metadata={ + "logging": [ + { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": { + "langfuse_public_key": "pk-key", + "langfuse_secret_key": "sk-key", + "langfuse_host": "http://key.local", + "excluded_services": "redis,postgres", + }, + } + ] + } + ) + + (destination,) = resolve_tenant_otel_destinations(auth) + + assert destination.excluded_db_systems == frozenset({"redis", "postgresql"}) + + @pytest.mark.parametrize( + ("value", "expected"), + [ + (None, None), + (17, None), + ("none", frozenset()), + ("redis,postgres", frozenset({"redis", "postgresql"})), + ], + ) + def test_parse_excluded_services_returns_only_valid_string_overrides( + self, value: object, expected: frozenset[str] | None + ) -> None: + assert parse_excluded_services(value) == expected + @pytest.mark.usefixtures("allow_test_hosts") @pytest.mark.parametrize("setting", ["event_only", "span_and_event"]) def test_a_stored_event_mode_is_skipped_like_any_invalid_entry( @@ -4805,6 +4978,24 @@ class TestCaptureMessageContent: assert flattened.callback_vars == {"langfuse_public_key": "pk"} + def test_a_team_wide_callback_vars_map_cannot_carry_excluded_services(self) -> None: + with pytest.raises(ValueError, match="Invalid callback variable: excluded_services"): + TeamCallbackMetadata( + success_callback=["langfuse_otel", "signoz"], callback_vars={"excluded_services": "redis"} + ) + + def test_flattening_excluded_services_leaves_it_on_its_own_entry(self) -> None: + flattened: Final = convert_key_logging_metadata_to_callback( + AddTeamCallback( + callback_name="langfuse_otel", + callback_type="success", + callback_vars={"langfuse_public_key": "pk", "excluded_services": "redis"}, + ), + None, + ) + + assert flattened.callback_vars == {"langfuse_public_key": "pk"} + class TestArizeProjectRouting: """Arize rejects an export that names no project (400: set x-project-name, diff --git a/tests/unit/proxy/common_utils/test_callback_config_validation.py b/tests/unit/proxy/common_utils/test_callback_config_validation.py index a6544e56c57..f42387d3e76 100644 --- a/tests/unit/proxy/common_utils/test_callback_config_validation.py +++ b/tests/unit/proxy/common_utils/test_callback_config_validation.py @@ -5,7 +5,7 @@ import pytest from litellm.proxy.common_utils.callback_config_validation import ( callback_config_error, - conflicting_capture_error, + conflicting_callback_var_error, conflicting_span_scope_error, cross_entry_family_error, logging_metadata_config_error, @@ -282,6 +282,13 @@ def test_capture_message_content_is_accepted_on_every_otel_v2_destination(callba assert callback_config_error(callback_name, {"capture_message_content": value}) is None +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.parametrize("callback_name", ["langfuse_otel", "arize", "weave_otel", "newrelic"]) +@pytest.mark.parametrize("value", ["redis,postgres", "none", " REDIS , postgresql "]) +def test_excluded_services_is_accepted_on_every_otel_v2_destination(callback_name: str, value: str) -> None: + assert callback_config_error(callback_name, {"excluded_services": value}) is None + + def test_capture_message_content_is_rejected_while_otel_v2_is_off(monkeypatch: pytest.MonkeyPatch) -> None: from litellm.integrations.otel.model.config import is_otel_v2_enabled @@ -294,12 +301,31 @@ def test_capture_message_content_is_rejected_while_otel_v2_is_off(monkeypatch: p assert error is not None and "capture_message_content" in error and "LITELLM_OTEL_V2" in error +def test_excluded_services_is_rejected_while_otel_v2_is_off(monkeypatch: pytest.MonkeyPatch) -> None: + from litellm.integrations.otel.model.config import is_otel_v2_enabled + + monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) + is_otel_v2_enabled.cache_clear() + try: + error: Final = callback_config_error("langfuse_otel", {"excluded_services": "redis"}) + finally: + is_otel_v2_enabled.cache_clear() + assert error is not None and "excluded_services" in error and "LITELLM_OTEL_V2" in error + + @pytest.mark.parametrize("callback_name", ["langfuse", "datadog", "otel", "arize_phoenix", None]) def test_capture_message_content_is_rejected_where_it_would_never_take_effect(callback_name: str | None) -> None: error: Final = callback_config_error(callback_name, {"capture_message_content": "no_content"}) assert error is not None and "capture_message_content" in error and "langfuse_otel" in error +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.parametrize("callback_name", ["langfuse", "datadog", "otel", "arize_phoenix", None]) +def test_excluded_services_is_rejected_where_it_would_never_take_effect(callback_name: str | None) -> None: + error: Final = callback_config_error(callback_name, {"excluded_services": "redis"}) + assert error is not None and "excluded_services" in error and "langfuse_otel" in error + + @pytest.mark.usefixtures("otel_v2_on") @pytest.mark.parametrize("value", ["full", "event_only", "span_and_event"]) def test_an_unsupported_capture_message_content_is_rejected_on_key_logging_metadata(value: str) -> None: @@ -317,6 +343,38 @@ def test_an_unsupported_capture_message_content_is_rejected_on_key_logging_metad assert error is not None and "Invalid capture_message_content" in error +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.parametrize("callback_var", ["capture_message_content", "excluded_services"]) +def test_key_logging_metadata_rejects_non_string_destination_callback_vars(callback_var: str) -> None: + error: Final = logging_metadata_config_error( + { + "logging": [ + { + "callback_name": "newrelic", + "callback_type": "success_and_failure", + "callback_vars": {callback_var: None}, + } + ] + } + ) + assert error == f"{callback_var} must be a string, got NoneType" + + +@pytest.mark.usefixtures("otel_v2_on") +def test_key_logging_metadata_with_non_string_callback_var_key_does_not_raise() -> None: + error: Final = logging_metadata_config_error( + { + "logging": [ + { + "callback_name": "langfuse_otel", + "callback_vars": {1: "x", "excluded_services": "redis"}, + } + ] + } + ) + assert error is None + + @pytest.mark.usefixtures("otel_v2_on") def test_each_entry_keeps_its_own_capture_message_content() -> None: error: Final = logging_metadata_config_error( @@ -357,7 +415,27 @@ def test_entries_for_one_backend_share_one_capture_message_content( stored: list[tuple[str, dict[str, str]]], rejected: bool ) -> None: """The backend's entries merge into one destination, so a second value would silently win or lose.""" - error: Final = conflicting_capture_error("langfuse_otel", {"capture_message_content": "span_only"}, stored) + error: Final = conflicting_callback_var_error( + "langfuse_otel", "capture_message_content", {"capture_message_content": "span_only"}, stored + ) + assert (error is not None) is rejected + + +@pytest.mark.parametrize( + ("stored", "rejected"), + [ + ([("langfuse_otel", {"excluded_services": "none"})], True), + ([("langfuse_otel", {"excluded_services": "redis"})], False), + ([("langfuse_otel", {"langfuse_public_key": "pk"})], False), + ([("newrelic", {"excluded_services": "none"})], False), + ], +) +def test_entries_for_one_backend_share_one_excluded_services_value( + stored: list[tuple[str, dict[str, str]]], rejected: bool +) -> None: + error: Final = conflicting_callback_var_error( + "langfuse_otel", "excluded_services", {"excluded_services": "redis"}, stored + ) assert (error is not None) is rejected @@ -374,3 +452,50 @@ def test_key_logging_entries_for_one_backend_may_not_disagree_on_capture_message {"logging": [entry("success", "no_content"), entry("success_and_failure", "span_only")]} ) assert error is not None and "already set to 'no_content' by another langfuse_otel entry" in error + + +@pytest.mark.usefixtures("otel_v2_on") +def test_key_logging_entries_for_one_backend_may_not_disagree_on_excluded_services() -> None: + def entry(callback_vars: dict[str, str]) -> dict[str, str | dict[str, str]]: + return {"callback_name": "langfuse_otel", "callback_type": "success", "callback_vars": callback_vars} + + error: Final = logging_metadata_config_error( + {"logging": [entry({"excluded_services": "redis"}), entry({"excluded_services": "none"})]} + ) + assert error is not None and "already set to 'redis' by another langfuse_otel entry" in error + + +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.parametrize( + ("stored", "incoming", "conflicts"), + [ + ("redis,postgres", "postgres,redis", False), + ("redis", "REDIS", False), + ("postgres", "postgresql", False), + ("none", "NONE", False), + ("redis", "postgres", True), + ("none", "redis", True), + ], +) +def test_key_logging_entries_compare_excluded_services_by_parsed_value( + stored: str, incoming: str, conflicts: bool +) -> None: + def entry(callback_type: str, excluded_services: str) -> dict[str, str | dict[str, str]]: + return { + "callback_name": "newrelic", + "callback_type": callback_type, + "callback_vars": {"newrelic_api_key": "license-key", "excluded_services": excluded_services}, + } + + error: Final = logging_metadata_config_error( + {"logging": [entry("success", stored), entry("success_and_failure", incoming)]} + ) + assert (error is not None) is conflicts + if conflicts: + assert error is not None and f"already set to {stored!r}" in error + + +@pytest.mark.usefixtures("otel_v2_on") +def test_a_failure_only_entry_rejects_excluded_services() -> None: + error: Final = callback_config_error("langfuse_otel", {"excluded_services": "redis"}, "failure") + assert error is not None and "excluded_services" in error and "success_and_failure" in error diff --git a/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py b/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py index ad5b6903ece..ac78085ca47 100644 --- a/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py +++ b/tests/unit/proxy/management_endpoints/test_team_callback_endpoints.py @@ -8,11 +8,12 @@ redacted audit rows for callback mutations. import json from collections.abc import Iterator -from typing import Final +from typing import Final, cast from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest from fastapi import HTTPException, Request +from pydantic import ValidationError import litellm from litellm.proxy._types import ( @@ -21,6 +22,7 @@ from litellm.proxy._types import ( LitellmUserRoles, UserAPIKeyAuth, ) +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth from litellm.proxy.common_utils.callback_config_validation import cross_entry_family_error from litellm.proxy.management.teams.authz import TeamAccess from litellm.proxy.management_endpoints.team_callback_endpoints import ( @@ -28,6 +30,7 @@ from litellm.proxy.management_endpoints.team_callback_endpoints import ( delete_team_callback, disable_team_logging, get_team_callbacks, + router, ) @@ -1434,7 +1437,7 @@ async def test_delete_team_callback_route_accepts_team_ids_containing_slashes(): } mock_prisma = _patch_prisma(_team_row(team_id=team_id, metadata=metadata)) - app = FastAPI() + app: Final = FastAPI() app.include_router(router) app.dependency_overrides[user_api_key_auth] = _admin_auth @@ -1683,6 +1686,72 @@ async def test_a_second_entry_may_not_flip_the_span_scope(patched_prisma, caller patched_prisma.db.litellm_teamtable.update.assert_awaited_once() +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.asyncio +async def test_a_second_newrelic_entry_may_not_change_excluded_services(patched_prisma) -> None: + patched_prisma.get_data = AsyncMock( + return_value=_team_row( + metadata={ + "logging": [ + { + "callback_name": "newrelic", + "callback_type": "success", + "callback_vars": {"newrelic_api_key": "license-key", "excluded_services": "redis"}, + } + ] + } + ) + ) + data: Final = AddTeamCallback( + callback_name="newrelic", + callback_type="success_and_failure", + callback_vars={"newrelic_api_key": "license-key", "excluded_services": "postgres"}, + ) + with pytest.raises(HTTPException) as exc: + await add_team_callbacks( + data=data, + http_request=Mock(spec=Request), + team_id="team-victim", + user_api_key_dict=_admin_auth(), + ) + assert exc.value.status_code == 400 + assert "excluded_services" in str(exc.value.detail) + patched_prisma.db.litellm_teamtable.update.assert_not_called() + + +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.asyncio +async def test_a_second_newrelic_entry_accepts_equivalent_excluded_services(patched_prisma) -> None: + patched_prisma.get_data = AsyncMock( + return_value=_team_row( + metadata={ + "logging": [ + { + "callback_name": "newrelic", + "callback_type": "success", + "callback_vars": { + "newrelic_api_key": "license-key", + "excluded_services": "redis,postgres", + }, + } + ] + } + ) + ) + data: Final = AddTeamCallback( + callback_name="newrelic", + callback_type="success_and_failure", + callback_vars={"newrelic_api_key": "license-key", "excluded_services": "postgres,redis"}, + ) + await add_team_callbacks( + data=data, + http_request=Mock(spec=Request), + team_id="team-victim", + user_api_key_dict=_admin_auth(), + ) + patched_prisma.db.litellm_teamtable.update.assert_awaited_once() + + @pytest.mark.asyncio async def test_add_team_callbacks_rejects_out_of_range_arize_sampling_rate(patched_prisma): data = AddTeamCallback( @@ -1772,3 +1841,59 @@ async def test_add_team_callbacks_stores_capture_message_content_in_the_existing patched_prisma.db.litellm_teamtable.update.assert_awaited_once() stored: Final = json.loads(patched_prisma.db.litellm_teamtable.update.await_args.kwargs["data"]["metadata"]) assert [entry["callback_vars"].get("capture_message_content") for entry in stored["logging"]] == [capture] + + +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.asyncio +async def test_add_team_callbacks_stores_excluded_services_in_the_existing_callback_metadata(patched_prisma) -> None: + data: Final = AddTeamCallback( + callback_name="langfuse_otel", + callback_type="success", + callback_vars={ + "langfuse_public_key": "pk", + "langfuse_secret_key": "sk", + "excluded_services": "redis,postgres", + }, + ) + await add_team_callbacks( + data=data, + http_request=Mock(spec=Request), + team_id="team-victim", + user_api_key_dict=_admin_auth(), + ) + stored: Final = json.loads(patched_prisma.db.litellm_teamtable.update.await_args.kwargs["data"]["metadata"]) + assert [entry["callback_vars"].get("excluded_services") for entry in stored["logging"]] == ["redis,postgres"] + + +def test_add_team_callback_returns_http_422_for_unknown_excluded_services() -> None: + from fastapi import FastAPI + from fastapi.testclient import TestClient + + app = FastAPI() + app.include_router(router) + app.dependency_overrides[user_api_key_auth] = _admin_auth + + response: Final = TestClient(app).post( + "/team/team-victim/callback", + json={ + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": { + "langfuse_public_key": "pk", + "langfuse_secret_key": "sk", + "excluded_services": "auth", + }, + }, + ) + + assert response.status_code == 422 + assert "excluded_services" in response.text + + +@pytest.mark.parametrize("callback_var", ["capture_message_content", "excluded_services"]) +@pytest.mark.parametrize("value", [None, 1]) +def test_add_team_callback_rejects_non_string_destination_callback_vars(callback_var: str, value: object) -> None: + callback_vars: Final = cast("dict[str, str]", {callback_var: value}) + with pytest.raises(ValidationError) as exc: + AddTeamCallback(callback_name="newrelic", callback_vars=callback_vars) + assert f"{callback_var} must be a string, got {type(value).__name__}" in str(exc.value) diff --git a/tests/unit/types/test_utils.py b/tests/unit/types/test_utils.py index 09063d894e7..32479c62372 100644 --- a/tests/unit/types/test_utils.py +++ b/tests/unit/types/test_utils.py @@ -1,8 +1,15 @@ import pytest +from litellm.litellm_core_utils.initialize_dynamic_callback_params import validate_excluded_services_value from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VALUES, CaptureMessageContent, captures_span_content +@pytest.mark.parametrize("value", ["", " ", "auth", "none,redis", "redis,", ",postgres"]) +def test_invalid_excluded_services_values_are_rejected(value: str) -> None: + with pytest.raises(ValueError, match=r"Invalid excluded_services.*none"): + validate_excluded_services_value(value) + + @pytest.mark.parametrize( "mode", [ diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index fc2cc73ff10..3a509fdd395 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -17299,6 +17299,7 @@ export interface paths { * - capture_message_content: For OTel v2 callbacks, "no_content" or "span_only". * An explicit value overrides the global setting for this destination; omitted follows the proxy's own callback for that backend, else the global setting. * "span_only" puts prompt and response content on the destination's spans, "no_content" leaves it out + * - excluded_services: For OTel v2 callbacks, a comma-separated list of datastore names such as "redis,postgres", or "none" to exclude nothing. An explicit value replaces the global exclusions; omitted follows them * - gcs_bucket_name: The name of the GCS bucket * - gcs_path_service_account: The path to the GCS service account * - langsmith_api_key: The API key for the Langsmith callback