From 997aa50939f55a97aac21a8d1af91801fd272728 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 9 Oct 2026 08:29:52 -0400 Subject: [PATCH] feat(alerts): alerts that arrive together go out as one email (#5465) #5465 "One email per burst of alerts". When a switch reboots, every host's agent-offline rule fires at once and each account got one email per rule. Alert email now waits 30 s from the first alert: each recipient then gets one message listing every alert in the window, oldest first, in their own zone ("[Steward] 3 alerts: 2 firing, 1 resolved"). One alert alone keeps its usual email. Webhooks still post per alert, at once. Every alert's event records the email's outcome. A burst held when Steward stops is not sent; the alerts stay on the record and on the Alerts page. Co-Authored-By: Claude Opus 5.5 --- steward/core/alert_burst.py | 105 +++++++++++++++++++++++++++++ steward/core/alerts.py | 36 +++++----- steward/core/mail.py | 15 +++++ steward/templates/email/burst.html | 31 +++++++++ steward/templates/email/burst.txt | 18 +++++ tests/core/test_alert_burst.py | 79 ++++++++++++++++++++++ 6 files changed, 264 insertions(+), 20 deletions(-) create mode 100644 steward/core/alert_burst.py create mode 100644 steward/templates/email/burst.html create mode 100644 steward/templates/email/burst.txt create mode 100644 tests/core/test_alert_burst.py diff --git a/steward/core/alert_burst.py b/steward/core/alert_burst.py new file mode 100644 index 0000000..0ba3547 --- /dev/null +++ b/steward/core/alert_burst.py @@ -0,0 +1,105 @@ +"""Alert emails gathered into one message per burst (#5465). + +When the network drops or a switch reboots, every host's agent-offline rule +fires together, and one email per rule per account turns one incident into a +dozen messages. So alert email waits: the first alert opens a window of +BURST_WINDOW_S, and when it closes each recipient gets one email for every +alert that arrived in it. An alert on its own still reads as it always has. + +Webhooks are not held here: each alert posts on its own, at once, since an +integration expects one alert per post (core.alerts._dispatch_notification). +Every alert's event still records the email's outcome. + +A burst held when Steward stops is not sent; the alerts stay on the record +and on the Alerts page. +""" +from __future__ import annotations + +import asyncio +import logging +from typing import TYPE_CHECKING + +from sqlalchemy import select + +if TYPE_CHECKING: + from quart import Quart + +logger = logging.getLogger(__name__) + +# Long enough to gather what one incident sets off (the agent-offline check +# runs each minute, agents report every 10-60 s), short enough that a lone +# alert is still news when it lands. +BURST_WINDOW_S = 30.0 + +_held: list[tuple[dict, str]] = [] # (alert data, event id), oldest first +_flusher: asyncio.Task | None = None + + +def hold(app: "Quart", alert: dict, event_id: str) -> None: + """Add an alert to the email burst, opening a window if none is open.""" + global _flusher + _held.append((alert, event_id)) + if _flusher is None or _flusher.done(): + _flusher = asyncio.create_task(_after_window(app)) + + +async def _after_window(app: "Quart") -> None: + """Wait out the window, then send what it gathered. Alerts that arrive + while the burst is being sent open the next window.""" + global _flusher + try: + await asyncio.sleep(BURST_WINDOW_S) + await flush(app) + except Exception: + logger.exception("Alert email burst failed") + finally: + _flusher = None + if _held: + _flusher = asyncio.create_task(_after_window(app)) + + +async def flush(app: "Quart") -> None: + """Send every held alert now: one email per recipient, then record the + outcome on each alert's event.""" + from steward.core.mail import recipients + items = _held[:] + _held.clear() + if not items: + return + smtp = app.config.get("SMTP", {}) + if not smtp.get("host"): + return + async with app.db_sessionmaker() as db: + to = await recipients(db, "alerts", smtp) + result = await email_burst(smtp, [alert for alert, _ in items], to) + if result is not None: + await _record(app, [event_id for _, event_id in items], result) + + +async def email_burst(cfg: dict, alerts: list[dict], to: list) -> dict | None: + """Email `alerts` to each recipient as one message, in their own zone: + the usual alert email for one alert, a list for several. Returns + {"sent", "error"}, or None when there is nobody to send to.""" + from steward.core.mail import alert_email, burst_email, message, send + if not to or not alerts: + return None + try: + messages = [] + for r in to: + parts = alert_email(cfg, alerts[0], r) if len(alerts) == 1 else burst_email(cfg, alerts, r) + messages.append(message(cfg, r.address, *parts)) + await send(cfg, messages) + return {"sent": True, "error": None} + except Exception as exc: + logger.error("Alert email failed: %s", exc) + return {"sent": False, "error": str(exc)} + + +async def _record(app: "Quart", event_ids: list[str], result: dict) -> None: + from steward.models.alerts import AlertEvent + async with app.db_sessionmaker() as db: + async with db.begin(): + for event in (await db.execute( + select(AlertEvent).where(AlertEvent.id.in_(event_ids)))).scalars(): + event.email_sent = result["sent"] + event.email_error = result.get("error") diff --git a/steward/core/alerts.py b/steward/core/alerts.py index 2e01dc1..0475708 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -495,7 +495,9 @@ async def _dispatch_notification( event_id: str, note: str | None = None, ) -> None: - """Run after the transaction commits. Sends notifications and updates the event row.""" + """Run after the transaction commits. Holds the alert's email for its + burst (core.alert_burst), posts its webhook, and records the webhook's + outcome on the event; the burst records the email's.""" from steward.core.notifications import dispatch_notifications alert_data = { @@ -511,29 +513,23 @@ async def _dispatch_notification( "timestamp": datetime.now(timezone.utc).isoformat(), } - from steward.core.mail import recipients - smtp_cfg = app.config.get("SMTP", {}) - webhook_cfg = app.config.get("WEBHOOK", {}) - async with app.db_sessionmaker() as db: - to = await recipients(db, "alerts", smtp_cfg) if smtp_cfg.get("host") else [] - results = await dispatch_notifications(smtp_cfg, webhook_cfg, alert_data, to) + # Email waits a moment so alerts that arrive together go out as one + # message per recipient (#5465); the webhook posts this alert now. + from steward.core import alert_burst + if app.config.get("SMTP", {}).get("host"): + alert_burst.hold(app, alert_data, event_id) + results = await dispatch_notifications({}, app.config.get("WEBHOOK", {}), alert_data) - # Update event row with delivery results in a new transaction + webhook_r = results.get("webhook") + if not webhook_r: + return async with app.db_sessionmaker() as db: async with db.begin(): - event_result = await db.execute( - select(AlertEvent).where(AlertEvent.id == event_id) - ) - event = event_result.scalar_one_or_none() + event = (await db.execute( + select(AlertEvent).where(AlertEvent.id == event_id))).scalar_one_or_none() if event: - email_r = results.get("email") - if email_r: - event.email_sent = email_r["sent"] - event.email_error = email_r.get("error") - webhook_r = results.get("webhook") - if webhook_r: - event.webhook_sent = webhook_r["sent"] - event.webhook_error = webhook_r.get("error") + event.webhook_sent = webhook_r["sent"] + event.webhook_error = webhook_r.get("error") async def _run_ansible_action( diff --git a/steward/core/mail.py b/steward/core/mail.py index ee8169d..0a022e3 100644 --- a/steward/core/mail.py +++ b/steward/core/mail.py @@ -247,6 +247,21 @@ def alert_email(cfg: dict, alert: dict, to: Recipient) -> tuple[str, str, str]: return facts["subject"], text, html +def burst_email(cfg: dict, alerts: list[dict], to: Recipient) -> tuple[str, str, str]: + """(subject, text, html) for several alerts that arrived together + (core.alert_burst, #5465): how many and what they came to in the subject, + then each alert's headline, rule and time, oldest first, in the + recipient's zone.""" + from collections import Counter + facts = [alert_facts(a, to.tz, to.show_utc) for a in alerts] + counts = Counter(f["state"] for f in facts) + said = ", ".join(f"{n} {state.lower()}" for state, n in counts.most_common()) + base = (cfg.get("base_url") or "").rstrip("/") + text, html = _render("burst", alerts=facts, said=said, + link=f"{base}/alerts/" if base else "") + return f"[Steward] {len(facts)} alerts: {said}", text, html + + def report_email(cfg: dict, data: dict, to: Recipient) -> tuple[str, str, str]: """(subject, text, html) for the weekly report built by reports.build_report_data, its dates in the recipient's zone.""" diff --git a/steward/templates/email/burst.html b/steward/templates/email/burst.html new file mode 100644 index 0000000..a9bdaf6 --- /dev/null +++ b/steward/templates/email/burst.html @@ -0,0 +1,31 @@ +{# Several alerts that arrived together (core.alert_burst, #5465), as one + email's HTML part: alerts = [mail.alert_facts()], oldest first; said = how + many of each ("3 firing, 1 resolved"); link = the alerts page or "". #} +{% extends "_layout.html" %} +{% from "_parts.html" import badge, button with context %} +{% block title %}{{ alerts|length }} alerts: {{ said }}{% endblock %} +{% block body %} +{{ alerts|length }} alerts changed together: {{ said }}. +Gathered into one email because they arrived within moments of each other. + + +{% for a in alerts %} + +{% endfor %} +
+
{{ badge(a.tone, a.state) }}
+
{{ a.headline }}
+
{{ a.rule }} · fires when {{ a.condition }} · {{ a.when }}
+{% if a.note %}
{{ a.note }}
{% endif %} +
+ +{% if link %} +{{ button(link, "View in Steward") }} +{% else %} + +{% endif %} +{% endblock %} +{% block footer %} +Steward sends this to everyone on its email list when alert rules change state. +Who gets it is set in Settings › Notifications. +{% endblock %} diff --git a/steward/templates/email/burst.txt b/steward/templates/email/burst.txt new file mode 100644 index 0000000..2438ecb --- /dev/null +++ b/steward/templates/email/burst.txt @@ -0,0 +1,18 @@ +{# Several alerts that arrived together, plain text; the same facts as burst.html. #} +{{ alerts|length }} alerts changed together: {{ said }}. +{% for a in alerts %} + +{{ a.state }}: {{ a.headline }} + {{ a.rule }} · fires when {{ a.condition }} · {{ a.when }} +{%- if a.note %} + {{ a.note }} +{%- endif %} +{% endfor %} +{% if link %} + +View in Steward: {{ link }} +{% endif %} + +-- +Steward sends this to everyone on its email list when alert rules change +state. Who gets it is set in Settings › Notifications. diff --git a/tests/core/test_alert_burst.py b/tests/core/test_alert_burst.py new file mode 100644 index 0000000..eb281c3 --- /dev/null +++ b/tests/core/test_alert_burst.py @@ -0,0 +1,79 @@ +"""Alert emails gathered into one message per burst (#5465): alerts that +arrive within the window go out as one email per recipient listing each of +them; one alert alone keeps its usual email; a failed send says so for every +alert in the burst; an alert arriving after a window closes opens the next.""" +import asyncio +from email.message import EmailMessage +from zoneinfo import ZoneInfo + +from steward.core import alert_burst, mail +from steward.core import notifications as n + +NY = ZoneInfo("America/New_York") +ANN = mail.Recipient("ann@example.com", NY, False, "ann") +BOB = mail.Recipient("bob@example.com", NY, False, "bob") +CFG = {"host": "mx", "port": 587, "from": "alerts@example.com", "base_url": "https://steward.example/"} + + +def _alert(host: str, state: str = "FIRING", value: float = 400.0) -> dict: + return {"rule_name": f"Agent offline: {host}", "state": state, "metric": "seconds_since_seen", + "value": value, "operator": ">", "threshold": 180.0, "resource": host, + "source_module": "host_agent", "note": None, "timestamp": "2026-10-09T12:00:00+00:00"} + + +def _capture(monkeypatch) -> list[EmailMessage]: + out: list[EmailMessage] = [] + monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msgs: out.extend(msgs)) + return out + + +async def test_one_alert_alone_keeps_its_usual_email(monkeypatch): + sent = _capture(monkeypatch) + assert await alert_burst.email_burst(CFG, [_alert("web01")], [ANN]) == {"sent": True, "error": None} + assert len(sent) == 1 + assert sent[0]["Subject"] == mail.alert_email(CFG, _alert("web01"), ANN)[0] + + +async def test_a_burst_is_one_email_per_recipient_listing_each_alert(monkeypatch): + sent = _capture(monkeypatch) + burst = [_alert("web01"), _alert("web02"), _alert("db01", "RESOLVED", 12.0)] + assert (await alert_burst.email_burst(CFG, burst, [ANN, BOB]))["sent"] + assert [m["To"] for m in sent] == ["ann@example.com", "bob@example.com"] + assert sent[0]["Subject"] == "[Steward] 3 alerts: 2 firing, 1 resolved" + text = sent[0].get_body(("plain",)).get_content() + html = sent[0].get_body(("html",)).get_content() + for host in ("web01", "web02", "db01"): + assert f"Agent offline: {host}" in text and f"Agent offline: {host}" in html + assert text.index("web01") < text.index("web02") < text.index("db01") # oldest first + assert "Fri 9 Oct 2026 08:00 EDT" in text # the reader's zone + + +async def test_a_failed_send_is_said_for_the_burst(monkeypatch): + def boom(cfg, msgs): + raise OSError("connection refused") + monkeypatch.setattr(n, "_smtp_send_sync", boom) + got = await alert_burst.email_burst(CFG, [_alert("a"), _alert("b")], [ANN]) + assert got == {"sent": False, "error": "connection refused"} + + +async def test_nobody_to_send_to_sends_nothing(monkeypatch): + sent = _capture(monkeypatch) + assert await alert_burst.email_burst(CFG, [_alert("a")], []) is None and sent == [] + + +async def test_alerts_within_the_window_are_flushed_together(monkeypatch): + flushed: list[list[str]] = [] + + async def fake_flush(app): + flushed.append([a["resource"] for a, _ in alert_burst._held]) + alert_burst._held.clear() + + monkeypatch.setattr(alert_burst, "BURST_WINDOW_S", 0.05) + monkeypatch.setattr(alert_burst, "flush", fake_flush) + alert_burst._held.clear() + for host in ("a", "b", "c"): + alert_burst.hold(None, _alert(host), f"ev-{host}") + await asyncio.sleep(0.15) + alert_burst.hold(None, _alert("d"), "ev-d") # after the window: the next burst + await asyncio.sleep(0.15) + assert flushed == [["a", "b", "c"], ["d"]]