From 2df5b0730845d50a51eb4a40f51b1d0bb6c88e61 Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Fri, 9 Oct 2026 14:27:56 -0700 Subject: [PATCH] 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> --- backend/routes/allowlist.py | 3 +- .../migration.sql | 15 + .../migration.sql | 17 + .../litellm_proxy_extras/schema.prisma | 28 ++ litellm/constants.py | 5 + litellm/integrations/otel/model/spans.py | 2 + litellm/proxy/db/db_spend_update_writer.py | 24 ++ litellm/proxy/db/gateway_request_tracking.py | 246 +++++++++--- litellm/proxy/db/request_error_tracking.py | 350 ++++++++++++++++ .../gateway_request_endpoints.py | 61 ++- .../request_error_endpoints.py | 273 +++++++++++++ litellm/proxy/proxy_server.py | 29 ++ litellm/proxy/schema.prisma | 28 ++ litellm/types/proxy/gateway_requests.py | 11 + litellm/types/proxy/request_errors.py | 66 ++++ schema.prisma | 28 ++ .../proxy/db/test_db_spend_update_writer.py | 32 ++ .../proxy/db/test_gateway_request_tracking.py | 319 +++++++++++++-- .../proxy/db/test_request_error_tracking.py | 372 ++++++++++++++++++ .../test_gateway_request_endpoints.py | 97 ++++- .../test_request_error_endpoints.py | 147 +++++++ ...est_billable_request_metrics_middleware.py | 10 +- ui/litellm-dashboard/src/lib/http/schema.d.ts | 180 +++++++++ 23 files changed, 2218 insertions(+), 125 deletions(-) create mode 100644 litellm-proxy-extras/litellm_proxy_extras/migrations/20261008010000_add_daily_gateway_failed_requests/migration.sql create mode 100644 litellm-proxy-extras/litellm_proxy_extras/migrations/20261008120000_add_daily_request_errors/migration.sql create mode 100644 litellm/proxy/db/request_error_tracking.py create mode 100644 litellm/proxy/management_endpoints/request_error_endpoints.py create mode 100644 litellm/types/proxy/request_errors.py create mode 100644 tests/unit/proxy/db/test_request_error_tracking.py create mode 100644 tests/unit/proxy/management_endpoints/test_request_error_endpoints.py diff --git a/backend/routes/allowlist.py b/backend/routes/allowlist.py index 44541276be0..83e2221e866 100644 --- a/backend/routes/allowlist.py +++ b/backend/routes/allowlist.py @@ -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 diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261008010000_add_daily_gateway_failed_requests/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261008010000_add_daily_gateway_failed_requests/migration.sql new file mode 100644 index 00000000000..7933761526b --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261008010000_add_daily_gateway_failed_requests/migration.sql @@ -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"); diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261008120000_add_daily_request_errors/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261008120000_add_daily_request_errors/migration.sql new file mode 100644 index 00000000000..a1037334d7b --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261008120000_add_daily_request_errors/migration.sql @@ -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"); diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index 59ffb037177..35d026bb6f6 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -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()) diff --git a/litellm/constants.py b/litellm/constants.py index 290749e562f..aee101dabb5 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -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" diff --git a/litellm/integrations/otel/model/spans.py b/litellm/integrations/otel/model/spans.py index 762efa46ebb..627466ba565 100644 --- a/litellm/integrations/otel/model/spans.py +++ b/litellm/integrations/otel/model/spans.py @@ -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", diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py index 5bafe19cd0b..da8a7fb0f15 100644 --- a/litellm/proxy/db/db_spend_update_writer.py +++ b/litellm/proxy/db/db_spend_update_writer.py @@ -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, diff --git a/litellm/proxy/db/gateway_request_tracking.py b/litellm/proxy/db/gateway_request_tracking.py index 62f2b45a7c0..2f2fd6b004b 100644 --- a/litellm/proxy/db/gateway_request_tracking.py +++ b/litellm/proxy/db/gateway_request_tracking.py @@ -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 diff --git a/litellm/proxy/db/request_error_tracking.py b/litellm/proxy/db/request_error_tracking.py new file mode 100644 index 00000000000..bfc9a90226c --- /dev/null +++ b/litellm/proxy/db/request_error_tracking.py @@ -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, + ) diff --git a/litellm/proxy/management_endpoints/gateway_request_endpoints.py b/litellm/proxy/management_endpoints/gateway_request_endpoints.py index c7ae50df91c..4052e6f714e 100644 --- a/litellm/proxy/management_endpoints/gateway_request_endpoints.py +++ b/litellm/proxy/management_endpoints/gateway_request_endpoints.py @@ -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), ) diff --git a/litellm/proxy/management_endpoints/request_error_endpoints.py b/litellm/proxy/management_endpoints/request_error_endpoints.py new file mode 100644 index 00000000000..7b54542a1fc --- /dev/null +++ b/litellm/proxy/management_endpoints/request_error_endpoints.py @@ -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), + ) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 67fec38e39e..4eab87cc84e 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -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) diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index 59ffb037177..35d026bb6f6 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -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()) diff --git a/litellm/types/proxy/gateway_requests.py b/litellm/types/proxy/gateway_requests.py index 77607899dde..cec198210a1 100644 --- a/litellm/types/proxy/gateway_requests.py +++ b/litellm/types/proxy/gateway_requests.py @@ -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, ...] = () diff --git a/litellm/types/proxy/request_errors.py b/litellm/types/proxy/request_errors.py new file mode 100644 index 00000000000..d9a70eddb1a --- /dev/null +++ b/litellm/types/proxy/request_errors.py @@ -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, ...] = () diff --git a/schema.prisma b/schema.prisma index 59ffb037177..35d026bb6f6 100644 --- a/schema.prisma +++ b/schema.prisma @@ -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()) diff --git a/tests/unit/proxy/db/test_db_spend_update_writer.py b/tests/unit/proxy/db/test_db_spend_update_writer.py index 4a508fcd272..1ec65a4fdcd 100644 --- a/tests/unit/proxy/db/test_db_spend_update_writer.py +++ b/tests/unit/proxy/db/test_db_spend_update_writer.py @@ -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 + } diff --git a/tests/unit/proxy/db/test_gateway_request_tracking.py b/tests/unit/proxy/db/test_gateway_request_tracking.py index 5566a5cb9bc..1491f3f0664 100644 --- a/tests/unit/proxy/db/test_gateway_request_tracking.py +++ b/tests/unit/proxy/db/test_gateway_request_tracking.py @@ -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) + ), } diff --git a/tests/unit/proxy/db/test_request_error_tracking.py b/tests/unit/proxy/db/test_request_error_tracking.py new file mode 100644 index 00000000000..43700490a1f --- /dev/null +++ b/tests/unit/proxy/db/test_request_error_tracking.py @@ -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() == {} diff --git a/tests/unit/proxy/management_endpoints/test_gateway_request_endpoints.py b/tests/unit/proxy/management_endpoints/test_gateway_request_endpoints.py index 24323d06d78..7153fb909a8 100644 --- a/tests/unit/proxy/management_endpoints/test_gateway_request_endpoints.py +++ b/tests/unit/proxy/management_endpoints/test_gateway_request_endpoints.py @@ -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}], } diff --git a/tests/unit/proxy/management_endpoints/test_request_error_endpoints.py b/tests/unit/proxy/management_endpoints/test_request_error_endpoints.py new file mode 100644 index 00000000000..3999b3150dc --- /dev/null +++ b/tests/unit/proxy/management_endpoints/test_request_error_endpoints.py @@ -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") diff --git a/tests/unit/proxy/middleware/test_billable_request_metrics_middleware.py b/tests/unit/proxy/middleware/test_billable_request_metrics_middleware.py index f2515905707..8c81a2c9b36 100644 --- a/tests/unit/proxy/middleware/test_billable_request_metrics_middleware.py +++ b/tests/unit/proxy/middleware/test_billable_request_metrics_middleware.py @@ -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)) } diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index fa90213a486..a3fae7a2f4f 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -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;