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

@@ -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.

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

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,

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):

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,
     )

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",

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
 
 

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,
     )
 
 

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)),
     )
 
 

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

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):

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(

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,

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

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)

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",
     [

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