feat(proxy): track failed requests by HTTP status and caller (#45244)

Counts failed gateway requests by HTTP status at the ASGI edge
(LiteLLM_DailyGatewayFailedRequests) and rolls failures up per virtual key,
team, user and model group with the status the logging callbacks recorded
(LiteLLM_DailyRequestErrors, fed from the spend writer, flushed like the
gateway counters, directly or through Redis with a pod lease, and on
shutdown). GET /gateway/daily/activity gains by_status_code; the new
admin-only GET /gateway/errors/activity returns per-day totals with
per-status counts, failures by status code, and keys, teams, users and
model groups ranked by failures with their top status. Dashboard schema.d.ts
regenerated.

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Yassin Kortam 2026-10-09 14:27:56 -07:00 • committed by GitHub
parent 43f07d0bd8
commit 2df5b07308
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 2218 additions and 125 deletions

View file

@ -91,9 +91,10 @@ BACKEND_PATH_PREFIXES: tuple[str, ...] = (
"/user_agent",
"/usage/",
"/daily/",
# Deployment-wide gateway request counts. Scoped to the analytics read rather
# Deployment-wide gateway request and error counts. Scoped to the analytics reads rather
# than all of /gateway/, which stays free for data-plane routes.
"/gateway/daily/",
"/gateway/errors/",
# CloudZero cost-export admin (init / settings / export / dry-run / delete)
"/cloudzero/",
# Caching admin

View file

@ -0,0 +1,15 @@
-- CreateTable
CREATE TABLE IF NOT EXISTS "LiteLLM_DailyGatewayFailedRequests" (
"date" TEXT NOT NULL,
"category" TEXT NOT NULL,
"route" TEXT NOT NULL,
"status_code" INTEGER NOT NULL,
"failed_requests" BIGINT NOT NULL DEFAULT 0,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "LiteLLM_DailyGatewayFailedRequests_pkey" PRIMARY KEY ("date","category","route","status_code")
);
-- CreateIndex
CREATE INDEX IF NOT EXISTS "LiteLLM_DailyGatewayFailedRequests_date_idx" ON "LiteLLM_DailyGatewayFailedRequests"("date");

View file

@ -0,0 +1,17 @@
-- CreateTable
CREATE TABLE IF NOT EXISTS "LiteLLM_DailyRequestErrors" (
"date" TEXT NOT NULL,
"api_key" TEXT NOT NULL,
"team_id" TEXT NOT NULL DEFAULT '',
"user_id" TEXT NOT NULL DEFAULT '',
"model_group" TEXT NOT NULL DEFAULT '',
"status_code" INTEGER NOT NULL,
"failed_requests" BIGINT NOT NULL DEFAULT 0,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "LiteLLM_DailyRequestErrors_pkey" PRIMARY KEY ("date","api_key","team_id","user_id","model_group","status_code")
);
-- CreateIndex
CREATE INDEX IF NOT EXISTS "LiteLLM_DailyRequestErrors_date_idx" ON "LiteLLM_DailyRequestErrors"("date");

View file

@ -1361,6 +1361,34 @@ model LiteLLM_DailyGatewayRequests {
@@index([date])
}
model LiteLLM_DailyGatewayFailedRequests {
date String
category String
route String
status_code Int
failed_requests BigInt @default(0)
created_at DateTime @default(now())
updated_at DateTime @updatedAt
@@id([date, category, route, status_code])
@@index([date])
}
model LiteLLM_DailyRequestErrors {
date String
api_key String
team_id String @default("")
user_id String @default("")
model_group String @default("")
status_code Int
failed_requests BigInt @default(0)
created_at DateTime @default(now())
updated_at DateTime @updatedAt
@@id([date, api_key, team_id, user_id, model_group, status_code])
@@index([date])
}
// Prompt table for storing prompt configurations
model LiteLLM_PromptTable {
id String @id @default(uuid())

View file

@ -404,6 +404,11 @@ DEFAULT_SSL_CIPHERS: Final = os.getenv(
########### v2 Architecture constants for managing writing updates to the database ###########
REDIS_UPDATE_BUFFER_KEY: Final = "litellm_spend_update_buffer"
REDIS_GATEWAY_REQUESTS_BUFFER_KEY: Final = "litellm_gateway_requests_buffer"
REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2: Final = "litellm_gateway_requests_buffer_v2"
REDIS_REQUEST_ERRORS_BUFFER_KEY: Final = "litellm_request_errors_buffer"
REQUEST_ERRORS_UNKNOWN_STATUS_CODE: Final = 0
REQUEST_ERRORS_MAX_ROWS_PER_UPSERT: Final = 1000
GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT: Final = 1000
REDIS_DAILY_SPEND_UPDATE_BUFFER_KEY: Final = "litellm_daily_spend_update_buffer"
REDIS_DAILY_TEAM_SPEND_UPDATE_BUFFER_KEY: Final = "litellm_daily_team_spend_update_buffer"
REDIS_DAILY_ORG_SPEND_UPDATE_BUFFER_KEY: Final = "litellm_daily_org_spend_update_buffer"

View file

@ -320,6 +320,8 @@ _PRISMA_MODELS: Final[frozenset[str]] = frozenset(
"LiteLLM_DailyToolSpend",
"LiteLLM_DailyModelUsage",
"LiteLLM_DailyGatewayRequests",
"LiteLLM_DailyGatewayFailedRequests",
"LiteLLM_DailyRequestErrors",
"LiteLLM_PromptTable",
"LiteLLM_HealthCheckTable",
"LiteLLM_SearchToolsTable",

View file

@ -75,6 +75,7 @@ from litellm.proxy.db.db_transaction_queue.window_spend_update_queue import (
)
from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
from litellm.proxy.db.model_usage_rollup import build_model_usage_transaction
from litellm.proxy.db.request_error_tracking import request_error_accumulator
from litellm.proxy.db.rollup_lock_timeout import apply_rollup_lock_timeout
from litellm.proxy.route_llm_request import ROUTE_ENDPOINT_MAPPING
from litellm.proxy.spend_tracking.compression_savings import (
@ -1173,6 +1174,13 @@ class DBSpendUpdateWriter:
"_batch_database_updates: add_spend_log_transaction_to_daily_team_transaction failed: %s",
traceback.format_exc(),
)
try:
self._record_request_error(payload=payload_copy, prisma_client=prisma_client)
except Exception: # noqa: BLE001 # the rollup must never skip the sibling spend writes
verbose_proxy_logger.debug(
"_batch_database_updates: _record_request_error failed: %s",
traceback.format_exc(),
)
try:
await self.add_spend_log_transaction_to_daily_org_transaction(
@ -2735,6 +2743,22 @@ class DBSpendUpdateWriter:
entity_id_field="tag",
)
@staticmethod
def _record_request_error(*, payload: SpendLogsPayload, prisma_client: PrismaClient | None) -> None:
if prisma_client is None:
return
start_time: Final = payload["startTime"]
date: Final = start_time.isoformat() if isinstance(start_time, datetime) else str(start_time or "")
if not date:
return
metadata: Final[SpendLogsMetadata] = json.loads(payload["metadata"])
request_error_accumulator.record(
payload=payload,
request_status=prisma_client.get_request_status(payload), # pyright: ignore[reportUnknownMemberType] # legacy payload union; this caller supplies a typed spend payload
date=date.split("T")[0],
is_internal_call=bool(metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY)), # pyright: ignore[reportUnknownMemberType] # TypedDict.get overloads carry Any defaults
)
async def _common_add_spend_log_transaction_to_daily_transaction(
self,
payload: dict | SpendLogsPayload,

View file

@ -1,28 +1,29 @@
"""
Accumulates gateway request counts (SGR) recorded at the ASGI edge and commits
them to ``LiteLLM_DailyGatewayRequests``.
them to ``LiteLLM_DailyGatewayRequests``. Failed requests are also tracked by
HTTP status in ``LiteLLM_DailyGatewayFailedRequests``.
Unlike the spend queues this keeps no per-request item. A count is a pure
aggregate, so requests fold into an in-memory map as they finish. Every
dimension of the key is server-chosen and drawn from a fixed set: the date, the
category, and a route that the classifier maps to one of a closed list of
strings rather than passing the raw path through. Nothing a caller sends can
add a key, so the fold and the table it commits to are bounded by (days x
routes) however much traffic arrives, and the response path carries no
unbounded queue that would block once full.
dimension of the key is bounded: the date, the category, the route that the
classifier maps to one of a closed list of strings rather than passing the raw
path through, and the response status code. The fold and the tables it commits
to are bounded by (days x categories x routes x status codes) however much
traffic arrives, and the response path carries no unbounded queue that would
block once full.
A flush commits its whole snapshot as one multi-row ``INSERT ... ON CONFLICT DO
UPDATE`` rather than one upsert per key, so a worker costs the primary one
statement per interval however many routes it served. With
``use_redis_transaction_buffer`` on, workers instead push their snapshot to a
Redis list and one lock-holding pod folds every entry and writes the table, so
Redis list and one lock-holding pod folds every entry and writes the tables, so
the deployment as a whole costs the primary one statement per interval.
"""
import json
from collections.abc import AsyncIterator, Callable, Iterable
from collections.abc import AsyncIterator, Callable, Iterable, Iterator
from datetime import datetime
from itertools import chain
from itertools import chain, groupby
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, TypeAlias
@ -31,7 +32,12 @@ from pydantic import TypeAdapter
from litellm._internal_context import with_service_target
from litellm._logging import verbose_proxy_logger
from litellm.caching import RedisCache
from litellm.constants import MAX_REDIS_BUFFER_DEQUEUE_COUNT, REDIS_GATEWAY_REQUESTS_BUFFER_KEY
from litellm.constants import (
GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT,
MAX_REDIS_BUFFER_DEQUEUE_COUNT,
REDIS_GATEWAY_REQUESTS_BUFFER_KEY,
REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2,
)
from litellm.proxy.db.db_span import db_span
from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
from litellm.proxy.middleware.billable_request_metrics_middleware import BillableCategory
@ -49,12 +55,17 @@ if TYPE_CHECKING:
_EMPTY: Final = GatewayRequestCounts(successful_requests=0, failed_requests=0)
_TABLE: Final = '"LiteLLM_DailyGatewayRequests"'
_FAILURE_TABLE: Final = '"LiteLLM_DailyGatewayFailedRequests"'
_COLUMNS_PER_ROW: Final = 5
_UTC_NOW: Final = "(NOW() AT TIME ZONE 'UTC')"
GATEWAY_REQUESTS_JOB_NAME: Final = "update_gateway_requests_job"
_BufferedRows: TypeAlias = tuple[tuple[str, str, str, int, int], ...]
_BufferedRow: TypeAlias = tuple[str, str, str, int | None, int, int]
_LegacyBufferedRow: TypeAlias = tuple[str, str, str, int, int]
_BufferedRows: TypeAlias = tuple[_BufferedRow, ...]
_LegacyBufferedRows: TypeAlias = tuple[_LegacyBufferedRow, ...]
_BUFFERED_ROWS: Final = TypeAdapter(_BufferedRows)
_LEGACY_BUFFERED_ROWS: Final = TypeAdapter(_LegacyBufferedRows)
_BUFFERED_ENTRIES: Final = TypeAdapter(tuple[str | bytes, ...])
_NO_COUNTS: Final[GatewayRequestSnapshot] = MappingProxyType({})
@ -67,7 +78,9 @@ class GatewayRequestAccumulator:
self._counts: dict[GatewayRequestKey, GatewayRequestCounts] = {} # mutable-ok: bounded fold, drained per flush
def record(self, *, category: BillableCategory, route: str, status_code: int) -> None:
key: Final = GatewayRequestKey(date=self._clock().strftime("%Y-%m-%d"), category=category.value, route=route)
key: Final = GatewayRequestKey(
date=self._clock().strftime("%Y-%m-%d"), category=category.value, route=route, status_code=status_code
)
self._counts[key] = self._counts.get(key, _EMPTY).plus(succeeded=200 <= status_code < 300)
def drain(self) -> GatewayRequestSnapshot:
@ -82,7 +95,7 @@ class GatewayRequestAccumulator:
A dropped flush would silently undercount the metric the dashboard now
treats as the source of truth. Merging cannot grow without bound: keys
collapse on collision, so the fold stays bounded by (date x category x
route) however long the database is unreachable.
route and status code) however long the database is unreachable.
This buys at-least-once, not exactly-once, and the cost is worth stating.
The statement commits on the server before its acknowledgement is read, so
@ -97,7 +110,7 @@ class GatewayRequestAccumulator:
def fold_counts(items: Iterable[tuple[GatewayRequestKey, GatewayRequestCounts]]) -> GatewayRequestSnapshot:
"""Sum counts key-wise; the result stays bounded by (date x category x route)."""
"""Sum counts key-wise; the result stays bounded by (date x category x route x status code)."""
folded: Final[dict[GatewayRequestKey, GatewayRequestCounts]] = {} # mutable-ok: local fold returned once
for key, counts in items:
existing = folded.get(key, _EMPTY)
@ -108,18 +121,64 @@ def fold_counts(items: Iterable[tuple[GatewayRequestKey, GatewayRequestCounts]])
return folded
def _decode_v2_entries(entries: Iterable[str | bytes]) -> Iterator[_BufferedRow]:
for entry in entries:
yield from _BUFFERED_ROWS.validate_json(entry)
def _decode_legacy_entries(entries: Iterable[str | bytes]) -> Iterator[_BufferedRow]:
for entry in entries:
for date, category, route, succeeded, failed in _LEGACY_BUFFERED_ROWS.validate_json(entry):
yield date, category, route, None, succeeded, failed
def _buffered_counts(rows: Iterable[_BufferedRow]) -> Iterator[tuple[GatewayRequestKey, GatewayRequestCounts]]:
for date, category, route, status_code, succeeded, failed in rows:
yield (
GatewayRequestKey(date=date, category=category, route=route, status_code=status_code),
GatewayRequestCounts(successful_requests=succeeded, failed_requests=failed),
)
def build_gateway_requests_upsert(snapshot: GatewayRequestSnapshot) -> tuple[str, tuple[str | int, ...]]:
"""
One ``INSERT ... ON CONFLICT DO UPDATE`` that increments every (date, category,
route) in the snapshot. Rows are ordered by the conflict key so concurrent
writers lock rows in the same order and cannot deadlock.
One statement that increments main-table counts across status codes and
failure-table counts by status code. Rows are ordered by their conflict
keys so concurrent writers lock rows in the same order and cannot deadlock.
"""
ordered: Final = sorted(snapshot.items(), key=lambda item: (item[0].date, item[0].category, item[0].route))
main_snapshot: Final = fold_counts(
(
GatewayRequestKey(date=key.date, category=key.category, route=key.route),
counts,
)
for key, counts in snapshot.items()
)
ordered: Final = sorted(main_snapshot.items(), key=lambda item: (item[0].date, item[0].category, item[0].route))
failed_ordered: Final = tuple(
sorted(
(
(key.date, key.category, key.route, key.status_code, counts.failed_requests)
for key, counts in snapshot.items()
if key.status_code is not None and counts.failed_requests > 0
),
key=lambda row: (row[0], row[1], row[2], row[3] or 0),
)
)
failed_rows: Final = ", ".join(
f"(${base + 1}::text, ${base + 2}::text, "
f"${base + 3}::text, ${base + 4}::integer, "
f"${base + 5}::bigint, {_UTC_NOW})"
for base in range(0, len(failed_ordered) * _COLUMNS_PER_ROW, _COLUMNS_PER_ROW)
)
rows: Final = ", ".join(
f"(${base + 1}::text, ${base + 2}::text, ${base + 3}::text, ${base + 4}::bigint, ${base + 5}::bigint, {_UTC_NOW})"
f"(${base + len(failed_ordered) * _COLUMNS_PER_ROW + 1}::text, "
f"${base + len(failed_ordered) * _COLUMNS_PER_ROW + 2}::text, "
f"${base + len(failed_ordered) * _COLUMNS_PER_ROW + 3}::text, "
f"${base + len(failed_ordered) * _COLUMNS_PER_ROW + 4}::bigint, "
f"${base + len(failed_ordered) * _COLUMNS_PER_ROW + 5}::bigint, {_UTC_NOW})"
for base in range(0, len(ordered) * _COLUMNS_PER_ROW, _COLUMNS_PER_ROW)
)
sql: Final = (
main_sql: Final = (
f'INSERT INTO {_TABLE} ("date", "category", "route", "successful_requests", "failed_requests", "updated_at")\n'
f"VALUES {rows}\n"
'ON CONFLICT ("date", "category", "route") DO UPDATE SET\n'
@ -127,30 +186,76 @@ def build_gateway_requests_upsert(snapshot: GatewayRequestSnapshot) -> tuple[str
f' "failed_requests" = {_TABLE}."failed_requests" + EXCLUDED."failed_requests",\n'
f' "updated_at" = {_UTC_NOW}'
)
params: Final[tuple[str | int, ...]] = tuple(
failed_params: Final[tuple[str | int, ...]] = tuple(chain.from_iterable(failed_ordered))
main_params: Final[tuple[str | int, ...]] = tuple(
value
for key, counts in ordered
for value in (key.date, key.category, key.route, counts.successful_requests, counts.failed_requests)
)
return sql, params
if not failed_ordered:
return main_sql, main_params
failure_cte: Final = (
f"WITH failed AS (INSERT INTO {_FAILURE_TABLE} "
'("date", "category", "route", "status_code", "failed_requests", "updated_at")\n'
f"VALUES {failed_rows}\n"
'ON CONFLICT ("date", "category", "route", "status_code") DO UPDATE SET\n'
f' "failed_requests" = {_FAILURE_TABLE}."failed_requests" + EXCLUDED."failed_requests",\n'
f' "updated_at" = {_UTC_NOW}\n'
'RETURNING "status_code")\n'
)
return f"{failure_cte}{main_sql}", failed_params + main_params
def _route_of(item: tuple[GatewayRequestKey, GatewayRequestCounts]) -> tuple[str, str, str]:
return (item[0].date, item[0].category, item[0].route)
def _statement_chunks(snapshot: GatewayRequestSnapshot) -> Iterator[GatewayRequestSnapshot]:
"""Bounded statements for a backlog; a route's total and its status rows always share a chunk."""
ordered: Final = sorted(snapshot.items(), key=_route_of)
groups: Final = tuple(tuple(rows) for _, rows in groupby(ordered, key=_route_of))
pending: tuple[tuple[GatewayRequestKey, GatewayRequestCounts], ...] = () # rebind-ok: the chunk under construction
for rows in groups:
if pending and len(pending) + len(rows) > GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT:
yield dict(pending)
pending = rows
else:
pending = pending + rows
if pending:
yield dict(pending)
async def commit_gateway_requests_to_db(
*,
prisma_client: "PrismaClient",
snapshot: GatewayRequestSnapshot,
) -> None:
"""Increment every (date, category, route) in the snapshot with a single statement."""
if not snapshot:
return
) -> GatewayRequestSnapshot:
"""
Increment request totals and failed status-code counts in bounded statements.
sql, params = build_gateway_requests_upsert(snapshot)
async with db_span("commit_gateway_requests", "LiteLLM_DailyGatewayRequests"):
await prisma_client.db.execute_raw(sql, *params) # pyright: ignore[reportAny] # untyped prisma client
verbose_proxy_logger.debug(
"Gateway request tracking - committed %d aggregated rows in one statement", len(snapshot)
)
Returns the rows that were not written (the failed chunk and everything after it) so the
caller retries only those; replaying the whole snapshot would count the landed chunks twice.
"""
chunks: Final = tuple(_statement_chunks(snapshot))
for index, chunk in enumerate(chunks):
try:
sql, params = build_gateway_requests_upsert(chunk)
async with db_span("commit_gateway_requests", "LiteLLM_DailyGatewayRequests"):
await prisma_client.db.execute_raw(sql, *params) # pyright: ignore[reportAny] # untyped prisma client
except Exception: # noqa: BLE001 -- keep the chunks already written, hand back the rest for retry
verbose_proxy_logger.warning(
"Gateway request tracking - statement %d of %d failed, %d rows go back for retry",
index + 1,
len(chunks),
sum(len(remaining) for remaining in chunks[index:]),
exc_info=True,
)
return fold_counts(chain.from_iterable(remaining.items() for remaining in chunks[index:]))
verbose_proxy_logger.debug(
"Gateway request tracking - committed %d aggregated rows in one statement", len(chunk)
)
return _NO_COUNTS
class GatewayRequestRedisBuffer:
@ -174,37 +279,53 @@ class GatewayRequestRedisBuffer:
if not snapshot:
return
rows: Final[_BufferedRows] = tuple(
(key.date, key.category, key.route, counts.successful_requests, counts.failed_requests)
(key.date, key.category, key.route, key.status_code, counts.successful_requests, counts.failed_requests)
for key, counts in snapshot.items()
)
await self._redis_cache.async_rpush(key=REDIS_GATEWAY_REQUESTS_BUFFER_KEY, values=(json.dumps(rows),))
await self._redis_cache.async_rpush( # pyright: ignore[reportUnknownMemberType] # redis cache span parameter is untyped
key=REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2, values=(json.dumps(rows),)
)
@with_service_target(_GATEWAY_REQUEST_QUEUE_TARGET)
async def _pop_batch(self) -> tuple[str | bytes, ...]:
async def _pop_batch(self, key: str) -> tuple[str | bytes, ...]:
popped: Final[object] = await self._redis_cache.async_lpop( # pyright: ignore[reportAny] # redis returns Any
key=REDIS_GATEWAY_REQUESTS_BUFFER_KEY, count=MAX_REDIS_BUFFER_DEQUEUE_COUNT
key=key, count=MAX_REDIS_BUFFER_DEQUEUE_COUNT
)
if not popped:
return ()
return _BUFFERED_ENTRIES.validate_python(popped if isinstance(popped, list) else (popped,))
async def _pop_all(self) -> AsyncIterator[str | bytes]:
async def _pop_all(self, key: str) -> AsyncIterator[str | bytes]:
while True:
batch = await self._pop_batch()
batch = await self._pop_batch(key)
for entry in batch:
yield entry
if len(batch) < MAX_REDIS_BUFFER_DEQUEUE_COUNT:
return
async def pop(self) -> GatewayRequestSnapshot:
entries: Final = tuple([entry async for entry in self._pop_all()])
return fold_counts(
(
GatewayRequestKey(date=date, category=category, route=route),
GatewayRequestCounts(successful_requests=succeeded, failed_requests=failed),
async def _pop_until_failure(self, key: str) -> AsyncIterator[str | bytes]:
try:
async for entry in self._pop_all(key):
yield entry
except Exception: # noqa: BLE001 -- fold what was already popped; unread entries stay queued in Redis
verbose_proxy_logger.warning(
"Gateway request tracking - Redis read of %s failed, folding the entries already popped",
key,
exc_info=True,
)
async def pop(self) -> GatewayRequestSnapshot:
entries_v2: Final = tuple(
[entry async for entry in self._pop_until_failure(REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2)]
)
legacy_entries: Final = tuple(
[entry async for entry in self._pop_until_failure(REDIS_GATEWAY_REQUESTS_BUFFER_KEY)]
)
return fold_counts(
chain(
_buffered_counts(_decode_v2_entries(entries_v2)),
_buffered_counts(_decode_legacy_entries(legacy_entries)),
)
for entry in entries
for date, category, route, succeeded, failed in _BUFFERED_ROWS.validate_json(entry)
)
async def commit_if_leader(self, prisma_client: "PrismaClient") -> GatewayRequestSnapshot:
@ -221,16 +342,13 @@ class GatewayRequestRedisBuffer:
if not await self._pod_lock_manager.acquire_lock(cronjob_id=GATEWAY_REQUESTS_JOB_NAME):
return _NO_COUNTS
buffered: Final = await self.pop()
try:
await commit_gateway_requests_to_db(prisma_client=prisma_client, snapshot=buffered)
except Exception: # noqa: BLE001 -- a failed commit must not stop the scheduler
verbose_proxy_logger.warning(
"Gateway request tracking - failed to commit %d buffered rows, re-queuing to Redis for the next flush",
len(buffered),
exc_info=True,
)
return await self._requeue(buffered)
return _NO_COUNTS
uncommitted: Final = await commit_gateway_requests_to_db(prisma_client=prisma_client, snapshot=buffered)
if not uncommitted:
return _NO_COUNTS
verbose_proxy_logger.warning(
"Gateway request tracking - re-queuing %d uncommitted rows to Redis for the next flush", len(uncommitted)
)
return await self._requeue(uncommitted)
async def _requeue(self, snapshot: GatewayRequestSnapshot) -> GatewayRequestSnapshot:
try:
@ -262,21 +380,19 @@ async def flush_gateway_requests(
the process is about to discard.
"""
snapshot: Final = accumulator.drain()
if redis_buffer is None:
accumulator.restore(await commit_gateway_requests_to_db(prisma_client=prisma_client, snapshot=snapshot))
return
try:
if redis_buffer is None:
await commit_gateway_requests_to_db(prisma_client=prisma_client, snapshot=snapshot)
else:
await redis_buffer.push(snapshot)
await redis_buffer.push(snapshot)
except Exception: # noqa: BLE001 -- a failed flush must not stop the scheduler
accumulator.restore(snapshot)
verbose_proxy_logger.warning(
"Gateway request tracking - failed to commit %d rows, retrying on the next flush",
"Gateway request tracking - failed to buffer %d rows, retrying on the next flush",
len(snapshot),
exc_info=True,
)
return
if redis_buffer is None:
return
try:
accumulator.restore(await redis_buffer.commit_if_leader(prisma_client))
except Exception: # noqa: BLE001 -- entries still in Redis are drained by the next flush

View file

@ -0,0 +1,350 @@
"""
Accumulates failed request counts per key, team, user and model group by HTTP
status and commits them to ``LiteLLM_DailyRequestErrors``.
The edge counters in ``LiteLLM_DailyGatewayFailedRequests`` say how many requests
failed with which status, but carry no caller dimension. This rollup is fed from
the spend writer, which sees the resolved key, team, user and model group of
every logged request together with the status code the logging callbacks
recorded, so the dashboard can answer who the failures land on. Only failures are
recorded, so the table grows with (days x failing callers x status codes), never
with successful traffic.
Flushing mirrors ``gateway_request_tracking``: one multi-row upsert per interval
per worker, or with ``use_redis_transaction_buffer`` one statement per interval
deployment-wide written by the lease holder.
"""
import asyncio
import json
from collections.abc import Iterable, Iterator, Mapping
from itertools import chain
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal, TypeAlias
from pydantic import TypeAdapter
from litellm._internal_context import with_service_target
from litellm._logging import verbose_proxy_logger
from litellm.caching import RedisCache
from litellm.constants import (
MAX_REDIS_BUFFER_DEQUEUE_COUNT,
REDIS_REQUEST_ERRORS_BUFFER_KEY,
REQUEST_ERRORS_MAX_ROWS_PER_UPSERT,
REQUEST_ERRORS_UNKNOWN_STATUS_CODE,
)
from litellm.proxy.db.db_span import db_span
from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
from litellm.types.proxy.request_errors import RequestErrorKey, RequestErrorSnapshot
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient
_REQUEST_ERROR_QUEUE_TARGET: Final = "request_error_queue"
_TABLE: Final = '"LiteLLM_DailyRequestErrors"'
_COLUMNS_PER_ROW: Final = 7
_UTC_NOW: Final = "(NOW() AT TIME ZONE 'UTC')"
REQUEST_ERRORS_JOB_NAME: Final = "update_request_errors_job"
_BufferedRow: TypeAlias = tuple[str, str, str, str, str, int, int]
_BUFFERED_ROWS: Final = TypeAdapter(tuple[_BufferedRow, ...])
_BUFFERED_ENTRIES: Final = TypeAdapter(tuple[str | bytes, ...])
_NO_COUNTS: Final[RequestErrorSnapshot] = MappingProxyType({})
_METADATA: Final = TypeAdapter(Mapping[str, object])
def status_code_from_metadata(metadata: Mapping[str, object]) -> int:
"""The HTTP status the logging callbacks recorded, or 0 when none was."""
error_information: Final = metadata.get("error_information")
if not isinstance(error_information, Mapping):
return REQUEST_ERRORS_UNKNOWN_STATUS_CODE
raw_code: Final = _METADATA.validate_python(error_information).get("error_code")
try:
code: Final = int(str(raw_code))
except (TypeError, ValueError):
return REQUEST_ERRORS_UNKNOWN_STATUS_CODE
return code if 100 <= code <= 599 else REQUEST_ERRORS_UNKNOWN_STATUS_CODE
def _text(value: object) -> str:
return value if isinstance(value, str) else ""
def _metadata(raw_metadata: object) -> Mapping[str, object]:
"""Spend logs carry metadata as a JSON string; accept an already-decoded mapping too."""
if isinstance(raw_metadata, Mapping):
return _METADATA.validate_python(raw_metadata)
if not isinstance(raw_metadata, str):
return {}
try:
return _METADATA.validate_json(raw_metadata)
except ValueError:
return {}
class RequestErrorAccumulator:
"""Sink for the spend writer. ``record`` is sync and never awaits."""
def __init__(self) -> None:
self._counts: dict[RequestErrorKey, int] = {} # mutable-ok: bounded fold, drained per flush
def record(
self,
*,
payload: Mapping[str, object],
request_status: Literal["success", "failure"],
date: str,
is_internal_call: bool,
) -> None:
if request_status == "success" or is_internal_call:
return
key: Final = RequestErrorKey(
date=date,
api_key=_text(payload.get("api_key")),
team_id=_text(payload.get("team_id")),
user_id=_text(payload.get("user")),
model_group=_text(payload.get("model_group")) or _text(payload.get("model")),
status_code=status_code_from_metadata(_metadata(payload.get("metadata"))),
)
self._counts[key] = self._counts.get(key, 0) + 1
def drain(self) -> RequestErrorSnapshot:
drained: Final = self._counts
self._counts = {}
return drained
def restore(self, snapshot: RequestErrorSnapshot) -> None:
"""Merge un-committed counts back so the next flush retries them (at-least-once)."""
self._counts = dict(fold_counts(chain(self._counts.items(), snapshot.items())))
request_error_accumulator: Final = RequestErrorAccumulator()
def fold_counts(items: Iterable[tuple[RequestErrorKey, int]]) -> RequestErrorSnapshot:
folded: Final[dict[RequestErrorKey, int]] = {} # mutable-ok: local fold returned once
for key, count in items:
folded[key] = folded.get(key, 0) + count
return folded
def _buffered_counts(entries: Iterable[str | bytes]) -> Iterator[tuple[RequestErrorKey, int]]:
for entry in entries:
for date, api_key, team_id, user_id, model_group, status_code, failed in _BUFFERED_ROWS.validate_json(entry):
yield (
RequestErrorKey(
date=date,
api_key=api_key,
team_id=team_id,
user_id=user_id,
model_group=model_group,
status_code=status_code,
),
failed,
)
def build_request_errors_upsert(snapshot: RequestErrorSnapshot) -> tuple[str, tuple[str | int, ...]]:
"""One statement; rows ordered by conflict key so concurrent writers cannot deadlock."""
ordered: Final = sorted(
((key, count) for key, count in snapshot.items() if count > 0),
key=lambda item: (
item[0].date,
item[0].api_key,
item[0].team_id,
item[0].user_id,
item[0].model_group,
item[0].status_code,
),
)
rows: Final = ", ".join(
f"(${base + 1}::text, ${base + 2}::text, ${base + 3}::text, ${base + 4}::text, "
f"${base + 5}::text, ${base + 6}::integer, ${base + 7}::bigint, {_UTC_NOW})"
for base in range(0, len(ordered) * _COLUMNS_PER_ROW, _COLUMNS_PER_ROW)
)
sql: Final = (
f"INSERT INTO {_TABLE} "
'("date", "api_key", "team_id", "user_id", "model_group", "status_code", "failed_requests", "updated_at")\n'
f"VALUES {rows}\n"
'ON CONFLICT ("date", "api_key", "team_id", "user_id", "model_group", "status_code") DO UPDATE SET\n'
f' "failed_requests" = {_TABLE}."failed_requests" + EXCLUDED."failed_requests",\n'
f' "updated_at" = {_UTC_NOW}'
)
params: Final[tuple[str | int, ...]] = tuple(
chain.from_iterable(
(key.date, key.api_key, key.team_id, key.user_id, key.model_group, key.status_code, count)
for key, count in ordered
)
)
return sql, params
def _statement_chunks(snapshot: RequestErrorSnapshot) -> Iterator[RequestErrorSnapshot]:
"""Bounded statements: a backlog of many callers must not exceed the bind-parameter limit."""
items: Final = tuple((key, count) for key, count in snapshot.items() if count > 0)
for start in range(0, len(items), REQUEST_ERRORS_MAX_ROWS_PER_UPSERT):
yield dict(items[start : start + REQUEST_ERRORS_MAX_ROWS_PER_UPSERT])
class RequestErrorCommitCancelled(asyncio.CancelledError):
"""Cancellation that interrupted a commit, carrying the rows that were not written yet."""
def __init__(self, uncommitted: RequestErrorSnapshot) -> None:
super().__init__()
self.uncommitted: Final = uncommitted
def _rows_from(chunks: tuple[RequestErrorSnapshot, ...], index: int) -> RequestErrorSnapshot:
return fold_counts(chain.from_iterable(remaining.items() for remaining in chunks[index:]))
async def commit_request_errors_to_db(
*, prisma_client: "PrismaClient", snapshot: RequestErrorSnapshot
) -> RequestErrorSnapshot:
"""One bounded statement per chunk, stopping at the first failure.
Returns the rows that were not written (the failed chunk and everything after it) so the
caller retries only those; replaying the whole snapshot would count the landed chunks twice.
"""
chunks: Final = tuple(_statement_chunks(snapshot))
for index, chunk in enumerate(chunks):
try:
sql, params = build_request_errors_upsert(chunk)
async with db_span("commit_request_errors", "LiteLLM_DailyRequestErrors"):
await prisma_client.db.execute_raw(sql, *params) # pyright: ignore[reportAny] # untyped prisma client
except asyncio.CancelledError:
raise RequestErrorCommitCancelled(_rows_from(chunks, index)) from None
except Exception: # noqa: BLE001 -- keep the chunks already written, hand back the rest for retry
verbose_proxy_logger.warning(
"Request error tracking - statement %d of %d failed, %d rows go back for retry",
index + 1,
len(chunks),
sum(len(remaining) for remaining in chunks[index:]),
exc_info=True,
)
return _rows_from(chunks, index)
verbose_proxy_logger.debug("Request error tracking - committed %d aggregated rows in one statement", len(chunk))
return _NO_COUNTS
class RequestErrorRedisBuffer:
"""Folds every worker's snapshot through one Redis list so a single pod per interval writes the table."""
def __init__(self, *, redis_cache: RedisCache, pod_lock_manager: PodLockManager) -> None:
self._redis_cache: Final = redis_cache
self._pod_lock_manager: Final = pod_lock_manager
@with_service_target(_REQUEST_ERROR_QUEUE_TARGET)
async def push(self, snapshot: RequestErrorSnapshot) -> None:
if not snapshot:
return
rows: Final[tuple[_BufferedRow, ...]] = tuple(
(key.date, key.api_key, key.team_id, key.user_id, key.model_group, key.status_code, count)
for key, count in snapshot.items()
)
await self._rpush((json.dumps(rows),))
@with_service_target(_REQUEST_ERROR_QUEUE_TARGET)
async def _rpush(self, entries: tuple[str | bytes, ...]) -> None:
if not entries:
return
await self._redis_cache.async_rpush( # pyright: ignore[reportUnknownMemberType] # redis cache span parameter is untyped
key=REDIS_REQUEST_ERRORS_BUFFER_KEY, values=entries
)
@with_service_target(_REQUEST_ERROR_QUEUE_TARGET)
async def _pop_batch(self) -> tuple[str | bytes, ...]:
popped: Final[object] = await self._redis_cache.async_lpop( # pyright: ignore[reportAny, reportUnknownMemberType] # redis returns Any; span parameter is untyped
key=REDIS_REQUEST_ERRORS_BUFFER_KEY, count=MAX_REDIS_BUFFER_DEQUEUE_COUNT
)
if not popped:
return ()
return _BUFFERED_ENTRIES.validate_python(popped if isinstance(popped, list) else (popped,))
async def pop(self) -> RequestErrorSnapshot:
"""Drains the Redis list; a cancellation mid-drain puts the popped entries back before propagating."""
entries: Final[list[str | bytes]] = [] # mutable-ok: bounded drain, folded once per call
try:
while True:
batch = await self._pop_batch()
entries.extend(batch)
if len(batch) < MAX_REDIS_BUFFER_DEQUEUE_COUNT:
break
except asyncio.CancelledError:
try:
await asyncio.shield(self._rpush(tuple(entries)))
except Exception: # noqa: BLE001 -- Redis refused them, so the caller keeps the rows in memory
raise RequestErrorCommitCancelled(fold_counts(_buffered_counts(entries))) from None
raise
except Exception: # noqa: BLE001 -- fold what was already popped; unread entries stay queued in Redis
verbose_proxy_logger.warning(
"Request error tracking - Redis read failed, folding the entries already popped", exc_info=True
)
return fold_counts(_buffered_counts(entries))
async def commit_if_leader(self, prisma_client: "PrismaClient") -> RequestErrorSnapshot:
"""Drain and write on the lease holder only; returns rows that could be neither committed nor re-queued."""
if not await self._pod_lock_manager.acquire_lock(cronjob_id=REQUEST_ERRORS_JOB_NAME):
return _NO_COUNTS
try:
buffered: Final = await self.pop()
uncommitted: Final = await commit_request_errors_to_db(prisma_client=prisma_client, snapshot=buffered)
except RequestErrorCommitCancelled as cancelled:
raise RequestErrorCommitCancelled(await asyncio.shield(self._requeue(cancelled.uncommitted))) from None
return await self._requeue(uncommitted)
async def _requeue(self, uncommitted: RequestErrorSnapshot) -> RequestErrorSnapshot:
"""Pushes unwritten rows back to Redis; returns them when Redis refuses so they stay in memory."""
if not uncommitted:
return _NO_COUNTS
try:
await self.push(uncommitted)
except Exception: # noqa: BLE001 -- the rows go back to the caller's accumulator instead
verbose_proxy_logger.warning(
"Request error tracking - Redis re-queue failed, keeping %d rows in memory",
len(uncommitted),
exc_info=True,
)
return uncommitted
verbose_proxy_logger.warning(
"Request error tracking - re-queued %d uncommitted rows to Redis", len(uncommitted)
)
return _NO_COUNTS
async def flush_request_errors(
prisma_client: "PrismaClient",
accumulator: RequestErrorAccumulator,
redis_buffer: RequestErrorRedisBuffer | None = None,
) -> None:
"""Scheduler entrypoint. Only cancellation propagates: a metering failure must not kill the job."""
snapshot: Final = accumulator.drain()
if redis_buffer is None:
try:
accumulator.restore(await commit_request_errors_to_db(prisma_client=prisma_client, snapshot=snapshot))
except RequestErrorCommitCancelled as cancelled:
accumulator.restore(cancelled.uncommitted)
raise
return
try:
await redis_buffer.push(snapshot)
except Exception: # noqa: BLE001 -- a failed push must not stop the scheduler
accumulator.restore(snapshot)
verbose_proxy_logger.warning(
"Request error tracking - failed to buffer %d rows, retrying on the next flush",
len(snapshot),
exc_info=True,
)
return
try:
accumulator.restore(await redis_buffer.commit_if_leader(prisma_client))
except RequestErrorCommitCancelled as cancelled:
accumulator.restore(cancelled.uncommitted)
raise
except Exception: # noqa: BLE001 -- entries still in Redis are drained by the next flush
verbose_proxy_logger.warning(
"Request error tracking - leader drain failed, buffered rows stay in Redis for the next flush",
exc_info=True,
)

View file

@ -13,10 +13,10 @@ and the endpoint is restricted to proxy admin roles.
from collections.abc import Sequence
from datetime import datetime, timedelta, timezone
from typing import Annotated, Final
from typing import Annotated, Final, Literal
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import TypeAdapter
from pydantic import ConfigDict, Field, TypeAdapter
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import CommonProxyErrors, LitellmUserRoles, UserAPIKeyAuth
@ -26,28 +26,38 @@ from litellm.types.proxy.gateway_requests import (
GatewayRequestActivityResponse,
GatewayRequestBreakdownEntry,
GatewayRequestDailyEntry,
GatewayRequestStatusCodeEntry,
)
router: Final = APIRouter()
_DEFAULT_LOOKBACK_DAYS: Final = 30
_AGGREGATE_SQL: Final = """
_ACTIVITY_SQL: Final = """
SELECT
date,
category,
route,
'route' AS kind, date, category, route,
SUM(successful_requests)::bigint AS successful_requests,
SUM(failed_requests)::bigint AS failed_requests
SUM(failed_requests)::bigint AS failed_requests,
NULL::integer AS status_code
FROM "LiteLLM_DailyGatewayRequests"
WHERE date >= $1 AND date <= $2
GROUP BY date, category, route
UNION ALL
SELECT
'status' AS kind, NULL, NULL, NULL,
0::bigint,
SUM(failed_requests)::bigint,
status_code
FROM "LiteLLM_DailyGatewayFailedRequests"
WHERE date >= $1 AND date <= $2
GROUP BY status_code
"""
class _AggregateRow(LiteLLMBaseModel):
"""Validates one query_raw row so the handler works with typed values, not Any."""
kind: Literal["route"]
date: str
category: str
route: str
@ -55,7 +65,17 @@ class _AggregateRow(LiteLLMBaseModel):
failed_requests: int
_ROWS_ADAPTER: Final = TypeAdapter(tuple[_AggregateRow, ...])
class _StatusCodeAggregateRow(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
kind: Literal["status"]
status_code: int
failed_requests: int
_ACTIVITY_ROWS_ADAPTER: Final = TypeAdapter(
tuple[Annotated[_AggregateRow | _StatusCodeAggregateRow, Field(discriminator="kind")], ...]
)
def _default_range() -> tuple[str, str]:
@ -92,6 +112,15 @@ def _fold_by_route(rows: Sequence[_AggregateRow]) -> tuple[GatewayRequestBreakdo
return tuple(sorted(entries, key=lambda entry: entry.successful_requests, reverse=True))
def _fold_by_status_code(
rows: Sequence[_StatusCodeAggregateRow],
) -> tuple[GatewayRequestStatusCodeEntry, ...]:
return tuple(
GatewayRequestStatusCodeEntry(status_code=row.status_code, failed_requests=row.failed_requests)
for row in sorted(rows, key=lambda row: (-row.failed_requests, row.status_code))
)
@router.get(
"/gateway/daily/activity",
tags=["Budget & Spend Tracking"],
@ -123,13 +152,18 @@ async def get_gateway_daily_activity(
raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
default_start, default_end = _default_range()
selected_start: Final = start_date or default_start
selected_end: Final = end_date or default_end
raw_rows: Final = await prisma_client.db.query_raw( # pyright: ignore[reportAny] # untyped prisma client
_AGGREGATE_SQL,
start_date or default_start,
end_date or default_end,
_ACTIVITY_SQL,
selected_start,
selected_end,
)
activity_rows: Final = _ACTIVITY_ROWS_ADAPTER.validate_python(raw_rows or ())
rows: Final[tuple[_AggregateRow, ...]] = tuple(row for row in activity_rows if isinstance(row, _AggregateRow))
status_code_rows: Final[tuple[_StatusCodeAggregateRow, ...]] = tuple(
row for row in activity_rows if isinstance(row, _StatusCodeAggregateRow)
)
# Every downstream use is typed: the adapter returns _AggregateRow or raises.
rows: Final = _ROWS_ADAPTER.validate_python(raw_rows or ())
verbose_proxy_logger.debug("/gateway/daily/activity - aggregated %d rows", len(rows))
return GatewayRequestActivityResponse(
@ -137,4 +171,5 @@ async def get_gateway_daily_activity(
total_failed_requests=sum(row.failed_requests for row in rows),
by_date=_fold_by_date(rows),
by_route=_fold_by_route(rows),
by_status_code=_fold_by_status_code(status_code_rows),
)

View file

@ -0,0 +1,273 @@
"""
FAILED REQUEST ANALYTICS
GET /gateway/errors/activity - failure rate over time, failures by HTTP status, and
the keys, teams, users and model groups the failures land on.
Totals and the per-status time series come from the edge counters
(LiteLLM_DailyGatewayRequests, LiteLLM_DailyGatewayFailedRequests), which count what
the proxy answered. The caller breakdown reads the daily spend rollups for request
volume and LiteLLM_DailyRequestErrors for the status each caller failed with.
Deployment-wide, so admin-only.
"""
from collections.abc import Mapping, Sequence
from datetime import datetime, timedelta, timezone
from typing import Annotated, Final, Literal, TypeAlias
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import ConfigDict, TypeAdapter
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import CommonProxyErrors, LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.proxy.request_errors import (
RequestErrorActivityResponse,
RequestErrorDailyEntry,
RequestErrorEntityEntry,
RequestErrorStatusCodeEntry,
)
router: Final = APIRouter()
_DEFAULT_LOOKBACK_DAYS: Final = 30
_ENTITY_LIMIT: Final = 25
EntityKind: TypeAlias = Literal["key", "team", "user", "model"]
_DATE_SQL: Final = """
SELECT date, NULL::integer AS status_code,
SUM(successful_requests)::bigint AS successful_requests,
SUM(failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyGatewayRequests"
WHERE date >= $1 AND date <= $2
GROUP BY date
UNION ALL
SELECT date, status_code, 0::bigint, SUM(failed_requests)::bigint
FROM "LiteLLM_DailyGatewayFailedRequests"
WHERE date >= $1 AND date <= $2
GROUP BY date, status_code
"""
_ENTITY_SQL: Final = f"""
SELECT * FROM (
SELECT 'key' AS kind, u.api_key AS id, MAX(v.key_alias) AS label,
SUM(u.api_requests)::bigint AS api_requests, SUM(u.failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyUserSpend" u
LEFT JOIN "LiteLLM_VerificationToken" v ON v.token = u.api_key
WHERE u.date >= $1 AND u.date <= $2
GROUP BY u.api_key HAVING SUM(u.failed_requests) > 0
ORDER BY failed_requests DESC, id LIMIT {_ENTITY_LIMIT}
) keys
UNION ALL
SELECT * FROM (
SELECT 'team' AS kind, t.team_id AS id, MAX(tt.team_alias) AS label,
SUM(t.api_requests)::bigint, SUM(t.failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyTeamSpend" t
LEFT JOIN "LiteLLM_TeamTable" tt ON tt.team_id = t.team_id
WHERE t.date >= $1 AND t.date <= $2 AND t.team_id IS NOT NULL AND t.team_id <> ''
GROUP BY t.team_id HAVING SUM(t.failed_requests) > 0
ORDER BY failed_requests DESC, id LIMIT {_ENTITY_LIMIT}
) teams
UNION ALL
SELECT * FROM (
SELECT 'user' AS kind, u.user_id AS id, MAX(ut.user_email) AS label,
SUM(u.api_requests)::bigint, SUM(u.failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyUserSpend" u
LEFT JOIN "LiteLLM_UserTable" ut ON ut.user_id = u.user_id
WHERE u.date >= $1 AND u.date <= $2 AND u.user_id IS NOT NULL AND u.user_id <> ''
GROUP BY u.user_id HAVING SUM(u.failed_requests) > 0
ORDER BY failed_requests DESC, id LIMIT {_ENTITY_LIMIT}
) users
UNION ALL
SELECT * FROM (
SELECT 'model' AS kind, u.model_group AS id, NULL::text AS label,
SUM(u.api_requests)::bigint, SUM(u.failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyUserSpend" u
WHERE u.date >= $1 AND u.date <= $2 AND u.model_group IS NOT NULL AND u.model_group <> ''
GROUP BY u.model_group HAVING SUM(u.failed_requests) > 0
ORDER BY failed_requests DESC, id LIMIT {_ENTITY_LIMIT}
) models
"""
_ENTITY_STATUS_SQL: Final = f"""
SELECT * FROM (
SELECT DISTINCT ON (api_key) 'key' AS kind, api_key AS id, status_code, SUM(failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyRequestErrors" WHERE date >= $1 AND date <= $2
AND api_key IN (SELECT u.api_key FROM "LiteLLM_DailyUserSpend" u WHERE u.date >= $1 AND u.date <= $2 GROUP BY u.api_key HAVING SUM(u.failed_requests) > 0 ORDER BY SUM(u.failed_requests) DESC, u.api_key LIMIT {_ENTITY_LIMIT})
GROUP BY api_key, status_code ORDER BY api_key, failed_requests DESC, status_code
) key_status
UNION ALL
SELECT * FROM (
SELECT DISTINCT ON (team_id) 'team' AS kind, team_id AS id, status_code, SUM(failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyRequestErrors" WHERE date >= $1 AND date <= $2 AND team_id <> ''
AND team_id IN (SELECT t.team_id FROM "LiteLLM_DailyTeamSpend" t WHERE t.date >= $1 AND t.date <= $2 AND t.team_id IS NOT NULL AND t.team_id <> '' GROUP BY t.team_id HAVING SUM(t.failed_requests) > 0 ORDER BY SUM(t.failed_requests) DESC, t.team_id LIMIT {_ENTITY_LIMIT})
GROUP BY team_id, status_code ORDER BY team_id, failed_requests DESC, status_code
) team_status
UNION ALL
SELECT * FROM (
SELECT DISTINCT ON (user_id) 'user' AS kind, user_id AS id, status_code, SUM(failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyRequestErrors" WHERE date >= $1 AND date <= $2 AND user_id <> ''
AND user_id IN (SELECT u.user_id FROM "LiteLLM_DailyUserSpend" u WHERE u.date >= $1 AND u.date <= $2 AND u.user_id IS NOT NULL AND u.user_id <> '' GROUP BY u.user_id HAVING SUM(u.failed_requests) > 0 ORDER BY SUM(u.failed_requests) DESC, u.user_id LIMIT {_ENTITY_LIMIT})
GROUP BY user_id, status_code ORDER BY user_id, failed_requests DESC, status_code
) user_status
UNION ALL
SELECT * FROM (
SELECT DISTINCT ON (model_group) 'model' AS kind, model_group AS id, status_code, SUM(failed_requests)::bigint AS failed_requests
FROM "LiteLLM_DailyRequestErrors" WHERE date >= $1 AND date <= $2 AND model_group <> ''
AND model_group IN (SELECT u.model_group FROM "LiteLLM_DailyUserSpend" u WHERE u.date >= $1 AND u.date <= $2 AND u.model_group IS NOT NULL AND u.model_group <> '' GROUP BY u.model_group HAVING SUM(u.failed_requests) > 0 ORDER BY SUM(u.failed_requests) DESC, u.model_group LIMIT {_ENTITY_LIMIT})
GROUP BY model_group, status_code ORDER BY model_group, failed_requests DESC, status_code
) model_status
"""
class _DateRow(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
date: str
status_code: int | None = None
successful_requests: int = 0
failed_requests: int = 0
class _EntityRow(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
kind: EntityKind
id: str
label: str | None = None
api_requests: int = 0
failed_requests: int = 0
class _EntityStatusRow(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
kind: EntityKind
id: str
status_code: int
failed_requests: int = 0
_DATE_ROWS: Final = TypeAdapter(tuple[_DateRow, ...])
_ENTITY_ROWS: Final = TypeAdapter(tuple[_EntityRow, ...])
_ENTITY_STATUS_ROWS: Final = TypeAdapter(tuple[_EntityStatusRow, ...])
def _default_range() -> tuple[str, str]:
end: Final = datetime.now(timezone.utc)
start: Final = end - timedelta(days=_DEFAULT_LOOKBACK_DAYS)
return start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d")
def _is_client_error(status_code: int | None) -> bool:
return status_code is not None and 400 <= status_code <= 499
def _is_server_error(status_code: int | None) -> bool:
return status_code is not None and 500 <= status_code <= 599
def _day_entry(date: str, rows: Sequence[_DateRow]) -> RequestErrorDailyEntry:
day_rows: Final = tuple(row for row in rows if row.date == date)
return RequestErrorDailyEntry(
date=date,
successful_requests=sum(row.successful_requests for row in day_rows),
failed_requests=sum(row.failed_requests for row in day_rows if row.status_code is None),
client_errors=sum(row.failed_requests for row in day_rows if _is_client_error(row.status_code)),
server_errors=sum(row.failed_requests for row in day_rows if _is_server_error(row.status_code)),
by_status_code=fold_by_status_code(day_rows),
)
def fold_by_date(rows: Sequence[_DateRow]) -> tuple[RequestErrorDailyEntry, ...]:
"""Edge totals per day with the 4xx and 5xx share of the failures and the per-status counts."""
return tuple(_day_entry(date, rows) for date in sorted(frozenset(row.date for row in rows)))
def fold_by_status_code(rows: Sequence[_DateRow]) -> tuple[RequestErrorStatusCodeEntry, ...]:
codes: Final = frozenset(row.status_code for row in rows if row.status_code is not None)
entries: Final = tuple(
RequestErrorStatusCodeEntry(
status_code=code, failed_requests=sum(row.failed_requests for row in rows if row.status_code == code)
)
for code in codes
)
return tuple(sorted(entries, key=lambda entry: (-entry.failed_requests, entry.status_code)))
def _entity_entry(row: _EntityRow, top_statuses: Mapping[tuple[str, str], _EntityStatusRow]) -> RequestErrorEntityEntry:
top: Final = top_statuses.get((row.kind, row.id))
return RequestErrorEntityEntry(
id=row.id,
label=row.label,
api_requests=row.api_requests,
failed_requests=row.failed_requests,
top_status_code=top.status_code if top is not None else None,
top_status_code_requests=top.failed_requests if top is not None else 0,
)
def fold_entities(
kind: EntityKind, rows: Sequence[_EntityRow], status_rows: Sequence[_EntityStatusRow]
) -> tuple[RequestErrorEntityEntry, ...]:
"""Callers of one kind ranked by failures, each with the status they failed with most (one row per caller)."""
top_statuses: Final = {(status.kind, status.id): status for status in status_rows}
entries: Final = tuple(_entity_entry(row, top_statuses) for row in filter(lambda row: row.kind == kind, rows))
return tuple(sorted(entries, key=lambda entry: (-entry.failed_requests, entry.id)))
@router.get(
"/gateway/errors/activity",
tags=["Budget & Spend Tracking"],
response_model=RequestErrorActivityResponse,
)
async def get_request_error_activity(
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
start_date: str | None = Query(default=None, description="Start date in YYYY-MM-DD format"),
end_date: str | None = Query(default=None, description="End date in YYYY-MM-DD format"),
) -> RequestErrorActivityResponse:
"""
Failed requests over time by HTTP status, and the keys, teams, users and
model groups they land on. Deployment-wide, so admin-only.
"""
from litellm.proxy.proxy_server import prisma_client
if user_api_key_dict.user_role not in (LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
raise HTTPException(
status_code=403, detail="Only proxy admin roles can view failed requests across the deployment"
)
if prisma_client is None:
raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
default_start, default_end = _default_range()
selected_start: Final = start_date or default_start
selected_end: Final = end_date or default_end
db: Final = prisma_client.db
date_rows: Final = _DATE_ROWS.validate_python(
await db.query_raw(_DATE_SQL, selected_start, selected_end) or () # pyright: ignore[reportAny] # untyped prisma client
)
entity_rows: Final = _ENTITY_ROWS.validate_python(
await db.query_raw(_ENTITY_SQL, selected_start, selected_end) or () # pyright: ignore[reportAny] # untyped prisma client
)
status_rows: Final = _ENTITY_STATUS_ROWS.validate_python(
await db.query_raw(_ENTITY_STATUS_SQL, selected_start, selected_end) or () # pyright: ignore[reportAny] # untyped prisma client
)
verbose_proxy_logger.debug(
"/gateway/errors/activity - %d date rows, %d entity rows, %d status rows",
len(date_rows),
len(entity_rows),
len(status_rows),
)
totals: Final = tuple(row for row in date_rows if row.status_code is None)
return RequestErrorActivityResponse(
total_successful_requests=sum(row.successful_requests for row in totals),
total_failed_requests=sum(row.failed_requests for row in totals),
by_date=fold_by_date(date_rows),
by_status_code=fold_by_status_code(date_rows),
by_key=fold_entities("key", entity_rows, status_rows),
by_team=fold_entities("team", entity_rows, status_rows),
by_user=fold_entities("user", entity_rows, status_rows),
by_model=fold_entities("model", entity_rows, status_rows),
)

View file

@ -559,6 +559,11 @@ from litellm.proxy.db.proxy_worker_heartbeat import (
PROXY_WORKER_HEARTBEAT_INTERVAL_SECONDS,
ProxyWorkerHeartbeat,
)
from litellm.proxy.db.request_error_tracking import (
RequestErrorRedisBuffer,
flush_request_errors,
request_error_accumulator,
)
from litellm.proxy.db.spend_counter_reseed import END_USER_COUNTER_PREFIX, SpendCounterReseed
from litellm.proxy.discovery_endpoints import (
agent_skills_discovery_router,
@ -696,6 +701,9 @@ from litellm.proxy.management_endpoints.password_endpoints import (
from litellm.proxy.management_endpoints.prompt_caching_requests import (
router as prompt_caching_requests_router,
)
from litellm.proxy.management_endpoints.request_error_endpoints import (
router as request_error_router,
)
from litellm.proxy.management_endpoints.router_settings_endpoints import (
router as router_settings_router,
)
@ -1191,6 +1199,7 @@ async def proxy_shutdown_event(worker_heartbeat: ProxyWorkerHeartbeat | None = N
# ClientNotConnectedError rather than persisting anything. Ordering this
# inside the same guard is what keeps the two from drifting apart.
await flush_gateway_requests(prisma_client, gateway_request_accumulator)
await flush_request_errors(prisma_client, request_error_accumulator)
verbose_proxy_logger.debug("Disconnecting from Prisma")
await prisma_client.disconnect()
@ -2850,6 +2859,14 @@ def _gateway_request_redis_buffer() -> GatewayRequestRedisBuffer | None:
return GatewayRequestRedisBuffer(redis_cache=redis_cache, pod_lock_manager=writer.pod_lock_manager)
def _request_error_redis_buffer() -> RequestErrorRedisBuffer | None:
writer: Final = proxy_logging_obj.db_spend_update_writer
redis_cache: Final = writer.redis_update_buffer.redis_cache
if redis_cache is None or not writer.redis_update_buffer.should_commit_spend_updates_to_redis():
return None
return RequestErrorRedisBuffer(redis_cache=redis_cache, pod_lock_manager=writer.pod_lock_manager)
### REDIS QUEUE ###
async_result: Final = None
celery_app_conn: Final = None
@ -10982,6 +10999,17 @@ class ProxyStartupEvent:
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
)
### UPDATE FAILED REQUEST COUNTS BY CALLER AND STATUS ###
scheduler.add_job( # pyright: ignore[reportUnknownMemberType] # apscheduler is untyped
flush_request_errors,
"interval",
seconds=batch_writing_interval,
args=(prisma_client, request_error_accumulator, _request_error_redis_buffer()),
id="update_request_errors_job",
replace_existing=True,
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
)
### MONITOR SPEND LOGS QUEUE (queue-size-based job) ###
if general_settings.get("disable_spend_logs", False) is False:
from litellm.proxy.utils import monitor_spend_logs_queue
@ -20342,6 +20370,7 @@ app.include_router(cache_settings_router)
app.include_router(coordination_redis_settings_router)
app.include_router(user_agent_analytics_router)
app.include_router(gateway_request_router)
app.include_router(request_error_router)
app.include_router(enterprise_router)
app.include_router(ui_discovery_endpoints_router)
app.include_router(agent_skills_discovery_router)

View file

@ -1361,6 +1361,34 @@ model LiteLLM_DailyGatewayRequests {
@@index([date])
}
model LiteLLM_DailyGatewayFailedRequests {
date String
category String
route String
status_code Int
failed_requests BigInt @default(0)
created_at DateTime @default(now())
updated_at DateTime @updatedAt
@@id([date, category, route, status_code])
@@index([date])
}
model LiteLLM_DailyRequestErrors {
date String
api_key String
team_id String @default("")
user_id String @default("")
model_group String @default("")
status_code Int
failed_requests BigInt @default(0)
created_at DateTime @default(now())
updated_at DateTime @updatedAt
@@id([date, api_key, team_id, user_id, model_group, status_code])
@@index([date])
}
// Prompt table for storing prompt configurations
model LiteLLM_PromptTable {
id String @id @default(uuid())

View file

@ -4,6 +4,8 @@ from collections.abc import Mapping
from dataclasses import dataclass
from typing import TypeAlias
from pydantic import ConfigDict
from litellm.types.llms.base import LiteLLMBaseModel
@ -12,6 +14,7 @@ class GatewayRequestKey:
date: str
category: str
route: str
status_code: int | None = None
@dataclass(frozen=True, slots=True)
@ -42,6 +45,13 @@ class GatewayRequestDailyEntry(LiteLLMBaseModel):
failed_requests: int = 0
class GatewayRequestStatusCodeEntry(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
status_code: int
failed_requests: int = 0
class GatewayRequestActivityResponse(LiteLLMBaseModel):
"""Response for GET /gateway/daily/activity."""
@ -49,3 +59,4 @@ class GatewayRequestActivityResponse(LiteLLMBaseModel):
total_failed_requests: int = 0
by_date: tuple[GatewayRequestDailyEntry, ...] = ()
by_route: tuple[GatewayRequestBreakdownEntry, ...] = ()
by_status_code: tuple[GatewayRequestStatusCodeEntry, ...] = ()

View file

@ -0,0 +1,66 @@
"""Types for per-entity failed request counts by HTTP status, rolled up daily."""
from collections.abc import Mapping
from dataclasses import dataclass
from typing import TypeAlias
from pydantic import ConfigDict
from litellm.types.llms.base import LiteLLMBaseModel
@dataclass(frozen=True, slots=True)
class RequestErrorKey:
date: str
api_key: str
team_id: str
user_id: str
model_group: str
status_code: int
RequestErrorSnapshot: TypeAlias = Mapping[RequestErrorKey, int]
class RequestErrorStatusCodeEntry(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
status_code: int
failed_requests: int = 0
class RequestErrorDailyEntry(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
date: str
successful_requests: int = 0
failed_requests: int = 0
client_errors: int = 0
server_errors: int = 0
by_status_code: tuple[RequestErrorStatusCodeEntry, ...] = ()
class RequestErrorEntityEntry(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
id: str
label: str | None = None
api_requests: int = 0
failed_requests: int = 0
top_status_code: int | None = None
top_status_code_requests: int = 0
class RequestErrorActivityResponse(LiteLLMBaseModel):
"""Response for GET /gateway/errors/activity."""
model_config = ConfigDict(frozen=True)
total_successful_requests: int = 0
total_failed_requests: int = 0
by_date: tuple[RequestErrorDailyEntry, ...] = ()
by_status_code: tuple[RequestErrorStatusCodeEntry, ...] = ()
by_key: tuple[RequestErrorEntityEntry, ...] = ()
by_team: tuple[RequestErrorEntityEntry, ...] = ()
by_user: tuple[RequestErrorEntityEntry, ...] = ()
by_model: tuple[RequestErrorEntityEntry, ...] = ()

View file

@ -1361,6 +1361,34 @@ model LiteLLM_DailyGatewayRequests {
@@index([date])
}
model LiteLLM_DailyGatewayFailedRequests {
date String
category String
route String
status_code Int
failed_requests BigInt @default(0)
created_at DateTime @default(now())
updated_at DateTime @updatedAt
@@id([date, category, route, status_code])
@@index([date])
}
model LiteLLM_DailyRequestErrors {
date String
api_key String
team_id String @default("")
user_id String @default("")
model_group String @default("")
status_code Int
failed_requests BigInt @default(0)
created_at DateTime @default(now())
updated_at DateTime @updatedAt
@@id([date, api_key, team_id, user_id, model_group, status_code])
@@index([date])
}
// Prompt table for storing prompt configurations
model LiteLLM_PromptTable {
id String @id @default(uuid())

View file

@ -5115,3 +5115,35 @@ async def test_daily_spend_rows_survive_a_lock_timeout_for_the_next_flush():
assert len(prisma_client.db.statements) == 1
assert list(daily_spend_transactions) == ["key"]
def test_record_request_error_feeds_the_failure_rollup_with_the_logged_status() -> None:
from litellm.proxy.db import db_spend_update_writer as writer_module
from litellm.proxy.db.request_error_tracking import RequestErrorAccumulator
from litellm.types.proxy.request_errors import RequestErrorKey
accumulator: Final = RequestErrorAccumulator()
prisma_client: Final = MagicMock()
prisma_client.get_request_status.side_effect = lambda payload: json.loads(payload["metadata"])["status"]
failed_payload: Final[dict[str, object]] = {
"startTime": datetime(2026, 10, 8, 23, 59, tzinfo=timezone.utc),
"api_key": "hash-1",
"team_id": "team-a",
"user": "user-a",
"model_group": "gpt-4o",
"metadata": json.dumps({"status": "failure", "error_information": {"error_code": "503"}}),
}
ok_payload: Final[dict[str, object]] = {**failed_payload, "metadata": json.dumps({"status": "success"})}
with patch.object(writer_module, "request_error_accumulator", accumulator):
DBSpendUpdateWriter._record_request_error(payload=failed_payload, prisma_client=prisma_client)
DBSpendUpdateWriter._record_request_error(payload=ok_payload, prisma_client=prisma_client)
assert accumulator.drain() == {
RequestErrorKey(
date="2026-10-08",
api_key="hash-1",
team_id="team-a",
user_id="user-a",
model_group="gpt-4o",
status_code=503,
): 1
}

View file

@ -4,13 +4,21 @@ LiteLLM_DailyGatewayRequests.
"""
import asyncio
import json
from collections.abc import Awaitable, Callable
from datetime import datetime, timezone
from itertools import chain
from typing import Final
from unittest.mock import AsyncMock, MagicMock
import pytest
from litellm.constants import MAX_REDIS_BUFFER_DEQUEUE_COUNT, REDIS_GATEWAY_REQUESTS_BUFFER_KEY
from litellm.constants import (
GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT,
MAX_REDIS_BUFFER_DEQUEUE_COUNT,
REDIS_GATEWAY_REQUESTS_BUFFER_KEY,
REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2,
)
from litellm.proxy.db.gateway_request_tracking import (
GATEWAY_REQUESTS_JOB_NAME,
GatewayRequestAccumulator,
@ -18,10 +26,9 @@ from litellm.proxy.db.gateway_request_tracking import (
commit_gateway_requests_to_db,
flush_gateway_requests,
)
from litellm.proxy.db.log_db_metrics import record_db_io
from litellm.proxy.middleware.billable_request_metrics_middleware import BillableCategory
from litellm.types.proxy.gateway_requests import GatewayRequestCounts, GatewayRequestKey
from litellm.proxy.db.log_db_metrics import record_db_io
_NOW: Final = datetime(2026, 3, 14, 12, 0, tzinfo=timezone.utc)
_DAY: Final = _NOW.strftime("%Y-%m-%d")
@ -50,9 +57,27 @@ def test_folds_repeated_requests_into_one_key():
snapshot = acc.drain()
assert snapshot == {
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions"): (
GatewayRequestCounts(successful_requests=3, failed_requests=1)
)
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=200): (
GatewayRequestCounts(successful_requests=3, failed_requests=0)
),
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=500): (
GatewayRequestCounts(successful_requests=0, failed_requests=1)
),
}
def test_folds_failed_status_codes_into_distinct_keys_for_the_same_route():
acc: Final = _accumulator()
_record(acc, 429)
_record(acc, 500)
assert acc.drain() == {
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=429): (
GatewayRequestCounts(successful_requests=0, failed_requests=1)
),
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=500): (
GatewayRequestCounts(successful_requests=0, failed_requests=1)
),
}
@ -69,12 +94,12 @@ def test_records_each_request_under_the_date_of_its_clock_read():
_record(accumulator, 200)
assert accumulator.drain() == {
GatewayRequestKey(date="2026-01-31", category="llm", route="/chat/completions"): GatewayRequestCounts(
successful_requests=1, failed_requests=0
),
GatewayRequestKey(date="2026-02-01", category="llm", route="/chat/completions"): GatewayRequestCounts(
successful_requests=1, failed_requests=0
),
GatewayRequestKey(
date="2026-01-31", category="llm", route="/chat/completions", status_code=200
): GatewayRequestCounts(successful_requests=1, failed_requests=0),
GatewayRequestKey(
date="2026-02-01", category="llm", route="/chat/completions", status_code=200
): GatewayRequestCounts(successful_requests=1, failed_requests=0),
}
@ -132,7 +157,32 @@ class FakePrismaClient:
def _rows_written(client: FakePrismaClient) -> list[tuple[object, ...]]:
"""Every (date, category, route, successful, failed) tuple the database received, in statement order."""
return [params[i : i + 5] for _, params in client.db.statements for i in range(0, len(params), 5)]
return list(chain.from_iterable(_main_rows_for_statement(query, params) for query, params in client.db.statements))
def _main_param_offset(query: str) -> int:
main_query: Final[str] = query[query.index('INSERT INTO "LiteLLM_DailyGatewayRequests"') :]
first_placeholder: Final[int] = main_query.index("$")
placeholder_type: Final[int] = main_query.index("::", first_placeholder)
offset: Final[int] = int(main_query[first_placeholder + 1 : placeholder_type]) - 1
return offset
def _main_rows_for_statement(query: str, params: tuple[object, ...]) -> list[tuple[object, ...]]:
offset: Final[int] = _main_param_offset(query)
return [params[offset + i : offset + i + 5] for i in range(0, len(params) - offset, 5)]
def _failure_rows_written(client: FakePrismaClient) -> list[tuple[object, ...]]:
return list(
chain.from_iterable(_failure_rows_for_statement(query, params) for query, params in client.db.statements)
)
def _failure_rows_for_statement(query: str, params: tuple[object, ...]) -> list[tuple[object, ...]]:
if 'INSERT INTO "LiteLLM_DailyGatewayFailedRequests"' not in query:
return []
return [params[i : i + 5] for i in range(0, _main_param_offset(query), 5)]
def test_commit_increments_with_a_single_statement_for_the_whole_snapshot():
@ -198,6 +248,70 @@ def test_commit_placeholders_line_up_with_params():
assert params == ("2026-08-01", "llm", "/chat/completions", 1, 0, "2026-08-01", "mcp", "/mcp", 0, 3)
def test_commit_upserts_per_status_and_sums_main_table_counts_in_one_statement():
client: Final = FakePrismaClient()
accumulator: Final = _accumulator()
_record(accumulator, 200)
_record(accumulator, 201)
_record(accumulator, 429)
_record(accumulator, 500)
asyncio.run(commit_gateway_requests_to_db(prisma_client=client, snapshot=accumulator.drain()))
assert len(client.db.statements) == 1
sql, params = client.db.statements[0]
assert sql.startswith('WITH failed AS (INSERT INTO "LiteLLM_DailyGatewayFailedRequests"')
assert sql.index('INSERT INTO "LiteLLM_DailyGatewayFailedRequests"') < sql.index(
'INSERT INTO "LiteLLM_DailyGatewayRequests"'
)
assert "$1::text, $2::text, $3::text, $4::integer, $5::bigint" in sql
assert "$6::text, $7::text, $8::text, $9::integer, $10::bigint" in sql
assert "$11::text, $12::text, $13::text, $14::bigint, $15::bigint" in sql
assert "$16" not in sql
assert params == (
_DAY,
"llm",
"/chat/completions",
429,
1,
_DAY,
"llm",
"/chat/completions",
500,
1,
_DAY,
"llm",
"/chat/completions",
2,
2,
)
assert _failure_rows_written(client) == [
(_DAY, "llm", "/chat/completions", 429, 1),
(_DAY, "llm", "/chat/completions", 500, 1),
]
assert _rows_written(client) == [(_DAY, "llm", "/chat/completions", 2, 2)]
def test_success_only_snapshot_uses_only_the_existing_table_upsert():
client: Final = FakePrismaClient()
accumulator: Final = _accumulator()
_record(accumulator, 200)
snapshot: Final = accumulator.drain()
assert snapshot == {
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=200): (
GatewayRequestCounts(successful_requests=1, failed_requests=0)
)
}
asyncio.run(commit_gateway_requests_to_db(prisma_client=client, snapshot=snapshot))
sql, _ = client.db.statements[0]
assert sql.startswith('INSERT INTO "LiteLLM_DailyGatewayRequests"')
assert "LiteLLM_DailyGatewayFailedRequests" not in sql
assert _failure_rows_written(client) == []
def test_commit_is_deterministically_ordered():
"""Concurrent writers must touch rows in the same order or they deadlock."""
client = FakePrismaClient()
@ -214,6 +328,65 @@ def test_commit_is_deterministically_ordered():
assert written_order == [("2026-08-01", "llm"), ("2026-08-01", "mcp"), ("2026-08-02", "llm")]
def _backlog(size: int) -> dict[GatewayRequestKey, GatewayRequestCounts]:
return {
GatewayRequestKey(date=_DAY, category="llm", route=f"/route-{index:05d}", status_code=200): (
GatewayRequestCounts(successful_requests=1, failed_requests=0)
)
for index in range(size)
}
def test_commit_splits_a_backlog_into_bounded_statements():
client: Final = MagicMock()
client.db.execute_raw = AsyncMock()
uncommitted: Final = asyncio.run(
commit_gateway_requests_to_db(prisma_client=client, snapshot=_backlog(GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT + 1))
)
assert uncommitted == {}
assert client.db.execute_raw.await_count == 2
assert len(client.db.execute_raw.await_args_list[0].args) - 1 == GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT * 5
def test_commit_keeps_a_route_total_and_its_status_rows_in_one_statement():
client: Final = MagicMock()
client.db.execute_raw = AsyncMock()
snapshot: Final = {
**_backlog(GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT - 1),
GatewayRequestKey(date=_DAY, category="llm", route="/last", status_code=429): (
GatewayRequestCounts(successful_requests=0, failed_requests=1)
),
GatewayRequestKey(date=_DAY, category="llm", route="/last", status_code=500): (
GatewayRequestCounts(successful_requests=0, failed_requests=1)
),
}
asyncio.run(commit_gateway_requests_to_db(prisma_client=client, snapshot=snapshot))
assert client.db.execute_raw.await_count == 2
with_last: Final = [
awaited.args for awaited in client.db.execute_raw.await_args_list if "/last" in awaited.args[1:]
]
assert len(with_last) == 1
assert with_last[0][1:].count("/last") == 3
def test_commit_hands_back_only_the_chunks_that_did_not_land():
client: Final = MagicMock()
client.db.execute_raw = AsyncMock(side_effect=(None, RuntimeError("db down")))
snapshot: Final = _backlog(GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT + 1)
uncommitted: Final = asyncio.run(commit_gateway_requests_to_db(prisma_client=client, snapshot=snapshot))
last_key: Final = GatewayRequestKey(
date=_DAY, category="llm", route=f"/route-{GATEWAY_REQUESTS_MAX_ROWS_PER_UPSERT:05d}", status_code=200
)
assert uncommitted == {last_key: snapshot[last_key]}
assert client.db.execute_raw.await_count == 2
def test_commit_skips_the_database_entirely_when_nothing_accumulated():
client = FakePrismaClient()
asyncio.run(commit_gateway_requests_to_db(prisma_client=client, snapshot={}))
@ -373,7 +546,37 @@ def test_non_leader_workers_push_to_redis_and_never_touch_the_database():
asyncio.run(flush_gateway_requests(client, acc, buffer))
assert client.db.statements == []
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY]) == 3
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2]) == 3
assert redis.lists.get(REDIS_GATEWAY_REQUESTS_BUFFER_KEY, []) == []
def test_push_writes_six_field_rows_to_the_v2_key():
redis: Final = FakeRedis()
accumulator: Final = _accumulator()
_record(accumulator, 429)
buffer, _ = _buffer(redis, leader=False)
asyncio.run(buffer.push(accumulator.drain()))
assert json.loads(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2][0]) == [
[_DAY, "llm", "/chat/completions", 429, 0, 1]
]
assert redis.lists.get(REDIS_GATEWAY_REQUESTS_BUFFER_KEY, []) == []
def test_leader_combines_legacy_and_v2_buffer_rows():
redis: Final = FakeRedis()
redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY] = [json.dumps(((_DAY, "llm", "/chat/completions", 2, 1),))]
redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] = [json.dumps(((_DAY, "llm", "/chat/completions", 429, 0, 2),))]
client: Final = FakePrismaClient()
leader, _ = _buffer(redis, leader=True)
asyncio.run(flush_gateway_requests(client, _accumulator(), leader))
assert _rows_written(client) == [(_DAY, "llm", "/chat/completions", 2, 3)]
assert _failure_rows_written(client) == [(_DAY, "llm", "/chat/completions", 429, 2)]
assert redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY] == []
assert redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] == []
def test_leader_folds_every_workers_snapshot_into_one_statement():
@ -397,7 +600,7 @@ def test_leader_folds_every_workers_snapshot_into_one_statement():
(_DAY, "llm", "/chat/completions", 51, 0),
(_DAY, "llm", "/responses", 0, 50),
]
assert redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY] == []
assert redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] == []
assert lock.held == [GATEWAY_REQUESTS_JOB_NAME]
assert lock.released == []
@ -421,7 +624,7 @@ def test_leader_keeps_the_lease_so_staggered_pods_cost_one_statement_per_interva
assert lease.holder == "pod-0"
assert len(client.db.statements) == 3
assert [row[3] for row in _rows_written(client)] == [1, 4, 4]
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY]) == 3
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2]) == 3
def test_leader_drains_a_backlog_deeper_than_one_capped_pop():
@ -440,7 +643,7 @@ def test_leader_drains_a_backlog_deeper_than_one_capped_pop():
assert len(client.db.statements) == 1
assert _rows_written(client) == [(_DAY, "llm", "/chat/completions", workers, 0)]
assert redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY] == []
assert redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] == []
def test_leader_with_nothing_buffered_writes_nothing():
@ -464,7 +667,7 @@ def test_leader_requeues_to_redis_when_the_database_commit_fails():
asyncio.run(flush_gateway_requests(ExplodingClient(), acc, leader))
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY]) == 1
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2]) == 1
assert lock.released == []
assert acc.drain() == {}
@ -484,6 +687,20 @@ class UnreadableRedis(FakeRedis):
raise RuntimeError("redis gone mid-flush")
class LegacyKeyReadFailureRedis(FakeRedis):
async def async_lpop(self, key: str, count: int) -> list[str] | None:
if key == REDIS_GATEWAY_REQUESTS_BUFFER_KEY:
raise RuntimeError("legacy redis key unreadable")
return await super().async_lpop(key, count)
class FailureAfterFullV2BatchRedis(FakeRedis):
async def async_lpop(self, key: str, count: int) -> list[str] | None:
if key == REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2 and not self.lists.get(key):
raise RuntimeError("redis read failed after full batch")
return await super().async_lpop(key, count)
class UnwritableRedis(FakeRedis):
"""Pops succeed, pushes fail: a Redis that went read-only between the leader's pop and its re-queue."""
@ -505,7 +722,7 @@ def test_leader_keeps_popped_counts_in_memory_when_both_the_database_and_the_req
leader_acc = _accumulator()
leader, _ = _buffer(degraded, leader=True)
asyncio.run(flush_gateway_requests(ExplodingClient(), leader_acc, leader))
assert degraded.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY] == []
assert degraded.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] == []
client = FakePrismaClient()
retry, _ = _buffer(redis, leader=True)
@ -525,7 +742,60 @@ def test_leader_whose_redis_read_fails_leaves_the_pushed_rows_for_the_next_flush
assert client.db.statements == []
assert acc.drain() == {}
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY]) == 1
assert len(redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2]) == 1
def test_pop_keeps_v2_counts_when_legacy_key_read_fails():
redis: Final = LegacyKeyReadFailureRedis()
redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] = [
json.dumps(((_DAY, "llm", "/chat/completions", 500, 0, 2),)),
json.dumps(((_DAY, "llm", "/embeddings", 429, 0, 3),)),
]
buffer: Final = _buffer(redis, leader=True)[0]
snapshot: Final = asyncio.run(buffer.pop())
assert snapshot == {
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=500): (
GatewayRequestCounts(successful_requests=0, failed_requests=2)
),
GatewayRequestKey(date=_DAY, category="llm", route="/embeddings", status_code=429): (
GatewayRequestCounts(successful_requests=0, failed_requests=3)
),
}
def test_commit_if_leader_writes_v2_counts_when_legacy_key_read_fails():
redis: Final = LegacyKeyReadFailureRedis()
redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] = [json.dumps(((_DAY, "llm", "/chat/completions", 500, 0, 2),))]
buffer: Final = _buffer(redis, leader=True)[0]
client: Final = FakePrismaClient()
asyncio.run(buffer.commit_if_leader(client))
assert _rows_written(client) == [(_DAY, "llm", "/chat/completions", 0, 2)]
assert _failure_rows_written(client) == [(_DAY, "llm", "/chat/completions", 500, 2)]
def test_pop_keeps_v2_first_batch_when_a_later_batch_read_fails(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr("litellm.proxy.db.gateway_request_tracking.MAX_REDIS_BUFFER_DEQUEUE_COUNT", 2)
redis: Final = FailureAfterFullV2BatchRedis()
redis.lists[REDIS_GATEWAY_REQUESTS_BUFFER_KEY_V2] = [
json.dumps(((_DAY, "llm", "/chat/completions", 500, 0, 2),)),
json.dumps(((_DAY, "llm", "/chat/completions", 429, 0, 3),)),
]
buffer: Final = _buffer(redis, leader=True)[0]
snapshot: Final = asyncio.run(buffer.pop())
assert snapshot == {
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=500): (
GatewayRequestCounts(successful_requests=0, failed_requests=2)
),
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=429): (
GatewayRequestCounts(successful_requests=0, failed_requests=3)
),
}
def test_failed_redis_push_keeps_counts_locally_for_the_next_flush():
@ -538,9 +808,12 @@ def test_failed_redis_push_keeps_counts_locally_for_the_next_flush():
assert lock.held == []
assert acc.drain() == {
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions"): (
GatewayRequestCounts(successful_requests=1, failed_requests=1)
)
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=200): (
GatewayRequestCounts(successful_requests=1, failed_requests=0)
),
GatewayRequestKey(date=_DAY, category="llm", route="/chat/completions", status_code=500): (
GatewayRequestCounts(successful_requests=0, failed_requests=1)
),
}

