scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

refactor(alerts): remove unused incident notification fallback

Browse source
This commit is contained in:
moe-berri 2026-11-07 12:10:49 +08:00
parent 73c4a52d4e
commit e7ff81b297
6 changed files with 54 additions and 115 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
litellm/integrations/SlackAlerting/usage_alerts.py
View file

@ -13,7 +13,7 @@ from pydantic import TypeAdapter
from litellm.types.integrations.slack_alerting import AlertType
from litellm.types.proxy.issues import IssuePendingDelivery
from litellm.types.proxy.usage_alerts import UsageAlert, UsageAlertChannel, UsageAlertRule
from litellm.types.proxy.usage_alerts import UsageAlert, UsageAlertChannel
if TYPE_CHECKING:
from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
@ -60,16 +60,6 @@ class UsageAlertDelivery:
def __init__(self, slack: "SlackAlerting") -> None:
self._slack = slack
async def send(self, alert: UsageAlert, rule: UsageAlertRule, channel: UsageAlertChannel) -> bool:
url: Final = activity_url(alert, os.getenv("PROXY_BASE_URL"))
return await self._send(
channel,
rule.email_recipients,
incident_message(alert, None),
f"LiteLLM performance alert: {alert.rule_name}",
(("View activity", url),),
)
async def send_issue(self, pending: IssuePendingDelivery) -> bool:
path: Final = "/alerts?" + urlencode({"issue": pending.issue.id})
issue_url: Final = activity_url(

56
litellm/proxy/spend_tracking/usage_alert_job.py
View file

@ -9,7 +9,6 @@ from itertools import chain, groupby, takewhile
from typing import (
TYPE_CHECKING,
Final,
Protocol,
cast, # noqa: TID251 # generated Prisma delegate supplies the repository protocol
)
from uuid import uuid4
@ -32,10 +31,7 @@ from litellm.repositories.usage_alerts import UsageAlertsRepository
from litellm.types.integrations.slack_alerting import AlertType
from litellm.types.proxy.usage_alert_defaults import USAGE_ALERT_SCOPES, default_usage_rules
from litellm.types.proxy.usage_alerts import (
UsageAlert,
UsageAlertChannel,
UsageAlertEvaluationStatus,
UsageAlertNotification,
UsageAlertObservation,
UsageAlertProgress,
UsageAlertRule,
@ -115,27 +111,21 @@ def _observations(
return _Observations(evaluated, tuple(changed))
class IncidentDelivery(Protocol):
async def send(self, alert: UsageAlert, rule: UsageAlertRule, channel: UsageAlertChannel) -> bool: ...
class UsageAlertJob:
def __init__(
self,
store: UsageAlertsRepository,
metrics: UsageAlertMetricsRepository,
health: UsageAlertHealthRepository,
delivery: IncidentDelivery,
defaults: tuple[UsageAlertRuleInput, ...] | None = None,
*,
deprecation_projection: Callable[[datetime], Awaitable[None]] | None = None,
performance_projection: PerformanceIssueProjector | None = None,
performance_projection: PerformanceIssueProjector,
work_budget_seconds: float = 60,
) -> None:
self._store = store
self._metrics = metrics
self._health = health
self._delivery = delivery
self._defaults = default_usage_rules() if defaults is None else defaults
self._defaults_registered = False
self._deprecation_projection = deprecation_projection
@ -160,7 +150,6 @@ class UsageAlertJob:
UsageAlertsRepository(database),
metrics,
UsageAlertHealthRepository(database),
delivery,
default_usage_rules(
("slack",)
if "slack" in configured_destinations(slack) and AlertType.usage_alerts in slack.alert_types
@ -224,9 +213,7 @@ class UsageAlertJob:
await self._evaluate(
target, asyncio.get_running_loop().time() + max(0, deadline - asyncio.get_running_loop().time()) * 0.65
)
if self._performance_projection is None:
await self._notify(deadline)
elif (work := await self._store.pending_projection_work()) is None:
if (work := await self._store.pending_projection_work()) is None:
await self._performance_projection.notify(deadline)
elif await self._performance_projection.run_pending(now, work, self._store, deadline):
await self._store.finish_projection_work(work)
@ -431,42 +418,3 @@ class UsageAlertJob:
if observations.changed:
await self._store.record_observations(rule, observations.changed)
return observations.evaluated
async def _deliver_channel(
self, alert: UsageAlert, rule: UsageAlertRule, channel: UsageAlertChannel, deadline: float
) -> str | None:
try:
delivered: Final = await asyncio.wait_for(
self._delivery.send(alert, rule, channel),
timeout=min(30, max(0, deadline - asyncio.get_running_loop().time() - 1)),
)
if not delivered:
return "transport_not_configured"
await self._store.mark_notified(alert.id, channel)
except Exception: # noqa: BLE001 # retry only failed channels; persisted acknowledgements survive restarts
verbose_proxy_logger.warning("Usage alert delivery failed for incident %s via %s", alert.id, channel)
return "delivery_failed"
return None
async def _notify(self, deadline: float) -> None:
while asyncio.get_running_loop().time() + 2 < deadline and (
page := await self._store.pending_notifications(limit=10)
):
for pending in page:
if asyncio.get_running_loop().time() + 2 >= deadline:
return
await self._notify_incident(pending, deadline)
async def _notify_incident(self, pending: UsageAlertNotification, deadline: float) -> None:
current: Final = await self._store.get_rule(pending.rule.id)
if current is None or not current.enabled or current.updated_at != pending.rule.updated_at:
return
failures: Final = tuple(
[
await self._deliver_channel(pending.incident, pending.rule, channel, deadline)
for channel in pending.pending_channels
]
)
if any(failures):
error: Final = "delivery_failed" if "delivery_failed" in failures else "transport_not_configured"
await self._store.defer_notification(pending.incident.id, error)

14
litellm/repositories/usage_alerts.py
View file

@ -1002,17 +1002,3 @@ class UsageAlertsRepository:
for rule_id, group in groupby(rows, lambda row: row.rule_id)
}
)
async def defer_notification(self, incident_id: str, error: str) -> None:
await self.database.query_raw(
"""UPDATE "LiteLLM_UsageAlertIncident" SET updated_at=CURRENT_TIMESTAMP,
record=record || jsonb_build_object(
'notification_error',$2::text,
'notification_attempts',COALESCE((record->>'notification_attempts')::int,0)+1,
'last_notification_attempt',CURRENT_TIMESTAMP,
'next_notification_attempt',CURRENT_TIMESTAMP + make_interval(secs =>
LEAST(86400,3600*power(2,LEAST(5,COALESCE((record->>'notification_attempts')::int,0)))::int))
) WHERE id=$1 RETURNING id""",
incident_id,
error,
)

4
tests/integration/database/test_performance_issues.py
View file

@ -417,7 +417,9 @@ async def test_late_custom_key_rule_only_overrides_that_keys_default_signal(db:
)
job: Final = UsageAlertJob.from_prisma(prisma, SlackAlerting(alerting=[], alert_types=[]))
# A rule can change after the daily observations, before Issue projection.
await UsageAlertJob(job._store, job._metrics, job._health, job._delivery)._run(_NOW)
await job._store.ensure_default_rules(job._defaults)
await job._store.ensure_projection_work(_DAY)
await job._evaluate(_DAY, asyncio.get_running_loop().time() + 60)
issues: Final = IssuesRepository.from_prisma(prisma)
original: Final = await issues.active_signals(_DAY, scope_type="key", metric="spend")
assert len(original) == 2

68
tests/unit/integrations/SlackAlerting/test_usage_alerts.py
View file

@ -45,6 +45,31 @@ _ALERT: Final = UsageAlert(
)
_ISSUE: Final = Issue(
id="issue-example",
cohort_key="cohort-example",
kind="performance",
family="traffic",
title="Traffic increased",
cohort=IssueCohort(scope_type="key", scope_id="reporting-key"),
observed_on=_ALERT.day,
observed_at=_NOW,
range_start=_ALERT.baseline_start,
range_end=_ALERT.day,
grouping_reason="same_resource",
status="open",
first_seen=_NOW,
last_seen=_NOW,
updated_at=_NOW,
)
_PENDING: Final = IssuePendingDelivery(
issue=_ISSUE,
rule=_RULE,
alert=_ALERT,
delivery=IssueDelivery(id="delivery-example", issue_id=_ISSUE.id, channel="slack"),
)
def test_activity_link_retains_dashboard_mount_and_entity_query() -> None:
assert activity_url(_ALERT, "https://gateway.example.com/") == (
"https://gateway.example.com/ui/api-keys?key=reporting-key&detail_tab=activity"
@ -76,10 +101,10 @@ async def test_slack_ack_requires_successful_http_response(status: int, monkeypa
delivery: Final = UsageAlertDelivery(slack)
try:
if status == 200:
assert await delivery.send(_ALERT, _RULE, "slack") is True
assert await delivery.send_issue(_PENDING) is True
else:
with pytest.raises(httpx.HTTPStatusError):
await delivery.send(_ALERT, _RULE, "slack")
await delivery.send_issue(_PENDING)
assert len(requests) == 1
assert (
b"https://gateway.example.com/ui/api-keys?key=reporting-key&amp;detail_tab=activity" in requests[0].content
@ -99,35 +124,12 @@ async def test_issue_delivery_links_to_investigation_and_preserves_the_bound_sou
requests.append(request)
return httpx.Response(200)
issue: Final = Issue(
id="issue-example",
cohort_key="cohort-example",
kind="performance",
family="traffic",
title="Traffic increased",
cohort=IssueCohort(scope_type="key", scope_id="reporting-key"),
observed_on=_ALERT.day,
observed_at=_NOW,
range_start=_ALERT.baseline_start,
range_end=_ALERT.day,
grouping_reason="same_resource",
status="open",
first_seen=_NOW,
last_seen=_NOW,
updated_at=_NOW,
)
handler: Final = AsyncHTTPHandler(transport=httpx.MockTransport(transport))
slack: Final = SlackAlerting(
alerting=["slack"], default_webhook_url="https://hooks.example.com/test", async_http_handler=handler
)
pending: Final = IssuePendingDelivery(
issue=issue,
rule=_RULE,
alert=_ALERT,
delivery=IssueDelivery(id="delivery-example", issue_id=issue.id, channel="slack"),
)
try:
assert await UsageAlertDelivery(slack).send_issue(pending)
assert await UsageAlertDelivery(slack).send_issue(_PENDING)
assert len(requests) == 1
assert b"https://gateway.example.com/ui/alerts?issue=issue-example" in requests[0].content
assert (
@ -151,12 +153,14 @@ async def test_slack_source_labels_cannot_inject_mentions_or_links() -> None:
alerting=["slack"], default_webhook_url="https://hooks.example.com/test", async_http_handler=handler
)
try:
assert await UsageAlertDelivery(slack).send(
_ALERT.model_copy(
update={"rule_name": "<!channel> & <https://example.com|fake>", "scope_id": "<@someone>"}
),
_RULE,
"slack",
assert await UsageAlertDelivery(slack).send_issue(
_PENDING.model_copy(
update={
"alert": _ALERT.model_copy(
update={"rule_name": "<!channel> & <https://example.com|fake>", "scope_id": "<@someone>"}
)
}
)
)
assert b"<!channel>" not in requests[0].content
assert b"&lt;!channel&gt; &amp; &lt;https://example.com|fake&gt;" in requests[0].content

15
tests/unit/proxy/spend_tracking/test_daily_usage_alerts.py
View file

@ -10,13 +10,17 @@ from typing import Final
import pytest
from litellm.proxy.spend_tracking.daily_usage_alerts import DailyComparison, compare_daily_usage
from litellm.proxy.spend_tracking.performance_issues import PerformanceIssueProjector
from litellm.proxy.spend_tracking.usage_alert_job import UsageAlertJob, completed_day
from litellm.repositories.analytics_budget import AnalyticsTransaction, BoundedAnalyticsDatabase
from litellm.repositories.issues import IssuesRepository
from litellm.repositories.provider_latency import ProviderLatencyRepository
from litellm.repositories.usage_alert_health import UsageAlertHealthRepository
from litellm.repositories.usage_alert_metrics import DailyAlertMetrics, UsageAlertMetricsRepository
from litellm.repositories.usage_alerts import UsageAlertsRepository
from litellm.types.proxy.usage_alert_defaults import default_usage_rule_id, default_usage_rules
from litellm.types.proxy.usage_alerts import UsageAlert, UsageAlertChannel, UsageAlertRule, UsageAlertRuleInput
from litellm.types.proxy.issues import IssuePendingDelivery
from litellm.types.proxy.usage_alerts import UsageAlertRule, UsageAlertRuleInput
_DAY: Final = date(2026, 10, 20)
@ -89,7 +93,7 @@ class _BudgetDatabase:
class _UnexpectedDelivery:
async def send(self, alert: UsageAlert, rule: UsageAlertRule, channel: UsageAlertChannel) -> bool:
async def send_issue(self, pending: IssuePendingDelivery) -> bool:
raise AssertionError("stalled evaluation must not deliver alerts")
@ -108,8 +112,13 @@ def test_job_budget_includes_query_cancellation_and_lease_cleanup() -> None:
UsageAlertsRepository(database),
UsageAlertMetricsRepository(database),
UsageAlertHealthRepository(database),
_UnexpectedDelivery(),
defaults=(),
performance_projection=PerformanceIssueProjector(
IssuesRepository(database),
UsageAlertMetricsRepository(database),
ProviderLatencyRepository(database),
_UnexpectedDelivery(),
),
deprecation_projection=projection,
work_budget_seconds=60,
)