mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-11 03:38:38 +00:00
feat(otel)!: let a team capture_message_content override the global setting for its OTel v2 destination (#44264)
* feat(otel): let a team strip captured message content from its own OTel v2 destination Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * refactor(proxy): split the shared callback option conflict check and sync dashboard API types Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): address review feedback on team capture_message_content Classify capture_message_content as NotSecret in the credential slot registry, inject test exporters through register_exporter_factory instead of patching _exporter_from_spec, type the test helpers, and stop reassigning processor in build_tracer_provider Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(otel): add integration cells for team capture_message_content Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(otel): cover weave_otel and newrelic team destinations in capture_message_content cells Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): stop rejecting client-sent capture_message_content at the request body boundary Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * feat(otel): let a team capture_message_content override the global setting in both directions Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): keep a team's capture_message_content on its own OTel v2 destination The setting no longer rides the flattened team callback vars, so a sibling destination and a routed provider follow the global mode. Request bodies get the base 400 back, a failure-only entry rejects the setting, and entries for one backend must agree Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(otel): type the capture_message_content callback validation tests Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(otel): type the remaining capture_message_content test signatures Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(otel): mark the new capture_message_content test locals Final Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): default tenant capture to callback_settings.otel and keep CaptureMessageContent formatting bare Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(otel): keep CaptureMessageContent a plain str like main The str-to-Enum switch changed the public type: the config field's type, repr and model_dump, a constructor that now raised on unknown values, and unknown env values coerced to no_content. CaptureMessageContent is the plain str subclass main ships again, the field is a str, and a captures_span_content helper replaces the enum property. A test pins that a published preset's capture mode stays the default for an unowned destination, which is what main forwards. * test(otel): drop the module-level CaptureMessageContent import the rewritten test no longer uses * fix(lint): keep the merged branch within main's type budgets Freeze the capture-mode settings model (LIT015) and read stored logging entries through one typed sequence helper, so the conflicting-capture checks add no unknown types. * fix(otel): reject a destination capture_message_content while OTel v2 is off The legacy OTel callbacks never read the per-destination setting, so a team could register no_content and still export prompts and replies under the global capture mode. Registration now refuses the var unless the proxy runs with LITELLM_OTEL_V2=true, the same rule the Arize transport and New Relic routing vars follow. * test(otel): type the OTel v2 fixtures and the v2-off regression test * fix(otel): accept only no_content and span_only on a team or key destination OTel v2 writes captured content only to span attributes, so event_only behaved like no_content and span_and_event like span_only on a destination. The proxy-wide setting keeps all four values * docs(otel): say an omitted capture setting follows the proxy's own callback for that backend * test(otel): describe the general capture fallback in the unowned destination test --------- Co-authored-by: mrinal <mrinal@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: yucheng-berri <yucheng@berri.ai> Co-authored-by: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
3981228d70
commit
dc95de5209
25 changed files with 2019 additions and 85 deletions
|
|
@ -295,7 +295,9 @@ nothing here imports outside it:
|
|||
- [`config.py`](./model/config.py) — `OpenTelemetryV2Config`, a pydantic-settings
|
||||
model that reads `OTEL_*` / `LITELLM_OTEL_*` env vars, plus the feature gate.
|
||||
`capture_span_content` gates whether prompt/response bodies may be written as
|
||||
span attributes; it defaults **off** (`no_content`). The Baggage allowlists are
|
||||
span attributes; it defaults **off** (`no_content`). A team or key can override
|
||||
the global mode per OTel v2 destination with `capture_message_content`; omitted
|
||||
values follow the proxy's own callback for that backend, else the global mode. The Baggage allowlists are
|
||||
configurable, not hard-coded: set `LITELLM_OTEL_BAGGAGE_PROMOTED_KEYS` /
|
||||
`LITELLM_OTEL_BAGGAGE_METADATA_KEYS` /
|
||||
`LITELLM_OTEL_BAGGAGE_TEAM_METADATA_KEYS` (comma-separated) as env vars, or
|
||||
|
|
|
|||
|
|
@ -51,6 +51,10 @@ class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2):
|
|||
response: "AsyncIterator[ModelResponseStream]",
|
||||
request_data: Mapping[str, object],
|
||||
) -> "AsyncGenerator[ModelResponseStream, None]":
|
||||
if not self._capture_span_content():
|
||||
async for chunk in response:
|
||||
yield chunk
|
||||
return
|
||||
relayed: Final[list[ModelResponseStream]] = [] # mutable-ok: relayed as they arrive, assembled at end of stream
|
||||
async for chunk in response:
|
||||
relayed.append(chunk)
|
||||
|
|
@ -58,6 +62,8 @@ class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2):
|
|||
self._stamp_root_io(request_data, lambda: stream_output(tuple(relayed), request_data))
|
||||
|
||||
def _stamp_root_io(self, data: Mapping[str, object], render_output: Callable[[], str | None]) -> None:
|
||||
if not self._capture_span_content():
|
||||
return
|
||||
root: Final = request_root_span()
|
||||
if root is None or not root.is_recording():
|
||||
return
|
||||
|
|
|
|||
|
|
@ -58,6 +58,7 @@ from litellm.integrations.otel.plumbing.context import (
|
|||
is_recordable_span,
|
||||
mcp_message_transport_span,
|
||||
post_response_root,
|
||||
request_destinations,
|
||||
request_root_http_route,
|
||||
request_root_span,
|
||||
resolve_internal_call_span_context,
|
||||
|
|
@ -82,6 +83,7 @@ from litellm.integrations.otel.plumbing.providers import (
|
|||
resolve_meter_provider,
|
||||
)
|
||||
from litellm.integrations.otel.plumbing.routing import TenantTracerCache
|
||||
from litellm.types.utils import captures_span_content
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from opentelemetry.metrics import MeterProvider
|
||||
|
|
@ -254,6 +256,11 @@ class OpenTelemetryV2(CustomLogger):
|
|||
return None
|
||||
return GenAIEventRecorder(get_event_logger(provider, LITELLM_TRACER_NAME), provider.resource)
|
||||
|
||||
def _capture_span_content(self) -> bool:
|
||||
return self.config.capture_span_content or any(
|
||||
captures_span_content(destination.capture_message_content) for destination in request_destinations()
|
||||
)
|
||||
|
||||
# ====================================================================== #
|
||||
# Proxy global registration
|
||||
# ====================================================================== #
|
||||
|
|
@ -433,7 +440,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
return False
|
||||
payload: Final = cast("StandardLoggingPayload", raw_payload)
|
||||
data: Final = MCPToolCallSpanData.from_standard_logging_payload(
|
||||
payload, capture_content=self.config.capture_span_content
|
||||
payload, capture_content=self._capture_span_content()
|
||||
)
|
||||
# A stray LLM carrier from a ``pre_call`` that mis-fired for this id would
|
||||
# otherwise linger until evicted; drop it so it's neither leaked nor closed
|
||||
|
|
@ -477,7 +484,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
return False
|
||||
payload: Final = cast("StandardLoggingPayload", raw_payload)
|
||||
data: Final = MCPListToolsSpanData.from_standard_logging_payload(
|
||||
payload, capture_content=self.config.capture_span_content
|
||||
payload, capture_content=self._capture_span_content()
|
||||
)
|
||||
if data.identity.call_id:
|
||||
self._release_carrier(self._open_llm_calls.pop(data.identity.call_id, None))
|
||||
|
|
@ -569,7 +576,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
return None
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
payload,
|
||||
capture_content=self.config.capture_span_content,
|
||||
capture_content=self._capture_span_content(),
|
||||
time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
|
||||
request_route=request_root_http_route(),
|
||||
request_purpose=call.purpose,
|
||||
|
|
@ -1076,6 +1083,6 @@ def build_otel_v2_logger(
|
|||
def _logger_class(config: OpenTelemetryV2Config) -> type[OpenTelemetryV2]:
|
||||
if "langfuse" not in config.mapper_names:
|
||||
return OpenTelemetryV2
|
||||
from litellm.integrations.otel.langfuse_logger import LangfuseContentOpenTelemetryV2, LangfuseOpenTelemetryV2
|
||||
from litellm.integrations.otel.langfuse_logger import LangfuseContentOpenTelemetryV2
|
||||
|
||||
return LangfuseContentOpenTelemetryV2 if config.capture_span_content else LangfuseOpenTelemetryV2
|
||||
return LangfuseContentOpenTelemetryV2
|
||||
|
|
|
|||
|
|
@ -16,19 +16,17 @@ from litellm.integrations.otel.model.baggage import (
|
|||
)
|
||||
from litellm.integrations.otel.model.spans import POSTGRESQL, db_system
|
||||
from litellm.types.llms.base import LiteLLMBaseModel
|
||||
from litellm.types.utils import OtelSpanScope
|
||||
from litellm.types.utils import (
|
||||
CAPTURE_MESSAGE_CONTENT_VALUES,
|
||||
CaptureMessageContent,
|
||||
OtelSpanScope,
|
||||
captures_span_content,
|
||||
)
|
||||
|
||||
#: Master feature-flag env var. The logger is inert until this is truthy.
|
||||
OTEL_V2_ENV: Final = "LITELLM_OTEL_V2"
|
||||
|
||||
|
||||
class CaptureMessageContent(str):
|
||||
NO_CONTENT = "no_content"
|
||||
SPAN_ONLY = "span_only"
|
||||
EVENT_ONLY = "event_only"
|
||||
SPAN_AND_EVENT = "span_and_event"
|
||||
|
||||
|
||||
class ExporterOwner(str, Enum):
|
||||
"""The preset that contributed an exporter. Values match the callback names
|
||||
in ``presets.PRESET_BY_CALLBACK`` so per-request dynamic-credential routing
|
||||
|
|
@ -297,9 +295,7 @@ class OpenTelemetryV2Config(BaseSettings):
|
|||
here are lower_snake_case; normalizing at the boundary keeps both
|
||||
spellings working and lets every downstream comparison stay exact.
|
||||
"""
|
||||
if isinstance(value, str):
|
||||
return value.lower()
|
||||
return value
|
||||
return _normalized_capture_message_content(value)
|
||||
|
||||
@field_validator("langfuse_span_scope", mode="before")
|
||||
@classmethod
|
||||
|
|
@ -371,22 +367,39 @@ class OpenTelemetryV2Config(BaseSettings):
|
|||
|
||||
@property
|
||||
def capture_span_content(self) -> bool:
|
||||
"""Whether prompt/response content may be stamped as span attributes.
|
||||
|
||||
Defaults off (``no_content``): an operator must opt in before message
|
||||
bodies leave the process, so a user request can never force its prompt
|
||||
or completion into the configured backend while capture is disabled.
|
||||
"""
|
||||
return self.capture_message_content in (
|
||||
CaptureMessageContent.SPAN_ONLY,
|
||||
CaptureMessageContent.SPAN_AND_EVENT,
|
||||
)
|
||||
"""Whether the global setting permits prompt/response span attributes."""
|
||||
return captures_span_content(self.capture_message_content)
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> "OpenTelemetryV2Config":
|
||||
return cls()
|
||||
|
||||
|
||||
def _normalized_capture_message_content(value: object) -> object:
|
||||
if isinstance(value, str):
|
||||
return value.lower()
|
||||
return value
|
||||
|
||||
|
||||
class _CaptureMessageContentSettings(BaseSettings):
|
||||
model_config = SettingsConfigDict(extra="ignore", frozen=True)
|
||||
|
||||
capture_message_content: str = Field(
|
||||
default=CaptureMessageContent.NO_CONTENT,
|
||||
validation_alias=AliasChoices("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
|
||||
)
|
||||
|
||||
@field_validator("capture_message_content", mode="before")
|
||||
@classmethod
|
||||
def _normalize_capture_message_content(cls, value: object) -> object:
|
||||
return _normalized_capture_message_content(value)
|
||||
|
||||
|
||||
def capture_message_content_from_env() -> str:
|
||||
"""The global capture mode, read without building a settings model that rereads every other env var"""
|
||||
return _CaptureMessageContentSettings().capture_message_content
|
||||
|
||||
|
||||
_EXCLUDED_SERVICES_INPUT: Final[TypeAdapter[str | tuple[object, ...]]] = TypeAdapter(str | tuple[object, ...])
|
||||
|
||||
|
||||
|
|
@ -395,6 +408,10 @@ def excluded_db_systems_from(value: object) -> frozenset[str]:
|
|||
return _normalize_excluded_services(excluded_service_names(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
|
||||
|
||||
|
||||
def excluded_service_names(value: object) -> frozenset[str]:
|
||||
"""Read a YAML list or comma-separated string of service names, logging and dropping unusable input
|
||||
so a malformed value cannot stop the OTel logger from being built"""
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ from urllib.parse import quote
|
|||
from pydantic import ConfigDict, Field
|
||||
|
||||
from litellm.types.llms.base import LiteLLMBaseModel
|
||||
from litellm.types.utils import OtelSpanScope
|
||||
from litellm.types.utils import OtelSpanScope, captures_span_content
|
||||
|
||||
|
||||
class OtelDestination(LiteLLMBaseModel):
|
||||
|
|
@ -50,6 +50,17 @@ class OtelDestination(LiteLLMBaseModel):
|
|||
default=None,
|
||||
description="Same, for the requests with a failed span in their tree; ``None`` forwards every one.",
|
||||
)
|
||||
capture_message_content: str | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"An explicit mode overrides the global capture policy for this destination. "
|
||||
"Omitted follows the global setting."
|
||||
),
|
||||
)
|
||||
|
||||
def captures_content(self, default: str) -> bool:
|
||||
setting: Final = self.capture_message_content
|
||||
return captures_span_content(default if setting is None else setting)
|
||||
|
||||
def header_string(self) -> str:
|
||||
"""Render headers as the ``k=v,k2=v2`` form an ``ExporterSpec`` expects.
|
||||
|
|
@ -64,9 +75,9 @@ 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`` and the sampling rates are left out on purpose: they decide which
|
||||
spans reach the processor, not how the processor exports them, so a full and an
|
||||
``llm_only`` view of the same account share 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.
|
||||
"""
|
||||
return (
|
||||
self.endpoint,
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import random
|
|||
import threading
|
||||
import time
|
||||
from collections import OrderedDict
|
||||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||||
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Final, Literal
|
||||
|
||||
|
|
@ -40,8 +40,19 @@ from opentelemetry.util.types import Attributes, AttributeValue
|
|||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm._version import version as litellm_version
|
||||
from litellm.integrations.otel.mappers.langfuse import LANGFUSE_TRACE_NAME
|
||||
from litellm.integrations.otel.model.config import ExporterOwner, ExporterSpec, OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.mappers.langfuse import (
|
||||
LANGFUSE_OBSERVATION_INPUT,
|
||||
LANGFUSE_OBSERVATION_OUTPUT,
|
||||
LANGFUSE_TRACE_NAME,
|
||||
)
|
||||
from litellm.integrations.otel.model.config import (
|
||||
CaptureMessageContent,
|
||||
ExporterOwner,
|
||||
ExporterSpec,
|
||||
OpenTelemetryV2Config,
|
||||
capture_message_content_from_env,
|
||||
captures_span_content,
|
||||
)
|
||||
from litellm.integrations.otel.model.semconv import (
|
||||
DB,
|
||||
MCP,
|
||||
|
|
@ -385,6 +396,26 @@ _CAPTURED_HEADER_PREFIXES: Final = ("http.request.header.", "http.response.heade
|
|||
# ``?key=`` query parameter.
|
||||
_URL_KEYS: Final = frozenset({"http.url", "http.target", "url.full"})
|
||||
_URL_QUERY_KEY: Final = "url.query"
|
||||
# Prompt and response content, in every vocabulary the mappers write it in once the
|
||||
# global capture policy collects it: GenAI semconv (MCP tool-call arguments and results
|
||||
# included), Langfuse, OpenInference's blobs and per-index messages, Langtrace, Weave.
|
||||
_MESSAGE_CONTENT_KEYS: Final = frozenset(
|
||||
{
|
||||
GenAI.INPUT_MESSAGES,
|
||||
GenAI.OUTPUT_MESSAGES,
|
||||
GenAI.SYSTEM_INSTRUCTIONS,
|
||||
GenAI.TOOL_CALL_ARGUMENTS,
|
||||
GenAI.TOOL_CALL_RESULT,
|
||||
LANGFUSE_OBSERVATION_INPUT,
|
||||
LANGFUSE_OBSERVATION_OUTPUT,
|
||||
"input.value",
|
||||
"output.value",
|
||||
"llm.prompts",
|
||||
"llm.completions",
|
||||
"weave.output",
|
||||
}
|
||||
)
|
||||
_INDEXED_MESSAGE_PREFIXES: Final = ("llm.input_messages.", "llm.output_messages.")
|
||||
|
||||
|
||||
class _SpanView(ReadableSpan):
|
||||
|
|
@ -415,6 +446,19 @@ class _SpanView(ReadableSpan):
|
|||
)
|
||||
|
||||
|
||||
def _is_message_content(key: str) -> bool:
|
||||
return key in _MESSAGE_CONTENT_KEYS or key.startswith(_INDEXED_MESSAGE_PREFIXES)
|
||||
|
||||
|
||||
def without_message_content(span: ReadableSpan) -> ReadableSpan:
|
||||
"""``span`` without its prompt and response content, everything else left as it was."""
|
||||
attributes: Final = span.attributes or _NO_ATTRIBUTES
|
||||
if not any(_is_message_content(key) for key in attributes):
|
||||
return span
|
||||
kept: Final = MappingProxyType({key: value for key, value in attributes.items() if not _is_message_content(key)})
|
||||
return _SpanView(span, span.resource, kept, span.events, span.status, parent=span.parent)
|
||||
|
||||
|
||||
def _is_database_span(attributes: Mapping[str, AttributeValue]) -> bool:
|
||||
return any(key in attributes for key in _DB_SYSTEM_KEYS)
|
||||
|
||||
|
|
@ -553,7 +597,11 @@ def _tenant_resource(resource: Resource, defaults: Mapping[str, str], extra: Map
|
|||
return filled.merge(Resource(extra)) if extra else filled
|
||||
|
||||
|
||||
def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> ReadableSpan:
|
||||
def _for_destination(
|
||||
span: ReadableSpan,
|
||||
destination: "OtelDestination",
|
||||
default_capture: str,
|
||||
) -> ReadableSpan:
|
||||
"""The view of ``span`` a tenant destination receives.
|
||||
|
||||
A span the tenant's own call produced keeps its error text. Every other span is
|
||||
|
|
@ -564,8 +612,10 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read
|
|||
exception it raised and names the operator's guardrail endpoint. Stack traces walk
|
||||
the operator's install and come off every span, as do the headers the operator
|
||||
captures on the server span, whose request side holds the caller's bearer token,
|
||||
and the query string of the request URL, which can hold the same key. The span
|
||||
itself stays, so the tenant still gets the whole trace tree.
|
||||
and the query string of the request URL, which can hold the same key. A
|
||||
destination whose effective setting disables span content loses the prompt and
|
||||
response content too. The span itself stays, so the tenant still gets the whole
|
||||
trace tree.
|
||||
"""
|
||||
extra: Final = destination.resource_attributes
|
||||
defaults: Final = destination.resource_defaults
|
||||
|
|
@ -573,11 +623,12 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read
|
|||
database: Final = _is_database_span(attributes)
|
||||
owned: Final = _is_tenant_owned_span(attributes)
|
||||
unreachable: Final = _guardrail_unreachable(attributes)
|
||||
redacted: Final = not destination.captures_content(default_capture)
|
||||
kept: Final = MappingProxyType(
|
||||
{
|
||||
key: _without_query(key, value)
|
||||
for key, value in attributes.items()
|
||||
if _tenant_visible(key, database, owned, unreachable)
|
||||
if _tenant_visible(key, database, owned, unreachable) and not (redacted and _is_message_content(key))
|
||||
}
|
||||
)
|
||||
recorded: Final = span.events
|
||||
|
|
@ -612,6 +663,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
|
|||
shutdown_drain_seconds: float = _SHUTDOWN_DRAIN_SECONDS,
|
||||
operator_sinks: 'Mapping[_SinkKey, "OtelSpanScope"]' = MappingProxyType({}),
|
||||
excluded_db_systems: frozenset[str] = frozenset(),
|
||||
default_capture: str = CaptureMessageContent.NO_CONTENT,
|
||||
default_capture_by_backend: Mapping[str, str] = MappingProxyType({}),
|
||||
pending_drains: int = _MAX_PENDING_DRAINS,
|
||||
drain_pool: _DrainPool | None = None,
|
||||
sampling_draw: Callable[[], float] | None = None,
|
||||
|
|
@ -619,6 +672,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
|
|||
self._operator_sinks: Final = operator_sinks
|
||||
self._draw: Final = sampling_draw if sampling_draw is not None else random.random
|
||||
self._excluded_db_systems: Final = excluded_db_systems
|
||||
self._default_capture: Final = default_capture
|
||||
self._default_capture_by_backend: Final = default_capture_by_backend
|
||||
self._drain_seconds: Final = shutdown_drain_seconds
|
||||
self._lock: Final = threading.Condition()
|
||||
self._closed = False # guarded by ``_lock``: an unlocked read races the teardown it gates
|
||||
|
|
@ -661,12 +716,22 @@ class TenantFanOutSpanProcessor(SpanProcessor):
|
|||
for held_span, held_destination in self._settle(trace_id, failed):
|
||||
self._forward(held_span, held_destination)
|
||||
|
||||
def _default_capture_for(self, destination: "OtelDestination") -> str:
|
||||
backend: Final = destination.callback_name
|
||||
return (
|
||||
self._default_capture_by_backend.get(backend, self._default_capture) if backend else self._default_capture
|
||||
)
|
||||
|
||||
def _forward(self, span: ReadableSpan, destination: "OtelDestination") -> None:
|
||||
processor: Final = self._acquire(destination)
|
||||
if processor is None:
|
||||
return
|
||||
try:
|
||||
processor.on_end(_scoped(_for_destination(span, destination), destination.span_scope))
|
||||
processor.on_end(
|
||||
_scoped(
|
||||
_for_destination(span, destination, self._default_capture_for(destination)), destination.span_scope
|
||||
)
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span
|
||||
verbose_logger.debug("OTel V2 fan-out: forwarding to %s failed: %s", destination.endpoint, exc)
|
||||
finally:
|
||||
|
|
@ -963,14 +1028,15 @@ class _OverriddenBackendFilter(SpanProcessor):
|
|||
Wrapping is the only place this works: ``SynchronousMultiSpanProcessor.on_end``
|
||||
ignores return values, so a sibling processor can never veto the export.
|
||||
|
||||
Under ``additive`` mode nothing is suppressed, so the wrapper passes every span
|
||||
straight through and the operator keeps its copy.
|
||||
Under ``additive`` mode spans are not suppressed, though global content policy
|
||||
still strips message content when disabled.
|
||||
|
||||
``scope`` narrows what the exporter receives independently of that: under
|
||||
``llm_only`` the model-call spans go through as trace roots and the rest of the
|
||||
tree is held back, unless a destination of the request names ``sink``, the account
|
||||
this exporter writes to, with a wider scope: the fan-out then delivers the rest of
|
||||
the tree there and the model call keeps its place in it.
|
||||
the tree there and the model call keeps its place in it. A destination naming
|
||||
``sink`` sets this exporter's content policy; otherwise it follows global capture.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
|
|
@ -979,11 +1045,15 @@ class _OverriddenBackendFilter(SpanProcessor):
|
|||
owner: str | None,
|
||||
scope: "OtelSpanScope" = "full",
|
||||
sink: _SinkKey | None = None,
|
||||
global_captures: bool = True,
|
||||
filter_content: bool = True,
|
||||
) -> None:
|
||||
self._inner: Final = inner
|
||||
self._owner: Final = owner
|
||||
self._scope: Final = scope
|
||||
self._sink: Final = sink
|
||||
self._global_captures: Final = global_captures
|
||||
self._filter_content: Final = filter_content
|
||||
|
||||
def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
|
||||
self._inner.on_start(span, parent_context)
|
||||
|
|
@ -991,7 +1061,28 @@ class _OverriddenBackendFilter(SpanProcessor):
|
|||
def on_end(self, span: ReadableSpan) -> None:
|
||||
if self._owner in suppressed_backends() or not _in_scope(span, self._scope):
|
||||
return
|
||||
self._inner.on_end(_scoped(span, self._account_scope()))
|
||||
scoped: Final = _scoped(span, self._account_scope())
|
||||
self._inner.on_end(
|
||||
without_message_content(scoped) if self._filter_content and not self._account_captures_content() else scoped
|
||||
)
|
||||
|
||||
def _account_captures_content(self) -> bool:
|
||||
"""Whether a destination of this request that names ``sink`` captures content.
|
||||
|
||||
The fan-out skips a destination this exporter already writes to, so this copy is
|
||||
the one that account gets and has to honor the destination's content mode.
|
||||
"""
|
||||
explicit: Final = next(
|
||||
(
|
||||
destination.capture_message_content
|
||||
for destination in request_destinations()
|
||||
if self._sink is not None
|
||||
and destination.capture_message_content is not None
|
||||
and _sink_key(destination.endpoint, destination.headers) == self._sink
|
||||
),
|
||||
None,
|
||||
)
|
||||
return captures_span_content(explicit) if explicit is not None else self._global_captures
|
||||
|
||||
def _account_scope(self) -> "OtelSpanScope":
|
||||
if self._scope == "full" or self._sink is None:
|
||||
|
|
@ -1010,6 +1101,25 @@ class _OverriddenBackendFilter(SpanProcessor):
|
|||
return self._inner.force_flush(timeout_millis)
|
||||
|
||||
|
||||
class _MessageContentFilter(SpanProcessor):
|
||||
"""Strip prompt and response content from every span before ``inner`` exports it."""
|
||||
|
||||
def __init__(self, inner: SpanProcessor) -> None:
|
||||
self._inner: Final = inner
|
||||
|
||||
def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
|
||||
self._inner.on_start(span, parent_context)
|
||||
|
||||
def on_end(self, span: ReadableSpan) -> None:
|
||||
self._inner.on_end(without_message_content(span))
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self._inner.shutdown()
|
||||
|
||||
def force_flush(self, timeout_millis: int = 30000) -> bool:
|
||||
return self._inner.force_flush(timeout_millis)
|
||||
|
||||
|
||||
def build_span_exporter(config: OpenTelemetryV2Config) -> SpanExporter:
|
||||
"""Build a single exporter from the top-level config fields.
|
||||
|
||||
|
|
@ -1261,12 +1371,23 @@ def build_resource(config: OpenTelemetryV2Config) -> Resource:
|
|||
return Resource.create(attributes)
|
||||
|
||||
|
||||
def _spec_processor(
|
||||
spec: ExporterSpec, use_simple_processor: bool | None, strip_message_content: bool
|
||||
) -> SpanProcessor:
|
||||
exporting: Final = _processor_for(
|
||||
_exporter_from_spec(spec),
|
||||
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
|
||||
)
|
||||
return _MessageContentFilter(exporting) if strip_message_content else exporting
|
||||
|
||||
|
||||
def build_tracer_provider(
|
||||
config: OpenTelemetryV2Config,
|
||||
exporter: SpanExporter | None = None,
|
||||
baggage_processor: SpanProcessor | None = None,
|
||||
use_simple_processor: bool | None = None,
|
||||
tenant_overrides: bool = False,
|
||||
strip_message_content: bool = False,
|
||||
) -> TracerProvider:
|
||||
"""Build the shared :class:`TracerProvider`.
|
||||
|
||||
|
|
@ -1285,6 +1406,9 @@ def build_tracer_provider(
|
|||
|
||||
``config.langfuse_span_scope`` narrows the exporter owned by ``langfuse_otel``
|
||||
alone; a collector or any other backend in the same config keeps the full tree.
|
||||
``strip_message_content`` removes prompt and response content before every
|
||||
exporter, for a routed provider whose spans can carry content another
|
||||
destination asked for.
|
||||
"""
|
||||
provider: Final = TracerProvider(resource=build_resource(config))
|
||||
if baggage_processor is None:
|
||||
|
|
@ -1300,17 +1424,23 @@ def build_tracer_provider(
|
|||
for spec in config.exporters:
|
||||
if spec.requires_headers and not spec.headers:
|
||||
continue
|
||||
exp = _exporter_from_spec(spec)
|
||||
processor = _processor_for(
|
||||
exp,
|
||||
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
|
||||
)
|
||||
processor = _spec_processor(spec, use_simple_processor, strip_message_content)
|
||||
owner = spec.owner.value if tenant_overrides and spec.owner is not None else None
|
||||
scope = _operator_scope(config, spec)
|
||||
sink = _sink_key(spec.endpoint, parse_headers(spec.headers)) if _exports_to_the_wire(spec) else None
|
||||
provider.add_span_processor(
|
||||
_OverriddenBackendFilter(processor, owner, scope, sink)
|
||||
if owner is not None or scope != "full"
|
||||
_OverriddenBackendFilter(
|
||||
processor,
|
||||
owner,
|
||||
scope,
|
||||
sink,
|
||||
config.capture_span_content,
|
||||
tenant_overrides,
|
||||
)
|
||||
if owner is not None
|
||||
or scope != "full"
|
||||
or sink is not None
|
||||
or (tenant_overrides and not config.capture_span_content)
|
||||
else processor
|
||||
)
|
||||
return provider
|
||||
|
|
@ -1340,10 +1470,29 @@ def attach_tenant_fan_out(
|
|||
TenantFanOutSpanProcessor(
|
||||
operator_sinks=operator_sink_scopes(*configs),
|
||||
excluded_db_systems=excluded_db_systems,
|
||||
default_capture=_operator_capture_mode(configs),
|
||||
default_capture_by_backend=MappingProxyType(dict(_owned_capture_modes(configs))),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _operator_capture_mode(configs: Sequence[OpenTelemetryV2Config]) -> str:
|
||||
"""The mode of the operator's generic ``otel`` callback, so ``callback_settings.otel`` counts as the global default"""
|
||||
return next(
|
||||
(config.capture_message_content for config in configs if any(spec.owner is None for spec in config.exporters)),
|
||||
capture_message_content_from_env(),
|
||||
)
|
||||
|
||||
|
||||
def _owned_capture_modes(
|
||||
configs: Sequence[OpenTelemetryV2Config],
|
||||
) -> Iterator[tuple[str, str]]:
|
||||
for config in configs:
|
||||
for spec in config.exporters:
|
||||
if spec.owner is not None:
|
||||
yield spec.owner.value, config.capture_message_content
|
||||
|
||||
|
||||
def deliverable_destinations(
|
||||
destinations: Iterable["OtelDestination"],
|
||||
provider: trace.TracerProvider | None = None,
|
||||
|
|
|
|||
|
|
@ -282,7 +282,8 @@ class TenantTracerCache:
|
|||
self._providers.move_to_end(cache_key)
|
||||
return cached
|
||||
built: Final = build_tracer_provider(
|
||||
self._routed_config(credential_headers, project_headers, endpoint, service_name)
|
||||
self._routed_config(credential_headers, project_headers, endpoint, service_name),
|
||||
strip_message_content=not self._config.capture_span_content,
|
||||
)
|
||||
self._providers[cache_key] = built
|
||||
return built
|
||||
|
|
|
|||
|
|
@ -187,6 +187,7 @@ def destination_for(
|
|||
callback_name: str,
|
||||
params: StandardCallbackDynamicParams,
|
||||
service_name: str | None = None,
|
||||
capture_message_content: str | None = None,
|
||||
) -> OtelDestination | None:
|
||||
"""The destination ``params`` names for ``callback_name``, or ``None``.
|
||||
|
||||
|
|
@ -217,6 +218,7 @@ def destination_for(
|
|||
callback_name=callback_name,
|
||||
protocol=protocol,
|
||||
span_scope=_span_scope(callback_name, params),
|
||||
capture_message_content=capture_message_content,
|
||||
success_sampling_rate=success_sampling_rate,
|
||||
error_sampling_rate=error_sampling_rate,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
|
|||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.integrations.otel.model.config import (
|
||||
CaptureMessageContent,
|
||||
ExporterOwner,
|
||||
ExporterSpec,
|
||||
OpenTelemetryV2Config,
|
||||
|
|
@ -66,7 +67,11 @@ def newrelic_preset(
|
|||
# New Relic ingests the OTLP GenAI semantic conventions natively.
|
||||
"mapper_names": ensure_mappers(base.mapper_names, "genai"),
|
||||
**(
|
||||
{"capture_message_content": ("span_only" if settings.record_content else "no_content")}
|
||||
{
|
||||
"capture_message_content": (
|
||||
CaptureMessageContent.SPAN_ONLY if settings.record_content else CaptureMessageContent.NO_CONTENT
|
||||
)
|
||||
}
|
||||
if settings.record_content is not None
|
||||
else {}
|
||||
),
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ from typing import Any, Final
|
|||
|
||||
from litellm.types.utils import (
|
||||
ARIZE_OTLP_PROTOCOLS,
|
||||
CAPTURE_MESSAGE_CONTENT_VALUES,
|
||||
OTEL_SPAN_SCOPES,
|
||||
TRUSTED_CALLBACK_VARS_FIELD,
|
||||
StandardCallbackDynamicParams,
|
||||
|
|
@ -91,6 +92,13 @@ def validate_arize_otlp_protocol_value(value: str) -> None:
|
|||
raise ValueError(f"Invalid arize_otlp_protocol {value!r}: must be one of {sorted(ARIZE_OTLP_PROTOCOLS)}")
|
||||
|
||||
|
||||
def validate_capture_message_content_value(value: str) -> None:
|
||||
if value not in CAPTURE_MESSAGE_CONTENT_VALUES:
|
||||
raise ValueError(
|
||||
f"Invalid capture_message_content {value!r}: must be one of {sorted(CAPTURE_MESSAGE_CONTENT_VALUES)}"
|
||||
)
|
||||
|
||||
|
||||
# Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict
|
||||
_supported_callback_params: Final[tuple[str, ...]] = (
|
||||
"langfuse_public_key",
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ from litellm._uuid import uuid
|
|||
from litellm.constants import DEFAULT_STAGGER_WINDOW_SECONDS, MCP_STDIO_ALLOWED_COMMANDS
|
||||
from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
|
||||
validate_arize_otlp_protocol_value,
|
||||
validate_capture_message_content_value,
|
||||
validate_langfuse_environment_value,
|
||||
validate_langfuse_span_scope_value,
|
||||
validate_no_callback_env_reference,
|
||||
|
|
@ -79,6 +80,7 @@ from litellm.types.router import AllowedModelRegion, RouterErrors, UpdateRouterC
|
|||
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,
|
||||
AzureSpillover,
|
||||
CallTypes,
|
||||
CostBreakdown,
|
||||
|
|
@ -2202,7 +2204,7 @@ class AddTeamCallback(LiteLLMPydanticObjectBase):
|
|||
@classmethod
|
||||
def validate_callback_vars(cls, values):
|
||||
callback_vars: Final = values.get("callback_vars", {})
|
||||
valid_keys: Final = set(StandardCallbackDynamicParams.__annotations__.keys())
|
||||
valid_keys: Final = {*StandardCallbackDynamicParams.__annotations__, CAPTURE_MESSAGE_CONTENT_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}")
|
||||
|
|
@ -2214,6 +2216,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase):
|
|||
validate_langfuse_span_scope_value(callback_vars[key])
|
||||
if key == "arize_otlp_protocol":
|
||||
validate_arize_otlp_protocol_value(callback_vars[key])
|
||||
if key == CAPTURE_MESSAGE_CONTENT_VAR:
|
||||
validate_capture_message_content_value(callback_vars[key])
|
||||
return values
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -10,6 +10,8 @@ from collections.abc import Mapping, Sequence
|
|||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VAR
|
||||
|
||||
_NEWRELIC_CALLBACK: Final = "newrelic"
|
||||
_NEWRELIC_VAR_PREFIX: Final = "newrelic_"
|
||||
_LANGFUSE_OTEL_CALLBACK: Final = "langfuse_otel"
|
||||
|
|
@ -36,6 +38,9 @@ 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
|
||||
if callback_name != _NEWRELIC_CALLBACK:
|
||||
return None
|
||||
return _newrelic_config_error(callback_vars)
|
||||
|
|
@ -80,6 +85,32 @@ 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
|
||||
) -> str | None:
|
||||
value: Final = callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)
|
||||
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}"
|
||||
if not is_otel_v2_enabled():
|
||||
return f"Per-destination {CAPTURE_MESSAGE_CONTENT_VAR} requires the proxy to run with LITELLM_OTEL_V2=true."
|
||||
try:
|
||||
validate_capture_message_content_value(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 None
|
||||
|
||||
|
||||
# Which credential family a dynamic variable belongs to. The families are the
|
||||
# integrations that share one account: every langfuse_* variable configures the
|
||||
# same Langfuse project whether it rides the classic callback or the OTel one,
|
||||
|
|
@ -190,20 +221,41 @@ def conflicting_span_scope_error(
|
|||
)
|
||||
|
||||
|
||||
def conflicting_capture_error(
|
||||
callback_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)
|
||||
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"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)
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
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:
|
||||
return None
|
||||
entries: Final = metadata.get("logging")
|
||||
if not isinstance(entries, Sequence) or isinstance(entries, (str, bytes)):
|
||||
entries: Final = _object_sequence(metadata.get("logging"))
|
||||
if entries is None:
|
||||
return None
|
||||
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))),
|
||||
)
|
||||
if error is not None
|
||||
),
|
||||
|
|
@ -211,6 +263,31 @@ 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``."""
|
||||
entries: Final = _object_sequence(logging_entries) or ()
|
||||
return tuple((_entry_callback_name(entry), _entry_callback_vars(entry)) for entry in entries)
|
||||
|
||||
|
||||
def _object_sequence(value: object) -> Sequence[object] | None:
|
||||
return value if isinstance(value, Sequence) and not isinstance(value, (str, bytes)) else None
|
||||
|
||||
|
||||
def _entry_callback_name(entry: object) -> str | None:
|
||||
match entry:
|
||||
case {"callback_name": str() as callback_name}:
|
||||
return callback_name
|
||||
case _:
|
||||
return None
|
||||
|
||||
|
||||
def _conflicting_entry_capture_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)
|
||||
|
||||
|
||||
def _entry_callback_vars(entry: object) -> Mapping[str, str]:
|
||||
callback_vars: Final = entry.get("callback_vars") if isinstance(entry, Mapping) else None
|
||||
if not isinstance(callback_vars, Mapping):
|
||||
|
|
|
|||
|
|
@ -227,6 +227,7 @@ from litellm.types.litellm_params import AGENTIC_LOOP_KWARG_NAMES
|
|||
from litellm.types.llms.anthropic import ANTHROPIC_API_HEADERS
|
||||
from litellm.types.services import ServiceTypes
|
||||
from litellm.types.utils import (
|
||||
CAPTURE_MESSAGE_CONTENT_VAR,
|
||||
CustomPricingLiteLLMParams,
|
||||
LlmProviders,
|
||||
ProviderSpecificHeader,
|
||||
|
|
@ -975,6 +976,8 @@ 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:
|
||||
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
|
||||
# put newrelic_* under a different callback never asked for New Relic and
|
||||
|
|
@ -1162,7 +1165,6 @@ def resolve_tenant_otel_destinations(
|
|||
span still routes to the tenant's credentials the way it did then.
|
||||
"""
|
||||
from litellm.integrations.otel.model.config import is_otel_v2_enabled
|
||||
from litellm.integrations.otel.presets.destinations import destination_for
|
||||
|
||||
if not is_otel_v2_enabled():
|
||||
return ()
|
||||
|
|
@ -1186,23 +1188,28 @@ def resolve_tenant_otel_destinations(
|
|||
for name in dict.fromkeys(
|
||||
callback.callback_name for callback in callbacks if callback.callback_type != "failure"
|
||||
)
|
||||
if (
|
||||
destination := destination_for(
|
||||
name,
|
||||
_tenant_otel_params(
|
||||
MappingProxyType(
|
||||
{
|
||||
var: value
|
||||
for callback in callbacks
|
||||
if callback.callback_name == name
|
||||
for var, value in callback.callback_vars.items()
|
||||
}
|
||||
)
|
||||
),
|
||||
_tenant_service_name(user_api_key_dict),
|
||||
if (destination := _tenant_destination(name, callbacks, _tenant_service_name(user_api_key_dict))) is not None
|
||||
)
|
||||
|
||||
|
||||
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.presets.destinations import destination_for
|
||||
|
||||
callback_vars: Final = MappingProxyType(
|
||||
dict(
|
||||
itertools.chain.from_iterable(
|
||||
callback.callback_vars.items() for callback in callbacks if callback.callback_name == name
|
||||
)
|
||||
)
|
||||
is not None
|
||||
)
|
||||
return destination_for(
|
||||
name,
|
||||
_tenant_otel_params(callback_vars),
|
||||
service_name,
|
||||
parse_capture_message_content(callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)),
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -31,8 +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_span_scope_error,
|
||||
cross_entry_family_error,
|
||||
stored_capture_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
|
||||
|
|
@ -288,6 +290,9 @@ async def add_team_callbacks(
|
|||
- langfuse_host: The host for the Langfuse callback
|
||||
- langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)
|
||||
- langfuse_span_scope: For langfuse_otel, "full" (default) sends the whole request trace, "llm_only" sends only the model-call spans
|
||||
- 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
|
||||
- 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
|
||||
|
|
@ -351,11 +356,19 @@ 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_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,
|
||||
)
|
||||
if capture_error is not None:
|
||||
raise _callback_config_error(capture_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
|
||||
|
|
|
|||
|
|
@ -3726,6 +3726,28 @@ ArizeOtlpProtocol = Literal["grpc", "http/protobuf"]
|
|||
ARIZE_OTLP_PROTOCOLS: Final[frozenset[str]] = frozenset(get_args(ArizeOtlpProtocol))
|
||||
|
||||
|
||||
CAPTURE_MESSAGE_CONTENT_VAR: Final = "capture_message_content"
|
||||
|
||||
|
||||
class CaptureMessageContent(str):
|
||||
NO_CONTENT = "no_content"
|
||||
SPAN_ONLY = "span_only"
|
||||
EVENT_ONLY = "event_only"
|
||||
SPAN_AND_EVENT = "span_and_event"
|
||||
|
||||
|
||||
# The values a team or key destination may set. OTel v2 writes content only to span attributes,
|
||||
# so the event modes stay global-only settings
|
||||
CAPTURE_MESSAGE_CONTENT_VALUES: Final[frozenset[str]] = frozenset(
|
||||
{CaptureMessageContent.NO_CONTENT, CaptureMessageContent.SPAN_ONLY}
|
||||
)
|
||||
|
||||
|
||||
def captures_span_content(mode: str | None) -> bool:
|
||||
"""Whether a capture mode puts prompt and response content on spans"""
|
||||
return mode in (CaptureMessageContent.SPAN_ONLY, CaptureMessageContent.SPAN_AND_EVENT)
|
||||
|
||||
|
||||
class StandardCallbackDynamicParams(TypedDict, total=False):
|
||||
# Langfuse dynamic params
|
||||
langfuse_public_key: str | None
|
||||
|
|
|
|||
|
|
@ -186,6 +186,7 @@ class _Handler(BaseHTTPRequestHandler):
|
|||
"count": len(recorded),
|
||||
"host": self.headers.get("host", ""),
|
||||
"headers": dict(self.headers),
|
||||
"span_ids": [span["span_id"] for span in recorded],
|
||||
}
|
||||
)
|
||||
self._send_json({"recorded": len(recorded)}, status=self.state.status)
|
||||
|
|
@ -466,9 +467,7 @@ def _await_sink(url: str) -> None:
|
|||
def owned_grpc_sink(directory: Path) -> Iterator[GrpcSink]:
|
||||
http_port: Final = _free_port()
|
||||
grpc_port: Final = _free_port()
|
||||
with _spawn_sink(
|
||||
directory, "otlp-grpc-sink.log", ["--port", str(http_port), "--grpc-port", str(grpc_port)]
|
||||
):
|
||||
with _spawn_sink(directory, "otlp-grpc-sink.log", ["--port", str(http_port), "--grpc-port", str(grpc_port)]):
|
||||
control_url: Final = f"http://127.0.0.1:{http_port}"
|
||||
_await_sink(control_url)
|
||||
yield GrpcSink(url=f"http://127.0.0.1:{grpc_port}", control_url=control_url)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,720 @@
|
|||
import json
|
||||
import re
|
||||
import uuid
|
||||
from collections.abc import Generator, Iterator, Mapping
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final, Literal
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import yaml
|
||||
from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment, object_value
|
||||
from integration._support.otlp_sink import (
|
||||
ConnectSink,
|
||||
Span,
|
||||
SpanSinks,
|
||||
configure_sink,
|
||||
owned_connect_sink,
|
||||
recorded_requests,
|
||||
recorded_spans,
|
||||
)
|
||||
from integration._support.process import OwnedProxy, owned_proxy_process
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
from pydantic import JsonValue, TypeAdapter
|
||||
|
||||
MARKER: Final = re.compile(rb"cmc-(?:tool-|fail-)?[0-9a-f]{32}")
|
||||
JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
|
||||
SERVER: Final = 2
|
||||
CALL_ID: Final = "litellm.call_id"
|
||||
OPERATION: Final = "gen_ai.operation.name"
|
||||
INPUT_TOKENS: Final = "gen_ai.usage.input_tokens"
|
||||
Endpoint = Literal["chat", "responses", "messages"]
|
||||
Capture = Literal["no_content", "span_only"]
|
||||
|
||||
|
||||
def _marker(kind: str = "") -> str:
|
||||
return f"cmc-{kind}{uuid.uuid4().hex}"
|
||||
|
||||
|
||||
def _usage() -> dict[str, JsonValue]:
|
||||
return {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}
|
||||
|
||||
|
||||
def _chat_reply(marker: str, stream: bool) -> Reply:
|
||||
identity: Final = f"chatcmpl-{uuid.uuid4().hex}"
|
||||
if marker.startswith("cmc-tool-"):
|
||||
call: Final[dict[str, JsonValue]] = {
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": {"name": "lookup", "arguments": json.dumps({"query": marker})},
|
||||
}
|
||||
message: Final[dict[str, JsonValue]] = {"role": "assistant", "content": None, "tool_calls": [call]}
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": identity,
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o-mini",
|
||||
"choices": [{"index": 0, "message": message, "finish_reason": "tool_calls"}],
|
||||
"usage": _usage(),
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
if not stream:
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": identity,
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o-mini",
|
||||
"choices": [
|
||||
{"index": 0, "message": {"role": "assistant", "content": marker}, "finish_reason": "stop"}
|
||||
],
|
||||
"usage": _usage(),
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
chunk: Final = {"id": identity, "object": "chat.completion.chunk", "created": 1, "model": "gpt-4o-mini"}
|
||||
deltas: Final[tuple[dict[str, JsonValue], ...]] = (
|
||||
{**chunk, "choices": [{"index": 0, "delta": {"role": "assistant", "content": marker}}]},
|
||||
{**chunk, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
|
||||
{**chunk, "choices": [], "usage": _usage()},
|
||||
)
|
||||
return Reply(
|
||||
content_type="text/event-stream",
|
||||
chunks=(*(b"data: " + json.dumps(delta).encode() + b"\n\n" for delta in deltas), b"data: [DONE]\n\n"),
|
||||
)
|
||||
|
||||
|
||||
def _responses_reply(marker: str, stream: bool) -> Reply:
|
||||
identity: Final = uuid.uuid4().hex
|
||||
response: Final[dict[str, JsonValue]] = {
|
||||
"id": f"resp_{identity}",
|
||||
"object": "response",
|
||||
"created_at": 1,
|
||||
"status": "completed",
|
||||
"model": "gpt-4o-mini",
|
||||
"output": [
|
||||
{
|
||||
"id": f"msg_{identity}",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"status": "completed",
|
||||
"content": [{"type": "output_text", "text": marker, "annotations": []}],
|
||||
}
|
||||
],
|
||||
"usage": {"input_tokens": 7, "output_tokens": 2, "total_tokens": 9},
|
||||
}
|
||||
if not stream:
|
||||
return Reply(body=json.dumps(response).encode())
|
||||
events: Final[tuple[dict[str, JsonValue], ...]] = (
|
||||
{
|
||||
"type": "response.created",
|
||||
"sequence_number": 0,
|
||||
"response": {**response, "status": "in_progress", "output": []},
|
||||
},
|
||||
{
|
||||
"type": "response.output_text.delta",
|
||||
"sequence_number": 1,
|
||||
"item_id": f"msg_{identity}",
|
||||
"output_index": 0,
|
||||
"content_index": 0,
|
||||
"delta": marker,
|
||||
},
|
||||
{"type": "response.completed", "sequence_number": 2, "response": response},
|
||||
)
|
||||
return Reply(
|
||||
content_type="text/event-stream",
|
||||
chunks=tuple(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() for event in events),
|
||||
)
|
||||
|
||||
|
||||
def _upstream(request: Request) -> Reply:
|
||||
found: Final = MARKER.search(request.body)
|
||||
if found is None:
|
||||
return Reply(status=404, body=b'{"error":"no marker"}')
|
||||
marker: Final = found.group(0).decode()
|
||||
if marker.startswith("cmc-fail-"):
|
||||
return Reply(status=500, body=b'{"error":{"message":"scripted upstream failure","type":"server_error"}}')
|
||||
stream: Final = object_value(JSON.validate_json(request.body)).get("stream") is True
|
||||
if request.target.endswith("/responses"):
|
||||
return _responses_reply(marker, stream)
|
||||
return _chat_reply(marker, stream)
|
||||
|
||||
|
||||
def _body(model: str, endpoint: Endpoint, marker: str, stream: bool) -> tuple[str, dict[str, JsonValue]]:
|
||||
if endpoint == "responses":
|
||||
return "/v1/responses", {"model": model, "input": marker, "stream": stream}
|
||||
messages: Final[list[JsonValue]] = [{"role": "user", "content": marker}]
|
||||
if endpoint == "messages":
|
||||
return "/v1/messages", {"model": model, "max_tokens": 16, "messages": messages, "stream": stream}
|
||||
return "/v1/chat/completions", {"model": model, "messages": messages, "stream": stream}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Sent:
|
||||
marker: str
|
||||
trace_id: str
|
||||
call_id: str
|
||||
status: int
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Cursors:
|
||||
operator: int
|
||||
tenant: int
|
||||
arize: int
|
||||
newrelic: int
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Rig:
|
||||
proxy: Gateway
|
||||
owned: OwnedProxy
|
||||
scenario: Scenario
|
||||
model: str
|
||||
upstream: Wire
|
||||
sinks: SpanSinks
|
||||
newrelic: ConnectSink
|
||||
|
||||
def cursors(self) -> Cursors:
|
||||
self.upstream.drain()
|
||||
return Cursors(
|
||||
recorded_spans(self.sinks.operator)[0],
|
||||
recorded_spans(self.sinks.tenant)[0],
|
||||
recorded_spans(self.sinks.arize)[0],
|
||||
recorded_spans(self.newrelic.control_url)[0],
|
||||
)
|
||||
|
||||
def upstream_hits(self, marker: str) -> int:
|
||||
return sum(1 for request in self.upstream.drain() if marker.encode() in request.body)
|
||||
|
||||
def langfuse(self, name: str, host: str | None = None) -> dict[str, str]:
|
||||
return {
|
||||
"langfuse_public_key": f"pk-lf-{name}",
|
||||
"langfuse_secret_key": f"sk-lf-{name}",
|
||||
"langfuse_host": host or self.sinks.tenant,
|
||||
}
|
||||
|
||||
def attach(self, team: str, callback: str, variables: Mapping[str, str], kind: str | None = None) -> httpx.Response:
|
||||
body: Final[dict[str, JsonValue]] = {"callback_name": callback, "callback_vars": dict(variables)}
|
||||
return self.proxy.request(
|
||||
"POST", f"/team/{team}/callback", body if kind is None else {**body, "callback_type": kind}
|
||||
)
|
||||
|
||||
def team_key(
|
||||
self,
|
||||
capture: Capture | None,
|
||||
name: str,
|
||||
*,
|
||||
callback: str = "langfuse_otel",
|
||||
variables: Mapping[str, str] | None = None,
|
||||
kind: str | None = None,
|
||||
) -> tuple[str, str]:
|
||||
team: Final = self.scenario.team()
|
||||
base: Final = dict(variables) if variables is not None else self.langfuse(name)
|
||||
attached: Final = self.attach(
|
||||
team, callback, base if capture is None else {**base, "capture_message_content": capture}, kind
|
||||
)
|
||||
assert attached.status_code == 200, attached.text
|
||||
return team, self.scenario.key(team_id=team)
|
||||
|
||||
def send(
|
||||
self,
|
||||
key: str,
|
||||
endpoint: Endpoint = "chat",
|
||||
stream: bool = False,
|
||||
*,
|
||||
marker: str | None = None,
|
||||
extra: Mapping[str, JsonValue] | None = None,
|
||||
) -> Sent:
|
||||
tagged: Final = marker or _marker()
|
||||
path, body = _body(self.model, endpoint, tagged, stream)
|
||||
trace_id: Final = uuid.uuid4().hex
|
||||
call_id: Final = str(uuid.uuid4())
|
||||
headers: Final = {
|
||||
"Authorization": f"Bearer {key}",
|
||||
"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01",
|
||||
"x-litellm-call-id": call_id,
|
||||
}
|
||||
with self.proxy.client.stream("POST", path, json={**body, **(extra or {})}, headers=headers) as response:
|
||||
text: Final = response.read().decode()
|
||||
return Sent(tagged, trace_id, call_id, response.status_code, text)
|
||||
|
||||
|
||||
def _served(sent: Sent) -> Sent:
|
||||
assert sent.status == 200, sent.text
|
||||
assert sent.marker in sent.text, sent.text
|
||||
return sent
|
||||
|
||||
|
||||
Credential = tuple[str, str]
|
||||
|
||||
|
||||
def _sent_with(url: str, credential: Credential) -> frozenset[str]:
|
||||
"""Span ids ``url`` received in export requests whose ``credential`` header matched."""
|
||||
header, value = credential
|
||||
return frozenset(
|
||||
span_id
|
||||
for request in recorded_requests(url)
|
||||
if {name.lower(): seen for name, seen in object_value(request.get("headers") or {}).items()}.get(header)
|
||||
== value
|
||||
for span_id in request.get("span_ids") or ()
|
||||
if isinstance(span_id, str)
|
||||
)
|
||||
|
||||
|
||||
def _of_request(url: str, since: int, sent: Sent, credential: Credential | None = None) -> tuple[Span, ...]:
|
||||
"""Spans of ``sent``'s trace at ``url``, narrowed to one account's export requests when ``credential`` is set."""
|
||||
spans: Final = recorded_spans(url, since)[1]
|
||||
traces: Final = {sent.trace_id} | {
|
||||
span["trace_id"] for span in spans if span["attributes"].get(CALL_ID) == sent.call_id
|
||||
}
|
||||
owned: Final = None if credential is None else _sent_with(url, credential)
|
||||
return tuple(span for span in spans if span["trace_id"] in traces and (owned is None or span["span_id"] in owned))
|
||||
|
||||
|
||||
def _carries(span: Span, marker: str) -> bool:
|
||||
return marker in json.dumps(span["attributes"])
|
||||
|
||||
|
||||
def _model_spans(spans: tuple[Span, ...]) -> tuple[Span, ...]:
|
||||
return tuple(span for span in spans if OPERATION in span["attributes"])
|
||||
|
||||
|
||||
def _names(spans: tuple[Span, ...]) -> list[str]:
|
||||
return sorted(span["name"] for span in spans)
|
||||
|
||||
|
||||
def _with_content(url: str, since: int, sent: Sent) -> tuple[Span, ...]:
|
||||
"""The copy of ``sent`` at ``url`` once its model span arrived and some span carries the marker."""
|
||||
return eventually(
|
||||
lambda: _of_request(url, since, sent),
|
||||
lambda spans: bool(_model_spans(spans)) and any(_carries(span, sent.marker) for span in spans),
|
||||
seconds=40,
|
||||
)
|
||||
|
||||
|
||||
def _arrived(
|
||||
url: str, since: int, sent: Sent, expected: frozenset[str], credential: Credential | None = None
|
||||
) -> tuple[Span, ...]:
|
||||
"""The copy of ``sent`` at ``url`` once every span name in ``expected`` and a usage-bearing model span landed."""
|
||||
spans: Final = eventually(
|
||||
lambda: _of_request(url, since, sent, credential),
|
||||
lambda found: (
|
||||
expected <= {span["name"] for span in found}
|
||||
and any(INPUT_TOKENS in span["attributes"] for span in _model_spans(found))
|
||||
),
|
||||
seconds=40,
|
||||
)
|
||||
assert all(
|
||||
span["attributes"].get(INPUT_TOKENS) == 7 for span in _model_spans(spans) if INPUT_TOKENS in span["attributes"]
|
||||
)
|
||||
return spans
|
||||
|
||||
|
||||
def _assert_redacted_twin(
|
||||
rig: Rig, sent: Sent, cursors: Cursors, sink: str, since: int, credential: Credential | None = None
|
||||
) -> tuple[Span, ...]:
|
||||
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
|
||||
carried: Final = frozenset(span["name"] for span in operator if _carries(span, sent.marker))
|
||||
tenant: Final = _arrived(sink, since, sent, carried, credential)
|
||||
leaked: Final = [span["name"] for span in tenant if _carries(span, sent.marker)]
|
||||
assert leaked == [], f"{leaked} carried content; tenant saw {_names(tenant)}"
|
||||
model: Final = _model_spans(tenant)[0]["attributes"]
|
||||
assert "gen_ai.input.messages" not in model and "gen_ai.output.messages" not in model, sorted(model)
|
||||
assert model.get("gen_ai.request.model"), sorted(model)
|
||||
return tenant
|
||||
|
||||
|
||||
def _assert_content_twin(rig: Rig, sent: Sent, cursors: Cursors) -> None:
|
||||
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
|
||||
carried: Final = frozenset(span["name"] for span in operator if _carries(span, sent.marker))
|
||||
tenant: Final = _arrived(rig.sinks.tenant, cursors.tenant, sent, carried)
|
||||
kept: Final = frozenset(span["name"] for span in tenant if _carries(span, sent.marker))
|
||||
assert carried <= kept, f"content lost on {sorted(carried - kept)}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("endpoint", "stream"),
|
||||
[("chat", False), ("chat", True), ("responses", False), ("messages", True)],
|
||||
ids=["chat-unary", "chat-stream", "responses-unary", "messages-stream"],
|
||||
)
|
||||
def test_no_content_team_keeps_its_trace_and_metadata_without_content(
|
||||
rig: Rig, endpoint: Endpoint, stream: bool
|
||||
) -> None:
|
||||
_, key = rig.team_key("no_content", "a")
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key, endpoint, stream))
|
||||
assert rig.upstream_hits(sent.marker) == 1
|
||||
_assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("capture", ["span_only", None], ids=["span_only", "omitted"])
|
||||
def test_span_only_and_omitted_teams_keep_the_globally_captured_content(rig: Rig, capture: Capture | None) -> None:
|
||||
_, key = rig.team_key(capture, "b")
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key))
|
||||
_assert_content_twin(rig, sent, cursors)
|
||||
|
||||
|
||||
def test_tool_call_arguments_never_reach_a_no_content_team(rig: Rig) -> None:
|
||||
_, key = rig.team_key("no_content", "a")
|
||||
cursors: Final = rig.cursors()
|
||||
tool: Final[dict[str, JsonValue]] = {
|
||||
"type": "function",
|
||||
"function": {"name": "lookup", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}},
|
||||
}
|
||||
sent: Final = rig.send(key, marker=_marker("tool-"), extra={"tools": [tool]})
|
||||
assert sent.status == 200 and sent.marker in sent.text, sent.text
|
||||
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
|
||||
assert any(sent.marker in str(span["attributes"].get("gen_ai.output.messages", "")) for span in operator), (
|
||||
"operator copy lacks the tool-call arguments"
|
||||
)
|
||||
tenant: Final = _assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
|
||||
assert "lookup" in json.dumps([span["attributes"] for span in tenant]), "tool definitions are metadata and stay"
|
||||
|
||||
|
||||
def test_cache_hit_twin_of_a_no_content_team_is_redacted(rig: Rig) -> None:
|
||||
_, key = rig.team_key("no_content", "a")
|
||||
marker: Final = _marker()
|
||||
_served(rig.send(key, marker=marker))
|
||||
assert rig.upstream_hits(marker) == 1
|
||||
cursors: Final = rig.cursors()
|
||||
hit: Final = eventually(
|
||||
lambda: _served(rig.send(key, marker=marker)), lambda _: rig.upstream_hits(marker) == 0, seconds=20
|
||||
)
|
||||
operator: Final = eventually(
|
||||
lambda: _of_request(rig.sinks.operator, cursors.operator, hit),
|
||||
lambda spans: any(span["kind"] == SERVER for span in spans),
|
||||
seconds=40,
|
||||
)
|
||||
expected: Final = frozenset(span["name"] for span in operator if span["kind"] == SERVER)
|
||||
tenant: Final = eventually(
|
||||
lambda: _of_request(rig.sinks.tenant, cursors.tenant, hit),
|
||||
lambda spans: expected <= {span["name"] for span in spans},
|
||||
seconds=40,
|
||||
)
|
||||
assert [span["name"] for span in tenant if _carries(span, marker)] == [], _names(tenant)
|
||||
|
||||
|
||||
def test_a_failure_only_entry_rejects_the_setting(rig: Rig) -> None:
|
||||
team: Final = rig.scenario.team()
|
||||
response: Final = rig.attach(
|
||||
team, "langfuse_otel", {**rig.langfuse("fail"), "capture_message_content": "no_content"}, "failure"
|
||||
)
|
||||
assert response.status_code == 400, response.text
|
||||
assert "success_and_failure" in response.text, response.text
|
||||
|
||||
|
||||
def test_key_level_logging_with_no_content_is_redacted(rig: Rig) -> None:
|
||||
logging: Final[list[JsonValue]] = [
|
||||
{
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": "success",
|
||||
"callback_vars": {**rig.langfuse("key"), "capture_message_content": "no_content"},
|
||||
}
|
||||
]
|
||||
key: Final = rig.scenario.key(metadata={"logging": logging})
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key))
|
||||
_assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
|
||||
|
||||
|
||||
def test_same_account_destination_redacts_the_shared_copy_only_for_that_team(rig: Rig) -> None:
|
||||
_, restricted = rig.team_key("no_content", "operator", variables=rig.langfuse("operator", host=rig.sinks.operator))
|
||||
_, other = rig.team_key("span_only", "b")
|
||||
cursors: Final = rig.cursors()
|
||||
shared: Final = _served(rig.send(restricted))
|
||||
assert rig.upstream_hits(shared.marker) == 1
|
||||
copy: Final = eventually(
|
||||
lambda: _of_request(rig.sinks.operator, cursors.operator, shared),
|
||||
lambda spans: any(INPUT_TOKENS in span["attributes"] for span in _model_spans(spans)),
|
||||
seconds=40,
|
||||
)
|
||||
assert [span["name"] for span in copy if _carries(span, shared.marker)] == [], _names(copy)
|
||||
assert len([span for span in copy if span["kind"] == SERVER and span["name"].startswith("POST ")]) == 1, _names(
|
||||
copy
|
||||
)
|
||||
_assert_content_twin(rig, _served(rig.send(other)), cursors)
|
||||
|
||||
|
||||
def test_arize_destination_drops_openinference_content(rig: Rig) -> None:
|
||||
_, key = rig.team_key(
|
||||
"no_content", "arize", callback="arize", variables={"arize_space_key": "space-a", "arize_api_key": "arize-a"}
|
||||
)
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key))
|
||||
operator: Final = _with_content(rig.sinks.operator, cursors.operator, sent)
|
||||
assert any(sent.marker in str(span["attributes"].get("input.value", "")) for span in operator), (
|
||||
"operator copy has no OpenInference content to strip"
|
||||
)
|
||||
tenant: Final = _assert_redacted_twin(
|
||||
rig, sent, cursors, rig.sinks.arize, cursors.arize, ("arize-space-id", "space-a")
|
||||
)
|
||||
model: Final = _model_spans(tenant)[0]["attributes"]
|
||||
assert not any(name.startswith(("llm.input_messages.", "llm.output_messages.")) for name in model), sorted(model)
|
||||
assert "input.value" not in model and "output.value" not in model, sorted(model)
|
||||
assert model.get("llm.model_name"), sorted(model)
|
||||
|
||||
|
||||
def test_weave_destination_drops_its_content(rig: Rig) -> None:
|
||||
_, key = rig.team_key(
|
||||
"no_content",
|
||||
"weave",
|
||||
callback="weave_otel",
|
||||
variables={"wandb_api_key": "wandb-a", "weave_project_id": "team/a"},
|
||||
)
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key))
|
||||
operator: Final = eventually(
|
||||
lambda: _of_request(rig.sinks.arize, cursors.arize, sent, ("project_id", "operator/weave")),
|
||||
lambda spans: any(sent.marker in str(span["attributes"].get("weave.output", "")) for span in spans),
|
||||
seconds=40,
|
||||
)
|
||||
assert operator, "operator weave copy has no weave.output to strip"
|
||||
tenant: Final = _assert_redacted_twin(rig, sent, cursors, rig.sinks.arize, cursors.arize, ("project_id", "team/a"))
|
||||
assert not any("weave.output" in span["attributes"] for span in tenant), _names(tenant)
|
||||
|
||||
|
||||
def test_newrelic_destination_drops_genai_content(rig: Rig) -> None:
|
||||
_, key = rig.team_key("no_content", "newrelic", callback="newrelic", variables={"newrelic_api_key": "nr-team-a"})
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key))
|
||||
tenant: Final = _assert_redacted_twin(
|
||||
rig, sent, cursors, rig.newrelic.control_url, cursors.newrelic, ("api-key", "nr-team-a")
|
||||
)
|
||||
hosts: Final = {
|
||||
object_value(request.get("headers") or {}).get("Host")
|
||||
for request in recorded_requests(rig.newrelic.control_url)
|
||||
if object_value(request.get("headers") or {}).get("api-key") == "nr-team-a"
|
||||
}
|
||||
assert hosts == {"otlp.nr-data.net"}, hosts
|
||||
assert tenant
|
||||
|
||||
|
||||
def test_span_only_team_lifts_global_no_content_only_for_its_destination(dark_rig: Rig) -> None:
|
||||
"""The sibling Arize entry gets a fan-out destination and the SigNoz entry rides the routed tracer."""
|
||||
team, key = dark_rig.team_key("span_only", "b")
|
||||
siblings: Final = (
|
||||
dark_rig.attach(team, "arize", {"arize_space_key": "space-sibling", "arize_api_key": "arize-sibling"}),
|
||||
dark_rig.attach(
|
||||
team,
|
||||
"signoz",
|
||||
{"signoz_ingestion_endpoint": dark_rig.sinks.arize + "/v1/traces", "signoz_ingestion_key": "signoz-team"},
|
||||
),
|
||||
)
|
||||
assert [response.status_code for response in siblings] == [200, 200], [response.text for response in siblings]
|
||||
cursors: Final = dark_rig.cursors()
|
||||
sent: Final = _served(dark_rig.send(key))
|
||||
assert dark_rig.upstream_hits(sent.marker) == 1
|
||||
operator: Final = _arrived(dark_rig.sinks.operator, cursors.operator, sent, frozenset())
|
||||
tenant: Final = _with_content(dark_rig.sinks.tenant, cursors.tenant, sent)
|
||||
sibling: Final = _arrived(
|
||||
dark_rig.sinks.arize,
|
||||
cursors.arize,
|
||||
sent,
|
||||
frozenset(),
|
||||
("arize-space-id", "space-sibling"),
|
||||
)
|
||||
signoz: Final = _arrived(
|
||||
dark_rig.sinks.arize,
|
||||
cursors.arize,
|
||||
sent,
|
||||
frozenset(),
|
||||
("signoz-ingestion-key", "signoz-team"),
|
||||
)
|
||||
assert any(_carries(span, sent.marker) for span in tenant), _names(tenant)
|
||||
assert not any(_carries(span, sent.marker) for span in operator), _names(operator)
|
||||
assert not any(_carries(span, sent.marker) for span in sibling), _names(sibling)
|
||||
assert not any(_carries(span, sent.marker) for span in signoz), _names(signoz)
|
||||
|
||||
|
||||
def test_an_unsupported_value_fails_registration(rig: Rig) -> None:
|
||||
team: Final = rig.scenario.team()
|
||||
response: Final = rig.attach(team, "langfuse_otel", {**rig.langfuse("a"), "capture_message_content": "bogus"})
|
||||
assert response.status_code == 422, response.text
|
||||
assert "Invalid capture_message_content 'bogus'" in response.text, response.text
|
||||
|
||||
|
||||
def test_classic_langfuse_rejects_the_setting(rig: Rig) -> None:
|
||||
team: Final = rig.scenario.team()
|
||||
response: Final = rig.attach(team, "langfuse", {**rig.langfuse("a"), "capture_message_content": "no_content"})
|
||||
assert response.status_code == 400, response.text
|
||||
assert "capture_message_content" in response.text, response.text
|
||||
|
||||
|
||||
def test_a_second_entry_for_the_same_backend_cannot_disagree(rig: Rig) -> None:
|
||||
team, _ = rig.team_key("no_content", "a")
|
||||
response: Final = rig.attach(team, "langfuse_otel", {"capture_message_content": "span_only"}, "success_and_failure")
|
||||
assert response.status_code == 400, response.text
|
||||
assert "already set to 'no_content' by another langfuse_otel entry" in response.text, response.text
|
||||
|
||||
|
||||
def test_replacing_the_registration_switches_the_team_to_content(rig: Rig) -> None:
|
||||
team, key = rig.team_key("no_content", "a")
|
||||
cursors: Final = rig.cursors()
|
||||
_assert_redacted_twin(rig, _served(rig.send(key)), cursors, rig.sinks.tenant, cursors.tenant)
|
||||
removed: Final = rig.proxy.request("DELETE", f"/team/{team}/callback/langfuse_otel")
|
||||
assert removed.status_code == 200, removed.text
|
||||
replaced: Final = rig.attach(team, "langfuse_otel", {**rig.langfuse("a"), "capture_message_content": "span_only"})
|
||||
assert replaced.status_code == 200, replaced.text
|
||||
|
||||
def attempt() -> bool:
|
||||
since: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key))
|
||||
operator: Final = _with_content(rig.sinks.operator, since.operator, sent)
|
||||
carried: Final = frozenset(span["name"] for span in operator if _carries(span, sent.marker))
|
||||
tenant: Final = _arrived(rig.sinks.tenant, since.tenant, sent, carried)
|
||||
return any(_carries(span, sent.marker) for span in tenant)
|
||||
|
||||
assert eventually(attempt, bool, seconds=60)
|
||||
|
||||
|
||||
def test_request_metadata_cannot_lift_the_team_restriction(rig: Rig) -> None:
|
||||
_, key = rig.team_key("no_content", "a")
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key, extra={"metadata": {"capture_message_content": "span_only"}}))
|
||||
assert rig.upstream_hits(sent.marker) == 1
|
||||
_assert_redacted_twin(rig, sent, cursors, rig.sinks.tenant, cursors.tenant)
|
||||
|
||||
|
||||
def test_stalled_tenant_during_a_mixed_burst_keeps_serving_and_each_teams_policy(rig: Rig) -> None:
|
||||
_, restricted = rig.team_key("no_content", "a")
|
||||
_, open_key = rig.team_key("span_only", "b")
|
||||
shapes: Final[tuple[tuple[Endpoint, bool], ...]] = (
|
||||
("chat", False),
|
||||
("chat", True),
|
||||
("responses", False),
|
||||
("messages", True),
|
||||
)
|
||||
plan: Final = [(restricted if index % 2 else open_key, *shapes[index % len(shapes)]) for index in range(30)]
|
||||
cursors: Final = rig.cursors()
|
||||
configure_sink(rig.sinks.tenant, paused=True)
|
||||
try:
|
||||
with ThreadPoolExecutor(max_workers=10) as pool:
|
||||
sent: Final = list(pool.map(lambda job: rig.send(job[0], job[1], job[2]), plan))
|
||||
finally:
|
||||
configure_sink(rig.sinks.tenant, paused=False)
|
||||
assert [result.status for result in sent] == [200] * 30, [result.text for result in sent if result.status != 200]
|
||||
for (key, _, _), result in zip(plan, sent, strict=True):
|
||||
roots = [
|
||||
span
|
||||
for span in _with_content(rig.sinks.operator, cursors.operator, result)
|
||||
if span["kind"] == SERVER and span["name"].startswith("POST ")
|
||||
]
|
||||
assert len(roots) == 1, roots
|
||||
if key == restricted:
|
||||
_assert_redacted_twin(rig, result, cursors, rig.sinks.tenant, cursors.tenant)
|
||||
else:
|
||||
_assert_content_twin(rig, result, cursors)
|
||||
|
||||
|
||||
def _write_config(directory: Path, sinks: SpanSinks, name: str, callbacks: tuple[str, ...]) -> Path:
|
||||
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
|
||||
config["litellm_settings"] = {
|
||||
**config.get("litellm_settings", {}),
|
||||
"callbacks": list(callbacks),
|
||||
"otel_tenant_destination_mode": "additive",
|
||||
"provider_url_destination_allowed_hosts": [
|
||||
urlparse(sinks.tenant).netloc,
|
||||
urlparse(sinks.operator).netloc,
|
||||
urlparse(sinks.arize).netloc,
|
||||
],
|
||||
}
|
||||
config["general_settings"] = {**config.get("general_settings", {}), "user_api_key_cache_ttl": 2}
|
||||
path: Final = directory / f"{name}.yaml"
|
||||
path.write_text(yaml.safe_dump(config))
|
||||
return path
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _started(
|
||||
provider: Wire,
|
||||
sinks: SpanSinks,
|
||||
newrelic: ConnectSink,
|
||||
directory: Path,
|
||||
capture: str | None,
|
||||
callbacks: tuple[str, ...] = ("langfuse_otel",),
|
||||
) -> Generator[Rig]:
|
||||
environment: Final = {
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
"OTEL_BSP_SCHEDULE_DELAY": "300",
|
||||
"MAPPER_NAMES": "genai,openinference",
|
||||
"LANGFUSE_HOST": sinks.operator,
|
||||
"LANGFUSE_PUBLIC_KEY": "pk-lf-operator",
|
||||
"LANGFUSE_SECRET_KEY": "sk-lf-operator",
|
||||
"ARIZE_SPACE_KEY": "space-operator",
|
||||
"ARIZE_API_KEY": "arize-operator",
|
||||
"ARIZE_HTTP_ENDPOINT": sinks.arize + "/v1/traces",
|
||||
"WANDB_API_KEY": "wandb-operator",
|
||||
"WANDB_PROJECT_ID": "operator/weave",
|
||||
"WANDB_HOST": sinks.arize,
|
||||
"SIGNOZ_INGESTION_ENDPOINT": sinks.arize + "/v1/traces",
|
||||
"SIGNOZ_INGESTION_KEY": "signoz-operator",
|
||||
"HTTPS_PROXY": newrelic.proxy_url,
|
||||
"NO_PROXY": "127.0.0.1,localhost",
|
||||
"REQUESTS_CA_BUNDLE": newrelic.ca_pem,
|
||||
**({} if capture is None else {"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": capture}),
|
||||
}
|
||||
with (
|
||||
gateway_from_environment() as gateway,
|
||||
owned_proxy_process(
|
||||
gateway,
|
||||
directory,
|
||||
environment,
|
||||
config=_write_config(directory, sinks, "capture", callbacks),
|
||||
remove_environment=(
|
||||
"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
|
||||
"LITELLM_OTEL_TENANT_DESTINATION_MODE",
|
||||
),
|
||||
workers=2,
|
||||
) as owned,
|
||||
owned.gateway.scenario() as scenario,
|
||||
):
|
||||
yield Rig(
|
||||
owned.gateway, owned, scenario, scenario.model(api_base=provider.url + "/v1"), provider, sinks, newrelic
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def provider() -> Iterator[Wire]:
|
||||
with wire_server(_upstream) as wire:
|
||||
yield wire
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def newrelic_sink(tmp_path_factory: pytest.TempPathFactory) -> Iterator[ConnectSink]:
|
||||
with owned_connect_sink(tmp_path_factory.mktemp("newrelic-sink")) as sink:
|
||||
yield sink
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def rig(
|
||||
provider: Wire, audit_sinks: SpanSinks, newrelic_sink: ConnectSink, tmp_path_factory: pytest.TempPathFactory
|
||||
) -> Iterator[Rig]:
|
||||
directory: Final = tmp_path_factory.mktemp("capture-span-only")
|
||||
with _started(provider, audit_sinks, newrelic_sink, directory, "span_only") as started:
|
||||
yield started
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def dark_rig(
|
||||
provider: Wire, audit_sinks: SpanSinks, newrelic_sink: ConnectSink, tmp_path_factory: pytest.TempPathFactory
|
||||
) -> Iterator[Rig]:
|
||||
with _started(
|
||||
provider,
|
||||
audit_sinks,
|
||||
newrelic_sink,
|
||||
tmp_path_factory.mktemp("capture-no-content"),
|
||||
None,
|
||||
("langfuse_otel", "signoz"),
|
||||
) as started:
|
||||
yield started
|
||||
|
|
@ -16,6 +16,7 @@ import litellm # noqa: E402
|
|||
from litellm.caching.dual_cache import DualCache # noqa: E402
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2, build_otel_v2_logger # noqa: E402
|
||||
from litellm.integrations.otel.model.config import OpenTelemetryV2Config, is_otel_v2_enabled # noqa: E402
|
||||
from litellm.integrations.otel.model.destination import OtelDestination # noqa: E402
|
||||
from litellm.integrations.otel.model.spans import LITELLM_PROXY_REQUEST_SPAN_NAME, SpanRole # noqa: E402
|
||||
from litellm.integrations.otel.plumbing import context as otel_context # noqa: E402
|
||||
from litellm.integrations.otel.plumbing import providers # noqa: E402
|
||||
|
|
@ -155,6 +156,25 @@ def test_chat_request_stamps_root_observation_input_and_output():
|
|||
assert [(turn["role"], turn["content"]) for turn in output] == [("assistant", "pong")]
|
||||
|
||||
|
||||
def test_content_on_destination_stamps_langfuse_root_io_when_global_capture_is_off() -> None:
|
||||
logger, exporter = _logger(capture="no_content")
|
||||
destination: Final = OtelDestination(
|
||||
endpoint="https://team.example.com/api/public/otel",
|
||||
callback_name="langfuse_otel",
|
||||
capture_message_content="span_only",
|
||||
)
|
||||
token: Final = otel_context.set_request_destinations((destination,))
|
||||
try:
|
||||
response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))])
|
||||
_run_request(logger, CHAT_DATA, "acompletion", response)
|
||||
finally:
|
||||
otel_context.reset_request_destinations(token)
|
||||
|
||||
attrs: Final = _root_attrs(exporter)
|
||||
assert json.loads(attrs[INPUT_ATTR]) == [{"role": "user", "content": "ping"}]
|
||||
assert json.loads(attrs[OUTPUT_ATTR]) == [{"role": "assistant", "content": "pong"}]
|
||||
|
||||
|
||||
def test_responses_request_folds_instructions_into_input_and_stamps_output_items():
|
||||
logger, exporter = _logger()
|
||||
data = {"model": "gpt-5.4-mini", "instructions": "be terse", "input": "ping"}
|
||||
|
|
@ -461,7 +481,9 @@ def test_a_request_without_trace_controls_stamps_none_of_them():
|
|||
logger, exporter = _logger()
|
||||
|
||||
root_attrs, generation_attrs = _run_named_request(
|
||||
logger, exporter, {"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}}
|
||||
logger,
|
||||
exporter,
|
||||
{"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}},
|
||||
)
|
||||
|
||||
assert set(TRACE_CONTROL_ATTRS).isdisjoint(root_attrs)
|
||||
|
|
@ -472,7 +494,7 @@ def test_a_request_without_trace_controls_stamps_none_of_them():
|
|||
("capture", "mappers"),
|
||||
[("no_content", ("genai", "langfuse")), ("span_only", ("genai",))],
|
||||
)
|
||||
def test_factory_keeps_the_base_logger_unless_langfuse_content_capture_is_on(capture, mappers):
|
||||
def test_langfuse_root_io_requires_global_or_destination_capture(capture: str, mappers: tuple[str, ...]) -> None:
|
||||
logger, exporter = _logger(capture=capture, mappers=mappers)
|
||||
|
||||
_run_request(logger, CHAT_DATA, "acompletion", ModelResponse())
|
||||
|
|
@ -484,13 +506,13 @@ def test_factory_keeps_the_base_logger_unless_langfuse_content_capture_is_on(cap
|
|||
("capture", "mappers", "relays_streams"),
|
||||
[
|
||||
("span_only", ("genai", "langfuse"), True),
|
||||
("no_content", ("genai", "langfuse"), False),
|
||||
("no_content", ("genai", "langfuse"), True),
|
||||
("span_only", ("genai",), False),
|
||||
],
|
||||
)
|
||||
def test_only_langfuse_content_capture_takes_proxy_streams_off_the_fast_path(
|
||||
monkeypatch, capture, mappers, relays_streams
|
||||
):
|
||||
def test_langfuse_mapper_relays_streams_for_per_request_content_capture(
|
||||
monkeypatch: pytest.MonkeyPatch, capture: str, mappers: tuple[str, ...], relays_streams: bool
|
||||
) -> None:
|
||||
logger, _ = _logger(capture=capture, mappers=mappers)
|
||||
monkeypatch.setattr(litellm, "callbacks", [logger])
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import contextvars
|
|||
import time
|
||||
from base64 import b64encode
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import replace
|
||||
from datetime import datetime, timezone
|
||||
from functools import reduce
|
||||
from types import MappingProxyType
|
||||
|
|
@ -13,13 +14,14 @@ from typing import Final
|
|||
import pytest
|
||||
from opentelemetry import trace as trace_api
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.trace import Status, StatusCode
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.integrations.otel import GenAIOperation
|
||||
from litellm.integrations.otel import logger as otel_logger
|
||||
from litellm.integrations.otel.logger import (
|
||||
OpenTelemetryV2,
|
||||
|
|
@ -27,13 +29,24 @@ from litellm.integrations.otel.logger import (
|
|||
fan_out_provider,
|
||||
publish_global_otel_v2_provider,
|
||||
)
|
||||
from litellm.integrations.otel.mappers import resolve_mappers
|
||||
from litellm.integrations.otel.model.config import (
|
||||
CaptureMessageContent,
|
||||
ExporterOwner,
|
||||
ExporterSpec,
|
||||
OpenTelemetryV2Config,
|
||||
is_otel_v2_enabled,
|
||||
)
|
||||
from litellm.integrations.otel.model.destination import OtelDestination
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
LLMCallSpanData,
|
||||
LLMRequestParams,
|
||||
LLMUsage,
|
||||
MCPToolCallSpanData,
|
||||
RequestIdentity,
|
||||
ServerInfo,
|
||||
ToolDefinition,
|
||||
)
|
||||
from litellm.integrations.otel.plumbing import providers as otel_providers
|
||||
from litellm.integrations.otel.plumbing.context import (
|
||||
destination_backends,
|
||||
|
|
@ -44,9 +57,11 @@ from litellm.integrations.otel.plumbing.providers import (
|
|||
TenantFanOutSpanProcessor,
|
||||
_OverriddenBackendFilter,
|
||||
_sink_key,
|
||||
attach_tenant_fan_out,
|
||||
build_tracer_provider,
|
||||
deliverable_destinations,
|
||||
operator_sink_scopes,
|
||||
register_exporter_factory,
|
||||
)
|
||||
from litellm.integrations.otel.plumbing.routing import TenantTracerCache, get_tracer
|
||||
from litellm.integrations.otel.presets.arize import arize_preset
|
||||
|
|
@ -55,7 +70,7 @@ from litellm.integrations.otel.presets.destinations import (
|
|||
destination_for,
|
||||
)
|
||||
from litellm.integrations.otel.presets.langfuse import langfuse_preset
|
||||
from litellm.proxy._types import AddTeamCallback, UserAPIKeyAuth
|
||||
from litellm.proxy._types import AddTeamCallback, TeamCallbackMetadata, UserAPIKeyAuth
|
||||
from litellm.proxy.litellm_pre_call_utils import (
|
||||
convert_key_logging_metadata_to_callback,
|
||||
resolve_tenant_otel_destinations,
|
||||
|
|
@ -4228,6 +4243,569 @@ class TestTenantHostSsrfGuard:
|
|||
assert sum("provider_url_destination_allowed_hosts" in record.message for record in caplog.records) == 1
|
||||
|
||||
|
||||
_ALL_MAPPERS: Final = ("genai", "legacy", "openinference", "langfuse", "weave", "langtrace")
|
||||
_SECRET: Final = "SECRET-MARKER"
|
||||
_MODEL_CALL_WITH_CONTENT: Final = LLMCallSpanData(
|
||||
operation=GenAIOperation.CHAT,
|
||||
provider="openai",
|
||||
request_model="gpt-4o",
|
||||
response_model="gpt-4o-2024",
|
||||
response_id="resp_1",
|
||||
request_params=LLMRequestParams(temperature=0.5, max_tokens=128),
|
||||
usage=LLMUsage(input_tokens=12, output_tokens=8, total_tokens=20),
|
||||
finish_reasons=("tool_calls",),
|
||||
error=None,
|
||||
response_cost=0.001,
|
||||
server=ServerInfo("api.openai.com", 443),
|
||||
identity=RequestIdentity(call_id="c1", team_id="t1"),
|
||||
tools=(ToolDefinition(name="lookup", description="Find a city", parameters_json='{"type":"object"}'),),
|
||||
messages_in=(
|
||||
{"role": "system", "content": f"system {_SECRET}"},
|
||||
{"role": "user", "content": f"prompt {_SECRET}"},
|
||||
),
|
||||
choices_out=(
|
||||
{
|
||||
"finish_reason": "tool_calls",
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": f"answer {_SECRET}",
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": {"name": "lookup", "arguments": f'{{"city": "{_SECRET}"}}'},
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
),
|
||||
)
|
||||
_MODEL_CALL_WITHOUT_CONTENT: Final = replace(_MODEL_CALL_WITH_CONTENT, messages_in=(), choices_out=())
|
||||
_TOOL_CALL_WITH_CONTENT: Final = MCPToolCallSpanData(
|
||||
operation=GenAIOperation.EXECUTE_TOOL,
|
||||
method="tools/call",
|
||||
tool_name="lookup",
|
||||
server_name="maps",
|
||||
server_address="maps.local",
|
||||
server_port=443,
|
||||
session_id="s1",
|
||||
arguments_json=f'{{"city": "{_SECRET}"}}',
|
||||
result_json=f'{{"weather": "{_SECRET}"}}',
|
||||
error=None,
|
||||
response_cost=None,
|
||||
identity=RequestIdentity(call_id="c2", team_id="t1"),
|
||||
)
|
||||
_TOOL_CALL_WITHOUT_CONTENT: Final = replace(_TOOL_CALL_WITH_CONTENT, arguments_json=None, result_json=None)
|
||||
|
||||
NO_CONTENT_DEST = OtelDestination(
|
||||
endpoint="http://team-a.local/api/public/otel",
|
||||
headers=MappingProxyType({"Authorization": "Basic YQ=="}),
|
||||
callback_name="langfuse_otel",
|
||||
capture_message_content="no_content",
|
||||
)
|
||||
INHERITING_DEST = OtelDestination(
|
||||
endpoint="http://team-b.local/v1",
|
||||
headers=MappingProxyType({"api_key": "k", "arize-space-id": "s"}),
|
||||
callback_name="arize",
|
||||
)
|
||||
|
||||
|
||||
def mapped(data: LLMCallSpanData | MCPToolCallSpanData) -> Mapping[str, object]:
|
||||
"""What the configured mappers write on the span, the way the logger stamps it."""
|
||||
return MappingProxyType(reduce(lambda acc, mapper: {**acc, **mapper.map(data)}, resolve_mappers(_ALL_MAPPERS), {}))
|
||||
|
||||
|
||||
def recorded(attributes: Mapping[str, object]) -> dict[str, object]:
|
||||
"""``attributes`` the way the SDK records them, sequences frozen into tuples."""
|
||||
return {key: tuple(value) if isinstance(value, list) else value for key, value in attributes.items()}
|
||||
|
||||
|
||||
def model_call_tree(provider: TracerProvider, attributes: Mapping[str, object]) -> None:
|
||||
tracer: Final = get_tracer(provider, "litellm")
|
||||
with tracer.start_as_current_span("POST /v1/chat/completions"):
|
||||
with tracer.start_as_current_span("chat gpt-4o") as llm:
|
||||
llm.set_attributes(attributes)
|
||||
llm.add_event("gen_ai.content.first_chunk", {"gen_ai.response.model": "gpt-4o-2024"})
|
||||
|
||||
|
||||
def by_name(exporter: InMemorySpanExporter) -> dict[str, ReadableSpan]:
|
||||
return {span.name: span for span in exporter.get_finished_spans()}
|
||||
|
||||
|
||||
def carries_content(span: ReadableSpan) -> bool:
|
||||
return any(_SECRET in str(value) for value in span.attributes.values())
|
||||
|
||||
|
||||
class TestCaptureMessageContent:
|
||||
"""A destination's explicit capture mode overrides the global default."""
|
||||
|
||||
@staticmethod
|
||||
def _fan_out(
|
||||
operator: InMemorySpanExporter,
|
||||
exporters: Mapping[str, InMemorySpanExporter],
|
||||
default_capture: CaptureMessageContent = CaptureMessageContent.SPAN_ONLY,
|
||||
) -> TracerProvider:
|
||||
provider: Final = TracerProvider()
|
||||
provider.add_span_processor(SimpleSpanProcessor(operator))
|
||||
provider.add_span_processor(
|
||||
TenantFanOutSpanProcessor(
|
||||
processor_factory=lambda d: SimpleSpanProcessor(exporters[d.endpoint]),
|
||||
default_capture=default_capture,
|
||||
)
|
||||
)
|
||||
return provider
|
||||
|
||||
@staticmethod
|
||||
def _run(
|
||||
provider: TracerProvider, destinations: tuple[OtelDestination, ...], attributes: Mapping[str, object]
|
||||
) -> None:
|
||||
def run() -> None:
|
||||
set_request_destinations(destinations)
|
||||
model_call_tree(provider, attributes)
|
||||
|
||||
in_fresh_context(run)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("setting", "content_exported"),
|
||||
[
|
||||
(None, True),
|
||||
("no_content", False),
|
||||
("span_only", True),
|
||||
("event_only", False),
|
||||
("span_and_event", True),
|
||||
],
|
||||
)
|
||||
def test_a_destination_setting_controls_its_copy_and_omission_uses_the_global_default(
|
||||
self, setting: str | None, content_exported: bool
|
||||
) -> None:
|
||||
attributes: Final = mapped(_MODEL_CALL_WITH_CONTENT)
|
||||
destination: Final = OtelDestination(
|
||||
endpoint="http://team.local/api/public/otel",
|
||||
headers=MappingProxyType({"Authorization": "Basic dA=="}),
|
||||
callback_name="langfuse_otel",
|
||||
capture_message_content=setting,
|
||||
)
|
||||
operator, tenant = InMemorySpanExporter(), InMemorySpanExporter()
|
||||
|
||||
self._run(self._fan_out(operator, {destination.endpoint: tenant}), (destination,), attributes)
|
||||
|
||||
exported: Final = by_name(tenant)["chat gpt-4o"]
|
||||
assert carries_content(exported) is content_exported
|
||||
assert carries_content(by_name(operator)["chat gpt-4o"])
|
||||
assert exported.attributes["gen_ai.request.model"] == "gpt-4o"
|
||||
assert exported.attributes["gen_ai.usage.input_tokens"] == 12
|
||||
assert exported.attributes["gen_ai.usage.output_tokens"] == 8
|
||||
|
||||
def test_a_span_only_team_lifts_global_no_content_only_for_its_destination(self) -> None:
|
||||
operator_exporter, team_exporter, sibling_exporter = (
|
||||
InMemorySpanExporter(),
|
||||
InMemorySpanExporter(),
|
||||
InMemorySpanExporter(),
|
||||
)
|
||||
kind: Final = "lit8244_global_no_content_capture"
|
||||
register_exporter_factory(kind, lambda _spec: operator_exporter)
|
||||
config: Final = OpenTelemetryV2Config(
|
||||
capture_message_content=CaptureMessageContent.NO_CONTENT,
|
||||
exporters=[ExporterSpec(kind=kind)],
|
||||
)
|
||||
provider: Final = build_tracer_provider(config, use_simple_processor=True, tenant_overrides=True)
|
||||
team: Final = OtelDestination(
|
||||
endpoint="http://team.local/api/public/otel",
|
||||
headers=MappingProxyType({"Authorization": "Basic dA=="}),
|
||||
callback_name="langfuse_otel",
|
||||
capture_message_content="span_only",
|
||||
)
|
||||
sibling: Final = INHERITING_DEST
|
||||
provider.add_span_processor(
|
||||
TenantFanOutSpanProcessor(
|
||||
processor_factory=lambda destination: SimpleSpanProcessor(
|
||||
team_exporter if destination.endpoint == team.endpoint else sibling_exporter
|
||||
),
|
||||
default_capture=config.capture_message_content,
|
||||
)
|
||||
)
|
||||
|
||||
self._run(
|
||||
provider,
|
||||
(team, sibling),
|
||||
mapped(_MODEL_CALL_WITH_CONTENT),
|
||||
)
|
||||
|
||||
assert carries_content(by_name(team_exporter)["chat gpt-4o"])
|
||||
assert not any(carries_content(span) for span in operator_exporter.get_finished_spans())
|
||||
assert not any(carries_content(span) for span in sibling_exporter.get_finished_spans())
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("with_content", "without_content"),
|
||||
[
|
||||
(_MODEL_CALL_WITH_CONTENT, _MODEL_CALL_WITHOUT_CONTENT),
|
||||
(_TOOL_CALL_WITH_CONTENT, _TOOL_CALL_WITHOUT_CONTENT),
|
||||
],
|
||||
ids=["model_call", "mcp_tool_call"],
|
||||
)
|
||||
def test_no_content_removes_exactly_what_global_capture_adds_in_every_mapper_vocabulary(
|
||||
self,
|
||||
with_content: LLMCallSpanData | MCPToolCallSpanData,
|
||||
without_content: LLMCallSpanData | MCPToolCallSpanData,
|
||||
) -> None:
|
||||
captured, uncaptured = mapped(with_content), mapped(without_content)
|
||||
assert frozenset(captured) - frozenset(uncaptured), "the fixture must exercise captured content"
|
||||
operator, tenant = InMemorySpanExporter(), InMemorySpanExporter()
|
||||
|
||||
self._run(self._fan_out(operator, {NO_CONTENT_DEST.endpoint: tenant}), (NO_CONTENT_DEST,), captured)
|
||||
|
||||
exported: Final = by_name(tenant)["chat gpt-4o"]
|
||||
assert dict(exported.attributes) == recorded(uncaptured), "only content goes, every other attribute stays"
|
||||
assert not carries_content(exported)
|
||||
|
||||
def test_restricting_one_destination_leaves_the_original_span_and_the_other_destination_alone(self) -> None:
|
||||
attributes: Final = mapped(_MODEL_CALL_WITH_CONTENT)
|
||||
operator, restricted, inheriting = InMemorySpanExporter(), InMemorySpanExporter(), InMemorySpanExporter()
|
||||
exporters: Final = {NO_CONTENT_DEST.endpoint: restricted, INHERITING_DEST.endpoint: inheriting}
|
||||
|
||||
self._run(self._fan_out(operator, exporters), (NO_CONTENT_DEST, INHERITING_DEST), attributes)
|
||||
|
||||
original: Final = by_name(operator)
|
||||
kept: Final = by_name(inheriting)
|
||||
stripped: Final = by_name(restricted)
|
||||
assert dict(original["chat gpt-4o"].attributes) == recorded(attributes)
|
||||
assert dict(kept["chat gpt-4o"].attributes) == recorded(attributes)
|
||||
assert not carries_content(stripped["chat gpt-4o"])
|
||||
for copy in (kept, stripped):
|
||||
for name, span in copy.items():
|
||||
assert span.context == original[name].context, "same trace id and span id"
|
||||
assert span.parent == original[name].parent, "same place in the tree"
|
||||
assert [e.name for e in copy["chat gpt-4o"].events] == [e.name for e in original["chat gpt-4o"].events]
|
||||
|
||||
def test_an_omitted_setting_and_span_only_export_the_same_span(self) -> None:
|
||||
attributes: Final = mapped(_MODEL_CALL_WITH_CONTENT)
|
||||
span_only: Final = INHERITING_DEST.model_copy(
|
||||
update={"capture_message_content": CaptureMessageContent.SPAN_ONLY}
|
||||
)
|
||||
omitted_exporter, span_only_exporter = InMemorySpanExporter(), InMemorySpanExporter()
|
||||
|
||||
for destination, exporter in ((INHERITING_DEST, omitted_exporter), (span_only, span_only_exporter)):
|
||||
self._run(
|
||||
self._fan_out(InMemorySpanExporter(), {destination.endpoint: exporter}), (destination,), attributes
|
||||
)
|
||||
|
||||
assert {n: dict(s.attributes) for n, s in by_name(omitted_exporter).items()} == {
|
||||
n: dict(s.attributes) for n, s in by_name(span_only_exporter).items()
|
||||
}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("globally_captured", "setting", "content_exported"),
|
||||
[
|
||||
(True, "no_content", False),
|
||||
(True, None, True),
|
||||
(False, "span_only", True),
|
||||
(False, None, False),
|
||||
],
|
||||
)
|
||||
def test_the_operators_exporter_on_the_same_account_honors_the_teams_setting_once(
|
||||
self,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
globally_captured: bool,
|
||||
setting: str | None,
|
||||
content_exported: bool,
|
||||
) -> None:
|
||||
"""The fan-out skips a destination the operator already writes to, so the operator's
|
||||
copy is the one that account receives and must not bypass the team's restriction."""
|
||||
monkeypatch.setattr(litellm, "otel_tenant_destination_mode", "additive", raising=False)
|
||||
shared: Final = InMemorySpanExporter()
|
||||
provider: Final = TracerProvider()
|
||||
provider.add_span_processor(
|
||||
_OverriddenBackendFilter(
|
||||
SimpleSpanProcessor(shared),
|
||||
"langfuse_otel",
|
||||
"full",
|
||||
TestRoutingMode.OPERATOR_SINK,
|
||||
global_captures=globally_captured,
|
||||
)
|
||||
)
|
||||
provider.add_span_processor(
|
||||
TenantFanOutSpanProcessor(
|
||||
processor_factory=lambda _d: SimpleSpanProcessor(shared),
|
||||
operator_sinks=MappingProxyType({TestRoutingMode.OPERATOR_SINK: "full"}),
|
||||
)
|
||||
)
|
||||
destination: Final = OtelDestination(
|
||||
endpoint=TestRoutingMode.SAME_ACCOUNT_ENDPOINT,
|
||||
headers=MappingProxyType({"Authorization": "Basic op"}),
|
||||
callback_name="langfuse_otel",
|
||||
capture_message_content=setting,
|
||||
)
|
||||
|
||||
self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
|
||||
|
||||
model_calls: Final = [s for s in shared.get_finished_spans() if s.name == "chat gpt-4o"]
|
||||
assert len(model_calls) == 1, "the same account received the span twice"
|
||||
assert carries_content(model_calls[0]) is content_exported
|
||||
|
||||
@pytest.mark.parametrize("mode", ["override", "additive"])
|
||||
def test_an_operator_collector_on_the_teams_account_does_not_bypass_no_content(
|
||||
self, monkeypatch: pytest.MonkeyPatch, mode: str
|
||||
) -> None:
|
||||
monkeypatch.setattr(litellm, "otel_tenant_destination_mode", mode, raising=False)
|
||||
shared: Final = InMemorySpanExporter()
|
||||
kind: Final = f"lit8244_collector_{mode}"
|
||||
register_exporter_factory(kind, lambda _spec: shared)
|
||||
config: Final = OpenTelemetryV2Config(
|
||||
exporters=[
|
||||
ExporterSpec(
|
||||
kind=kind,
|
||||
endpoint=TestRoutingMode.OPERATOR_SINK[0],
|
||||
headers="authorization=Basic op",
|
||||
)
|
||||
]
|
||||
)
|
||||
provider: Final = build_tracer_provider(config, use_simple_processor=True, tenant_overrides=True)
|
||||
provider.add_span_processor(
|
||||
TenantFanOutSpanProcessor(
|
||||
processor_factory=lambda _d: SimpleSpanProcessor(shared),
|
||||
operator_sinks=operator_sink_scopes(config),
|
||||
)
|
||||
)
|
||||
destination: Final = OtelDestination(
|
||||
endpoint=TestRoutingMode.SAME_ACCOUNT_ENDPOINT,
|
||||
headers=MappingProxyType({"Authorization": "Basic op"}),
|
||||
callback_name="langfuse_otel",
|
||||
capture_message_content="no_content",
|
||||
)
|
||||
|
||||
self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
|
||||
|
||||
model_calls: Final = [s for s in shared.get_finished_spans() if s.name == "chat gpt-4o"]
|
||||
assert model_calls, "the account must receive the model call before its content can be judged absent"
|
||||
assert not any(carries_content(span) for span in model_calls)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("global_capture", "content_exported"),
|
||||
[(CaptureMessageContent.NO_CONTENT, False), (CaptureMessageContent.SPAN_ONLY, True)],
|
||||
)
|
||||
def test_a_routed_provider_exports_content_only_when_the_global_setting_captures(
|
||||
self, global_capture: str, content_exported: bool
|
||||
) -> None:
|
||||
"""A sibling destination's span_only makes the logger collect content, and a request
|
||||
routed to this callback's team credentials must not carry it under a global no_content."""
|
||||
exporters: dict[ExporterOwner | None, InMemorySpanExporter] = {}
|
||||
kind: Final = f"lit8244_routed_{global_capture}"
|
||||
register_exporter_factory(kind, lambda spec: exporters.setdefault(spec.owner, InMemorySpanExporter()))
|
||||
config: Final = OpenTelemetryV2Config(
|
||||
capture_message_content=global_capture,
|
||||
exporters=[
|
||||
ExporterSpec(
|
||||
kind=kind,
|
||||
endpoint="http://op.local",
|
||||
owner=ExporterOwner.LANGFUSE_OTEL,
|
||||
use_simple_processor=True,
|
||||
),
|
||||
ExporterSpec(kind=kind),
|
||||
],
|
||||
)
|
||||
cache: Final = TenantTracerCache(config, "langfuse_otel", "litellm")
|
||||
|
||||
route: Final = cache.route_for(
|
||||
get_tracer(TracerProvider(), "litellm"),
|
||||
{"langfuse_public_key": "pk-team", "langfuse_secret_key": "sk-team"},
|
||||
)
|
||||
with route.tracer.start_as_current_span("chat gpt-4o") as llm:
|
||||
llm.set_attributes(mapped(_MODEL_CALL_WITH_CONTENT))
|
||||
cache.release(route.provider)
|
||||
|
||||
assert route.detached is True
|
||||
for exporter in (exporters[ExporterOwner.LANGFUSE_OTEL], exporters[None]):
|
||||
copy = by_name(exporter)["chat gpt-4o"]
|
||||
assert carries_content(copy) is content_exported
|
||||
assert copy.attributes["gen_ai.usage.input_tokens"] == 12
|
||||
|
||||
@pytest.mark.parametrize("first", ["newrelic", "langfuse_otel"])
|
||||
def test_an_omitted_setting_follows_its_own_callbacks_mode_whatever_the_callback_order(
|
||||
self, monkeypatch: pytest.MonkeyPatch, first: str
|
||||
) -> None:
|
||||
monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False)
|
||||
tenant: Final = InMemorySpanExporter()
|
||||
monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, "otlp_http", lambda _spec: tenant)
|
||||
langfuse: Final = OpenTelemetryV2Config(
|
||||
capture_message_content=CaptureMessageContent.SPAN_ONLY,
|
||||
exporters=[ExporterSpec(kind="in_memory", owner=ExporterOwner.LANGFUSE_OTEL)],
|
||||
)
|
||||
newrelic: Final = OpenTelemetryV2Config(
|
||||
capture_message_content=CaptureMessageContent.NO_CONTENT,
|
||||
exporters=[ExporterSpec(kind="in_memory", owner=ExporterOwner.NEWRELIC)],
|
||||
)
|
||||
provider: Final = TracerProvider()
|
||||
attach_tenant_fan_out(provider, *((newrelic, langfuse) if first == "newrelic" else (langfuse, newrelic)))
|
||||
destination: Final = OtelDestination(
|
||||
endpoint="http://team.local/api/public/otel",
|
||||
headers=MappingProxyType({"Authorization": "Basic dA=="}),
|
||||
callback_name="langfuse_otel",
|
||||
)
|
||||
|
||||
self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
|
||||
provider.force_flush()
|
||||
|
||||
assert carries_content(by_name(tenant)["chat gpt-4o"])
|
||||
|
||||
def test_an_omitted_setting_follows_the_capture_mode_set_in_callback_settings_otel(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False)
|
||||
tenant: Final = InMemorySpanExporter()
|
||||
monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, "otlp_http", lambda _spec: tenant)
|
||||
operator: Final = OpenTelemetryV2Config(**{"capture_message_content": "span_only", "exporter": "in_memory"})
|
||||
provider: Final = TracerProvider()
|
||||
attach_tenant_fan_out(provider, operator)
|
||||
destination: Final = OtelDestination(
|
||||
endpoint="http://team.local/api/public/otel",
|
||||
headers=MappingProxyType({"Authorization": "Basic dA=="}),
|
||||
callback_name="langfuse_otel",
|
||||
)
|
||||
|
||||
self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
|
||||
provider.force_flush()
|
||||
|
||||
assert carries_content(by_name(tenant)["chat gpt-4o"]), "the YAML value is the global default, not only the env"
|
||||
|
||||
def test_a_published_presets_capture_mode_is_the_default_for_an_unowned_destination(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""A destination for a backend the proxy does not export to itself falls back to the published logger's mode,
|
||||
here New Relic with record_content on, so it keeps the content."""
|
||||
monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False)
|
||||
tenant: Final = InMemorySpanExporter()
|
||||
monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, "otlp_http", lambda _spec: tenant)
|
||||
newrelic: Final = OpenTelemetryV2Config(
|
||||
capture_message_content=CaptureMessageContent.SPAN_ONLY,
|
||||
exporters=[
|
||||
ExporterSpec(kind="console"),
|
||||
ExporterSpec(kind="in_memory", owner=ExporterOwner.NEWRELIC),
|
||||
],
|
||||
)
|
||||
provider: Final = TracerProvider()
|
||||
attach_tenant_fan_out(provider, newrelic)
|
||||
destination: Final = OtelDestination(
|
||||
endpoint="http://team.local/api/public/otel",
|
||||
headers=MappingProxyType({"Authorization": "Basic dA=="}),
|
||||
callback_name="langfuse_otel",
|
||||
)
|
||||
|
||||
self._run(provider, (destination,), mapped(_MODEL_CALL_WITH_CONTENT))
|
||||
provider.force_flush()
|
||||
|
||||
assert carries_content(by_name(tenant)["chat gpt-4o"])
|
||||
|
||||
@pytest.mark.usefixtures("allow_test_hosts")
|
||||
@pytest.mark.parametrize("setting", ["no_content", "span_only", None])
|
||||
def test_the_setting_rides_the_teams_destination_and_omission_stays_omitted(
|
||||
self, monkeypatch: pytest.MonkeyPatch, setting: 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",
|
||||
}
|
||||
if setting is not None:
|
||||
callback_vars["capture_message_content"] = setting
|
||||
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"].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", ["event_only", "span_and_event"])
|
||||
def test_a_stored_event_mode_is_skipped_like_any_invalid_entry(
|
||||
self, monkeypatch: pytest.MonkeyPatch, setting: str
|
||||
) -> None:
|
||||
monkeypatch.setenv("LITELLM_OTEL_V2", "true")
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
auth: Final = UserAPIKeyAuth(
|
||||
team_metadata={
|
||||
"logging": [
|
||||
{
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": "success",
|
||||
"callback_vars": {
|
||||
"langfuse_public_key": "pk-team",
|
||||
"langfuse_secret_key": "sk-team",
|
||||
"langfuse_host": "http://team.local",
|
||||
"capture_message_content": setting,
|
||||
},
|
||||
},
|
||||
{
|
||||
"callback_name": "arize",
|
||||
"callback_type": "success",
|
||||
"callback_vars": {"arize_api_key": "k", "arize_space_id": "s"},
|
||||
},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
destinations: Final = [d.callback_name for d in resolve_tenant_otel_destinations(auth)]
|
||||
|
||||
assert destinations == ["arize"]
|
||||
|
||||
@pytest.mark.parametrize("value", ["full", "NO_CONTENT", "", "true", "event_only", "span_and_event"])
|
||||
def test_an_unsupported_value_fails_registration(self, value: str) -> None:
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match=r"Invalid capture_message_content .*\['no_content', 'span_only'\]",
|
||||
):
|
||||
AddTeamCallback(
|
||||
callback_name="langfuse_otel",
|
||||
callback_type="success",
|
||||
callback_vars={
|
||||
"langfuse_public_key": "pk",
|
||||
"langfuse_secret_key": "sk",
|
||||
"capture_message_content": value,
|
||||
},
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("value", ["no_content", "span_only"])
|
||||
def test_a_supported_value_is_stored_as_given(self, value: str) -> None:
|
||||
saved: Final = AddTeamCallback(
|
||||
callback_name="langfuse_otel",
|
||||
callback_type="success",
|
||||
callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "capture_message_content": value},
|
||||
)
|
||||
|
||||
assert saved.callback_vars["capture_message_content"] == value
|
||||
|
||||
def test_a_team_wide_callback_vars_map_cannot_carry_the_setting(self) -> None:
|
||||
"""The flattened map is shared by every callback, so a value there would apply to all of them."""
|
||||
with pytest.raises(ValueError, match="Invalid callback variable: capture_message_content"):
|
||||
TeamCallbackMetadata(
|
||||
success_callback=["langfuse_otel", "signoz"], callback_vars={"capture_message_content": "span_only"}
|
||||
)
|
||||
|
||||
def test_flattening_the_entries_leaves_the_setting_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", "capture_message_content": "span_only"},
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
assert flattened.callback_vars == {"langfuse_public_key": "pk"}
|
||||
|
||||
|
||||
class TestArizeProjectRouting:
|
||||
"""Arize rejects an export that names no project (400: set x-project-name,
|
||||
arize.project.name, openinference.project.name, or model_id), so an Arize
|
||||
|
|
|
|||
|
|
@ -9,11 +9,13 @@ hooks, proxy SERVER span lifecycle (start + setters), parent-context resolution
|
|||
import ast
|
||||
import asyncio
|
||||
import contextlib
|
||||
import contextvars
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
from unittest.mock import patch
|
||||
|
||||
|
|
@ -43,7 +45,8 @@ from litellm.integrations.otel import ( # noqa: E402
|
|||
OpenTelemetryV2Config,
|
||||
)
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402
|
||||
from litellm.integrations.otel.model.config import ExporterSpec # noqa: E402
|
||||
from litellm.integrations.otel.model.config import CaptureMessageContent, ExporterSpec # noqa: E402
|
||||
from litellm.integrations.otel.model.destination import OtelDestination # noqa: E402
|
||||
from litellm.integrations.otel.model.spans import ( # noqa: E402
|
||||
LITELLM_PROXY_REQUEST_SPAN_NAME,
|
||||
SpanRole,
|
||||
|
|
@ -55,6 +58,7 @@ from litellm.integrations.otel.plumbing.context import ( # noqa: E402
|
|||
reset_mcp_message_transport_span,
|
||||
set_mcp_message_trace_carrier,
|
||||
set_mcp_message_transport_span,
|
||||
set_request_destinations,
|
||||
set_request_root_span,
|
||||
)
|
||||
from litellm.types.utils import AUTOROUTER_CLASSIFIER_CALL_ORIGIN # noqa: E402
|
||||
|
|
@ -617,8 +621,6 @@ def _mcp_payload(**overrides):
|
|||
|
||||
|
||||
def _logger_capturing():
|
||||
from litellm.integrations.otel.model.config import CaptureMessageContent
|
||||
|
||||
cfg = OpenTelemetryV2Config(
|
||||
exporter="in_memory",
|
||||
legacy_compat=False,
|
||||
|
|
@ -684,6 +686,57 @@ def test_mcp_tool_call_captures_io_when_enabled():
|
|||
assert "21" in span.attributes["gen_ai.tool.call.result"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("capture_mode", "captures_content"),
|
||||
[
|
||||
(None, False),
|
||||
(CaptureMessageContent.NO_CONTENT, False),
|
||||
(CaptureMessageContent.SPAN_ONLY, True),
|
||||
(CaptureMessageContent.EVENT_ONLY, False),
|
||||
(CaptureMessageContent.SPAN_AND_EVENT, True),
|
||||
],
|
||||
)
|
||||
def test_only_a_request_destination_can_turn_on_llm_payload_content(
|
||||
capture_mode: CaptureMessageContent | None, captures_content: bool
|
||||
) -> None:
|
||||
config: Final = OpenTelemetryV2Config(
|
||||
exporter="in_memory",
|
||||
capture_message_content=CaptureMessageContent.NO_CONTENT,
|
||||
)
|
||||
exporter: Final = InMemorySpanExporter()
|
||||
logger: Final = OpenTelemetryV2(
|
||||
config=config, tracer_provider=providers.build_tracer_provider(config, exporter=exporter)
|
||||
)
|
||||
payload: Final = _payload(
|
||||
messages=[{"role": "user", "content": "prompt marker"}],
|
||||
response={
|
||||
"id": "resp_1",
|
||||
"model": "gpt-4o-2024",
|
||||
"choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "answer marker"}}],
|
||||
},
|
||||
)
|
||||
kwargs: Final = {
|
||||
**_kwargs(payload),
|
||||
"standard_callback_dynamic_params": {"capture_message_content": "span_only"},
|
||||
}
|
||||
destination: Final = OtelDestination(
|
||||
endpoint="http://team.local/api/public/otel",
|
||||
headers=MappingProxyType({"Authorization": "Basic dA=="}),
|
||||
callback_name="langfuse_otel",
|
||||
capture_message_content=capture_mode,
|
||||
)
|
||||
|
||||
def emit() -> None:
|
||||
set_request_destinations((destination,))
|
||||
_emit_llm(logger, kwargs)
|
||||
|
||||
contextvars.copy_context().run(emit)
|
||||
|
||||
(span,) = exporter.get_finished_spans()
|
||||
assert ("gen_ai.input.messages" in span.attributes) is captures_content
|
||||
assert ("gen_ai.output.messages" in span.attributes) is captures_content
|
||||
|
||||
|
||||
def test_mcp_tool_call_failure_marks_error():
|
||||
logger, exporter = _logger()
|
||||
payload = _mcp_payload(
|
||||
|
|
|
|||
|
|
@ -1692,6 +1692,19 @@ def test_capture_message_content_normalizer_only_touches_strings():
|
|||
OpenTelemetryV2Config(capture_message_content=123)
|
||||
|
||||
|
||||
def test_unknown_capture_message_content_env_is_kept_as_a_plain_string_that_captures_nothing(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "Invalid-Mode")
|
||||
|
||||
config: Final = OpenTelemetryV2Config()
|
||||
|
||||
assert type(config.capture_message_content) is str
|
||||
assert config.capture_message_content == "invalid-mode"
|
||||
assert config.capture_span_content is False
|
||||
assert config.model_dump()["capture_message_content"] == "invalid-mode"
|
||||
|
||||
|
||||
def test_v2_flag_is_off_by_default(monkeypatch):
|
||||
monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
|
|
|
|||
|
|
@ -1,7 +1,11 @@
|
|||
from collections.abc import Iterator
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.proxy.common_utils.callback_config_validation import (
|
||||
callback_config_error,
|
||||
conflicting_capture_error,
|
||||
conflicting_span_scope_error,
|
||||
cross_entry_family_error,
|
||||
logging_metadata_config_error,
|
||||
|
|
@ -259,3 +263,114 @@ def test_arize_sampling_rates_are_not_family_credentials():
|
|||
stored = [{"arize_api_key": "k1", "arize_success_sampling_rate": "0.5"}]
|
||||
assert cross_entry_family_error({"arize_success_sampling_rate": "0.1"}, stored) is None
|
||||
assert cross_entry_family_error({"arize_error_sampling_rate": "0.5"}, stored) is None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def otel_v2_on(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
||||
from litellm.integrations.otel.model.config import is_otel_v2_enabled
|
||||
|
||||
monkeypatch.setenv("LITELLM_OTEL_V2", "true")
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
yield
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("otel_v2_on")
|
||||
@pytest.mark.parametrize("callback_name", ["langfuse_otel", "arize", "weave_otel", "newrelic"])
|
||||
@pytest.mark.parametrize("value", ["no_content", "span_only"])
|
||||
def test_capture_message_content_is_accepted_on_every_otel_v2_destination(callback_name: str, value: str) -> None:
|
||||
assert callback_config_error(callback_name, {"capture_message_content": 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
|
||||
|
||||
monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
try:
|
||||
error: Final = callback_config_error("langfuse_otel", {"capture_message_content": "no_content"})
|
||||
finally:
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
assert error is not None and "capture_message_content" 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("value", ["full", "event_only", "span_and_event"])
|
||||
def test_an_unsupported_capture_message_content_is_rejected_on_key_logging_metadata(value: str) -> None:
|
||||
error: Final = logging_metadata_config_error(
|
||||
{
|
||||
"logging": [
|
||||
{
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": "success",
|
||||
"callback_vars": {"capture_message_content": value},
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
assert error is not None and "Invalid capture_message_content" in error
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("otel_v2_on")
|
||||
def test_each_entry_keeps_its_own_capture_message_content() -> None:
|
||||
error: Final = logging_metadata_config_error(
|
||||
{
|
||||
"logging": [
|
||||
{
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": "success",
|
||||
"callback_vars": {"capture_message_content": "span_only"},
|
||||
},
|
||||
{
|
||||
"callback_name": "arize",
|
||||
"callback_type": "success",
|
||||
"callback_vars": {"capture_message_content": "no_content"},
|
||||
},
|
||||
]
|
||||
}
|
||||
)
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("otel_v2_on")
|
||||
def test_a_failure_only_entry_rejects_capture_message_content() -> None:
|
||||
error: Final = callback_config_error("langfuse_otel", {"capture_message_content": "no_content"}, "failure")
|
||||
assert error is not None and "capture_message_content" in error and "success_and_failure" in error
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("stored", "rejected"),
|
||||
[
|
||||
([("langfuse_otel", {"capture_message_content": "no_content"})], True),
|
||||
([("langfuse_otel", {"capture_message_content": "span_only"})], False),
|
||||
([("langfuse_otel", {"langfuse_public_key": "pk"})], False),
|
||||
([("newrelic", {"capture_message_content": "no_content"})], False),
|
||||
],
|
||||
)
|
||||
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)
|
||||
assert (error is not None) is rejected
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("otel_v2_on")
|
||||
def test_key_logging_entries_for_one_backend_may_not_disagree_on_capture_message_content() -> None:
|
||||
def entry(callback_type: str, capture: str) -> dict[str, str | dict[str, str]]:
|
||||
return {
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": callback_type,
|
||||
"callback_vars": {"capture_message_content": capture},
|
||||
}
|
||||
|
||||
error: Final = logging_metadata_config_error(
|
||||
{"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
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ redacted audit rows for callback mutations.
|
|||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Iterator
|
||||
from typing import Final
|
||||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||
|
||||
|
|
@ -1712,3 +1713,62 @@ def test_add_team_callback_accepts_arize_sampling_rate_vars():
|
|||
)
|
||||
assert data.callback_vars["arize_success_sampling_rate"] == "0.5"
|
||||
assert data.callback_vars["arize_error_sampling_rate"] == "1.0"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_team_callbacks_rejects_capture_message_content_on_a_non_otel_v2_callback(patched_prisma):
|
||||
data: Final = AddTeamCallback(
|
||||
callback_name="langfuse",
|
||||
callback_type="success",
|
||||
callback_vars={
|
||||
"langfuse_public_key": "pk",
|
||||
"langfuse_secret_key": "sk",
|
||||
"capture_message_content": "no_content",
|
||||
},
|
||||
)
|
||||
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 "capture_message_content" in str(exc.value.detail)
|
||||
patched_prisma.db.litellm_teamtable.update.assert_not_called()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def otel_v2_on(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
||||
from litellm.integrations.otel.model.config import is_otel_v2_enabled
|
||||
|
||||
monkeypatch.setenv("LITELLM_OTEL_V2", "true")
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
yield
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("otel_v2_on")
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("capture", ["no_content", "span_only"])
|
||||
async def test_add_team_callbacks_stores_capture_message_content_in_the_existing_callback_metadata(
|
||||
patched_prisma, capture: str
|
||||
) -> None:
|
||||
data: Final = AddTeamCallback(
|
||||
callback_name="langfuse_otel",
|
||||
callback_type="success",
|
||||
callback_vars={
|
||||
"langfuse_public_key": "pk",
|
||||
"langfuse_secret_key": "sk",
|
||||
"capture_message_content": capture,
|
||||
},
|
||||
)
|
||||
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()
|
||||
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]
|
||||
|
|
|
|||
40
tests/unit/types/test_utils.py
Normal file
40
tests/unit/types/test_utils.py
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
import pytest
|
||||
|
||||
from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VALUES, CaptureMessageContent, captures_span_content
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mode",
|
||||
[
|
||||
CaptureMessageContent.NO_CONTENT,
|
||||
CaptureMessageContent.SPAN_ONLY,
|
||||
CaptureMessageContent.EVENT_ONLY,
|
||||
CaptureMessageContent.SPAN_AND_EVENT,
|
||||
],
|
||||
)
|
||||
def test_capture_message_content_members_are_plain_strings(mode: str) -> None:
|
||||
assert type(mode) is str
|
||||
assert (str(mode), f"{mode}", repr(mode)) == (mode, mode, repr(str(mode)))
|
||||
|
||||
|
||||
def test_a_destination_may_only_set_the_span_capture_modes() -> None:
|
||||
assert CAPTURE_MESSAGE_CONTENT_VALUES == {"no_content", "span_only"}
|
||||
|
||||
|
||||
def test_capture_message_content_constructor_accepts_any_string() -> None:
|
||||
assert CaptureMessageContent("custom") == "custom"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mode", "captures"),
|
||||
[
|
||||
("no_content", False),
|
||||
("span_only", True),
|
||||
("event_only", False),
|
||||
("span_and_event", True),
|
||||
("bogus", False),
|
||||
(None, False),
|
||||
],
|
||||
)
|
||||
def test_captures_span_content(mode: str | None, captures: bool) -> None:
|
||||
assert captures_span_content(mode) is captures
|
||||
3
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
3
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -17246,6 +17246,9 @@ export interface paths {
|
|||
* - langfuse_host: The host for the Langfuse callback
|
||||
* - langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)
|
||||
* - langfuse_span_scope: For langfuse_otel, "full" (default) sends the whole request trace, "llm_only" sends only the model-call spans
|
||||
* - 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
|
||||
* - 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