View file

@ -0,0 +1,372 @@
import asyncio
import json
from typing import Final
from unittest.mock import AsyncMock, MagicMock
import pytest
from litellm.constants import MAX_REDIS_BUFFER_DEQUEUE_COUNT, REQUEST_ERRORS_MAX_ROWS_PER_UPSERT
from litellm.proxy.db.request_error_tracking import (
RequestErrorAccumulator,
RequestErrorRedisBuffer,
build_request_errors_upsert,
commit_request_errors_to_db,
flush_request_errors,
fold_counts,
status_code_from_metadata,
)
from litellm.types.proxy.request_errors import RequestErrorKey
def _key(status_code: int = 429, api_key: str = "hash-1", date: str = "2026-10-08") -> RequestErrorKey:
return RequestErrorKey(
date=date, api_key=api_key, team_id="team-a", user_id="user-a", model_group="gpt-4o", status_code=status_code
)
def _payload(status: str = "failure", error_code: str | None = "429", **overrides: object) -> dict[str, object]:
error_information: Final = (
{} if error_code is None else {"error_information": {"error_code": error_code, "error_class": "RateLimitError"}}
)
metadata: Final[dict[str, object]] = {"status": status, **error_information}
return {
"api_key": "hash-1",
"team_id": "team-a",
"user": "user-a",
"model_group": "gpt-4o",
"model": "gpt-4o-2024-08-06",
"metadata": json.dumps(metadata),
**overrides,
}
@pytest.mark.parametrize(
("metadata", "expected"),
[
({"error_information": {"error_code": "429"}}, 429),
({"error_information": {"error_code": 503}}, 503),
({"error_information": {"error_code": ""}}, 0),
({"error_information": {"error_code": "RateLimitError"}}, 0),
({"error_information": {"error_code": "9999"}}, 0),
({"error_information": None}, 0),
({}, 0),
],
)
def test_status_code_from_metadata(metadata: dict[str, object], expected: int) -> None:
assert status_code_from_metadata(metadata) == expected
def test_record_counts_failures_by_caller_and_status() -> None:
accumulator: Final = RequestErrorAccumulator()
for _ in range(2):
accumulator.record(payload=_payload(), request_status="failure", date="2026-10-08", is_internal_call=False)
accumulator.record(
payload=_payload(error_code="500"), request_status="failure", date="2026-10-08", is_internal_call=False
)
assert accumulator.drain() == {_key(429): 2, _key(500): 1}
assert accumulator.drain() == {}
def test_record_skips_successes_and_internal_sub_calls() -> None:
accumulator: Final = RequestErrorAccumulator()
accumulator.record(
payload=_payload(status="success"), request_status="success", date="2026-10-08", is_internal_call=False
)
accumulator.record(payload=_payload(), request_status="failure", date="2026-10-08", is_internal_call=True)
assert accumulator.drain() == {}
def test_record_normalizes_missing_dimensions_to_empty_text() -> None:
accumulator: Final = RequestErrorAccumulator()
accumulator.record(
payload=_payload(error_code=None, team_id=None, user=None, model_group=None, model=None),
request_status="failure",
date="2026-10-08",
is_internal_call=False,
)
assert accumulator.drain() == {
RequestErrorKey(date="2026-10-08", api_key="hash-1", team_id="", user_id="", model_group="", status_code=0): 1
}
def test_record_falls_back_to_model_when_model_group_is_missing() -> None:
accumulator: Final = RequestErrorAccumulator()
accumulator.record(
payload=_payload(model_group=None), request_status="failure", date="2026-10-08", is_internal_call=False
)
(key,) = accumulator.drain()
assert key.model_group == "gpt-4o-2024-08-06"
def test_record_tolerates_unparseable_metadata() -> None:
accumulator: Final = RequestErrorAccumulator()
accumulator.record(
payload=_payload(metadata="{not json"), request_status="failure", date="2026-10-08", is_internal_call=False
)
assert accumulator.drain() == {_key(0): 1}
def test_restore_folds_into_pending_counts() -> None:
accumulator: Final = RequestErrorAccumulator()
accumulator.record(payload=_payload(), request_status="failure", date="2026-10-08", is_internal_call=False)
accumulator.restore({_key(429): 3, _key(500): 1})
assert accumulator.drain() == {_key(429): 4, _key(500): 1}
def test_fold_counts_sums_duplicate_keys() -> None:
assert fold_counts([(_key(), 1), (_key(), 2), (_key(500), 5)]) == {_key(): 3, _key(500): 5}
def test_upsert_orders_rows_by_conflict_key_and_increments_on_conflict() -> None:
sql, params = build_request_errors_upsert({_key(500): 1, _key(429, api_key="hash-0"): 2, _key(429): 0})
assert params == (
"2026-10-08", "hash-0", "team-a", "user-a", "gpt-4o", 429, 2,
"2026-10-08", "hash-1", "team-a", "user-a", "gpt-4o", 500, 1,
) # fmt: skip
assert sql.count("($") == 2
assert 'ON CONFLICT ("date", "api_key", "team_id", "user_id", "model_group", "status_code")' in sql
assert '"failed_requests" = "LiteLLM_DailyRequestErrors"."failed_requests" + EXCLUDED."failed_requests"' in sql
@pytest.mark.asyncio
async def test_flush_commits_one_statement_and_drains() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock()
accumulator: Final = RequestErrorAccumulator()
accumulator.record(payload=_payload(), request_status="failure", date="2026-10-08", is_internal_call=False)
await flush_request_errors(prisma_client, accumulator)
prisma_client.db.execute_raw.assert_awaited_once()
assert prisma_client.db.execute_raw.await_args.args[1:] == (
"2026-10-08", "hash-1", "team-a", "user-a", "gpt-4o", 429, 1,
) # fmt: skip
assert accumulator.drain() == {}
@pytest.mark.asyncio
async def test_flush_restores_counts_when_commit_fails() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock(side_effect=RuntimeError("db down"))
accumulator: Final = RequestErrorAccumulator()
accumulator.record(payload=_payload(), request_status="failure", date="2026-10-08", is_internal_call=False)
await flush_request_errors(prisma_client, accumulator)
assert accumulator.drain() == {_key(429): 1}
@pytest.mark.asyncio
async def test_commit_splits_a_large_backlog_into_bounded_statements() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock()
snapshot: Final = {_key(api_key=f"hash-{index:05d}"): 1 for index in range(REQUEST_ERRORS_MAX_ROWS_PER_UPSERT + 1)}
uncommitted: Final = await commit_request_errors_to_db(prisma_client=prisma_client, snapshot=snapshot)
assert uncommitted == {}
assert prisma_client.db.execute_raw.await_count == 2
first, second = prisma_client.db.execute_raw.await_args_list
assert len(first.args) - 1 == REQUEST_ERRORS_MAX_ROWS_PER_UPSERT * 7
assert len(second.args) - 1 == 7
@pytest.mark.asyncio
async def test_commit_hands_back_only_the_chunks_that_did_not_land() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock(side_effect=(None, RuntimeError("db down")))
snapshot: Final = {_key(api_key=f"hash-{index:05d}"): 1 for index in range(REQUEST_ERRORS_MAX_ROWS_PER_UPSERT + 1)}
uncommitted: Final = await commit_request_errors_to_db(prisma_client=prisma_client, snapshot=snapshot)
assert uncommitted == {_key(api_key=f"hash-{REQUEST_ERRORS_MAX_ROWS_PER_UPSERT:05d}"): 1}
assert prisma_client.db.execute_raw.await_count == 2
@pytest.mark.asyncio
async def test_commit_stops_at_the_first_failed_chunk_and_hands_back_the_rest() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock(side_effect=RuntimeError("db down"))
snapshot: Final = {_key(api_key=f"hash-{index:05d}"): 1 for index in range(REQUEST_ERRORS_MAX_ROWS_PER_UPSERT + 1)}
uncommitted: Final = await commit_request_errors_to_db(prisma_client=prisma_client, snapshot=snapshot)
assert uncommitted == snapshot
prisma_client.db.execute_raw.assert_awaited_once()
@pytest.mark.asyncio
async def test_cancelled_commit_restores_only_the_unwritten_chunks_and_propagates() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock(side_effect=(None, asyncio.CancelledError()))
accumulator: Final = RequestErrorAccumulator()
snapshot: Final = {_key(api_key=f"hash-{index:05d}"): 1 for index in range(REQUEST_ERRORS_MAX_ROWS_PER_UPSERT + 1)}
accumulator.restore(snapshot)
with pytest.raises(asyncio.CancelledError):
await flush_request_errors(prisma_client, accumulator)
assert accumulator.drain() == {_key(api_key=f"hash-{REQUEST_ERRORS_MAX_ROWS_PER_UPSERT:05d}"): 1}
assert prisma_client.db.execute_raw.await_count == 2
@pytest.mark.asyncio
async def test_cancelled_leader_commit_requeues_the_unwritten_chunks_to_redis() -> None:
redis_cache: Final = MagicMock()
stored: Final[list[str]] = [] # mutable-ok: stands in for the Redis list
async def rpush(key: str, values: tuple[str, ...]) -> int:
stored.extend(values)
return len(stored)
async def lpop(key: str, count: int) -> list[str]:
popped: Final = stored[:count]
del stored[:count]
return popped
redis_cache.async_rpush = AsyncMock(side_effect=rpush)
redis_cache.async_lpop = AsyncMock(side_effect=lpop)
pod_lock_manager: Final = MagicMock()
pod_lock_manager.acquire_lock = AsyncMock(return_value=True)
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock(side_effect=(None, asyncio.CancelledError()))
buffer: Final = RequestErrorRedisBuffer(redis_cache=redis_cache, pod_lock_manager=pod_lock_manager)
accumulator: Final = RequestErrorAccumulator()
accumulator.restore(
{_key(api_key=f"hash-{index:05d}"): 1 for index in range(REQUEST_ERRORS_MAX_ROWS_PER_UPSERT + 1)}
)
with pytest.raises(asyncio.CancelledError):
await flush_request_errors(prisma_client, accumulator, buffer)
assert accumulator.drain() == {}
assert await buffer.pop() == {_key(api_key=f"hash-{REQUEST_ERRORS_MAX_ROWS_PER_UPSERT:05d}"): 1}
assert prisma_client.db.execute_raw.await_count == 2
@pytest.mark.asyncio
async def test_cancelled_leader_commit_keeps_rows_in_memory_when_redis_refuses_them() -> None:
redis_cache: Final = MagicMock()
stored: Final[list[str]] = [] # mutable-ok: stands in for the Redis list
async def rpush_once(key: str, values: tuple[str, ...]) -> int:
if stored:
raise RuntimeError("redis down")
stored.extend(values)
return len(stored)
async def lpop(key: str, count: int) -> list[str]:
return list(stored)
redis_cache.async_rpush = AsyncMock(side_effect=rpush_once)
redis_cache.async_lpop = AsyncMock(side_effect=lpop)
pod_lock_manager: Final = MagicMock()
pod_lock_manager.acquire_lock = AsyncMock(return_value=True)
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock(side_effect=asyncio.CancelledError())
buffer: Final = RequestErrorRedisBuffer(redis_cache=redis_cache, pod_lock_manager=pod_lock_manager)
accumulator: Final = RequestErrorAccumulator()
accumulator.restore({_key(429): 2})
with pytest.raises(asyncio.CancelledError):
await flush_request_errors(prisma_client, accumulator, buffer)
assert accumulator.drain() == {_key(429): 2}
@pytest.mark.asyncio
async def test_cancelled_drain_puts_the_popped_entries_back_in_redis() -> None:
redis_cache: Final = MagicMock()
entry: Final = json.dumps([["2026-10-08", "hash-1", "team-a", "user-a", "gpt-4o", 429, 1]])
stored: Final[list[str]] = [entry] * MAX_REDIS_BUFFER_DEQUEUE_COUNT # mutable-ok: stands in for the Redis list
async def lpop(key: str, count: int) -> list[str]:
if not stored:
raise asyncio.CancelledError()
popped: Final = stored[:count]
del stored[:count]
return popped
async def rpush(key: str, values: tuple[str, ...]) -> int:
stored.extend(values)
return len(stored)
redis_cache.async_lpop = AsyncMock(side_effect=lpop)
redis_cache.async_rpush = AsyncMock(side_effect=rpush)
pod_lock_manager: Final = MagicMock()
pod_lock_manager.acquire_lock = AsyncMock(return_value=True)
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock()
buffer: Final = RequestErrorRedisBuffer(redis_cache=redis_cache, pod_lock_manager=pod_lock_manager)
accumulator: Final = RequestErrorAccumulator()
with pytest.raises(asyncio.CancelledError):
await flush_request_errors(prisma_client, accumulator, buffer)
assert len(stored) == MAX_REDIS_BUFFER_DEQUEUE_COUNT
prisma_client.db.execute_raw.assert_not_awaited()
assert accumulator.drain() == {}
@pytest.mark.asyncio
async def test_flush_skips_the_database_when_nothing_failed() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock()
await flush_request_errors(prisma_client, RequestErrorAccumulator())
prisma_client.db.execute_raw.assert_not_awaited()
@pytest.mark.asyncio
async def test_redis_buffer_round_trips_and_leader_commits() -> None:
redis_cache: Final = MagicMock()
stored: Final[list[str]] = [] # mutable-ok: stands in for the Redis list
async def rpush(key: str, values: tuple[str, ...]) -> int:
stored.extend(values)
return len(stored)
async def lpop(key: str, count: int) -> list[str]:
popped: Final = stored[:count]
del stored[:count]
return popped
redis_cache.async_rpush = AsyncMock(side_effect=rpush)
redis_cache.async_lpop = AsyncMock(side_effect=lpop)
pod_lock_manager: Final = MagicMock()
pod_lock_manager.acquire_lock = AsyncMock(return_value=True)
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock()
buffer: Final = RequestErrorRedisBuffer(redis_cache=redis_cache, pod_lock_manager=pod_lock_manager)
accumulator: Final = RequestErrorAccumulator()
accumulator.record(payload=_payload(), request_status="failure", date="2026-10-08", is_internal_call=False)
await buffer.push({_key(429): 2})
await flush_request_errors(prisma_client, accumulator, buffer)
prisma_client.db.execute_raw.assert_awaited_once()
assert prisma_client.db.execute_raw.await_args.args[1:] == (
"2026-10-08", "hash-1", "team-a", "user-a", "gpt-4o", 429, 3,
) # fmt: skip
assert stored == []
assert accumulator.drain() == {}
@pytest.mark.asyncio
async def test_redis_buffer_follower_leaves_rows_in_redis() -> None:
redis_cache: Final = MagicMock()
redis_cache.async_rpush = AsyncMock()
redis_cache.async_lpop = AsyncMock()
pod_lock_manager: Final = MagicMock()
pod_lock_manager.acquire_lock = AsyncMock(return_value=False)
prisma_client: Final = MagicMock()
prisma_client.db.execute_raw = AsyncMock()
buffer: Final = RequestErrorRedisBuffer(redis_cache=redis_cache, pod_lock_manager=pod_lock_manager)
accumulator: Final = RequestErrorAccumulator()
accumulator.record(payload=_payload(), request_status="failure", date="2026-10-08", is_internal_call=False)
await flush_request_errors(prisma_client, accumulator, buffer)
redis_cache.async_rpush.assert_awaited_once()
redis_cache.async_lpop.assert_not_awaited()
prisma_client.db.execute_raw.assert_not_awaited()
assert accumulator.drain() == {}

