fix(passthrough): emit otel guardrail span when a guardrail blocks (#29470)
* fix(passthrough): emit otel guardrail span when a guardrail blocks The otel_v2 logger emits guardrail spans from its post-call hooks by reading standard_logging_guardrail_information off the top-level metadata of the dict handed to those hooks. On passthrough, post-call guardrails run against a throwaway hook_data dict (metadata was already stripped off _parsed_body by _init_kwargs_for_pass_through_endpoint), so a deny that raises a non ModifyResponseException records its logging info on hook_data and then the generic failure handler forwards _parsed_body, which no longer carries it. The span was therefore present on allow but missing on block; the unified path keeps metadata on the same dict it passes to the failure hook, so its span always shows. Carry the guardrail logging entries recorded on hook_data over to the request_data forwarded to post_call_failure_hook so the failure path matches the unified path. Resolves LIT-3510 * test(passthrough): cover guardrail-logging carry helper; simplify helper Address review feedback on the guardrail-block span fix. Simplify _carry_guardrail_logging_info: the realistic failure path always builds fresh metadata on request_data, so the merge-into-existing-list branch was dead code. Use setdefault with a shallow-copied list so the carried entries never share the source hook_data list reference. Drop the module-level sys.modules proxy_server mock from the otel span test; pass_through_endpoints imports proxy_server lazily, so it is unnecessary and avoided the test-isolation risk of registering a mock under that key. Add pure unit tests for _carry_guardrail_logging_info (no otel dependency) that pin its contract: carries entries, copies the list, populates existing metadata without clobbering prior guardrail entries, and no-ops when there is nothing to carry. * test(passthrough): cover deny-path guardrail logging forwarding without otel The otel span regression test skips in coverage jobs that lack the optional opentelemetry package, leaving the failure-handler wiring (capturing hook_data and carrying its guardrail logging info) uncovered. Add an otel-independent regression that drives the real pass_through_request through a post-call deny and asserts post_call_failure_hook receives request_data carrying the standard_logging_guardrail_information. Fails on the pre-fix code. --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Yassin Kortam
GitHub
parent
b98a656254
commit
ce7b1fd29d
@ -667,6 +667,34 @@ class HttpPassThroughEndpointHelpers(BasePassthroughUtils):
|
||||
return stream
|
||||
|
||||
|
||||
def _carry_guardrail_logging_info(
|
||||
request_data: dict, guardrail_data: Optional[dict]
|
||||
) -> None:
|
||||
"""Copy guardrail logging entries from ``guardrail_data`` onto ``request_data``.
|
||||
|
||||
Post-call guardrails run against a throwaway ``hook_data`` dict (its
|
||||
``metadata`` is what ``_init_kwargs_for_pass_through_endpoint`` already
|
||||
stripped off ``_parsed_body``), so a block records the
|
||||
``standard_logging_guardrail_information`` there and not on the dict the
|
||||
failure handler forwards to ``post_call_failure_hook``. Without this the
|
||||
otel guardrail span is emitted on allow but missing on block. Carry the
|
||||
entries over so the failure path matches the unified path.
|
||||
"""
|
||||
if guardrail_data is None:
|
||||
return
|
||||
source_metadata = guardrail_data.get("metadata")
|
||||
if not isinstance(source_metadata, dict):
|
||||
return
|
||||
entries = source_metadata.get("standard_logging_guardrail_information")
|
||||
if not entries:
|
||||
return
|
||||
|
||||
metadata = request_data.get("metadata")
|
||||
if not isinstance(metadata, dict):
|
||||
metadata = request_data["metadata"] = {}
|
||||
metadata.setdefault("standard_logging_guardrail_information", list(entries))
|
||||
|
||||
|
||||
async def pass_through_request( # noqa: PLR0915
|
||||
request: Request,
|
||||
target: str,
|
||||
@ -718,6 +746,9 @@ async def pass_through_request( # noqa: PLR0915
|
||||
# kwargs for pass through endpoint, contains metadata, litellm_params, call_type, litellm_call_id, passthrough_logging_payload
|
||||
kwargs: Optional[dict] = None
|
||||
logging_obj: Optional[Logging] = None
|
||||
# the dict post-call guardrails wrote their logging info into; the failure
|
||||
# handler reuses it so a guardrail block still surfaces its span/logs
|
||||
post_call_guardrail_data: Optional[dict] = None
|
||||
|
||||
#########################################################
|
||||
try:
|
||||
@ -1160,6 +1191,7 @@ async def pass_through_request( # noqa: PLR0915
|
||||
**existing_metadata,
|
||||
"guardrails": guardrails_to_run,
|
||||
}
|
||||
post_call_guardrail_data = hook_data
|
||||
response_body = await proxy_logging_obj.post_call_success_hook(
|
||||
data=hook_data,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
@ -1343,6 +1375,8 @@ async def pass_through_request( # noqa: PLR0915
|
||||
if "custom_llm_provider" not in request_payload and custom_llm_provider:
|
||||
request_payload["custom_llm_provider"] = custom_llm_provider
|
||||
|
||||
_carry_guardrail_logging_info(request_payload, post_call_guardrail_data)
|
||||
|
||||
await proxy_logging_obj.post_call_failure_hook(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
original_exception=e,
|
||||
|
||||
@ -0,0 +1,68 @@
|
||||
"""Unit tests for ``_carry_guardrail_logging_info``.
|
||||
|
||||
This is the helper that lets a passthrough guardrail block still surface its otel
|
||||
span: it copies ``standard_logging_guardrail_information`` from the post-call
|
||||
guardrail's (otherwise discarded) ``hook_data`` onto the dict the failure handler
|
||||
forwards to ``post_call_failure_hook``. No otel dependency here, so these run
|
||||
everywhere and pin the helper's contract directly.
|
||||
"""
|
||||
|
||||
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
|
||||
_carry_guardrail_logging_info,
|
||||
)
|
||||
|
||||
_ENTRY = {"guardrail_name": "block-demo", "guardrail_status": "guardrail_intervened"}
|
||||
|
||||
|
||||
def _source(entries):
|
||||
return {"metadata": {"standard_logging_guardrail_information": entries}}
|
||||
|
||||
|
||||
def test_carries_entries_onto_request_without_metadata():
|
||||
request_data: dict = {}
|
||||
_carry_guardrail_logging_info(request_data, _source([_ENTRY]))
|
||||
assert request_data["metadata"]["standard_logging_guardrail_information"] == [
|
||||
_ENTRY
|
||||
]
|
||||
|
||||
|
||||
def test_carried_list_is_copied_not_shared():
|
||||
source = _source([_ENTRY])
|
||||
request_data: dict = {}
|
||||
_carry_guardrail_logging_info(request_data, source)
|
||||
carried = request_data["metadata"]["standard_logging_guardrail_information"]
|
||||
assert carried is not source["metadata"]["standard_logging_guardrail_information"]
|
||||
carried.append({"guardrail_name": "other"})
|
||||
assert source["metadata"]["standard_logging_guardrail_information"] == [_ENTRY]
|
||||
|
||||
|
||||
def test_existing_metadata_without_guardrail_key_is_populated():
|
||||
request_data: dict = {"metadata": {"user_api_key": "sk-x"}}
|
||||
_carry_guardrail_logging_info(request_data, _source([_ENTRY]))
|
||||
assert request_data["metadata"]["user_api_key"] == "sk-x"
|
||||
assert request_data["metadata"]["standard_logging_guardrail_information"] == [
|
||||
_ENTRY
|
||||
]
|
||||
|
||||
|
||||
def test_existing_guardrail_entries_are_not_clobbered():
|
||||
existing = [{"guardrail_name": "already-logged"}]
|
||||
request_data = {"metadata": {"standard_logging_guardrail_information": existing}}
|
||||
_carry_guardrail_logging_info(request_data, _source([_ENTRY]))
|
||||
assert (
|
||||
request_data["metadata"]["standard_logging_guardrail_information"] is existing
|
||||
)
|
||||
|
||||
|
||||
def test_noop_when_guardrail_data_is_none():
|
||||
request_data: dict = {}
|
||||
_carry_guardrail_logging_info(request_data, None)
|
||||
assert request_data == {}
|
||||
|
||||
|
||||
def test_noop_when_no_guardrail_entries():
|
||||
request_data: dict = {}
|
||||
_carry_guardrail_logging_info(request_data, {"metadata": {}})
|
||||
_carry_guardrail_logging_info(request_data, _source([]))
|
||||
_carry_guardrail_logging_info(request_data, {})
|
||||
assert request_data == {}
|
||||
@ -0,0 +1,189 @@
|
||||
"""Regression: a guardrail block on a passthrough endpoint must still emit the
|
||||
otel guardrail span.
|
||||
|
||||
Before the fix the post-call guardrail recorded its
|
||||
``standard_logging_guardrail_information`` onto a throwaway ``hook_data`` dict,
|
||||
which the failure handler discarded. So ``pass_through_request`` forwarded a
|
||||
``request_data`` without it to ``post_call_failure_hook`` and the otel guardrail
|
||||
span (emitted from that hook) was present on allow but missing on block. The
|
||||
unified path always has it. These tests drive the real ``pass_through_request``
|
||||
with a real ``ProxyLogging`` + a real otel V2 logger and assert the span is
|
||||
emitted on both allow and block.
|
||||
"""
|
||||
|
||||
import json
|
||||
from contextlib import ExitStack
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
pytest.importorskip("opentelemetry")
|
||||
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( # noqa: E402
|
||||
InMemorySpanExporter,
|
||||
)
|
||||
|
||||
import litellm # noqa: E402
|
||||
from litellm.caching.dual_cache import DualCache # noqa: E402
|
||||
from litellm.integrations.custom_guardrail import ( # noqa: E402
|
||||
CustomGuardrail,
|
||||
log_guardrail_information,
|
||||
)
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402
|
||||
from litellm.integrations.otel.model.config import OpenTelemetryV2Config # noqa: E402
|
||||
from litellm.integrations.otel.plumbing import providers # noqa: E402
|
||||
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache # noqa: E402
|
||||
from litellm.proxy.utils import ProxyLogging # noqa: E402
|
||||
from litellm.types.guardrails import GuardrailEventHooks # noqa: E402
|
||||
|
||||
_PT_MOD = "litellm.proxy.pass_through_endpoints.pass_through_endpoints"
|
||||
_COLLECT = (
|
||||
"litellm.proxy.pass_through_endpoints.passthrough_guardrails."
|
||||
"PassthroughGuardrailHandler.collect_guardrails"
|
||||
)
|
||||
_GUARDRAIL_SPAN = "execute_guardrail block-demo"
|
||||
_TRIGGER = "BLOCKME"
|
||||
|
||||
# pass_through_endpoints imports proxy_server lazily (inside the request
|
||||
# function), so importing this at module scope does not require the real
|
||||
# proxy_server and does not mutate sys.modules.
|
||||
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( # noqa: E402
|
||||
pass_through_request,
|
||||
)
|
||||
|
||||
|
||||
class _BlockOnTextGuardrail(CustomGuardrail):
|
||||
"""Denies (HTTP 400) when the response carries the trigger word; records its
|
||||
standard guardrail logging info on both allow and block via the decorator."""
|
||||
|
||||
@log_guardrail_information
|
||||
async def async_post_call_success_hook(self, data, user_api_key_dict, response):
|
||||
if _TRIGGER in json.dumps(response):
|
||||
raise HTTPException(
|
||||
status_code=400, detail={"error": "blocked by block-demo guardrail"}
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
def _user_api_key_dict():
|
||||
d = MagicMock()
|
||||
d.api_key = "sk-test"
|
||||
d.user_id = "user-1"
|
||||
d.team_id = "team-1"
|
||||
d.org_id = None
|
||||
d.metadata = {}
|
||||
d.team_metadata = {}
|
||||
d.parent_otel_span = None
|
||||
d.request_route = "/mock/echo"
|
||||
return d
|
||||
|
||||
|
||||
def _mock_request():
|
||||
r = MagicMock()
|
||||
r.method = "POST"
|
||||
r.query_params = {}
|
||||
r.url = "http://testserver/mock/echo"
|
||||
headers = MagicMock()
|
||||
headers.copy.return_value = {}
|
||||
r.headers = headers
|
||||
return r
|
||||
|
||||
|
||||
def _httpx_response(text: str) -> httpx.Response:
|
||||
body = {"candidates": [{"content": {"role": "model", "parts": [{"text": text}]}}]}
|
||||
return httpx.Response(
|
||||
status_code=200,
|
||||
headers={"content-type": "application/json"},
|
||||
content=json.dumps(body).encode("utf-8"),
|
||||
request=httpx.Request("POST", "https://upstream.example/echo"),
|
||||
)
|
||||
|
||||
|
||||
def _otel_logger_with_exporter():
|
||||
cfg = OpenTelemetryV2Config(exporter="in_memory")
|
||||
exporter = InMemorySpanExporter()
|
||||
tracer_provider = providers.build_tracer_provider(cfg, exporter=exporter)
|
||||
return OpenTelemetryV2(config=cfg, tracer_provider=tracer_provider), exporter
|
||||
|
||||
|
||||
def _guardrail_span_names(exporter):
|
||||
return [
|
||||
s.name
|
||||
for s in exporter.get_finished_spans()
|
||||
if s.name.startswith("execute_guardrail")
|
||||
]
|
||||
|
||||
|
||||
async def _drive(response_text: str):
|
||||
"""Run the real pass_through_request with the block-demo guardrail + otel V2
|
||||
logger registered, returning (status_code, guardrail_span_names)."""
|
||||
otel, exporter = _otel_logger_with_exporter()
|
||||
guardrail = _BlockOnTextGuardrail(
|
||||
guardrail_name="block-demo", event_hook=[GuardrailEventHooks.post_call]
|
||||
)
|
||||
proxy_logging = ProxyLogging(user_api_key_cache=UserApiKeyCache(DualCache()))
|
||||
|
||||
saved_callbacks = list(litellm.callbacks)
|
||||
litellm.callbacks = [guardrail, otel]
|
||||
|
||||
mock_async_client_obj = MagicMock()
|
||||
mock_async_client_obj.client = AsyncMock()
|
||||
mock_pt_logging = MagicMock()
|
||||
mock_pt_logging.pass_through_async_success_handler = AsyncMock()
|
||||
|
||||
patches = [
|
||||
patch(
|
||||
f"{_PT_MOD}.HttpPassThroughEndpointHelpers.non_streaming_http_request_handler",
|
||||
new_callable=AsyncMock,
|
||||
return_value=_httpx_response(response_text),
|
||||
),
|
||||
patch(f"{_PT_MOD}._is_streaming_response", return_value=False),
|
||||
patch("litellm.proxy.proxy_server.proxy_logging_obj", proxy_logging),
|
||||
patch("litellm.proxy.proxy_server.llm_router", None),
|
||||
patch(f"{_PT_MOD}.pass_through_endpoint_logging", mock_pt_logging),
|
||||
patch(f"{_PT_MOD}.get_async_httpx_client", return_value=mock_async_client_obj),
|
||||
patch(f"{_PT_MOD}._read_request_body", new_callable=AsyncMock, return_value={}),
|
||||
patch(f"{_PT_MOD}._safe_get_request_headers", return_value={}),
|
||||
patch(_COLLECT, return_value=["block-demo"]),
|
||||
]
|
||||
try:
|
||||
with ExitStack() as stack:
|
||||
for p in patches:
|
||||
stack.enter_context(p)
|
||||
try:
|
||||
result = await pass_through_request(
|
||||
request=_mock_request(),
|
||||
target="https://upstream.example/echo",
|
||||
custom_headers={"Content-Type": "application/json"},
|
||||
user_api_key_dict=_user_api_key_dict(),
|
||||
stream=False,
|
||||
)
|
||||
# A deny (HTTP 4xx) re-raises as ProxyException; an allow returns
|
||||
# the upstream Response.
|
||||
status_code = result.status_code
|
||||
except Exception as e:
|
||||
status_code = getattr(e, "code", None) or getattr(
|
||||
e, "status_code", None
|
||||
)
|
||||
return int(status_code), _guardrail_span_names(exporter)
|
||||
finally:
|
||||
litellm.callbacks = saved_callbacks
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_guardrail_block_emits_otel_guardrail_span():
|
||||
status_code, span_names = await _drive(f"{_TRIGGER} please")
|
||||
assert status_code == 400
|
||||
assert span_names == [_GUARDRAIL_SPAN], (
|
||||
"guardrail span must be emitted when a passthrough guardrail blocks, "
|
||||
f"got spans: {span_names}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_guardrail_allow_emits_otel_guardrail_span():
|
||||
status_code, span_names = await _drive("hello world")
|
||||
assert status_code == 200
|
||||
assert span_names == [_GUARDRAIL_SPAN]
|
||||
@ -12,6 +12,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from litellm.integrations.custom_guardrail import (
|
||||
CustomGuardrail,
|
||||
@ -216,6 +217,53 @@ class TestPassthroughPostCallGuardrails:
|
||||
assert body["error"]["guardrail_name"] == "rubrik"
|
||||
assert body["error"]["model"] == "gemini-2.0-flash"
|
||||
|
||||
@patch(_COLLECT, return_value=["rubrik"])
|
||||
async def test_deny_forwards_guardrail_logging_info_to_failure_hook(
|
||||
self,
|
||||
mock_collect,
|
||||
):
|
||||
"""A post-call guardrail deny (non-ModifyResponseException) records its
|
||||
standard_logging_guardrail_information on the hook_data dict; the failure
|
||||
handler must forward that info to post_call_failure_hook so downstream
|
||||
loggers (e.g. the otel guardrail span) still see it. Regression for the
|
||||
block path dropping it."""
|
||||
mock_response = _make_httpx_response(_GEMINI_RESPONSE)
|
||||
|
||||
def _block(*, data, user_api_key_dict, response):
|
||||
metadata = data.setdefault("metadata", {})
|
||||
metadata.setdefault("standard_logging_guardrail_information", []).append(
|
||||
{"guardrail_name": "rubrik", "guardrail_status": "guardrail_intervened"}
|
||||
)
|
||||
raise HTTPException(status_code=400, detail={"error": "blocked"})
|
||||
|
||||
captured = {}
|
||||
|
||||
async def _capture_failure(**kwargs):
|
||||
captured.update(kwargs)
|
||||
|
||||
mock_proxy_logging = MagicMock()
|
||||
mock_proxy_logging.pre_call_hook = AsyncMock(return_value={})
|
||||
mock_proxy_logging.post_call_success_hook = AsyncMock(side_effect=_block)
|
||||
mock_proxy_logging.post_call_failure_hook = AsyncMock(
|
||||
side_effect=_capture_failure
|
||||
)
|
||||
|
||||
with _common_patches(mock_proxy_logging, mock_response):
|
||||
with pytest.raises(Exception):
|
||||
await pass_through_request(
|
||||
request=_make_mock_request(),
|
||||
target="https://example.com/v1/generateContent",
|
||||
custom_headers={"Content-Type": "application/json"},
|
||||
user_api_key_dict=_make_user_api_key_dict(),
|
||||
stream=False,
|
||||
)
|
||||
|
||||
mock_proxy_logging.post_call_failure_hook.assert_awaited_once()
|
||||
entries = captured["request_data"]["metadata"][
|
||||
"standard_logging_guardrail_information"
|
||||
]
|
||||
assert any(e.get("guardrail_name") == "rubrik" for e in entries)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestUnifiedGuardrailCallTypeResolution:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user