mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-15 09:12:18 +00:00
fix(alerts): honor custom overrides for queued notifications
This commit is contained in:
moe-berri
parent
572eb24170
commit
ae04c3f161
2 changed files with 98 additions and 5 deletions
|
|
@ -381,14 +381,14 @@ class IssuesRepository:
|
|||
COALESCE(r.enabled AND NOT r.deleted
|
||||
AND r.scope_type=a.scope_type AND r.record->>'metric'=a.metric
|
||||
AND r.record->>'direction'=a.record->>'direction'
|
||||
AND (r.scope_id IS NULL OR r.scope_id=a.scope_id)
|
||||
AND (NOT(r.id=ANY($2::text[])) OR NOT EXISTS (
|
||||
AND (r.scope_id IS NULL OR r.scope_id=a.scope_id),false) AS eligible,
|
||||
COALESCE(r.id=ANY($2::text[]) AND EXISTS (
|
||||
SELECT 1 FROM "LiteLLM_UsageAlertRule" custom WHERE custom.enabled AND NOT custom.deleted
|
||||
AND COALESCE((custom.record->>'overrides_defaults')::boolean,false)
|
||||
AND NOT(custom.id=ANY($2::text[])) AND custom.scope_type=r.scope_type
|
||||
AND custom.record->>'metric'=r.record->>'metric'
|
||||
AND custom.record->>'direction'=r.record->>'direction'
|
||||
AND (custom.scope_id IS NULL OR custom.scope_id=a.scope_id))),false) AS eligible
|
||||
AND (custom.scope_id IS NULL OR custom.scope_id=a.scope_id)),false) AS suppressed
|
||||
FROM "LiteLLM_Issue" i JOIN "LiteLLM_IssueSignal" s ON s.issue_id=i.id
|
||||
LEFT JOIN "LiteLLM_UsageAlertIncident" a ON a.id=s.source_id
|
||||
LEFT JOIN "LiteLLM_UsageAlertRule" r ON r.id=a.rule_id
|
||||
|
|
@ -407,7 +407,7 @@ class IssuesRepository:
|
|||
FROM primary_signals
|
||||
), healthy AS (
|
||||
SELECT id FROM observations GROUP BY id
|
||||
HAVING bool_or(eligible AND healthy) AND bool_and(NOT eligible OR healthy)
|
||||
HAVING bool_or(eligible AND healthy) AND bool_and(NOT eligible OR suppressed OR healthy)
|
||||
), changed AS (
|
||||
UPDATE "LiteLLM_Issue" i SET active=false,
|
||||
status=CASE WHEN status='open' THEN 'resolved' ELSE status END,updated_at=CURRENT_TIMESTAMP
|
||||
|
|
@ -560,6 +560,13 @@ class IssuesRepository:
|
|||
AND a.scope_type=r.scope_type AND a.metric=r.record->>'metric'
|
||||
AND a.record->>'direction'=r.record->>'direction'
|
||||
AND (r.scope_id IS NULL OR a.scope_id=r.scope_id)
|
||||
AND (NOT(r.id=ANY($3::text[])) OR NOT EXISTS (
|
||||
SELECT 1 FROM "LiteLLM_UsageAlertRule" custom WHERE custom.enabled AND NOT custom.deleted
|
||||
AND COALESCE((custom.record->>'overrides_defaults')::boolean,false)
|
||||
AND NOT(custom.id=ANY($3::text[])) AND custom.scope_type=r.scope_type
|
||||
AND custom.record->>'metric'=r.record->>'metric'
|
||||
AND custom.record->>'direction'=r.record->>'direction'
|
||||
AND (custom.scope_id IS NULL OR custom.scope_id=a.scope_id)))
|
||||
AND EXISTS(SELECT 1 FROM "LiteLLM_IssueSignal" s WHERE s.issue_id=i.id
|
||||
AND s.source='usage_alert' AND s.source_id=a.id)
|
||||
ORDER BY b.rule_id,b.source_id LIMIT 1
|
||||
|
|
@ -569,6 +576,7 @@ class IssuesRepository:
|
|||
ORDER BY d.next_attempt,d.id LIMIT $1""",
|
||||
limit,
|
||||
delivery_id,
|
||||
tuple(DEFAULT_USAGE_RULE_IDS),
|
||||
)
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ from litellm.types.proxy.issues import (
|
|||
IssueSignalList,
|
||||
)
|
||||
from litellm.types.proxy.usage_alert_defaults import default_usage_rule_id, default_usage_rules
|
||||
from litellm.types.proxy.usage_alerts import UsageAlertObservation, UsageAlertRuleInput
|
||||
from litellm.types.proxy.usage_alerts import UsageAlert, UsageAlertObservation, UsageAlertRuleInput
|
||||
|
||||
|
||||
def _issue(cohort: IssueCohort) -> IssueInput:
|
||||
|
|
@ -623,6 +623,91 @@ async def _family_database(db: ManagementDatabase) -> AsyncGenerator[Prisma]:
|
|||
setup.execute(sql.SQL("DROP SCHEMA {} CASCADE").format(sql.Identifier(schema)))
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_late_default_override_stops_queued_delivery_and_recovers_on_fresh_evidence(
|
||||
db: ManagementDatabase,
|
||||
) -> None:
|
||||
async with _family_database(db) as database:
|
||||
borrowed: Final = cast(AlertDatabase, database) # cast-ok: real generated Prisma query delegate
|
||||
for table in (
|
||||
"LiteLLM_UsageAlertIncident",
|
||||
"LiteLLM_Issue",
|
||||
"LiteLLM_IssueSignal",
|
||||
"LiteLLM_IssueDelivery",
|
||||
"LiteLLM_IssueDeliveryRule",
|
||||
"LiteLLM_VerificationToken",
|
||||
"LiteLLM_TeamTable",
|
||||
"LiteLLM_UserTable",
|
||||
):
|
||||
await borrowed.query_raw(f'CREATE TABLE "{table}" (LIKE public."{table}" INCLUDING ALL)')
|
||||
alerts: Final = UsageAlertsRepository(borrowed)
|
||||
store: Final = IssuesRepository(borrowed)
|
||||
await alerts.ensure_default_rules(default_usage_rules(("slack",)))
|
||||
rule: Final = await alerts.get_rule(default_usage_rule_id("key", "requests", "increase"))
|
||||
assert rule is not None
|
||||
issues: Final = tuple(_issue(IssueCohort(scope_type="key", scope_id=key)) for key in ("key-a", "key-b"))
|
||||
observations: Final = tuple(
|
||||
UsageAlertObservation(
|
||||
entity_id=issue.cohort.scope_id,
|
||||
day=issue.observed_on,
|
||||
observed=420,
|
||||
baseline=120,
|
||||
threshold=360,
|
||||
breached=True,
|
||||
)
|
||||
for issue in issues
|
||||
)
|
||||
sources: Final = await alerts.record_observations(rule, observations)
|
||||
await store.upsert_issues(issues)
|
||||
for issue in issues:
|
||||
source: UsageAlert = next(source for source in sources if source.scope_id == issue.cohort.scope_id)
|
||||
await store.add_signals(
|
||||
issue.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),)
|
||||
)
|
||||
await store.queue_delivery(
|
||||
IssueDeliveryInput(id=issue.id, issue_id=issue.id, channel="slack"),
|
||||
rule.id,
|
||||
rule.updated_at.isoformat(),
|
||||
source.id,
|
||||
)
|
||||
assert await store.pending_delivery(issue.id) is not None
|
||||
definition: Final = UsageAlertRuleInput(
|
||||
name="Project A threshold",
|
||||
scope_type="key",
|
||||
scope_id="key-a",
|
||||
metric="requests",
|
||||
direction="increase",
|
||||
multiplier=100,
|
||||
)
|
||||
ordinary: Final = await alerts.save_rule(definition, "example-user")
|
||||
assert ordinary is not None and await store.pending_delivery(issues[0].id) is not None
|
||||
assert await alerts.save_rule(definition, "qa-admin", ordinary, overrides_defaults=True) is not None
|
||||
pending: Final = await store.pending_delivery(issues[0].id)
|
||||
unaffected: Final = await store.pending_delivery(issues[1].id)
|
||||
await store.recover_issues(issues[0].observed_on + timedelta(days=1), (issues[0].id,))
|
||||
awaiting_evidence: Final = await store.get_issue(issues[0].id)
|
||||
assert awaiting_evidence is not None and awaiting_evidence.active
|
||||
await alerts.record_observations(
|
||||
rule,
|
||||
(
|
||||
observations[0].model_copy(
|
||||
update={"day": observations[0].day + timedelta(days=1), "observed": 120, "breached": False}
|
||||
),
|
||||
),
|
||||
)
|
||||
await store.recover_issues(issues[0].observed_on + timedelta(days=2), (issues[0].id,))
|
||||
recovered: Final = await store.get_issue(issues[0].id)
|
||||
other: Final = await store.get_issue(issues[1].id)
|
||||
assert recovered is not None and other is not None
|
||||
assert (pending is None, unaffected is not None, recovered.active, recovered.status, other.active) == (
|
||||
True,
|
||||
True,
|
||||
False,
|
||||
"resolved",
|
||||
True,
|
||||
), f"overridden delivery eligible={pending is not None}; recovered issue active={recovered.active}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_family_settings_preserve_custom_deleted_rules_and_abort_concurrent_edits(db: ManagementDatabase) -> None:
|
||||
async with _family_database(db) as database:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue