mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
fix(ci): namespace claude session ids in tracing seeds and allowlist /v1/logs on backend (#44761)
* fix(ci): namespace claude session ids in tracing seeds and allowlist /v1/logs on backend Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(proxy): allowlist /v1/logs on the gateway alongside /v1/traces Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
parent
bc2df7cdcd
commit
ff5084687c
5 changed files with 127 additions and 2 deletions
|
|
@ -85,6 +85,7 @@ BACKEND_PATH_PREFIXES: tuple[str, ...] = (
|
|||
"/analytics/",
|
||||
"/lens/",
|
||||
"/v1/traces",
|
||||
"/v1/logs",
|
||||
"/global/",
|
||||
"/user_agent",
|
||||
"/usage/",
|
||||
|
|
|
|||
|
|
@ -76,6 +76,7 @@ GATEWAY_PATH_PREFIXES: tuple[str, ...] = (
|
|||
"/containers",
|
||||
"/v1/evals",
|
||||
"/v1/traces",
|
||||
"/v1/logs",
|
||||
"/v1/memory",
|
||||
"/queue/chat/",
|
||||
# Google data plane (v1beta is the Google AI Studio version)
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ SPEND_ROWS: Final = TypeAdapter(tuple[SpendLogRecord, ...])
|
|||
TRACE: Final = TypeAdapter(Trace)
|
||||
NANOSECOND_FIELDS: Final = frozenset({"startTimeUnixNano", "endTimeUnixNano", "timeUnixNano"})
|
||||
TRACE_ID_FIELDS: Final = frozenset({"traceId", "trace_id", "session_id"})
|
||||
TRACE_ID_ATTRIBUTES: Final = frozenset({"session.id"})
|
||||
SPAN_ID_FIELDS: Final = frozenset({"spanId", "parentSpanId", "span_id"})
|
||||
COPY_WINDOW_MS: Final = 24 * 60 * 60 * 1000
|
||||
LONG_SESSION_SOURCE: Final = "openai_agents_swarm"
|
||||
|
|
@ -191,6 +192,32 @@ def rebase(
|
|||
if isinstance(value, list):
|
||||
return [rebase(item, offset_ns, namespace, response_pattern) for item in value]
|
||||
if isinstance(value, dict):
|
||||
attribute_key: Final = value.get("key")
|
||||
attribute_value: Final = value.get("value")
|
||||
session_id: Final = (
|
||||
attribute_value.get("stringValue") if isinstance(attribute_value, dict) else None
|
||||
)
|
||||
if (
|
||||
isinstance(attribute_key, str)
|
||||
and attribute_key in TRACE_ID_ATTRIBUTES
|
||||
and isinstance(attribute_value, dict)
|
||||
and isinstance(session_id, str)
|
||||
):
|
||||
return {
|
||||
**{
|
||||
key: rebase(item, offset_ns, namespace, response_pattern, key)
|
||||
for key, item in value.items()
|
||||
if key != "value"
|
||||
},
|
||||
"value": {
|
||||
**{
|
||||
key: rebase(item, offset_ns, namespace, response_pattern, key)
|
||||
for key, item in attribute_value.items()
|
||||
if key != "stringValue"
|
||||
},
|
||||
"stringValue": seed_id(session_id, namespace, 32),
|
||||
},
|
||||
}
|
||||
return {key: rebase(item, offset_ns, namespace, response_pattern, key) for key, item in value.items()}
|
||||
return value
|
||||
|
||||
|
|
|
|||
|
|
@ -30,9 +30,10 @@ from collections.abc import AsyncGenerator, Mapping
|
|||
from contextlib import asynccontextmanager
|
||||
from functools import partial
|
||||
from typing import Final, Literal
|
||||
from unittest.mock import AsyncMock, MagicMock, call
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from starlette.applications import Starlette
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import JSONResponse
|
||||
|
|
@ -64,8 +65,14 @@ if _REPO_ROOT not in sys.path:
|
|||
|
||||
from backend.routes.allowlist import BACKEND_MOUNT_PATHS
|
||||
from gateway.routes.allowlist import GATEWAY_MOUNT_PATHS
|
||||
from litellm.proxy import tracing_endpoints
|
||||
from litellm.proxy._lazy_features import LazyFeature, attach_lazy_features
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
from litellm.proxy.auth.authorization_dependencies import get_log_team_lookup
|
||||
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
|
||||
from litellm.proxy.proxy_server import app
|
||||
from litellm.rust_bridge.trace.storage import ClickHouseStorage
|
||||
from litellm.tracing import Tenant, TraceReceiver
|
||||
from tests.test_litellm_rust.support.child_interpreter import run_child_interpreter
|
||||
|
||||
for _key, _previous in _PRE_EXISTING_ENV.items():
|
||||
|
|
@ -222,6 +229,78 @@ def test_composed_lifespan_propagates_lifecycle_failures(
|
|||
assert events == (["startup"] if phase == "startup" else ["startup", "serving", "shutdown"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"component_lifespan", (_gateway_lifespan, _backend_lifespan), ids=("gateway", "backend")
|
||||
)
|
||||
@pytest.mark.parametrize("endpoint", ("/v1/traces", "/v1/logs"), ids=("traces", "logs"))
|
||||
def test_otlp_ingest_routes_authenticate_and_isolate_tenants_on_each_component(
|
||||
component_lifespan: Lifespan[Starlette], endpoint: str
|
||||
) -> None:
|
||||
application: Final = FastAPI()
|
||||
application.include_router(tracing_endpoints.router)
|
||||
storage: Final = MagicMock(spec=ClickHouseStorage)
|
||||
storage.ingest = AsyncMock(return_value=1)
|
||||
application.dependency_overrides[tracing_endpoints.provide_receiver] = lambda: TraceReceiver(storage)
|
||||
|
||||
async def lookup(auth: UserAPIKeyAuth) -> tuple[str, ...]:
|
||||
return ()
|
||||
|
||||
application.dependency_overrides[get_log_team_lookup] = lambda: lookup
|
||||
|
||||
def authenticate(request: Request) -> UserAPIKeyAuth:
|
||||
match request.headers.get("Authorization"):
|
||||
case "Bearer team-a-key":
|
||||
return UserAPIKeyAuth(
|
||||
user_id="user-a",
|
||||
token="hashed-a",
|
||||
team_id="team-a",
|
||||
org_id="org-a",
|
||||
user_role=LitellmUserRoles.INTERNAL_USER,
|
||||
)
|
||||
case "Bearer team-b-key":
|
||||
return UserAPIKeyAuth(
|
||||
user_id="user-b",
|
||||
token="hashed-b",
|
||||
team_id="team-b",
|
||||
org_id="org-b",
|
||||
user_role=LitellmUserRoles.INTERNAL_USER,
|
||||
)
|
||||
case _:
|
||||
raise HTTPException(status_code=401, detail="Invalid API key")
|
||||
|
||||
application.dependency_overrides[user_api_key_auth] = authenticate
|
||||
application.router.lifespan_context = partial(
|
||||
component_lifespan, lifespan=application.router.lifespan_context
|
||||
)
|
||||
|
||||
body: Final = b'{"resourceLogs": []}'
|
||||
content_type: Final = "application/json"
|
||||
with TestClient(application) as client:
|
||||
unauthenticated: Final = client.post(endpoint, content=body, headers={"content-type": content_type})
|
||||
assert unauthenticated.status_code == 401, unauthenticated.text
|
||||
|
||||
team_a: Final = client.post(
|
||||
endpoint,
|
||||
content=body,
|
||||
headers={"Authorization": "Bearer team-a-key", "content-type": content_type},
|
||||
)
|
||||
assert team_a.status_code == 200, team_a.text
|
||||
|
||||
team_b: Final = client.post(
|
||||
endpoint,
|
||||
content=body,
|
||||
headers={"Authorization": "Bearer team-b-key", "content-type": content_type},
|
||||
)
|
||||
assert team_b.status_code == 200, team_b.text
|
||||
|
||||
tenant_a: Final = Tenant(team_id="team-a", api_key_hash="hashed-a", org_id="org-a", user_id="user-a")
|
||||
tenant_b: Final = Tenant(team_id="team-b", api_key_hash="hashed-b", org_id="org-b", user_id="user-b")
|
||||
assert storage.ingest.await_args_list == [
|
||||
call(body, content_type, tenant_a, endpoint == "/v1/logs"),
|
||||
call(body, content_type, tenant_b, endpoint == "/v1/logs"),
|
||||
]
|
||||
|
||||
|
||||
def test_gateway_plus_backend_covers_full_app():
|
||||
"""Every route on the proxy app must be served by gateway or backend.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from datetime import datetime
|
||||
|
|
@ -64,7 +65,23 @@ def test_all_fixture_replays_are_recent_and_preserve_spans(path: Path) -> None:
|
|||
trace_id, span_id, parent_id, timestamp = SPAN_IDENTITY.validate_python(
|
||||
(before["TraceId"], before["SpanId"], before["ParentSpanId"], before["Timestamp"])
|
||||
)
|
||||
assert after["TraceId"] == seed_id(trace_id, replay.namespace, 32)
|
||||
span_attributes: Final = before["SpanAttributes"]
|
||||
if isinstance(span_attributes, dict) and "lens.original_trace_id" in span_attributes:
|
||||
before_original_trace_id: Final = span_attributes["lens.original_trace_id"]
|
||||
before_session: Final = span_attributes["session.id"]
|
||||
assert isinstance(before_original_trace_id, str)
|
||||
assert isinstance(before_session, str)
|
||||
after_span_attributes: Final = after["SpanAttributes"]
|
||||
assert isinstance(after_span_attributes, dict)
|
||||
assert after_span_attributes["lens.original_trace_id"] == seed_id(
|
||||
before_original_trace_id, replay.namespace, 32
|
||||
)
|
||||
assert after["TraceId"] == hashlib.sha256(
|
||||
f"litellm.claude.session.v1\0{seed_id(before_session, replay.namespace, 32)}".encode()
|
||||
).hexdigest()[:32]
|
||||
assert after["TraceId"] != before["TraceId"]
|
||||
else:
|
||||
assert after["TraceId"] == seed_id(trace_id, replay.namespace, 32)
|
||||
assert after["SpanId"] == seed_id(span_id, replay.namespace, 16)
|
||||
assert after["ParentSpanId"] == seed_id(parent_id, replay.namespace, 16)
|
||||
assert after["Timestamp"] == timestamp + replay.offset_ms * 1_000_000
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue