From f82c79a6f05046c268b89c98fe2d18ba31b8a237 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Thu, 5 Nov 2026 11:41:02 +0800 Subject: [PATCH] feat: add scoped daily usage alert API and storage --- .../migration.sql | 24 + .../litellm_proxy_extras/schema.prisma | 40 ++ litellm/proxy/_types.py | 12 + .../proxy/alerting_endpoints/usage_alerts.py | 175 +++++++ litellm/proxy/proxy_server.py | 2 + litellm/proxy/schema.prisma | 40 ++ litellm/repositories/usage_alerts.py | 460 ++++++++++++++++++ litellm/types/proxy/usage_alerts.py | 116 +++++ schema.prisma | 40 ++ 9 files changed, 909 insertions(+) create mode 100644 litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_add_daily_usage_alerts/migration.sql create mode 100644 litellm/proxy/alerting_endpoints/usage_alerts.py create mode 100644 litellm/repositories/usage_alerts.py create mode 100644 litellm/types/proxy/usage_alerts.py diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_add_daily_usage_alerts/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_add_daily_usage_alerts/migration.sql new file mode 100644 index 00000000000..1e459468f84 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_add_daily_usage_alerts/migration.sql @@ -0,0 +1,24 @@ +CREATE TABLE "LiteLLM_UsageAlertRule" ( + "id" TEXT PRIMARY KEY, "scope_type" TEXT NOT NULL, "scope_id" TEXT, + "enabled" BOOLEAN NOT NULL DEFAULT true, "record" JSONB NOT NULL, + "deleted" BOOLEAN NOT NULL DEFAULT false +); +CREATE INDEX "LiteLLM_UsageAlertRule_enabled_deleted_id_idx" ON "LiteLLM_UsageAlertRule" ("enabled", "deleted", "id"); +CREATE INDEX "LiteLLM_UsageAlertRule_scope_type_scope_id_deleted_idx" ON "LiteLLM_UsageAlertRule" ("scope_type", "scope_id", "deleted"); +CREATE TABLE "LiteLLM_UsageAlertIncident" ( + "id" TEXT PRIMARY KEY, "rule_id" TEXT NOT NULL, "scope_key" TEXT NOT NULL, + "scope_type" TEXT NOT NULL, "scope_id" TEXT, "metric" TEXT NOT NULL, + "status" TEXT NOT NULL, "day" TEXT NOT NULL, "evaluation_day" TEXT NOT NULL, + "breached" BOOLEAN NOT NULL, "record" JSONB NOT NULL, + "notified_channels" TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[], + "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE INDEX "LiteLLM_UsageAlertIncident_rule_id_scope_key_evaluation_day_idx" ON "LiteLLM_UsageAlertIncident" ("rule_id", "scope_key", "evaluation_day"); +CREATE INDEX "LiteLLM_UsageAlertIncident_scope_type_scope_id_day_idx" ON "LiteLLM_UsageAlertIncident" ("scope_type", "scope_id", "day"); +CREATE INDEX "LiteLLM_UsageAlertIncident_status_day_idx" ON "LiteLLM_UsageAlertIncident" ("status", "day"); +CREATE INDEX "LiteLLM_UsageAlertIncident_updated_at_idx" ON "LiteLLM_UsageAlertIncident" ("updated_at"); +CREATE TABLE "LiteLLM_UsageAlertCheckpoint" ( + "id" TEXT PRIMARY KEY, "completed" BOOLEAN NOT NULL DEFAULT false, + "owner" TEXT, "expires_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE INDEX "LiteLLM_UsageAlertCheckpoint_expires_at_idx" ON "LiteLLM_UsageAlertCheckpoint" ("expires_at"); diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index 78f16d9f1fd..46ab28530a9 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -2109,3 +2109,43 @@ model LiteLLM_SubscriptionAccountTable { @@unique([custom_llm_provider, account_id], map: "LiteLLM_SubscriptionAccountTable_provider_account_key") } + +// Daily alerts consume existing aggregate tables; no request-path writes. +model LiteLLM_UsageAlertRule { + id String @id + scope_type String + scope_id String? + enabled Boolean @default(true) + record Json + deleted Boolean @default(false) + @@index([enabled, deleted, id]) + @@index([scope_type, scope_id, deleted]) +} + +model LiteLLM_UsageAlertIncident { + id String @id + rule_id String + scope_key String + scope_type String + scope_id String? + metric String + status String + day String + evaluation_day String + breached Boolean + record Json + notified_channels String[] @default([]) + updated_at DateTime @default(now()) + @@index([rule_id, scope_key, evaluation_day]) + @@index([scope_type, scope_id, day]) + @@index([status, day]) + @@index([updated_at]) +} + +model LiteLLM_UsageAlertCheckpoint { + id String @id + completed Boolean @default(false) + owner String? + expires_at DateTime @default(now()) + @@index([expires_at]) +} diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 25849634764..14e71a0f292 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -646,6 +646,10 @@ class LiteLLMRoutes(enum.Enum): management_routes = ( [ + "/usage/alert-rules", + "/usage/alert-rules/{rule_id}", + "/usage/alerts", + "/usage/alerts/{alert_id}", # user "/user/new", "/management/v1/users/bulk", @@ -855,7 +859,15 @@ class LiteLLMRoutes(enum.Enum): ] ) + usage_alert_routes = [ + "/usage/alert-rules", + "/usage/alert-rules/{rule_id}", + "/usage/alerts", + "/usage/alerts/{alert_id}", + ] + self_managed_routes = [ + *usage_alert_routes, "/lens", "/lens/{path:path}", "/telemetry/ui_events", diff --git a/litellm/proxy/alerting_endpoints/usage_alerts.py b/litellm/proxy/alerting_endpoints/usage_alerts.py new file mode 100644 index 00000000000..3a4c4112581 --- /dev/null +++ b/litellm/proxy/alerting_endpoints/usage_alerts.py @@ -0,0 +1,175 @@ +"""Public management API for daily usage alert rules and incidents.""" + +from datetime import date +from typing import Annotated, Final + +from fastapi import APIRouter, Depends, HTTPException, Query, Response + +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth +from litellm.proxy.deps.database import DatabaseDep +from litellm.repositories.daily_activity_buckets import api_key_from_bucket +from litellm.repositories.usage_alerts import UsageAlertsRepository +from litellm.types.proxy.auth.user_roles import LitellmUserRoles +from litellm.types.proxy.usage_alerts import ( + UsageAlert, + UsageAlertList, + UsageAlertMetric, + UsageAlertRule, + UsageAlertRuleInput, + UsageAlertRuleList, + UsageAlertScope, + UsageAlertStatus, + UsageAlertUpdate, +) + +router: Final = APIRouter(prefix="/usage", tags=["usage alerts"]) + + +def usage_alert_repository(database: DatabaseDep) -> UsageAlertsRepository: + if database is None: + raise HTTPException(status_code=503, detail="Usage alerts require a database connection") + return UsageAlertsRepository.from_prisma(database.prisma) + + +User = Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)] +Repository = Annotated[UsageAlertsRepository, Depends(usage_alert_repository)] +Limit = Annotated[int, Query(ge=1, le=100)] +Offset = Annotated[int, Query(ge=0, le=100000)] +ScopeId = Annotated[str | None, Query(min_length=1, max_length=1024)] + + +async def require_scope( + user: UserAPIKeyAuth, + repository: UsageAlertsRepository, + scope_type: UsageAlertScope | None, + scope_id: str | None, + *, + write: bool, +) -> None: + if user.key_type == "llm_api" or (write and user.key_type == "read_only"): + raise HTTPException(status_code=403, detail="A management-capable credential is required") + if user.user_role == LitellmUserRoles.PROXY_ADMIN: + return + if user.user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY: + if not write: + return + raise HTTPException(status_code=403, detail="Read-only users cannot manage usage alerts") + if user.user_role not in ( + LitellmUserRoles.INTERNAL_USER, + LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, + LitellmUserRoles.ORG_ADMIN, + LitellmUserRoles.TEAM, + ): + raise HTTPException(status_code=403, detail="Usage alerts require a management role") + if write and user.user_role == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY: + raise HTTPException(status_code=403, detail="Read-only users cannot manage usage alerts") + if scope_id is None: + raise HTTPException(status_code=403, detail="Specify a scope you can access") + if scope_type == "user" and user.user_id is not None and scope_id == user.user_id: + return + if scope_type == "team": + role: Final = await repository.team_role(scope_id, user.user_id) if user.user_id else None + if role == "admin" or ( + not write and (role == "user" or (user.user_role == LitellmUserRoles.TEAM and user.team_id == scope_id)) + ): + return + if scope_type == "key": + ownership: Final = await repository.key_ownership(api_key_from_bucket(scope_id)) + if ownership is not None: + owner_id, team_id = ownership + if user.user_id is not None and owner_id == user.user_id: + return + if user.user_id and team_id and await repository.team_role(team_id, user.user_id) == "admin": + return + raise HTTPException(status_code=403, detail="You cannot access this alert scope") + + +@router.get("/alert-rules", response_model=UsageAlertRuleList) +async def list_usage_alert_rules( + user: User, + repository: Repository, + scope_type: UsageAlertScope | None = None, + scope_id: ScopeId = None, + limit: Limit = 50, + offset: Offset = 0, +) -> UsageAlertRuleList: + await require_scope(user, repository, scope_type, scope_id, write=False) + return await repository.list_rules(scope_type, scope_id, limit, offset) + + +@router.post("/alert-rules", response_model=UsageAlertRule, status_code=201) +async def create_usage_alert_rule(data: UsageAlertRuleInput, user: User, repository: Repository) -> UsageAlertRule: + await require_scope(user, repository, data.scope_type, data.scope_id, write=True) + return await repository.save_rule(data, user.user_id) + + +@router.get("/alert-rules/{rule_id}", response_model=UsageAlertRule) +async def get_usage_alert_rule(rule_id: str, user: User, repository: Repository) -> UsageAlertRule: + rule: Final = await repository.get_rule(rule_id) + if rule is None: + raise HTTPException(status_code=404, detail="Alert rule not found") + await require_scope(user, repository, rule.scope_type, rule.scope_id, write=False) + return rule + + +@router.put("/alert-rules/{rule_id}", response_model=UsageAlertRule) +async def update_usage_alert_rule( + rule_id: str, data: UsageAlertRuleInput, user: User, repository: Repository +) -> UsageAlertRule: + rule: Final = await repository.get_rule(rule_id) + if rule is None: + raise HTTPException(status_code=404, detail="Alert rule not found") + await require_scope(user, repository, rule.scope_type, rule.scope_id, write=True) + await require_scope(user, repository, data.scope_type, data.scope_id, write=True) + return await repository.save_rule(data, user.user_id, existing=rule) + + +@router.delete("/alert-rules/{rule_id}", status_code=204) +async def delete_usage_alert_rule(rule_id: str, user: User, repository: Repository) -> Response: + rule: Final = await repository.get_rule(rule_id) + if rule is None: + raise HTTPException(status_code=404, detail="Alert rule not found") + await require_scope(user, repository, rule.scope_type, rule.scope_id, write=True) + await repository.delete_rule(rule_id) + return Response(status_code=204) + + +@router.get("/alerts", response_model=UsageAlertList) +async def list_usage_alerts( + user: User, + repository: Repository, + scope_type: UsageAlertScope | None = None, + scope_id: ScopeId = None, + metric: UsageAlertMetric | None = None, + status: UsageAlertStatus | None = None, + start_date: date | None = None, + end_date: date | None = None, + limit: Limit = 50, + offset: Offset = 0, +) -> UsageAlertList: + await require_scope(user, repository, scope_type, scope_id, write=False) + if start_date is not None and end_date is not None and start_date > end_date: + raise HTTPException(status_code=422, detail="start_date must not be after end_date") + return await repository.list_alerts(scope_type, scope_id, metric, status, start_date, end_date, limit, offset) + + +@router.get("/alerts/{alert_id}", response_model=UsageAlert) +async def get_usage_alert(alert_id: str, user: User, repository: Repository) -> UsageAlert: + alert: Final = await repository.get_alert(alert_id) + if alert is None: + raise HTTPException(status_code=404, detail="Usage alert not found") + await require_scope(user, repository, alert.scope_type, alert.scope_id, write=False) + return alert + + +@router.patch("/alerts/{alert_id}", response_model=UsageAlert) +async def update_usage_alert(alert_id: str, data: UsageAlertUpdate, user: User, repository: Repository) -> UsageAlert: + alert: Final = await repository.get_alert(alert_id) + if alert is None: + raise HTTPException(status_code=404, detail="Usage alert not found") + await require_scope(user, repository, alert.scope_type, alert.scope_id, write=True) + updated: Final = await repository.set_status(alert_id, data.status) + if updated is None: + raise HTTPException(status_code=404, detail="Usage alert not found") + return updated diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index debefec4a30..d876ce011a4 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -427,6 +427,7 @@ from litellm.proxy.agent_endpoints.auth.agent_access_groups import CeilingResolv from litellm.proxy.alerting_endpoints.model_deprecation_alerts import ( router as model_deprecation_alerts_router, ) +from litellm.proxy.alerting_endpoints.usage_alerts import router as usage_alerts_router from litellm.proxy.analytics_endpoints.analytics_endpoints import ( router as analytics_router, ) @@ -22438,6 +22439,7 @@ app.include_router(router_settings_router) app.include_router(fallback_management_router) app.include_router(cache_settings_router) app.include_router(model_deprecation_alerts_router) +app.include_router(usage_alerts_router) app.include_router(coordination_redis_settings_router) app.include_router(user_agent_analytics_router) app.include_router(gateway_request_router) diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index 78f16d9f1fd..46ab28530a9 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -2109,3 +2109,43 @@ model LiteLLM_SubscriptionAccountTable { @@unique([custom_llm_provider, account_id], map: "LiteLLM_SubscriptionAccountTable_provider_account_key") } + +// Daily alerts consume existing aggregate tables; no request-path writes. +model LiteLLM_UsageAlertRule { + id String @id + scope_type String + scope_id String? + enabled Boolean @default(true) + record Json + deleted Boolean @default(false) + @@index([enabled, deleted, id]) + @@index([scope_type, scope_id, deleted]) +} + +model LiteLLM_UsageAlertIncident { + id String @id + rule_id String + scope_key String + scope_type String + scope_id String? + metric String + status String + day String + evaluation_day String + breached Boolean + record Json + notified_channels String[] @default([]) + updated_at DateTime @default(now()) + @@index([rule_id, scope_key, evaluation_day]) + @@index([scope_type, scope_id, day]) + @@index([status, day]) + @@index([updated_at]) +} + +model LiteLLM_UsageAlertCheckpoint { + id String @id + completed Boolean @default(false) + owner String? + expires_at DateTime @default(now()) + @@index([expires_at]) +} diff --git a/litellm/repositories/usage_alerts.py b/litellm/repositories/usage_alerts.py new file mode 100644 index 00000000000..0a42657c478 --- /dev/null +++ b/litellm/repositories/usage_alerts.py @@ -0,0 +1,460 @@ +"""Bounded, writer-consistent storage for daily alert rules and episodes.""" + +import json +from datetime import date, datetime, timedelta, timezone +from typing import TYPE_CHECKING, Final, Protocol, cast +from urllib.parse import urlencode +from uuid import NAMESPACE_URL, uuid4, uuid5 + +from pydantic import BaseModel, TypeAdapter + +from litellm.types.proxy.usage_alerts import ( + UsageAlert, + UsageAlertChannel, + UsageAlertEvaluationStatus, + UsageAlertList, + UsageAlertMetric, + UsageAlertNotification, + UsageAlertObservation, + UsageAlertRule, + UsageAlertRuleInput, + UsageAlertRuleList, + UsageAlertScope, + UsageAlertStatus, +) + +if TYPE_CHECKING: + from litellm.proxy.utils import PrismaClient + + +class AlertDatabase(Protocol): + async def query_raw(self, query: str, *args: object) -> object: ... + + +class _Record(BaseModel): + record: dict[str, object] + + +class _Count(BaseModel): + count: int + + +class _Episode(_Record): + evaluation_day: date + breached: bool + + +class _Pending(_Record): + rule: dict[str, object] + notified_channels: tuple[UsageAlertChannel, ...] + + +class _Ownership(BaseModel): + user_id: str | None + team_id: str | None + + +def activity_path( + scope: UsageAlertScope, entity_id: str | None, day: date, alert_id: str, baseline_days: int = 7 +) -> str: + routes: Final = { + "key": ("/api-keys", "key"), + "team": ("/teams", "team"), + "user": ("/users", "user"), + "model": ("/models-and-endpoints", "activity_model"), + "deployment": ("/usage", ""), + } + route, parameter = routes[scope] + if scope == "deployment": + return ( + route + + "?" + + urlencode( + { + "range": "custom", + "from": (day - timedelta(days=baseline_days)).isoformat(), + "to": day.isoformat(), + "alert": alert_id, + } + ) + ) + query: Final = { + "activity_from": (day - timedelta(days=baseline_days)).isoformat(), + "activity_to": day.isoformat(), + "alert": alert_id, + } + if parameter and entity_id and not entity_id.startswith("sk-"): + query[parameter] = entity_id + query["detail_tab"] = "activity" + return route + "?" + urlencode(query) + + +class UsageAlertsRepository: + def __init__(self, database: AlertDatabase): + self.database: Final = database + + @classmethod + def from_prisma(cls, prisma: "PrismaClient") -> "UsageAlertsRepository": + return cls(cast(AlertDatabase, prisma.writer_db)) + + async def get_rule(self, rule_id: str) -> UsageAlertRule | None: + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw( + 'SELECT record FROM "LiteLLM_UsageAlertRule" WHERE id=$1 AND NOT deleted', rule_id + ) + ) + return UsageAlertRule.model_validate(rows[0].record) if rows else None + + async def save_rule( + self, data: UsageAlertRuleInput, created_by: str | None, existing: UsageAlertRule | None = None + ) -> UsageAlertRule: + now: Final = datetime.now(timezone.utc) + rule: Final = UsageAlertRule( + **data.model_dump(), + id=existing.id if existing else str(uuid4()), + created_by=existing.created_by if existing else created_by, + created_at=existing.created_at if existing else now, + updated_at=now, + ) + rows: Final = await self.database.query_raw( + """INSERT INTO "LiteLLM_UsageAlertRule" (id,scope_type,scope_id,enabled,record) VALUES ($1,$2,$3,$4,$5::jsonb) + ON CONFLICT (id) DO UPDATE SET scope_type=EXCLUDED.scope_type,scope_id=EXCLUDED.scope_id,enabled=EXCLUDED.enabled,record=EXCLUDED.record + WHERE NOT "LiteLLM_UsageAlertRule".deleted RETURNING record""", + rule.id, + rule.scope_type, + rule.scope_id, + rule.enabled, + rule.model_dump_json(), + ) + return UsageAlertRule.model_validate(TypeAdapter(tuple[_Record, ...]).validate_python(rows)[0].record) + + async def delete_rule(self, rule_id: str) -> None: + await self.database.query_raw( + 'UPDATE "LiteLLM_UsageAlertRule" SET deleted=true,enabled=false WHERE id=$1 RETURNING id', rule_id + ) + + async def list_rules( + self, scope_type: UsageAlertScope | None = None, scope_id: str | None = None, limit: int = 50, offset: int = 0 + ) -> UsageAlertRuleList: + args: Final = (scope_type, scope_id) + where: Final = "NOT deleted AND ($1::text IS NULL OR scope_type=$1) AND ($2::text IS NULL OR scope_id=$2)" + count: Final = TypeAdapter(tuple[_Count, ...]).validate_python( + await self.database.query_raw( + f'SELECT count(*)::int AS count FROM "LiteLLM_UsageAlertRule" WHERE {where}', *args + ) + ) + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw( + f'SELECT record FROM "LiteLLM_UsageAlertRule" WHERE {where} ORDER BY id LIMIT $3 OFFSET $4', + *args, + limit, + offset, + ) + ) + return UsageAlertRuleList( + data=tuple(UsageAlertRule.model_validate(row.record) for row in rows), + total=count[0].count, + limit=limit, + offset=offset, + ) + + async def list_enabled_rules(self, limit: int = 100, offset: int = 0) -> tuple[UsageAlertRule, ...]: + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw( + 'SELECT record FROM "LiteLLM_UsageAlertRule" WHERE enabled AND NOT deleted ORDER BY id LIMIT $1 OFFSET $2', + limit, + offset, + ) + ) + return tuple(UsageAlertRule.model_validate(row.record) for row in rows) + + async def get_alert(self, alert_id: str) -> UsageAlert | None: + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw('SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE id=$1', alert_id) + ) + return UsageAlert.model_validate(rows[0].record) if rows else None + + async def list_alerts( + self, + scope_type: UsageAlertScope | None = None, + scope_id: str | None = None, + metric: UsageAlertMetric | None = None, + status: UsageAlertStatus | None = None, + start_date: date | None = None, + end_date: date | None = None, + limit: int = 50, + offset: int = 0, + ) -> UsageAlertList: + args: Final = ( + scope_type, + scope_id, + metric, + status, + start_date.isoformat() if start_date else None, + end_date.isoformat() if end_date else None, + ) + where: Final = "($1::text IS NULL OR scope_type=$1) AND ($2::text IS NULL OR scope_id=$2) AND ($3::text IS NULL OR metric=$3) AND ($4::text IS NULL OR status=$4) AND ($5::text IS NULL OR day >= $5) AND ($6::text IS NULL OR day <= $6)" + count: Final = TypeAdapter(tuple[_Count, ...]).validate_python( + await self.database.query_raw( + f'SELECT count(*)::int AS count FROM "LiteLLM_UsageAlertIncident" WHERE {where}', *args + ) + ) + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw( + f'SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE {where} ORDER BY day DESC,id LIMIT $7 OFFSET $8', + *args, + limit, + offset, + ) + ) + return UsageAlertList( + data=tuple(UsageAlert.model_validate(row.record) for row in rows), + total=count[0].count, + limit=limit, + offset=offset, + ) + + async def set_status(self, alert_id: str, status: UsageAlertStatus) -> UsageAlert | None: + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw( + """UPDATE "LiteLLM_UsageAlertIncident" SET status=$2, record=jsonb_set(record,'{status}',to_jsonb($2::text)),updated_at=CURRENT_TIMESTAMP WHERE id=$1 RETURNING record""", + alert_id, + status, + ) + ) + return UsageAlert.model_validate(rows[0].record) if rows else None + + async def key_ownership(self, key_id: str) -> tuple[str | None, str | None] | None: + rows: Final = TypeAdapter(tuple[_Ownership, ...]).validate_python( + await self.database.query_raw( + 'SELECT user_id,team_id FROM "LiteLLM_VerificationToken" WHERE token=$1', key_id + ) + ) + return (rows[0].user_id, rows[0].team_id) if rows else None + + async def claim_job(self, owner: str, lease_seconds: int = 900) -> bool: + rows: Final = TypeAdapter(tuple[_Count, ...]).validate_python( + await self.database.query_raw( + """INSERT INTO "LiteLLM_UsageAlertCheckpoint" (id,owner,expires_at) VALUES ('worker',$1,CURRENT_TIMESTAMP+($2::int * INTERVAL '1 second')) + ON CONFLICT(id) DO UPDATE SET owner=EXCLUDED.owner,expires_at=EXCLUDED.expires_at WHERE "LiteLLM_UsageAlertCheckpoint".expires_at None: + await self.database.query_raw( + "DELETE FROM \"LiteLLM_UsageAlertCheckpoint\" WHERE id='worker' AND owner=$1 RETURNING id", owner + ) + + @staticmethod + def _checkpoint(rule: UsageAlertRule, day: date) -> str: + return f"{rule.id}:{rule.updated_at.isoformat()}:{day.isoformat()}" + + async def is_day_complete(self, rule: UsageAlertRule, day: date) -> bool: + rows: Final = TypeAdapter(tuple[_Count, ...]).validate_python( + await self.database.query_raw( + 'SELECT 1 AS count FROM "LiteLLM_UsageAlertCheckpoint" WHERE id=$1 AND completed', + self._checkpoint(rule, day), + ) + ) + return bool(rows) + + async def finish_day(self, rule: UsageAlertRule, day: date) -> None: + await self.database.query_raw( + 'INSERT INTO "LiteLLM_UsageAlertCheckpoint" (id,completed) VALUES ($1,true) ON CONFLICT(id) DO NOTHING RETURNING id', + self._checkpoint(rule, day), + ) + + async def ensure_default_rules(self, rules: tuple[UsageAlertRuleInput, ...]) -> None: + for rule in rules: + rule_id: Final = str( + uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.scope_type}:{rule.metric}:{rule.direction}") + ) + now: Final = datetime.now(timezone.utc) + record: Final = UsageAlertRule( + **rule.model_dump(), id=rule_id, created_at=now, updated_at=now, created_by=None + ) + await self.database.query_raw( + 'INSERT INTO "LiteLLM_UsageAlertRule" (id,scope_type,scope_id,enabled,record) VALUES ($1,$2,$3,$4,$5::jsonb) ON CONFLICT(id) DO NOTHING RETURNING id', + rule_id, + rule.scope_type, + rule.scope_id, + rule.enabled, + record.model_dump_json(), + ) + + async def list_breached_entity_ids(self, rule_id: str, limit: int = 500, offset: int = 0) -> tuple[str | None, ...]: + rows: Final = TypeAdapter(tuple[dict[str, str | None], ...]).validate_python( + await self.database.query_raw( + 'SELECT DISTINCT scope_id FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND breached ORDER BY scope_id NULLS FIRST LIMIT $2 OFFSET $3', + rule_id, + limit, + offset, + ) + ) + return tuple(row["scope_id"] for row in rows) + + async def record_evaluation_status( + self, rule: UsageAlertRule, day: date, status: UsageAlertEvaluationStatus + ) -> None: + await self.database.query_raw( + """UPDATE "LiteLLM_UsageAlertRule" SET record=record || jsonb_build_object('last_evaluated_day',$2::text,'evaluation_status',$3::text) WHERE id=$1 AND record->>'updated_at'=$4 AND NOT deleted RETURNING id""", + rule.id, + day.isoformat(), + status, + rule.model_dump(mode="json")["updated_at"], + ) + + async def record_observation( + self, + rule: UsageAlertRule, + entity_id: str | None, + day: date, + observed: float, + baseline: float, + threshold: float, + breached: bool, + ) -> UsageAlert | None: + observation: Final = UsageAlertObservation( + entity_id=entity_id, day=day, observed=observed, baseline=baseline, threshold=threshold, breached=breached + ) + changed: Final = await self.record_observations(rule, (observation,)) + if changed: + return changed[0] + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw( + 'SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND scope_key=$2 ORDER BY evaluation_day DESC,id DESC LIMIT 1', + rule.id, + entity_id or "", + ) + ) + return UsageAlert.model_validate(rows[0].record) if rows else None + + async def record_observations( + self, rule: UsageAlertRule, observations: tuple[UsageAlertObservation, ...] + ) -> tuple[UsageAlert, ...]: + if not observations: + return () + if len(observations) > 500 or len({item.entity_id for item in observations}) != len(observations): + raise ValueError("An observation batch must contain at most 500 distinct entities") + now: Final = datetime.now(timezone.utc) + records: Final = [] + for item in observations: + alert_id: Final = str( + uuid5(NAMESPACE_URL, f"litellm:usage-alert:{rule.id}:{item.entity_id or ''}:{item.day.isoformat()}") + ) + incident: Final = UsageAlert( + id=alert_id, + rule_id=rule.id, + rule_name=rule.name, + scope_type=rule.scope_type, + scope_id=item.entity_id, + metric=rule.metric, + direction=rule.direction, + status="open", + day=item.day, + baseline_start=item.day - timedelta(days=rule.baseline_days), + baseline_end=item.day - timedelta(days=1), + observed_value=item.observed, + baseline_value=item.baseline, + threshold=item.threshold, + first_seen=now, + last_seen=now, + activity_path=activity_path(rule.scope_type, item.entity_id, item.day, alert_id, rule.baseline_days), + ) + records.append( + { + "scope_key": item.entity_id or "", + "breached": item.breached, + "day": item.day.isoformat(), + "record": incident.model_dump(mode="json"), + } + ) + rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python( + await self.database.query_raw( + """ + WITH input AS (SELECT * FROM jsonb_to_recordset($2::jsonb) AS x(scope_key text, breached boolean, day text, record jsonb)), + prior AS ( + SELECT x.*,p.id AS prior_id,p.evaluation_day,p.breached AS prior_breached,p.record AS prior_record + FROM input x LEFT JOIN LATERAL ( + SELECT * FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND scope_key=x.scope_key ORDER BY evaluation_day DESC,id DESC LIMIT 1 + ) p ON true + WHERE EXISTS (SELECT 1 FROM "LiteLLM_UsageAlertRule" WHERE id=$1 AND enabled AND NOT deleted AND record->>'updated_at'=$3) + ), updated AS ( + UPDATE "LiteLLM_UsageAlertIncident" i SET + evaluation_day=p.day, breached=p.breached, + day=CASE WHEN p.breached THEN p.day ELSE i.day END, + record=(CASE WHEN p.breached THEN p.record || jsonb_build_object( + 'id',i.id,'first_seen',i.record->'first_seen', + 'activity_path',replace(p.record->>'activity_path',p.record->>'id',i.id)) ELSE i.record END) + || jsonb_build_object('status',CASE WHEN i.status='open' AND NOT p.breached THEN 'resolved' ELSE i.status END), + status=CASE WHEN i.status='open' AND NOT p.breached THEN 'resolved' ELSE i.status END, + updated_at=CURRENT_TIMESTAMP + FROM prior p WHERE i.id=p.prior_id AND p.evaluation_day>'id',$1,p.scope_key,p.record->>'scope_type',p.record->>'scope_id',p.record->>'metric','open',p.day,p.day,true,p.record + FROM prior p WHERE p.breached AND (p.prior_id IS NULL OR (NOT p.prior_breached AND p.evaluation_day tuple[UsageAlertNotification, ...]: + rows: Final = TypeAdapter(tuple[_Pending, ...]).validate_python( + await self.database.query_raw( + """SELECT i.record,r.record AS rule,i.notified_channels FROM "LiteLLM_UsageAlertIncident" i JOIN "LiteLLM_UsageAlertRule" r ON r.id=i.rule_id + WHERE i.status='open' AND r.enabled AND NOT r.deleted AND EXISTS ( + SELECT 1 FROM jsonb_array_elements_text(r.record->'notification_channels') channel WHERE NOT (channel=ANY(i.notified_channels))) + ORDER BY i.updated_at,i.id LIMIT $1""", + limit, + ) + ) + return tuple( + UsageAlertNotification( + incident=UsageAlert.model_validate(row.record), + rule=UsageAlertRule.model_validate(row.rule), + pending_channels=tuple( + channel + for channel in UsageAlertRule.model_validate(row.rule).notification_channels + if channel not in row.notified_channels + ), + ) + for row in rows + ) + + async def mark_notified(self, incident_id: str, channel: UsageAlertChannel) -> None: + await self.database.query_raw( + 'UPDATE "LiteLLM_UsageAlertIncident" SET notified_channels=array_append(notified_channels,$2::text) WHERE id=$1 AND NOT ($2::text=ANY(notified_channels)) RETURNING id', + incident_id, + channel, + ) + + async def prune(self, retention_days: int = 90, limit: int = 500) -> None: + await self.database.query_raw( + """DELETE FROM "LiteLLM_UsageAlertIncident" WHERE id IN (SELECT id FROM "LiteLLM_UsageAlertIncident" WHERE status<>'open' AND updated_at str | None: + rows: Final = TypeAdapter(tuple[dict[str, str], ...]).validate_python( + await self.database.query_raw( + """SELECT member->>'role' AS role FROM "LiteLLM_TeamTable", jsonb_array_elements(members_with_roles) member WHERE team_id=$1 AND member->>'user_id'=$2 LIMIT 1""", + team_id, + user_id, + ) + ) + return rows[0]["role"] if rows else None diff --git a/litellm/types/proxy/usage_alerts.py b/litellm/types/proxy/usage_alerts.py new file mode 100644 index 00000000000..b0bd8d148ea --- /dev/null +++ b/litellm/types/proxy/usage_alerts.py @@ -0,0 +1,116 @@ +"""Daily usage alert rules and incidents shared by the API and evaluator.""" + +from datetime import date, datetime +from typing import Literal, Self + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +UsageAlertScope = Literal["deployment", "key", "user", "team", "model"] +UsageAlertMetric = Literal["spend", "requests", "latency", "errors"] +UsageAlertDirection = Literal["increase", "decrease"] +UsageAlertStatus = Literal["open", "resolved", "dismissed"] +UsageAlertChannel = Literal["email", "slack"] +UsageAlertEvaluationStatus = Literal["pending", "complete", "insufficient_coverage", "insufficient_history"] + + +class UsageAlertRuleInput(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid", allow_inf_nan=False) + + name: str = Field(min_length=1, max_length=200) + scope_type: UsageAlertScope + scope_id: str | None = Field(default=None, min_length=1, max_length=1024) + metric: UsageAlertMetric + direction: UsageAlertDirection + multiplier: float = Field(default=3, gt=1, le=1000) + baseline_days: int = Field(default=7, ge=1, le=30) + min_history_days: int = Field(default=3, ge=1, le=30) + min_requests: int = Field(default=100, ge=0) + min_value: float = Field(default=0, ge=0, description="Minimum absolute change in the metric's units") + enabled: bool = True + notification_channels: tuple[UsageAlertChannel, ...] = Field(default=(), max_length=2) + email_recipients: tuple[str, ...] = Field(default=(), max_length=20) + + @model_validator(mode="after") + def validate_rule(self) -> Self: + if self.min_history_days > self.baseline_days: + raise ValueError("min_history_days must not exceed baseline_days") + if self.scope_type == "deployment" and self.scope_id is not None: + raise ValueError("Deployment alerts cover the whole gateway; omit scope_id") + if self.scope_type == "key" and self.scope_id is not None and self.scope_id.startswith("sk-"): + raise ValueError("Use the key's reporting ID, never the secret key") + if len(set(self.notification_channels)) != len(self.notification_channels): + raise ValueError("notification_channels must be unique") + if "email" in self.notification_channels and not self.email_recipients: + raise ValueError("Email notifications require at least one recipient") + if any( + "@" not in email or "\n" in email or "\r" in email or len(email) > 254 for email in self.email_recipients + ): + raise ValueError("Invalid email recipient") + return self + + +class UsageAlertRule(UsageAlertRuleInput): + id: str + created_at: datetime + updated_at: datetime + created_by: str | None + last_evaluated_day: date | None = None + evaluation_status: UsageAlertEvaluationStatus = "pending" + + +class UsageAlert(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid", allow_inf_nan=False) + + id: str + rule_id: str + rule_name: str + scope_type: UsageAlertScope + scope_id: str | None + metric: UsageAlertMetric + direction: UsageAlertDirection + status: UsageAlertStatus + day: date + baseline_start: date + baseline_end: date + observed_value: float + baseline_value: float + threshold: float + first_seen: datetime + last_seen: datetime + activity_path: str + + +class UsageAlertUpdate(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + status: UsageAlertStatus + + +class UsageAlertRuleList(BaseModel): + data: tuple[UsageAlertRule, ...] + total: int + limit: int + offset: int + + +class UsageAlertList(BaseModel): + data: tuple[UsageAlert, ...] + total: int + limit: int + offset: int + + +class UsageAlertNotification(BaseModel): + model_config = ConfigDict(frozen=True) + incident: UsageAlert + rule: UsageAlertRule + pending_channels: tuple[UsageAlertChannel, ...] + + +class UsageAlertObservation(BaseModel): + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + entity_id: str | None + day: date + observed: float + baseline: float + threshold: float + breached: bool diff --git a/schema.prisma b/schema.prisma index 78f16d9f1fd..46ab28530a9 100644 --- a/schema.prisma +++ b/schema.prisma @@ -2109,3 +2109,43 @@ model LiteLLM_SubscriptionAccountTable { @@unique([custom_llm_provider, account_id], map: "LiteLLM_SubscriptionAccountTable_provider_account_key") } + +// Daily alerts consume existing aggregate tables; no request-path writes. +model LiteLLM_UsageAlertRule { + id String @id + scope_type String + scope_id String? + enabled Boolean @default(true) + record Json + deleted Boolean @default(false) + @@index([enabled, deleted, id]) + @@index([scope_type, scope_id, deleted]) +} + +model LiteLLM_UsageAlertIncident { + id String @id + rule_id String + scope_key String + scope_type String + scope_id String? + metric String + status String + day String + evaluation_day String + breached Boolean + record Json + notified_channels String[] @default([]) + updated_at DateTime @default(now()) + @@index([rule_id, scope_key, evaluation_day]) + @@index([scope_type, scope_id, day]) + @@index([status, day]) + @@index([updated_at]) +} + +model LiteLLM_UsageAlertCheckpoint { + id String @id + completed Boolean @default(false) + owner String? + expires_at DateTime @default(now()) + @@index([expires_at]) +}