mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-14 04:09:09 +00:00
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>
This commit is contained in:
parent
988cc7dc94
commit
9447dc3773
17 changed files with 567 additions and 49 deletions
|
|
@ -314,7 +314,11 @@ nothing here imports outside it:
|
|||
`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` 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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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,7 +2048,11 @@ 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}")
|
||||
|
|
@ -2060,6 +2066,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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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 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,20 +242,21 @@ 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)) not in (None, incoming)
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
|
@ -255,7 +277,7 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str
|
|||
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 +285,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 +303,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]:
|
||||
|
|
|
|||
|
|
@ -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,10 @@ 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 +981,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 +1200,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 +1215,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)),
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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,25 @@ 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
|
||||
|
|
|
|||
|
|
@ -3727,6 +3727,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):
|
||||
|
|
|
|||
|
|
@ -16,8 +16,10 @@ from integration._support.client import (
|
|||
gateway_from_environment,
|
||||
)
|
||||
from integration._support.otlp_sink import (
|
||||
ConnectSink,
|
||||
Span,
|
||||
SpanSinks,
|
||||
owned_connect_sink,
|
||||
recorded_spans,
|
||||
spans_for_trace,
|
||||
)
|
||||
|
|
@ -61,19 +63,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(
|
||||
|
|
@ -260,6 +286,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(
|
||||
|
|
|
|||
|
|
@ -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,99 @@ 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 +4821,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 +4976,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,
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
@ -357,7 +383,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 +420,20 @@ 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")
|
||||
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
|
||||
|
|
|
|||
|
|
@ -21,6 +21,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 +29,7 @@ from litellm.proxy.management_endpoints.team_callback_endpoints import (
|
|||
delete_team_callback,
|
||||
disable_team_logging,
|
||||
get_team_callbacks,
|
||||
router,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -1434,7 +1436,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
|
||||
|
||||
|
|
@ -1772,3 +1774,50 @@ 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
|
||||
|
|
|
|||
|
|
@ -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="Invalid excluded_services.*none"):
|
||||
validate_excluded_services_value(value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mode",
|
||||
[
|
||||
|
|
|
|||
1
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
1
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -17271,6 +17271,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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue