From 0fb55f08095b1faf3f66f0a543f8b838f3b6f74a Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Thu, 9 Jul 2026 16:55:14 -0700 Subject: [PATCH] feat: add LiteLLM Relay collector logs --- gateway/routes/allowlist.py | 1 + litellm/constants.py | 6 + litellm/proxy/collector_endpoints/__init__.py | 1 + .../proxy/collector_endpoints/spend_logs.py | 396 ++++++++++++++++++ litellm/proxy/proxy_server.py | 8 + .../proxy/auth/test_route_checks.py | 54 +++ .../collector_endpoints/test_spend_logs.py | 362 ++++++++++++++++ .../scripts/gen-api-types.mjs | 4 +- .../LogDetailContent.test.tsx | 16 + .../LogDetailsDrawer/LogDetailContent.tsx | 82 +++- .../LogDetailsDrawer/LogDetailsDrawer.tsx | 41 +- .../components/view_logs/TypeBadges.test.tsx | 49 ++- .../src/components/view_logs/TypeBadges.tsx | 100 ++++- .../src/components/view_logs/columns.test.tsx | 63 +++ .../src/components/view_logs/columns.tsx | 75 +++- .../src/components/view_logs/constants.ts | 3 + .../src/components/view_logs/index.tsx | 20 +- .../src/components/view_logs/utils.ts | 7 +- 18 files changed, 1248 insertions(+), 40 deletions(-) create mode 100644 litellm/proxy/collector_endpoints/__init__.py create mode 100644 litellm/proxy/collector_endpoints/spend_logs.py create mode 100644 tests/test_litellm/proxy/collector_endpoints/test_spend_logs.py diff --git a/gateway/routes/allowlist.py b/gateway/routes/allowlist.py index 792a56a2cd8..132b3df05e6 100644 --- a/gateway/routes/allowlist.py +++ b/gateway/routes/allowlist.py @@ -118,6 +118,7 @@ GATEWAY_EXACT_PATHS: frozenset[str] = frozenset( "/docs/oauth2-redirect", "/redoc", "/test", + "/collector/spend-logs", } ) diff --git a/litellm/constants.py b/litellm/constants.py index 7423d9b2211..6e9b7e273b9 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -1626,3 +1626,9 @@ ADVISOR_TOOL_DESCRIPTION: str = ( "want to verify your reasoning, or face a complex decision. " "Describe your question or challenge clearly in the 'question' field." ) + +########################### LiteLLM Relay Collector Constants ########################### +LITELLM_RELAY_CALL_TYPE = "litellm-relay" +MAX_COLLECTOR_SPEND_LOGS = 1000 +MAX_COLLECTOR_SPEND_LOG_BYTES = 256 * 1024 +MAX_COLLECTOR_SPEND_LOG_BATCH_BYTES = 10 * 1024 * 1024 diff --git a/litellm/proxy/collector_endpoints/__init__.py b/litellm/proxy/collector_endpoints/__init__.py new file mode 100644 index 00000000000..9eafa9908e0 --- /dev/null +++ b/litellm/proxy/collector_endpoints/__init__.py @@ -0,0 +1 @@ +"""Collector endpoints for ingesting external LiteLLM Relay logs.""" diff --git a/litellm/proxy/collector_endpoints/spend_logs.py b/litellm/proxy/collector_endpoints/spend_logs.py new file mode 100644 index 00000000000..a69f9636c1e --- /dev/null +++ b/litellm/proxy/collector_endpoints/spend_logs.py @@ -0,0 +1,396 @@ +import hashlib +import hmac +import json +from datetime import datetime, timezone +from typing import Any, Optional, TypedDict + +from fastapi import APIRouter, Depends, HTTPException, Request, status + +from litellm.constants import ( + LITELLM_ASYNCIO_QUEUE_MAXSIZE, + LITELLM_RELAY_CALL_TYPE, + MAX_COLLECTOR_SPEND_LOG_BATCH_BYTES, + MAX_COLLECTOR_SPEND_LOG_BYTES, + MAX_COLLECTOR_SPEND_LOGS, +) +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + +class CollectorSpendLogRow(TypedDict, total=False): + request_id: str + call_type: str + api_key: str + spend: float + total_tokens: int + prompt_tokens: int + completion_tokens: int + startTime: datetime + endTime: datetime + model: str + api_base: str + custom_llm_provider: str + user: Optional[str] + team_id: Optional[str] + organization_id: Optional[str] + metadata: dict[str, Any] + cache_hit: str + cache_key: str + request_tags: str + messages: Any + response: Any + proxy_server_request: Any + status: str + + +class CollectorSpendLogsIngestResponse(TypedDict): + enqueued: int + + +async def _enqueue_collector_spend_logs( + prisma_client: Any, + spend_logs: list[CollectorSpendLogRow], +) -> None: + async with prisma_client._spend_log_transactions_lock: + queued_spend_logs = len(prisma_client.spend_log_transactions) + if queued_spend_logs + len(spend_logs) > LITELLM_ASYNCIO_QUEUE_MAXSIZE: + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail={ + "error": "Collector spend-log queue is full", + "queued": queued_spend_logs, + "limit": LITELLM_ASYNCIO_QUEUE_MAXSIZE, + }, + ) + prisma_client.spend_log_transactions.extend(spend_logs) + + +class CollectorSpendLogTransformer: + PASSTHROUGH_FIELDS = { + "total_tokens", + "prompt_tokens", + "completion_tokens", + "startTime", + "endTime", + "completionStartTime", + "model", + "model_id", + "model_group", + "mcp_namespaced_tool_name", + "agent_id", + "api_base", + "cache_hit", + "cache_key", + "end_user", + "requester_ip_address", + "messages", + "response", + "proxy_server_request", + "session_id", + "request_duration_ms", + "status", + } + + @staticmethod + def transform_collector_events_to_spend_logs( + logs: list[dict[str, Any]], + user_api_key_dict: Any, + now: datetime, + ) -> list[CollectorSpendLogRow]: + CollectorSpendLogTransformer._validate_raw_batch_size(logs) + spend_logs = [ + CollectorSpendLogTransformer.transform_collector_event_to_spend_log( + log=log, + user_api_key_dict=user_api_key_dict, + now=now, + ) + for log in logs + ] + CollectorSpendLogTransformer._validate_normalized_batch_size(spend_logs) + return spend_logs + + @staticmethod + def transform_collector_event_to_spend_log( + log: dict[str, Any], + user_api_key_dict: Any, + now: datetime, + ) -> CollectorSpendLogRow: + collector_request_id = log.get("request_id") + if ( + not isinstance(collector_request_id, str) + or not collector_request_id.strip() + ): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail={ + "error": "request_id is required for collector spend-log ingestion" + }, + ) + + key_hash = CollectorSpendLogTransformer._get_auth_key_hash(user_api_key_dict) + row: CollectorSpendLogRow = { + "request_id": CollectorSpendLogTransformer._collector_request_id_for( + key_hash, + collector_request_id, + ), + "call_type": LITELLM_RELAY_CALL_TYPE, + "api_key": key_hash or LITELLM_RELAY_CALL_TYPE, + "spend": 0.0, + "total_tokens": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "startTime": now, + "endTime": now, + "model": "local-ai-traffic", + "api_base": "", + "custom_llm_provider": "", + "user": getattr(user_api_key_dict, "user_id", None), + "team_id": getattr(user_api_key_dict, "team_id", None), + "organization_id": CollectorSpendLogTransformer._get_auth_organization_id( + user_api_key_dict + ), + "metadata": CollectorSpendLogTransformer._normalize_metadata( + log.get("metadata"), + user_api_key_dict, + collector_request_id, + ), + "cache_hit": "False", + "cache_key": "", + "request_tags": CollectorSpendLogTransformer._normalize_request_tags( + log.get("request_tags") + ), + "messages": {}, + "response": {}, + "proxy_server_request": {}, + "status": "success", + } + + for key in CollectorSpendLogTransformer.PASSTHROUGH_FIELDS: + if key in log: + row[key] = log[key] + + return CollectorSpendLogTransformer._sanitize_json_value(row) + + @staticmethod + def _get_auth_organization_id(user_api_key_dict: Any) -> Optional[str]: + return getattr(user_api_key_dict, "organization_id", None) or getattr( + user_api_key_dict, "org_id", None + ) + + @staticmethod + def _get_auth_key_hash(user_api_key_dict: Any) -> Optional[str]: + return getattr(user_api_key_dict, "api_key", None) or getattr( + user_api_key_dict, "token", None + ) + + @staticmethod + def _get_auth_key_alias(user_api_key_dict: Any) -> str: + return ( + getattr(user_api_key_dict, "key_alias", None) + or getattr(user_api_key_dict, "key_name", None) + or LITELLM_RELAY_CALL_TYPE + ) + + @staticmethod + def _get_auth_team_alias(user_api_key_dict: Any) -> Optional[str]: + return getattr(user_api_key_dict, "team_alias", None) or None + + @staticmethod + def _collector_request_id_for( + key_hash: Optional[str], collector_request_id: str + ) -> str: + digest = hmac.new( + (key_hash or LITELLM_RELAY_CALL_TYPE).encode(), + collector_request_id.encode(), + hashlib.sha256, + ).hexdigest() + return f"collector-{digest[:32]}" + + @staticmethod + def _normalize_metadata( + metadata: Any, + user_api_key_dict: Any, + collector_request_id: str, + ) -> dict[str, Any]: + if isinstance(metadata, dict): + normalized = dict(metadata) + elif metadata is None: + normalized = {} + else: + normalized = {"relay_raw_metadata": metadata} + + key_hash = CollectorSpendLogTransformer._get_auth_key_hash(user_api_key_dict) + normalized.update( + { + "source": LITELLM_RELAY_CALL_TYPE, + "collector_request_id": collector_request_id, + "relay_request_id": collector_request_id, + "user_api_key": key_hash, + "user_api_key_alias": CollectorSpendLogTransformer._get_auth_key_alias( + user_api_key_dict + ), + "user_api_key_user_id": getattr(user_api_key_dict, "user_id", None), + "user_api_key_team_id": getattr(user_api_key_dict, "team_id", None), + "user_api_key_team_alias": CollectorSpendLogTransformer._get_auth_team_alias( + user_api_key_dict + ), + "user_api_key_org_id": CollectorSpendLogTransformer._get_auth_organization_id( + user_api_key_dict + ), + } + ) + return normalized + + @staticmethod + def _normalize_request_tags(value: Any) -> str: + if isinstance(value, str): + try: + parsed = json.loads(value) + if not isinstance(parsed, list): + parsed = [parsed] + except json.JSONDecodeError: + parsed = [value] if value.strip() else [] + elif isinstance(value, list): + parsed = list(value) + elif value is None: + parsed = [] + else: + parsed = [value] + + if LITELLM_RELAY_CALL_TYPE not in parsed: + parsed.append(LITELLM_RELAY_CALL_TYPE) + return json.dumps(parsed, separators=(",", ":")) + + @staticmethod + def _sanitize_json_value(value: Any) -> Any: + if isinstance(value, str): + return value.replace("\x00", "") + if not isinstance(value, (dict, list)): + return value + + sanitized: Any = {} if isinstance(value, dict) else [] + stack = [(value, sanitized)] + while stack: + source, target = stack.pop() + if isinstance(source, dict): + for key, item in source.items(): + sanitized_key = str(key).replace("\x00", "") + if isinstance(item, str): + target[sanitized_key] = item.replace("\x00", "") + elif isinstance(item, dict): + child: dict[str, Any] = {} + target[sanitized_key] = child + stack.append((item, child)) + elif isinstance(item, list): + child_list: list[Any] = [] + target[sanitized_key] = child_list + stack.append((item, child_list)) + else: + target[sanitized_key] = item + else: + for item in source: + if isinstance(item, str): + target.append(item.replace("\x00", "")) + elif isinstance(item, dict): + child = {} + target.append(child) + stack.append((item, child)) + elif isinstance(item, list): + child_list = [] + target.append(child_list) + stack.append((item, child_list)) + else: + target.append(item) + return sanitized + + @staticmethod + def _json_size_bytes(value: Any) -> int: + return len(json.dumps(value, default=str, separators=(",", ":")).encode()) + + @staticmethod + def _validate_log_size(log: dict[str, Any]) -> int: + encoded_size = CollectorSpendLogTransformer._json_size_bytes(log) + if encoded_size > MAX_COLLECTOR_SPEND_LOG_BYTES: + raise HTTPException( + status_code=status.HTTP_413_CONTENT_TOO_LARGE, + detail={ + "error": f"Collector spend-log entry exceeds {MAX_COLLECTOR_SPEND_LOG_BYTES} bytes" + }, + ) + return encoded_size + + @staticmethod + def _validate_batch_size(total_bytes: int) -> None: + if total_bytes > MAX_COLLECTOR_SPEND_LOG_BATCH_BYTES: + raise HTTPException( + status_code=status.HTTP_413_CONTENT_TOO_LARGE, + detail={ + "error": f"Collector spend-log batch exceeds {MAX_COLLECTOR_SPEND_LOG_BATCH_BYTES} bytes" + }, + ) + + @staticmethod + def _validate_raw_batch_size(logs: list[dict[str, Any]]) -> None: + total_bytes = 0 + for log in logs: + total_bytes += CollectorSpendLogTransformer._json_size_bytes(log) + CollectorSpendLogTransformer._validate_batch_size(total_bytes) + + @staticmethod + def _validate_normalized_batch_size(spend_logs: list[dict[str, Any]]) -> None: + total_bytes = 0 + for spend_log in spend_logs: + total_bytes += CollectorSpendLogTransformer._validate_log_size(spend_log) + CollectorSpendLogTransformer._validate_batch_size(total_bytes) + + +router = APIRouter(include_in_schema=False) + + +@router.post( + "/collector/spend-logs", + tags=["Collector"], +) +async def ingest_collector_spend_logs( + payload: dict[str, list[dict[str, Any]]], + request: Request, + user_api_key_dict: Any = Depends(user_api_key_auth), +) -> CollectorSpendLogsIngestResponse: + """ + Ingest LiteLLM Relay captures into the existing spend-log batcher so they + appear in the Gateway Logs UI without replaying captured traffic. + """ + prisma_client = getattr(request.app.state, "prisma_client", None) + + if prisma_client is None: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail={"error": "Prisma Client is not initialized"}, + ) + if getattr(request.app.state, "proxy_logging_obj", None) is None: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail={"error": "Proxy logging is not initialized"}, + ) + + logs = payload.get("logs", []) + if len(logs) == 0: + return {"enqueued": 0} + if len(logs) > MAX_COLLECTOR_SPEND_LOGS: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail={ + "error": f"Collector spend-log ingestion is limited to {MAX_COLLECTOR_SPEND_LOGS} rows" + }, + ) + + spend_logs = CollectorSpendLogTransformer.transform_collector_events_to_spend_logs( + logs=logs, + user_api_key_dict=user_api_key_dict, + now=datetime.now(timezone.utc), + ) + await _enqueue_collector_spend_logs( + prisma_client=prisma_client, + spend_logs=spend_logs, + ) + + return {"enqueued": len(spend_logs)} diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 4114bda47c9..b8b13273653 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -468,6 +468,9 @@ from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( router as pass_through_router, ) +from litellm.proxy.collector_endpoints.spend_logs import ( + router as collector_spend_logs_router, +) from litellm.proxy.public_endpoints import router as public_endpoints_router from litellm.proxy.rag_endpoints.endpoints import router as rag_router from litellm.proxy.rerank_endpoints.endpoints import router as rerank_router @@ -913,6 +916,8 @@ async def proxy_startup_event(app: FastAPI): proxy_logging_obj=proxy_logging_obj, user_api_key_cache=user_api_key_cache, ) + if app is not None: + app.state.prisma_client = prisma_client if prisma_client is not None: @@ -1933,6 +1938,8 @@ store_model_in_db: bool = False open_telemetry_logger: Optional[OpenTelemetry] = None ### INITIALIZE GLOBAL LOGGING OBJECT ### proxy_logging_obj: ProxyLogging = ProxyLogging(user_api_key_cache=user_api_key_cache, premium_user=premium_user) +app.state.proxy_logging_obj = proxy_logging_obj +app.state.prisma_client = prisma_client ### REDIS QUEUE ### async_result = None celery_app_conn = None @@ -15792,6 +15799,7 @@ app.include_router(team_router) app.include_router(ui_sso_router) app.include_router(organization_router) app.include_router(customer_router) +app.include_router(collector_spend_logs_router) app.include_router(spend_management_router) app.include_router(caching_router) app.include_router(analytics_router) diff --git a/tests/test_litellm/proxy/auth/test_route_checks.py b/tests/test_litellm/proxy/auth/test_route_checks.py index d623149ff6a..9c050444e80 100644 --- a/tests/test_litellm/proxy/auth/test_route_checks.py +++ b/tests/test_litellm/proxy/auth/test_route_checks.py @@ -86,6 +86,60 @@ def test_compliance_routes_open_to_non_admin_roles(role, route): ) +def test_collector_spend_logs_requires_explicit_allowed_route_for_non_admin_key(): + """Collector ingestion is a write path and must not be open to every key.""" + user_obj = LiteLLM_UserTable( + user_id="relay_user", + user_email="relay@example.com", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + valid_token = UserAPIKeyAuth( + user_id="relay_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + request = MagicMock(spec=Request) + request.query_params = {} + + with pytest.raises(Exception) as exc_info: + RouteChecks.non_proxy_admin_allowed_routes_check( + user_obj=user_obj, + _user_role=LitellmUserRoles.INTERNAL_USER.value, + route="/collector/spend-logs", + request=request, + valid_token=valid_token, + request_data={}, + ) + + assert "Only proxy admin" in str(exc_info.value) + assert "Route=/collector/spend-logs" in str(exc_info.value) + + +@pytest.mark.parametrize("allowed_route", ["/collector/spend-logs", "/collector/*"]) +def test_collector_spend_logs_allows_explicit_allowed_route(allowed_route): + """Relay keys can be scoped to only the collector ingest endpoint.""" + user_obj = LiteLLM_UserTable( + user_id="relay_user", + user_email="relay@example.com", + user_role=LitellmUserRoles.INTERNAL_USER.value, + ) + valid_token = UserAPIKeyAuth( + user_id="relay_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + allowed_routes=[allowed_route], + ) + request = MagicMock(spec=Request) + request.query_params = {} + + RouteChecks.non_proxy_admin_allowed_routes_check( + user_obj=user_obj, + _user_role=LitellmUserRoles.INTERNAL_USER.value, + route="/collector/spend-logs", + request=request, + valid_token=valid_token, + request_data={}, + ) + + def test_proxy_admin_viewer_config_update_route_rejected(): """Test that proxy admin viewer users are rejected when trying to call /config/update""" diff --git a/tests/test_litellm/proxy/collector_endpoints/test_spend_logs.py b/tests/test_litellm/proxy/collector_endpoints/test_spend_logs.py new file mode 100644 index 00000000000..3e91230bcda --- /dev/null +++ b/tests/test_litellm/proxy/collector_endpoints/test_spend_logs.py @@ -0,0 +1,362 @@ +import asyncio +import json + +import pytest +from fastapi.testclient import TestClient + +import litellm.proxy.proxy_server as ps +import litellm.proxy.collector_endpoints.spend_logs as collector_spend_logs +from litellm.constants import MAX_COLLECTOR_SPEND_LOG_BYTES +from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth +from litellm.proxy.proxy_server import app + + +class MockPrismaClient: + def __init__(self): + self.spend_log_transactions = [] + self._spend_log_transactions_lock = asyncio.Lock() + + +def _set_collector_runtime(monkeypatch, prisma_client): + monkeypatch.setattr(ps, "prisma_client", prisma_client) + monkeypatch.setattr(app.state, "prisma_client", prisma_client, raising=False) + + +def test_collector_spend_logs_enqueues_batcher_rows(monkeypatch): + prisma_client = MockPrismaClient() + _set_collector_runtime(monkeypatch, prisma_client) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="admin_user", + api_key="hashed-admin-key", + team_id="team-1", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": "relay-test-request", + "model": "notion-ai", + "metadata": {"app": "notion", "host": "www.notion.so"}, + "proxy_server_request": { + "method": "POST", + "body_preview": "hi", + }, + "response": { + "status_code": 200, + "body_preview": "hello", + }, + "request_duration_ms": 42, + } + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 200, response.text + assert response.json() == {"enqueued": 1} + assert len(prisma_client.spend_log_transactions) == 1 + row = prisma_client.spend_log_transactions[0] + assert row["request_id"].startswith("collector-") + assert row["request_id"] != "relay-test-request" + assert row["call_type"] == "litellm-relay" + assert row["model"] == "notion-ai" + assert row["spend"] == 0.0 + assert row["team_id"] == "team-1" + assert row["metadata"]["source"] == "litellm-relay" + assert row["metadata"]["user_api_key_team_alias"] is None + assert row["metadata"]["collector_request_id"] == "relay-test-request" + assert row["metadata"]["app"] == "notion" + assert row["proxy_server_request"]["body_preview"] == "hi" + assert row["response"]["body_preview"] == "hello" + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_attributes_valid_virtual_key(monkeypatch): + prisma_client = MockPrismaClient() + _set_collector_runtime(monkeypatch, prisma_client) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="user_1", + api_key="hashed-virtual-key", + team_id="team_1", + team_alias="Relay Team", + key_alias="relay-key", + organization_id="org_1", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": "relay-test-request", + "api_key": "client-supplied-key-is-ignored", + "spend": 10, + "team_id": "client-team-is-ignored", + "organization_id": "client-org-is-ignored", + "user": "client-user-is-ignored", + "custom_llm_provider": "client-provider-is-ignored", + "metadata": { + "source": "client-source-is-ignored", + "user_api_key": "client-key-is-ignored", + }, + } + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 200, response.text + row = prisma_client.spend_log_transactions[0] + assert row["api_key"] == "hashed-virtual-key" + assert row["spend"] == 0.0 + assert row["team_id"] == "team_1" + assert row["organization_id"] == "org_1" + assert row["user"] == "user_1" + assert row["custom_llm_provider"] == "" + assert row["metadata"]["source"] == "litellm-relay" + assert row["metadata"]["user_api_key"] == "hashed-virtual-key" + assert row["metadata"]["user_api_key_alias"] == "relay-key" + assert row["metadata"]["user_api_key_team_alias"] == "Relay Team" + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_accepts_json_object_request_tags(monkeypatch): + prisma_client = MockPrismaClient() + _set_collector_runtime(monkeypatch, prisma_client) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="admin_user", + api_key="hashed-admin-key", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": "relay-json-tags-test", + "request_tags": '{"source":"notion"}', + } + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 200, response.text + row = prisma_client.spend_log_transactions[0] + assert json.loads(row["request_tags"]) == [ + {"source": "notion"}, + "litellm-relay", + ] + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_strips_nul_bytes(monkeypatch): + prisma_client = MockPrismaClient() + _set_collector_runtime(monkeypatch, prisma_client) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="user_1", + api_key="hashed-virtual-key", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": "relay-nul-test", + "proxy_server_request": {"body_preview": "hello\u0000world"}, + "response": {"body_preview": "ok\u0000"}, + } + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 200, response.text + row_text = json.dumps(prisma_client.spend_log_transactions[0], default=str) + assert "\u0000" not in row_text + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_rejects_oversized_log(monkeypatch): + prisma_client = MockPrismaClient() + _set_collector_runtime(monkeypatch, prisma_client) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="admin_user", + api_key="hashed-admin-key", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": "large-relay-request", + "proxy_server_request": { + "body_preview": "x" * (MAX_COLLECTOR_SPEND_LOG_BYTES + 1) + }, + } + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 413, response.text + assert prisma_client.spend_log_transactions == [] + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_rejects_normalized_row_over_size_limit(monkeypatch): + prisma_client = MockPrismaClient() + _set_collector_runtime(monkeypatch, prisma_client) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="admin_user", + api_key="hashed-admin-key", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": "normalized-large-relay-request", + "proxy_server_request": { + "body_preview": "x" * (MAX_COLLECTOR_SPEND_LOG_BYTES - 50) + }, + } + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 413, response.text + assert prisma_client.spend_log_transactions == [] + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_rejects_oversized_batch(monkeypatch): + prisma_client = MockPrismaClient() + _set_collector_runtime(monkeypatch, prisma_client) + monkeypatch.setattr( + collector_spend_logs, + "MAX_COLLECTOR_SPEND_LOG_BATCH_BYTES", + 900, + ) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="admin_user", + api_key="hashed-admin-key", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": f"batch-relay-request-{idx}", + "proxy_server_request": {"body_preview": "x" * 200}, + } + for idx in range(5) + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 413, response.text + assert prisma_client.spend_log_transactions == [] + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_rejects_when_queue_is_full(monkeypatch): + prisma_client = MockPrismaClient() + prisma_client.spend_log_transactions = [{} for _ in range(3)] + _set_collector_runtime(monkeypatch, prisma_client) + monkeypatch.setattr( + collector_spend_logs, + "LITELLM_ASYNCIO_QUEUE_MAXSIZE", + 3, + ) + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="admin_user", + api_key="hashed-admin-key", + ) + + try: + response = TestClient(app).post( + "/collector/spend-logs", + json={ + "logs": [ + { + "request_id": "queued-relay-request", + } + ] + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert response.status_code == 429, response.text + assert response.json()["detail"] == { + "error": "Collector spend-log queue is full", + "queued": 3, + "limit": 3, + } + assert len(prisma_client.spend_log_transactions) == 3 + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + +def test_collector_spend_logs_enqueue_is_capacity_checked_under_lock(monkeypatch): + prisma_client = MockPrismaClient() + monkeypatch.setattr( + collector_spend_logs, + "LITELLM_ASYNCIO_QUEUE_MAXSIZE", + 3, + ) + + async def enqueue_two_batches(): + await collector_spend_logs._enqueue_collector_spend_logs( + prisma_client=prisma_client, + spend_logs=[{"request_id": "one"}, {"request_id": "two"}], + ) + with pytest.raises(collector_spend_logs.HTTPException) as exc_info: + await collector_spend_logs._enqueue_collector_spend_logs( + prisma_client=prisma_client, + spend_logs=[{"request_id": "three"}, {"request_id": "four"}], + ) + return exc_info.value + + error = asyncio.run(enqueue_two_batches()) + + assert error.status_code == 429 + assert error.detail == { + "error": "Collector spend-log queue is full", + "queued": 2, + "limit": 3, + } + assert [row["request_id"] for row in prisma_client.spend_log_transactions] == [ + "one", + "two", + ] diff --git a/ui/litellm-dashboard/scripts/gen-api-types.mjs b/ui/litellm-dashboard/scripts/gen-api-types.mjs index 6b9f8581292..3bfcd79eb87 100644 --- a/ui/litellm-dashboard/scripts/gen-api-types.mjs +++ b/ui/litellm-dashboard/scripts/gen-api-types.mjs @@ -26,6 +26,8 @@ const python = (process.env.LITELLM_PYTHON ?? "python3").split(" "); // The dashboard calls internal UI routes that the public /openapi.json hides via // include_in_schema=False. Force them in so they get typed here; this mutates a // throwaway interpreter, so the spec the proxy actually serves is unchanged. +// Collector ingestion routes are machine-to-machine write APIs and should stay +// out of dashboard client types even though other hidden UI routes are included. // Python 3.13 strips a docstring's common leading indentation at compile time // while 3.12 keeps it, so the same model yields differently-indented descriptions // depending on the interpreter — enough to make this output non-reproducible @@ -37,7 +39,7 @@ const dumpSpec = [ "from fastapi.routing import APIRoute", "for route in app.routes:", " if isinstance(route, APIRoute):", - " route.include_in_schema = True", + " route.include_in_schema = not route.path.startswith('/collector/')", "app.openapi_schema = None", "def normalize(node):", " if isinstance(node, dict):", diff --git a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailContent.test.tsx b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailContent.test.tsx index 85a38e26977..009839ad7f2 100644 --- a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailContent.test.tsx +++ b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailContent.test.tsx @@ -56,6 +56,22 @@ describe("LogDetailContent", () => { expect(screen.getByText("completion")).toBeInTheDocument(); }); + it("should display relay source with logo in Request Details", () => { + render( + , + ); + + expect(screen.getByText("Source")).toBeInTheDocument(); + expect(screen.getByLabelText("Notion logo")).toBeInTheDocument(); + expect(screen.getByText("Notion")).toBeInTheDocument(); + }); + it("should display error alert when request has failed", () => { render( | undefined => { + if (!requestTags || Array.isArray(requestTags) || typeof requestTags !== "object") { + return undefined; + } + + return requestTags; +}; + export interface LogDetailContentProps { logEntry: LogEntry; /** When true, log details (messages/response) are still being lazy-loaded. */ @@ -53,8 +64,11 @@ export interface LogDetailContentProps { */ export function LogDetailContent({ logEntry, isLoadingDetails = false, accessToken }: LogDetailContentProps) { const metadata = logEntry.metadata || {}; + const requestTags = getObjectRequestTags(logEntry.request_tags); const hasError = metadata.status === "failure"; const errorInfo = hasError ? metadata.error_information : null; + const isRelayCapture = logEntry.call_type === "litellm-relay"; + const relaySource = getRelaySource(logEntry); const hasMessages = checkHasMessages(logEntry.messages); const hasResponse = checkHasResponse(logEntry.response); @@ -107,9 +121,7 @@ export function LogDetailContent({ logEntry, isLoadingDetails = false, accessTok )} {/* Tags */} - {logEntry.request_tags && Object.keys(logEntry.request_tags).length > 0 && ( - - )} + {requestTags && Object.keys(requestTags).length > 0 && } {/* Request Details */}
@@ -118,6 +130,11 @@ export function LogDetailContent({ logEntry, isLoadingDetails = false, accessTok {logEntry.model} {logEntry.custom_llm_provider || "-"} {logEntry.call_type} + {isRelayCapture && ( + + + + )} @@ -232,7 +249,7 @@ function ErrorDescription({ errorInfo }: { errorInfo: any }) { ); } -function TagsSection({ tags }: { tags: Record }) { +function TagsSection({ tags }: { tags: Record }) { return (
@@ -418,6 +435,7 @@ function RequestResponseSection({ : totalTokens > 0 ? (totalSpend * completionTokens) / totalTokens : 0; + const isRelayCapture = logEntry.call_type === "litellm-relay"; return (
@@ -448,6 +466,7 @@ function RequestResponseSection({ ), children: (
+ {isRelayCapture && } {viewMode === "pretty" ? ( + {value || "No body captured"} + + ); +} + +function RelayPayloadPreview({ request, response }: { request: any; response: any }) { + const requestBody = getPreviewText(request); + const responseBody = getPreviewText(response); + return ( +
+ +
+ + + + + + +
+
+ ); +} + export function GuardrailJumpLink({ guardrailEntries }: { guardrailEntries: any[] }) { const allPassed = guardrailEntries.every((e) => { const status = e?.guardrail_status || e?.status; diff --git a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailsDrawer.tsx b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailsDrawer.tsx index cc087a611b0..e3472c16a78 100644 --- a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailsDrawer.tsx +++ b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/LogDetailsDrawer.tsx @@ -1,9 +1,9 @@ import { useEffect, useMemo, useState } from "react"; import { Button, Drawer, Segmented } from "antd"; import { CheckOutlined, CopyOutlined, LeftOutlined, RightOutlined } from "@ant-design/icons"; -import { Bot, Sparkles, Wrench } from "lucide-react"; +import { Bot, Cable, Sparkles, Wrench } from "lucide-react"; import { LogEntry } from "../columns"; -import { AGENT_CALL_TYPES, MCP_CALL_TYPES } from "../constants"; +import { AGENT_CALL_TYPES, MCP_CALL_TYPES, RELAY_CALL_TYPES } from "../constants"; import { getEventDisplayName } from "../utils"; import { DrawerHeader } from "./DrawerHeader"; import { useKeyboardNavigation } from "./useKeyboardNavigation"; @@ -49,12 +49,22 @@ interface TraceEventRowProps { function TraceEventRow({ row, isSelected, onClick }: TraceEventRowProps) { const isMcp = MCP_CALL_TYPES.includes(row.call_type); const isAgent = AGENT_CALL_TYPES.includes(row.call_type); - const durationValue = - row.request_duration_ms != null - ? (row.request_duration_ms / 1000).toFixed(3) - : row.startTime && row.endTime - ? ((Date.parse(row.endTime) - Date.parse(row.startTime)) / 1000).toFixed(3) - : "-"; + const isRelay = RELAY_CALL_TYPES.includes(row.call_type); + let durationValue = "-"; + if (row.request_duration_ms != null) { + durationValue = (row.request_duration_ms / 1000).toFixed(3); + } else if (row.startTime && row.endTime) { + durationValue = ((Date.parse(row.endTime) - Date.parse(row.startTime)) / 1000).toFixed(3); + } + + let eventIcon = ; + if (isMcp) { + eventIcon = ; + } else if (isAgent) { + eventIcon = ; + } else if (isRelay) { + eventIcon = ; + } return (