diff --git a/litellm/integrations/SlackAlerting/usage_alerts.py b/litellm/integrations/SlackAlerting/usage_alerts.py index 24fbdd5a2d5..3ec02b18d39 100644 --- a/litellm/integrations/SlackAlerting/usage_alerts.py +++ b/litellm/integrations/SlackAlerting/usage_alerts.py @@ -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( diff --git a/litellm/proxy/spend_tracking/usage_alert_job.py b/litellm/proxy/spend_tracking/usage_alert_job.py index d0b403c4643..d26b463ea70 100644 --- a/litellm/proxy/spend_tracking/usage_alert_job.py +++ b/litellm/proxy/spend_tracking/usage_alert_job.py @@ -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) diff --git a/litellm/repositories/usage_alerts.py b/litellm/repositories/usage_alerts.py index 07042c067d9..d86a2552f3a 100644 --- a/litellm/repositories/usage_alerts.py +++ b/litellm/repositories/usage_alerts.py @@ -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, - ) diff --git a/tests/integration/database/test_performance_issues.py b/tests/integration/database/test_performance_issues.py index ed810840f4d..1ba63368a37 100644 --- a/tests/integration/database/test_performance_issues.py +++ b/tests/integration/database/test_performance_issues.py @@ -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 diff --git a/tests/unit/integrations/SlackAlerting/test_usage_alerts.py b/tests/unit/integrations/SlackAlerting/test_usage_alerts.py index cc8f608bfad..985fbff0328 100644 --- a/tests/unit/integrations/SlackAlerting/test_usage_alerts.py +++ b/tests/unit/integrations/SlackAlerting/test_usage_alerts.py @@ -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&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": " & ", "scope_id": "<@someone>"} - ), - _RULE, - "slack", + assert await UsageAlertDelivery(slack).send_issue( + _PENDING.model_copy( + update={ + "alert": _ALERT.model_copy( + update={"rule_name": " & ", "scope_id": "<@someone>"} + ) + } + ) ) assert b"" not in requests[0].content assert b"<!channel> & <https://example.com|fake>" in requests[0].content diff --git a/tests/unit/proxy/spend_tracking/test_daily_usage_alerts.py b/tests/unit/proxy/spend_tracking/test_daily_usage_alerts.py index 41fab29ef44..a7f2ef5c2e3 100644 --- a/tests/unit/proxy/spend_tracking/test_daily_usage_alerts.py +++ b/tests/unit/proxy/spend_tracking/test_daily_usage_alerts.py @@ -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, )