View file

@ -1,6 +1,8 @@
import os
from collections.abc import Mapping, Sequence
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock, patch
from typing import Final
from unittest.mock import AsyncMock, MagicMock, call, patch
from tests._master_key import MASTER_KEY
@ -24,10 +26,13 @@ from fastapi.testclient import TestClient
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management_endpoints.gateway_request_endpoints import (
_ACTIVITY_SQL,
_AggregateRow,
_default_range,
_fold_by_date,
_fold_by_route,
_fold_by_status_code,
_StatusCodeAggregateRow,
get_gateway_daily_activity,
router,
)
@ -64,6 +69,7 @@ def _row(
failed: int = 0,
) -> _AggregateRow:
return _AggregateRow(
kind="route",
date=date,
category=category,
route=route,
@ -76,10 +82,16 @@ def _admin() -> UserAPIKeyAuth:
return UserAPIKeyAuth(api_key="sk-test", user_role=LitellmUserRoles.PROXY_ADMIN)
def _prisma_returning(rows: list) -> MagicMock:
client = MagicMock()
def _prisma_returning(rows: Sequence[Mapping[str, str | int | None]] | None) -> MagicMock:
client: Final = MagicMock()
client.db = MagicMock()
client.db.query_raw = AsyncMock(return_value=rows)
async def query_raw(query: str, *_args: str) -> Sequence[Mapping[str, str | int | None]] | None:
if query != _ACTIVITY_SQL:
raise AssertionError(f"Unexpected gateway activity query: {query}")
return rows
client.db.query_raw = AsyncMock(side_effect=query_raw)
return client
@ -155,6 +167,18 @@ class TestFoldByRoute:
assert [entry.route for entry in _fold_by_route(tuple(reversed(rows)))] == expected
class TestFoldByStatusCode:
def test_orders_by_failed_count_then_status_code(self):
folded: Final = _fold_by_status_code(
(
_StatusCodeAggregateRow(kind="status", status_code=503, failed_requests=1),
_StatusCodeAggregateRow(kind="status", status_code=500, failed_requests=2),
_StatusCodeAggregateRow(kind="status", status_code=429, failed_requests=2),
)
)
assert [(entry.status_code, entry.failed_requests) for entry in folded] == [(429, 2), (500, 2), (503, 1)]
class TestGatewayDailyActivityEndpoint:
@pytest.mark.asyncio
@pytest.mark.parametrize(
@ -194,39 +218,66 @@ class TestGatewayDailyActivityEndpoint:
assert exc.value.status_code == 500
@pytest.mark.asyncio
async def test_totals_and_breakdowns_come_from_the_same_rows(self):
rows = [
async def test_mixed_route_and_status_rows_fold_into_totals_and_breakdowns(self):
rows: Final = (
{
"kind": "route",
"date": "2026-08-03",
"category": "llm",
"route": "/chat/completions",
"successful_requests": 5,
"failed_requests": 1,
"status_code": None,
},
{
"kind": "route",
"date": "2026-08-04",
"category": "llm",
"route": "/chat/completions",
"successful_requests": 7,
"failed_requests": 3,
"status_code": None,
},
{
"kind": "route",
"date": "2026-08-04",
"category": "llm",
"route": "/embeddings",
"successful_requests": 4,
"failed_requests": 0,
"status_code": None,
},
]
with patch("litellm.proxy.proxy_server.prisma_client", _prisma_returning(rows)):
response = await get_gateway_daily_activity(user_api_key_dict=_admin())
{
"kind": "status",
"date": None,
"category": None,
"route": None,
"successful_requests": 0,
"failed_requests": 3,
"status_code": 429,
},
{
"kind": "status",
"date": None,
"category": None,
"route": None,
"successful_requests": 0,
"failed_requests": 1,
"status_code": 500,
},
)
prisma: Final = _prisma_returning(rows)
with patch("litellm.proxy.proxy_server.prisma_client", prisma):
response: Final = await get_gateway_daily_activity(user_api_key_dict=_admin())
assert response.total_successful_requests == 16
assert response.total_failed_requests == 4
assert sum(entry.successful_requests for entry in response.by_date) == 16
assert sum(entry.successful_requests for entry in response.by_route) == 16
assert [(entry.status_code, entry.failed_requests) for entry in response.by_status_code] == [(429, 3), (500, 1)]
assert [entry.date for entry in response.by_date] == ["2026-08-03", "2026-08-04"]
assert [entry.route for entry in response.by_route] == ["/chat/completions", "/embeddings"]
assert prisma.db.query_raw.call_count == 1
@pytest.mark.asyncio
async def test_a_null_result_set_is_not_an_error(self):
@ -236,6 +287,8 @@ class TestGatewayDailyActivityEndpoint:
assert response.total_successful_requests == 0
assert response.by_date == ()
assert response.by_route == ()
assert response.by_status_code == ()
class TestGatewayDailyActivityRoute:
"""
@ -258,8 +311,7 @@ class TestGatewayDailyActivityRoute:
params={"start_date": "2026-01-01", "end_date": "2026-01-31"},
)
assert response.status_code == 200
_, start, end = prisma.db.query_raw.call_args.args
assert (start, end) == ("2026-01-01", "2026-01-31")
assert prisma.db.query_raw.call_args_list == [call(_ACTIVITY_SQL, "2026-01-01", "2026-01-31")]
def test_omitted_dates_fall_back_to_the_default_window(self, frozen_clock):
prisma = _prisma_returning([])
@ -269,26 +321,36 @@ class TestGatewayDailyActivityRoute:
with patch("litellm.proxy.proxy_server.prisma_client", prisma):
response = TestClient(app).get("/gateway/daily/activity")
assert response.status_code == 200
_, start, end = prisma.db.query_raw.call_args.args
assert (start, end) == _FROZEN_RANGE
assert prisma.db.query_raw.call_args_list == [call(_ACTIVITY_SQL, *_FROZEN_RANGE)]
def test_serialized_response_carries_the_documented_shape(self):
prisma = _prisma_returning(
prisma: Final = _prisma_returning(
[
{
"kind": "route",
"date": "2026-08-04",
"category": "llm",
"route": "/chat/completions",
"successful_requests": 7,
"failed_requests": 3,
}
"status_code": None,
},
{
"kind": "status",
"date": None,
"category": None,
"route": None,
"successful_requests": 0,
"failed_requests": 3,
"status_code": 500,
},
]
)
app = FastAPI()
app: Final = FastAPI()
app.include_router(router)
app.dependency_overrides[user_api_key_auth] = _admin
with patch("litellm.proxy.proxy_server.prisma_client", prisma):
body = TestClient(app).get("/gateway/daily/activity").json()
body: Final = TestClient(app).get("/gateway/daily/activity").json()
assert body == {
"total_successful_requests": 7,
@ -302,4 +364,5 @@ class TestGatewayDailyActivityRoute:
"failed_requests": 3,
}
],
"by_status_code": [{"status_code": 500, "failed_requests": 3}],
}

View file

@ -0,0 +1,147 @@
import os
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
from tests._master_key import MASTER_KEY
_THROWAWAY_ENV: Final = {"DATABASE_URL": "sqlite:///:memory:", "LITELLM_MASTER_KEY": MASTER_KEY}
_PRE_EXISTING_ENV: Final = {key: os.environ.get(key) for key in _THROWAWAY_ENV}
for _key, _value in _THROWAWAY_ENV.items():
os.environ.setdefault(_key, _value)
import pytest
from fastapi import HTTPException
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.request_error_endpoints import (
_DateRow,
_EntityRow,
_EntityStatusRow,
fold_by_date,
fold_by_status_code,
fold_entities,
get_request_error_activity,
)
for _key, _previous in _PRE_EXISTING_ENV.items():
if _previous is None:
os.environ.pop(_key, None)
else:
os.environ[_key] = _previous # test-quality-ok: import-time restore of the pre-existing value
_DATE_ROWS: Final = (
_DateRow(date="2026-10-07", successful_requests=90, failed_requests=10),
_DateRow(date="2026-10-07", status_code=429, failed_requests=7),
_DateRow(date="2026-10-07", status_code=500, failed_requests=2),
_DateRow(date="2026-10-08", successful_requests=50, failed_requests=5),
_DateRow(date="2026-10-08", status_code=429, failed_requests=5),
)
def test_fold_by_date_splits_failures_into_client_and_server_errors() -> None:
by_date: Final = fold_by_date(_DATE_ROWS)
assert [entry.model_dump(exclude={"by_status_code"}) for entry in by_date] == [
{
"date": "2026-10-07",
"successful_requests": 90,
"failed_requests": 10,
"client_errors": 7,
"server_errors": 2,
},
{"date": "2026-10-08", "successful_requests": 50, "failed_requests": 5, "client_errors": 5, "server_errors": 0},
]
assert [entry.model_dump() for entry in by_date[0].by_status_code] == [
{"status_code": 429, "failed_requests": 7},
{"status_code": 500, "failed_requests": 2},
]
assert [entry.model_dump() for entry in by_date[1].by_status_code] == [{"status_code": 429, "failed_requests": 5}]
def test_fold_by_status_code_sums_across_days_and_ranks_by_count() -> None:
assert [entry.model_dump() for entry in fold_by_status_code(_DATE_ROWS)] == [
{"status_code": 429, "failed_requests": 12},
{"status_code": 500, "failed_requests": 2},
]
def test_fold_entities_attaches_top_status_and_keeps_kinds_apart() -> None:
rows: Final = (
_EntityRow(kind="key", id="hash-1", label="prod-key", api_requests=100, failed_requests=10),
_EntityRow(kind="key", id="hash-2", label=None, api_requests=40, failed_requests=12),
_EntityRow(kind="team", id="team-a", label="Team A", api_requests=140, failed_requests=22),
)
status_rows: Final = (
_EntityStatusRow(kind="key", id="hash-1", status_code=429, failed_requests=6),
_EntityStatusRow(kind="team", id="hash-1", status_code=401, failed_requests=99),
)
by_key: Final = fold_entities("key", rows, status_rows)
assert [entry.model_dump() for entry in by_key] == [
{
"id": "hash-2",
"label": None,
"api_requests": 40,
"failed_requests": 12,
"top_status_code": None,
"top_status_code_requests": 0,
},
{
"id": "hash-1",
"label": "prod-key",
"api_requests": 100,
"failed_requests": 10,
"top_status_code": 429,
"top_status_code_requests": 6,
},
]
assert [entry.id for entry in fold_entities("team", rows, status_rows)] == ["team-a"]
assert fold_entities("user", rows, status_rows) == ()
def _auth(role: LitellmUserRoles) -> UserAPIKeyAuth:
return UserAPIKeyAuth(api_key="sk-test", user_role=role)
@pytest.mark.asyncio
@pytest.mark.parametrize("role", [LitellmUserRoles.INTERNAL_USER, LitellmUserRoles.TEAM])
async def test_non_admin_roles_are_refused(role: LitellmUserRoles) -> None:
prisma_client: Final = MagicMock()
prisma_client.db.query_raw = AsyncMock()
with patch("litellm.proxy.proxy_server.prisma_client", prisma_client):
with pytest.raises(HTTPException) as refused:
await get_request_error_activity(user_api_key_dict=_auth(role))
assert refused.value.status_code == 403
prisma_client.db.query_raw.assert_not_awaited()
@pytest.mark.asyncio
async def test_missing_database_is_a_500() -> None:
with patch("litellm.proxy.proxy_server.prisma_client", None):
with pytest.raises(HTTPException) as refused:
await get_request_error_activity(user_api_key_dict=_auth(LitellmUserRoles.PROXY_ADMIN))
assert refused.value.status_code == 500
@pytest.mark.asyncio
async def test_admin_gets_totals_series_and_caller_breakdown() -> None:
prisma_client: Final = MagicMock()
prisma_client.db.query_raw = AsyncMock(
side_effect=[
[row.model_dump() for row in _DATE_ROWS],
[{"kind": "team", "id": "team-a", "label": "Team A", "api_requests": 140, "failed_requests": 15}],
[{"kind": "team", "id": "team-a", "status_code": 429, "failed_requests": 12}],
]
)
with patch("litellm.proxy.proxy_server.prisma_client", prisma_client):
response: Final = await get_request_error_activity(
user_api_key_dict=_auth(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY),
start_date="2026-10-01",
end_date="2026-10-08",
)
assert response.total_successful_requests == 140
assert response.total_failed_requests == 15
assert [entry.date for entry in response.by_date] == ["2026-10-07", "2026-10-08"]
assert response.by_status_code[0].status_code == 429
assert response.by_team[0].top_status_code == 429
assert response.by_key == ()
for awaited in prisma_client.db.query_raw.await_args_list:
assert awaited.args[1:] == ("2026-10-01", "2026-10-08")

View file

@ -37,9 +37,7 @@ class FakeRecorder:
self.calls: List[dict] = []
def record(self, *, category: BillableCategory, route: str, status_code: int, model_id: Optional[str]) -> None:
self.calls.append(
{"category": category, "route": route, "status_code": status_code, "model_id": model_id}
)
self.calls.append({"category": category, "route": route, "status_code": status_code, "model_id": model_id})
def _make_app(recorder: Optional[FakeRecorder], status_code: int = 200, model_id: Optional[str] = None) -> Starlette:
@ -515,9 +513,9 @@ def test_varying_model_ids_fold_into_a_single_persisted_key():
snapshot: Final = accumulator.drain()
assert snapshot == {
GatewayRequestKey(date=frozen_now.strftime("%Y-%m-%d"), category="llm", route="/chat/completions"): (
GatewayRequestCounts(successful_requests=3, failed_requests=0)
)
GatewayRequestKey(
date=frozen_now.strftime("%Y-%m-%d"), category="llm", route="/chat/completions", status_code=200
): (GatewayRequestCounts(successful_requests=3, failed_requests=0))
}

View file

@ -5589,6 +5589,27 @@ export interface paths {
patch?: never;
trace?: never;
};
"/gateway/errors/activity": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/**
* Get Request Error Activity
* @description Failed requests over time by HTTP status, and the keys, teams, users and
* model groups they land on. Deployment-wide, so admin-only.
*/
get: operations["get_request_error_activity_gateway_errors_activity_get"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/gemini/{endpoint}": {
parameters: {
query?: never;
@ -33067,6 +33088,11 @@ export interface components {
* @default []
*/
by_route: components["schemas"]["GatewayRequestBreakdownEntry"][];
/**
* By Status Code
* @default []
*/
by_status_code: components["schemas"]["GatewayRequestStatusCodeEntry"][];
/**
* Total Failed Requests
* @default 0
@ -33110,6 +33136,16 @@ export interface components {
*/
successful_requests: number;
};
/** GatewayRequestStatusCodeEntry */
GatewayRequestStatusCodeEntry: {
/**
* Failed Requests
* @default 0
*/
failed_requests: number;
/** Status Code */
status_code: number;
};
/** GenerateKeyRequest */
GenerateKeyRequest: {
/** Access Group Ids */
@ -43213,6 +43249,116 @@ export interface components {
} & {
[key: string]: unknown;
};
/**
* RequestErrorActivityResponse
* @description Response for GET /gateway/errors/activity.
*/
RequestErrorActivityResponse: {
/**
* By Date
* @default []
*/
by_date: components["schemas"]["RequestErrorDailyEntry"][];
/**
* By Key
* @default []
*/
by_key: components["schemas"]["RequestErrorEntityEntry"][];
/**
* By Model
* @default []
*/
by_model: components["schemas"]["RequestErrorEntityEntry"][];
/**
* By Status Code
* @default []
*/
by_status_code: components["schemas"]["RequestErrorStatusCodeEntry"][];
/**
* By Team
* @default []
*/
by_team: components["schemas"]["RequestErrorEntityEntry"][];
/**
* By User
* @default []
*/
by_user: components["schemas"]["RequestErrorEntityEntry"][];
/**
* Total Failed Requests
* @default 0
*/
total_failed_requests: number;
/**
* Total Successful Requests
* @default 0
*/
total_successful_requests: number;
};
/** RequestErrorDailyEntry */
RequestErrorDailyEntry: {
/**
* By Status Code
* @default []
*/
by_status_code: components["schemas"]["RequestErrorStatusCodeEntry"][];
/**
* Client Errors
* @default 0
*/
client_errors: number;
/** Date */
date: string;
/**
* Failed Requests
* @default 0
*/
failed_requests: number;
/**
* Server Errors
* @default 0
*/
server_errors: number;
/**
* Successful Requests
* @default 0
*/
successful_requests: number;
};
/** RequestErrorEntityEntry */
RequestErrorEntityEntry: {
/**
* Api Requests
* @default 0
*/
api_requests: number;
/**
* Failed Requests
* @default 0
*/
failed_requests: number;
/** Id */
id: string;
/** Label */
label?: string | null;
/** Top Status Code */
top_status_code?: number | null;
/**
* Top Status Code Requests
* @default 0
*/
top_status_code_requests: number;
};
/** RequestErrorStatusCodeEntry */
RequestErrorStatusCodeEntry: {
/**
* Failed Requests
* @default 0
*/
failed_requests: number;
/** Status Code */
status_code: number;
};
/**
* RequestType
* @description Fixed v0 taxonomy. User-extensible types come in v1.
@ -60027,6 +60173,40 @@ export interface operations {
};
};
};
get_request_error_activity_gateway_errors_activity_get: {
parameters: {
query?: {
/** @description Start date in YYYY-MM-DD format */
start_date?: string | null;
/** @description End date in YYYY-MM-DD format */
end_date?: string | null;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["RequestErrorActivityResponse"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
gemini_proxy_route_gemini__endpoint__get: {
parameters: {
query?: never;