mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-14 09:03:16 +00:00
fix(alerts): project bounded resource metadata
This commit is contained in:
parent
2f898fa2de
commit
01e1198a88
3 changed files with 146 additions and 10 deletions
|
|
@ -57,7 +57,7 @@ Replace `issue-example` with an ID returned by `/issues`. Activity uses the issu
|
|||
|
||||
Alert Activity returns exact daily totals in `results` and exact range totals in `metadata`. It does not return the general Usage API's per-day dimension dictionaries. Key issues also return `model_breakdown`, with up to 20 models by range spend by default. Set `model_limit` between 1 and 100 to change that limit. `total_models` counts every matching model; `other` contains the combined metrics and `model_count` of all remaining models, or null when none remain. It is an aggregate, not a model identifier. Adding the displayed models and `other` recovers the full totals. Model identifiers longer than 1,024 characters are included in `other` without truncation. Non-key issues return null for `model_breakdown`. Activity queries use a bounded execution budget. A busy or over-budget query returns `503` with `Retry-After: 5`; retry later or choose a shorter date range.
|
||||
|
||||
`/impact` reports distinct resource counts and requests within that same Activity range. These counts mean that resources used the cohort, not that each resource suffered a regression. Saved `impact.measured_keys` and `impact.measured_models`, when present, describe cohorts that met the detector's comparison criteria. Historical team aggregates do not retain user membership, so a team issue has no `/users` view and its user count is null. User issues similarly have no historical team count. Missing attribution remains unknown.
|
||||
`/impact` reports distinct resource counts and requests within that same Activity range. These counts mean that resources used the cohort, not that each resource suffered a regression. Saved `impact.measured_keys` and `impact.measured_models`, when present, describe cohorts that met the detector's comparison criteria. Resource pages fetch labels only for the selected page, using the latest deleted-key label when needed. They do not load arbitrary key or user metadata. Display aliases are limited to 256 characters; email labels longer than 320 characters are omitted. Resource IDs remain unchanged. Historical team aggregates do not retain user membership, so a team issue has no `/users` view and its user count is null. User issues similarly have no historical team count. Missing attribution remains unknown.
|
||||
|
||||
Supporting signals have their own saved comparisons and relationship explanations. Their baseline windows can differ, so do not derive a combined error rate or unit cost by adding their displayed values. Administrators and admin viewers can inspect broader supporting signals. A scoped reader receives only signals for the exact key, user, or team they can access. That restriction applies before the signal count and pagination; broader totals are not disclosed.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
"""Issue investigations reuse the same aggregate filters as entity Activity."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import asdict, replace
|
||||
from datetime import date
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Final, cast # noqa: TID251 # generated Prisma structural database interface
|
||||
|
||||
from pydantic import BaseModel, TypeAdapter
|
||||
|
|
@ -38,6 +40,17 @@ class _Counts(BaseModel, frozen=True):
|
|||
requests: int
|
||||
|
||||
|
||||
class _KeyLabel(BaseModel, frozen=True):
|
||||
api_key: str
|
||||
metadata: KeyMetadata
|
||||
|
||||
|
||||
class _UserLabel(BaseModel, frozen=True):
|
||||
user_id: str
|
||||
user_alias: str | None
|
||||
user_email: str | None
|
||||
|
||||
|
||||
def issue_activity_scope(cohort: IssueCohort, start: date, end: date) -> DailyActivityScope:
|
||||
if start > end or (end - start).days > 365:
|
||||
raise ValueError("Choose an ordered date range of at most 366 days")
|
||||
|
|
@ -79,18 +92,16 @@ class IssueActivityRepository:
|
|||
async def keys(
|
||||
self, cohort: IssueCohort, start: date, end: date, limit: int, offset: int
|
||||
) -> DailyActivityKeyPageResponse:
|
||||
if not 1 <= limit <= 100 or offset < 0:
|
||||
raise ValueError("Read between one and 100 keys with a nonnegative offset")
|
||||
page: Final = await self.activity.key_page(issue_activity_scope(cohort, start, end), limit=limit, offset=offset)
|
||||
metadata: Final = await self.activity.key_metadata(
|
||||
frozenset(row.api_key for row in page.rows if row.api_key), None
|
||||
)
|
||||
metadata: Final = await self._key_labels(tuple(row.api_key for row in page.rows if row.api_key))
|
||||
return DailyActivityKeyPageResponse(
|
||||
api_keys=[
|
||||
KeySpendActivityRow(
|
||||
api_key=api_key_bucket(row.api_key),
|
||||
metrics=KeySpendMetrics.model_validate(asdict(row)),
|
||||
metadata=KeyMetadata.model_validate(asdict(metadata[row.api_key]))
|
||||
if row.api_key in metadata
|
||||
else KeyMetadata(),
|
||||
metadata=metadata.get(row.api_key, KeyMetadata()),
|
||||
)
|
||||
for row in page.rows
|
||||
],
|
||||
|
|
@ -102,19 +113,23 @@ class IssueActivityRepository:
|
|||
async def users(
|
||||
self, cohort: IssueCohort, start: date, end: date, limit: int, offset: int
|
||||
) -> DailyActivityUserPageResponse:
|
||||
if not 1 <= limit <= 100 or offset < 0:
|
||||
raise ValueError("Read between one and 100 users with a nonnegative offset")
|
||||
if cohort.scope_type == "team":
|
||||
raise ValueError("Historical team aggregates do not retain user membership")
|
||||
page: Final = await self.activity.user_page(
|
||||
issue_activity_scope(cohort, start, end), limit=limit, offset=offset
|
||||
)
|
||||
metadata: Final = await self.activity.user_metadata(frozenset(row.user_id for row in page.rows if row.user_id))
|
||||
metadata: Final = await self._user_labels(tuple(row.user_id for row in page.rows if row.user_id))
|
||||
return DailyActivityUserPageResponse(
|
||||
users=tuple(
|
||||
UserActivityRow.model_validate(
|
||||
{
|
||||
**asdict(row),
|
||||
**(
|
||||
asdict(metadata[row.user_id]) if row.user_id is not None and row.user_id in metadata else {}
|
||||
metadata[row.user_id].model_dump()
|
||||
if row.user_id is not None and row.user_id in metadata
|
||||
else {}
|
||||
),
|
||||
}
|
||||
)
|
||||
|
|
@ -125,6 +140,47 @@ class IssueActivityRepository:
|
|||
offset=offset,
|
||||
)
|
||||
|
||||
async def _key_labels(self, ids: tuple[str, ...]) -> Mapping[str, KeyMetadata]:
|
||||
if len(ids) > 100:
|
||||
raise ValueError("Read at most 100 key labels")
|
||||
if not ids:
|
||||
return MappingProxyType({})
|
||||
rows: Final = TypeAdapter(tuple[_KeyLabel, ...]).validate_python(
|
||||
await self.database.query_raw(
|
||||
"""SELECT w.api_key,jsonb_build_object(
|
||||
'key_alias',LEFT(COALESCE(k.key_alias,d.key_alias),256),
|
||||
'team_id',COALESCE(k.team_id,d.team_id),'user_id',COALESCE(k.user_id,d.user_id),
|
||||
'user_email',CASE WHEN length(u.user_email)<=320 THEN u.user_email END,
|
||||
'key_exists',CASE WHEN k.token IS NOT NULL THEN true WHEN d.id IS NOT NULL THEN false END
|
||||
) AS metadata
|
||||
FROM unnest($1::text[]) w(api_key)
|
||||
LEFT JOIN "LiteLLM_VerificationToken" k ON k.token=w.api_key
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT id,key_alias,team_id,user_id FROM "LiteLLM_DeletedVerificationToken"
|
||||
WHERE token=w.api_key AND k.token IS NULL ORDER BY deleted_at DESC,id DESC LIMIT 1
|
||||
) d ON true
|
||||
LEFT JOIN "LiteLLM_UserTable" u ON u.user_id=COALESCE(k.user_id,d.user_id)
|
||||
LIMIT 100""",
|
||||
list(ids),
|
||||
)
|
||||
)
|
||||
return MappingProxyType({row.api_key: row.metadata for row in rows})
|
||||
|
||||
async def _user_labels(self, ids: tuple[str, ...]) -> Mapping[str, _UserLabel]:
|
||||
if len(ids) > 100:
|
||||
raise ValueError("Read at most 100 user labels")
|
||||
if not ids:
|
||||
return MappingProxyType({})
|
||||
rows: Final = TypeAdapter(tuple[_UserLabel, ...]).validate_python(
|
||||
await self.database.query_raw(
|
||||
"""SELECT user_id,LEFT(user_alias,256) AS user_alias,
|
||||
CASE WHEN length(user_email)<=320 THEN user_email END AS user_email
|
||||
FROM "LiteLLM_UserTable" WHERE user_id=ANY($1::text[]) LIMIT 100""",
|
||||
list(ids),
|
||||
)
|
||||
)
|
||||
return MappingProxyType({row.user_id: row for row in rows})
|
||||
|
||||
async def _counts(self, scope: DailyActivityScope) -> _Counts:
|
||||
where, params = build_where_clause(scope)
|
||||
rows: Final = TypeAdapter(tuple[_Counts, ...]).validate_python(
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import asyncio
|
||||
import json
|
||||
from collections.abc import AsyncGenerator, Mapping
|
||||
from collections.abc import AsyncGenerator, Mapping, Sequence
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from hashlib import sha256
|
||||
|
|
@ -17,6 +17,7 @@ from prisma.types import DatasourceOverride
|
|||
from psycopg import sql
|
||||
from pydantic import BaseModel, TypeAdapter, ValidationError
|
||||
|
||||
from litellm.repositories.daily_activity_repository import DailyActivityQueryDatabase
|
||||
from litellm.repositories.daily_activity_sql import build_bounded_daily_activity_sql
|
||||
from litellm.repositories.issue_activity import IssueActivityRepository
|
||||
from litellm.repositories.issues import IssuesRepository
|
||||
|
|
@ -889,3 +890,82 @@ async def test_issue_prune_removes_expired_children_in_bounded_passes(
|
|||
'DELETE FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 RETURNING id',
|
||||
rule.id,
|
||||
)
|
||||
|
||||
|
||||
class _MeasuredActivityDatabase:
|
||||
def __init__(self, database: DailyActivityQueryDatabase) -> None:
|
||||
self.database = database
|
||||
self.outputs: tuple[tuple[int, int], ...] = ()
|
||||
|
||||
async def query_raw(self, query: str, *args: object) -> Sequence[Mapping[str, object]] | None:
|
||||
result: Final = await self.database.query_raw(query, *args)
|
||||
self.outputs = (*self.outputs, (len(result or ()), len(json.dumps(result, default=str))))
|
||||
return result
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_issue_resource_metadata_bounds_labels_and_deleted_history(
|
||||
gateway: Gateway, db: ManagementDatabase
|
||||
) -> None:
|
||||
store: Final = IssuesRepository.from_prisma(db.prisma)
|
||||
activity: Final = IssueActivityRepository.from_prisma(db.prisma)
|
||||
measured: Final = _MeasuredActivityDatabase(activity.database)
|
||||
repository: Final = IssueActivityRepository(activity.activity, measured)
|
||||
identity: Final = uuid4().hex
|
||||
reporting_key: Final = sha256(identity.encode()).hexdigest()
|
||||
day: Final = date.today() - timedelta(days=3)
|
||||
with gateway.scenario() as scenario:
|
||||
owner: Final = scenario.user(user_role="internal_user")
|
||||
issue: Final = _issue(IssueCohort(scope_type="user", scope_id=owner)).model_copy(
|
||||
update={"range_start": day, "range_end": day}
|
||||
)
|
||||
try:
|
||||
await _query(
|
||||
db,
|
||||
"""UPDATE "LiteLLM_UserTable" SET user_alias=repeat('u',1024),
|
||||
user_email=repeat('m',400)||'@example.com' WHERE user_id=$1""",
|
||||
owner,
|
||||
)
|
||||
await _query(
|
||||
db,
|
||||
"""INSERT INTO "LiteLLM_DeletedVerificationToken"
|
||||
(id,token,key_alias,user_id,models,metadata,deleted_at)
|
||||
SELECT $1::text||'-'||n,$2::text,'archived-'||n||'-'||repeat('x',1024),$3::text,
|
||||
ARRAY[]::text[],jsonb_build_object('padding',repeat('x',65536)),
|
||||
CURRENT_TIMESTAMP+n*INTERVAL '1 second' FROM generate_series(1,400) n""",
|
||||
identity,
|
||||
reporting_key,
|
||||
owner,
|
||||
)
|
||||
await _query(
|
||||
db,
|
||||
"""INSERT INTO "LiteLLM_DailyUserSpend"
|
||||
(id,user_id,date,api_key,model,api_requests,successful_requests,updated_at)
|
||||
VALUES($1,$2,$3,$4,'example-model',10,10,CURRENT_TIMESTAMP)""",
|
||||
identity,
|
||||
owner,
|
||||
day.isoformat(),
|
||||
reporting_key,
|
||||
)
|
||||
await store.upsert_issue(issue)
|
||||
keys: Final = await repository.keys(issue.cohort, day, day, 1, 0)
|
||||
assert keys.total_api_keys == 1 and len(keys.api_keys) == 1
|
||||
key: Final = keys.api_keys[0]
|
||||
assert key.api_key == reporting_key and key.metadata.key_exists is False
|
||||
assert key.metadata.key_alias is not None and key.metadata.key_alias.startswith("archived-400-")
|
||||
assert len(key.metadata.key_alias) == 256 and key.metadata.user_id == owner
|
||||
assert key.metadata.user_email is None
|
||||
assert measured.outputs[-1][0] == 1 and measured.outputs[-1][1] < 2048
|
||||
users: Final = await repository.users(issue.cohort, day, day, 1, 0)
|
||||
assert users.total_users == 1 and len(users.users) == 1
|
||||
assert users.users[0].user_id == owner and users.users[0].user_alias == "u" * 256
|
||||
assert users.users[0].user_email is None
|
||||
assert measured.outputs[-1][0] == 1 and measured.outputs[-1][1] < 2048
|
||||
for endpoint in ("keys", "users"):
|
||||
response: Final = gateway.request("GET", f"/issues/{issue.id}/{endpoint}", params={"limit": "1"})
|
||||
assert response.status_code == 200 and len(response.content) < 2048
|
||||
assert "padding" not in response.text
|
||||
finally:
|
||||
await _remove_issues(db, (issue.id,))
|
||||
await _query(db, 'DELETE FROM "LiteLLM_DailyUserSpend" WHERE id=$1', identity)
|
||||
await _query(db, 'DELETE FROM "LiteLLM_DeletedVerificationToken" WHERE token=$1', reporting_key)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue