From 5237ffc663c5e590d70973371411538fbc283bbb Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 9 Oct 2026 15:59:17 -0400 Subject: [PATCH] feat(alerts): alert email waits 2 minutes in the outbox and each alert records its outcome (#5507) #5507 "Alert emails through the outbox, gathered for 2 minutes" (milestone 501). Replaces the in-memory core/alert_burst.py. - outbox.hold_alert: an alert joins each recipient's held alert email, or opens one due 2 minutes later (System 156 ruling). A Postgres advisory lock keeps two alerts at once from opening two windows for one reader. - email_outbox_events.alert_json (migration 0043) keeps the alert as it fired, so the email written when the window closes says what it said. - write_held writes it then, in the reader's zone: the usual alert email for one alert, the list for several (outbox.alert_words). - after_settle records on each AlertEvent, once every email carrying it has settled: email_sent, or email_error naming each reader it didn't reach and the server's reason. - A held email survives a restart and goes out when its window closes. - A refusal of a message's one recipient no longer repeats the address. Co-Authored-By: Claude Opus 5.5 --- docs/core/alerting.md | 27 ++- steward/core/alert_burst.py | 105 ----------- steward/core/alerts.py | 18 +- steward/core/mail.py | 2 +- steward/core/outbox.py | 105 ++++++++++- .../versions/0043_outbox_alert_words.py | 26 +++ steward/models/outbox.py | 3 + steward/templates/email/burst.html | 2 +- tests/core/test_alert_burst.py | 79 -------- tests/core/test_outbox.py | 30 +++- tests/integration/test_alert_email_outbox.py | 169 ++++++++++++++++++ tests/integration/test_email_outbox.py | 2 +- 12 files changed, 356 insertions(+), 212 deletions(-) delete mode 100644 steward/core/alert_burst.py create mode 100644 steward/migrations/versions/0043_outbox_alert_words.py delete mode 100644 tests/core/test_alert_burst.py create mode 100644 tests/integration/test_alert_email_outbox.py diff --git a/docs/core/alerting.md b/docs/core/alerting.md index cd49490..5ee6a14 100644 --- a/docs/core/alerting.md +++ b/docs/core/alerting.md @@ -162,13 +162,26 @@ Configured at `/settings/` under SMTP. Settings: Email is skipped if `smtp.host` is empty. -Alerts that arrive together go out as one email (`core/alert_burst.py`). The -first alert opens a 30-second window. When it closes, each recipient gets one -message listing every alert in the window, oldest first, in their own zone, -with a subject like "[Steward] 3 alerts: 2 firing, 1 resolved". One alert on -its own keeps its usual email. Every alert's event records whether the email -went. A burst still waiting when Steward stops is not sent; its alerts stay on -the Alerts page and in the history. +Alerts that arrive together go out as one email. The first alert opens a +2-minute window. When it closes, each recipient gets one message listing every +alert in the window, oldest first, in their own zone, with a subject like +"[Steward] 3 alerts: 2 firing, 1 resolved". One alert on its own keeps its +usual email. + +Email waits in the database (`email_outbox`, `core/outbox.py`) until the mail +server accepts it, so nothing waiting is lost when Steward restarts: the +sender picks it up at boot. The sender runs every 15 seconds and settles each +email on the server's answer: + +- **Accepted:** sent. +- **Refused** (an SMTP 5xx, such as "no such mailbox"): failed at once, with + the server's words. Trying again would only repeat the refusal. +- **Temporary** (no connection, a timeout, a 4xx): tried again after 30 s, + 1, 2, 5, 10 and then every 15 minutes. An email the server still hasn't + taken an hour after its first try is failed with the last error. + +Once every email carrying an alert has settled, that alert's event records +whether the email went, and to whom it didn't. ### Webhook diff --git a/steward/core/alert_burst.py b/steward/core/alert_burst.py deleted file mode 100644 index 0ba3547..0000000 --- a/steward/core/alert_burst.py +++ /dev/null @@ -1,105 +0,0 @@ -"""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 0475708..e8a1bf8 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -495,9 +495,10 @@ async def _dispatch_notification( event_id: str, note: str | None = None, ) -> None: - """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.""" + """Run after the transaction commits. Queues the alert's email in the + outbox, gathered for 2 minutes with any others (core.outbox.hold_alert), + posts its webhook, and records the webhook's outcome on the event; the + outbox records the email's once it settles.""" from steward.core.notifications import dispatch_notifications alert_data = { @@ -513,11 +514,14 @@ async def _dispatch_notification( "timestamp": datetime.now(timezone.utc).isoformat(), } - # 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 + # Email waits so alerts that arrive together go out as one message per + # recipient (#5465, #5507); the webhook posts this alert now. + from steward.core import outbox if app.config.get("SMTP", {}).get("host"): - alert_burst.hold(app, alert_data, event_id) + try: + await outbox.hold_alert(app, alert_data, event_id) + except Exception: + logger.exception("Alert %s: its email could not be queued", event_id) results = await dispatch_notifications({}, app.config.get("WEBHOOK", {}), alert_data) webhook_r = results.get("webhook") diff --git a/steward/core/mail.py b/steward/core/mail.py index 03511f7..7c65f7c 100644 --- a/steward/core/mail.py +++ b/steward/core/mail.py @@ -252,7 +252,7 @@ def alert_email(cfg: dict, alert: dict, to: Recipient) -> tuple[str, str, str]: 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, + (core.outbox.alert_words, #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 diff --git a/steward/core/outbox.py b/steward/core/outbox.py index 973d979..4f460d3 100644 --- a/steward/core/outbox.py +++ b/steward/core/outbox.py @@ -16,21 +16,28 @@ How a row settles: by then, it is failed with the last error. - **a test** (retry=False): settles on its first answer either way. +Alert email (#5507) is held: the first alert opens a row per recipient with +send_after ALERT_WINDOW_S out, alerts arriving before it closes join the +same rows, and the email is written when the window closes: the usual alert +email for one, a list for several. Each alert's event records the outcome +once every email carrying it has settled. + Nothing is kept in memory. A row left behind by a restart is still pending (or held, or claimed by a run that died), and the next tick picks it up. """ from __future__ import annotations import asyncio +import json import logging import smtplib from datetime import datetime, timedelta, timezone from email.message import EmailMessage from typing import TYPE_CHECKING -from sqlalchemy import or_, select, update +from sqlalchemy import insert, or_, select, text, update -from steward.models.outbox import EmailOutbox +from steward.models.outbox import EmailOutbox, email_outbox_events if TYPE_CHECKING: from quart import Quart @@ -47,6 +54,10 @@ GIVE_UP_S = 3600 # A claim older than this belongs to a run that died mid-send: back to pending. STALE_CLAIM_S = 300 BATCH = 50 +# Alerts that fire within this of the first go out as one email (System 156 +# ruling: "change the timing on 3 to be 2 minutes"). One incident's alerts +# arrive over a minute or so: the agent-offline check runs each minute. +ALERT_WINDOW_S = 120 def backoff(attempts: int) -> int: @@ -76,6 +87,9 @@ def reason(exc: BaseException) -> str: text = msg.decode(errors="replace") if isinstance(msg, bytes) else str(msg) return f"{code} {' '.join(text.split())}" if isinstance(exc, smtplib.SMTPRecipientsRefused): + # Each message has one reader, whose row already names them. + if len(exc.recipients) == 1: + return said(*next(iter(exc.recipients.values()))) return "; ".join(f"{to}: {said(c, m)}" for to, (c, m) in exc.recipients.items()) if isinstance(exc, smtplib.SMTPResponseException): return said(exc.smtp_code, exc.smtp_error) @@ -148,10 +162,61 @@ def deliver_sync(cfg: dict, messages: list[EmailMessage]) -> list[BaseException return outcomes + [smtplib.SMTPServerDisconnected("connection lost")] * (len(messages) - len(outcomes)) -async def write_held(app: "Quart", row: EmailOutbox) -> None: - """Fill in a row queued without its words: an alert burst, written when - its window closes (#5507). Raises when it can't be written.""" - raise ValueError(f"no words to send for a {row.kind} email") +async def hold_alert(app: "Quart", alert: dict, event_id: str, + now: datetime | None = None) -> int: + """Queue an alert's email: join each recipient's open alert email, or + open one that waits ALERT_WINDOW_S. Returns how many it went to. The + event must be committed first (the link refers to it).""" + from steward.core.mail import recipients + cfg = app.config.get("SMTP", {}) + now = now or datetime.now(timezone.utc) + async with app.db_sessionmaker() as db: + async with db.begin(): + # One holder at a time, so two alerts at once can't each open a + # window for the same reader. + await db.execute(text("SELECT pg_advisory_xact_lock(hashtext('steward.email_outbox.alert'))")) + to = await recipients(db, "alerts", cfg) + if not to: + return 0 + open_rows = {r.recipient: r for r in (await db.execute( + select(EmailOutbox).where(EmailOutbox.kind == "alert", EmailOutbox.state == "held", + EmailOutbox.recipient.in_([r.address for r in to])) + .with_for_update())).scalars()} + rows = [open_rows.get(r.address) or enqueue( + db, kind="alert", to=r, send_after=now + timedelta(seconds=ALERT_WINDOW_S), now=now) + for r in to] + await db.flush() + await db.execute(insert(email_outbox_events), [ + {"outbox_id": row.id, "event_id": event_id, "alert_json": json.dumps(alert)} + for row in rows]) + return len(to) + + +def alert_words(cfg: dict, alerts: list[dict], to: "Recipient") -> tuple[str, str, str]: + """(subject, text, html) for the alerts one email carries, oldest first: + the usual alert email for one, a list of them for several (#5465).""" + from steward.core.mail import alert_email, burst_email + return alert_email(cfg, alerts[0], to) if len(alerts) == 1 else burst_email(cfg, alerts, to) + + +async def write_held(app: "Quart", db, row: EmailOutbox) -> None: + """Fill in a row queued without its words: an alert email, written when + its window closes, in its reader's zone. Raises when it can't be.""" + from zoneinfo import ZoneInfo + + from steward.core.mail import Recipient + from steward.models.alerts import AlertEvent + if row.kind != "alert": + raise ValueError(f"no words to send for a {row.kind} email") + alerts = [json.loads(a) for a in (await db.execute( + select(email_outbox_events.c.alert_json) + .join(AlertEvent, AlertEvent.id == email_outbox_events.c.event_id) + .where(email_outbox_events.c.outbox_id == row.id) + .order_by(AlertEvent.transitioned_at, AlertEvent.id))).scalars()] + if not alerts: + raise ValueError("The alerts this email carried were deleted") + to = Recipient(row.recipient, ZoneInfo(row.tz), row.show_utc) + row.subject, row.text, row.html = alert_words(app.config.get("SMTP", {}), alerts, to) async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) -> dict[str, str]: @@ -172,7 +237,7 @@ async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) -> continue try: if row.subject is None: - await write_held(app, row) + await write_held(app, db, row) sendable.append((row.id, message(cfg, row.recipient, row.subject, row.text, row.html))) except Exception as exc: logger.exception("email %s: could not be written", row.id) @@ -196,8 +261,30 @@ async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) -> async def after_settle(db, rows: list[EmailOutbox]) -> None: - """What else follows from these rows' new states, in the same - transaction: an alert event's email outcome (#5507).""" + """Record on each alert event its email's outcome, once every email + carrying it has settled: sent when all were, else failed naming each + reader it didn't reach and why. Same transaction as the settling.""" + from steward.models.alerts import AlertEvent + done = [r.id for r in rows if r.kind == "alert" and r.state in ("sent", "failed")] + if not done: + return + link = email_outbox_events.c + touched = select(link.event_id).where(link.outbox_id.in_(done)) + emails: dict[str, list] = {} + for event_id, state, recipient, error in (await db.execute( + select(link.event_id, EmailOutbox.state, EmailOutbox.recipient, EmailOutbox.last_error) + .join(EmailOutbox, EmailOutbox.id == link.outbox_id) + .where(link.event_id.in_(touched)))).all(): + emails.setdefault(event_id, []).append((state, recipient, error)) + settled = {eid: es for eid, es in emails.items() + if all(state in ("sent", "failed") for state, _, _ in es)} + if not settled: + return + for event in (await db.execute( + select(AlertEvent).where(AlertEvent.id.in_(settled)))).scalars(): + failed = sorted(f"{rcpt}: {err}" for state, rcpt, err in settled[event.id] if state == "failed") + event.email_sent = not failed + event.email_error = "; ".join(failed) or None async def claim_due(app: "Quart", now: datetime, limit: int = BATCH) -> list[str]: diff --git a/steward/migrations/versions/0043_outbox_alert_words.py b/steward/migrations/versions/0043_outbox_alert_words.py new file mode 100644 index 0000000..5592185 --- /dev/null +++ b/steward/migrations/versions/0043_outbox_alert_words.py @@ -0,0 +1,26 @@ +"""The alert each queued email carries, as it was when it fired (#5507), so +an email written when its 2-minute window closes says what the alert said. + +Revision ID: 0043_outbox_alert_words +Revises: 0042_email_outbox +Create Date: 2026-10-09 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0043_outbox_alert_words" +down_revision: Union[str, None] = "0042_email_outbox" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # No row links an alert yet (#5506 shipped the table unused), so the + # empty default is only for the ALTER. + op.add_column("email_outbox_events", + sa.Column("alert_json", sa.Text, nullable=False, server_default="{}")) + + +def downgrade() -> None: + op.drop_column("email_outbox_events", "alert_json") diff --git a/steward/models/outbox.py b/steward/models/outbox.py index 3494272..318e1b8 100644 --- a/steward/models/outbox.py +++ b/steward/models/outbox.py @@ -65,4 +65,7 @@ email_outbox_events = Table( primary_key=True), Column("event_id", String(36), ForeignKey("alert_events.id", ondelete="CASCADE"), primary_key=True, index=True), + # The alert as it fired (notifications' alert dict), so the email written + # when the window closes says what it said then, whatever the rule says now. + Column("alert_json", Text, nullable=False, server_default="{}"), ) diff --git a/steward/templates/email/burst.html b/steward/templates/email/burst.html index a9bdaf6..139203c 100644 --- a/steward/templates/email/burst.html +++ b/steward/templates/email/burst.html @@ -1,4 +1,4 @@ -{# Several alerts that arrived together (core.alert_burst, #5465), as one +{# Several alerts that arrived together (core.outbox.alert_words, #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" %} diff --git a/tests/core/test_alert_burst.py b/tests/core/test_alert_burst.py deleted file mode 100644 index eb281c3..0000000 --- a/tests/core/test_alert_burst.py +++ /dev/null @@ -1,79 +0,0 @@ -"""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"]] diff --git a/tests/core/test_outbox.py b/tests/core/test_outbox.py index 092bab6..792ac77 100644 --- a/tests/core/test_outbox.py +++ b/tests/core/test_outbox.py @@ -46,7 +46,9 @@ def test_reason_is_what_the_server_said(): assert outbox.reason(smtplib.SMTPDataError(554, b"5.7.1 rejected\r\n as spam")) \ == "554 5.7.1 rejected as spam" assert outbox.reason(smtplib.SMTPRecipientsRefused({"a@x": (550, b"no such user")})) \ - == "a@x: 550 no such user" + == "550 no such user" + assert outbox.reason(smtplib.SMTPRecipientsRefused({"a@x": (550, b"no"), "b@x": (450, b"busy")})) \ + == "a@x: 550 no; b@x: 450 busy" assert outbox.reason(TimeoutError()) == "The mail server didn't answer in time" @@ -61,7 +63,7 @@ def test_a_refusal_fails_at_once_with_the_servers_words(): row = _row() outbox.settle(row, smtplib.SMTPRecipientsRefused({"ann@example.com": (550, b"no such user")}), NOW) assert row.state == "failed" and row.next_attempt_at is None - assert row.last_error == "ann@example.com: 550 no such user" + assert row.last_error == "550 no such user" def test_a_temporary_failure_waits_and_tries_again(): @@ -160,3 +162,27 @@ def test_no_connection_is_every_messages_outcome(monkeypatch): monkeypatch.setattr(notifications, "smtp_open", _open) assert outbox.deliver_sync({}, _msgs(2)) == [refused, refused] + + +CFG = {"host": "mx", "port": 587, "from": "alerts@example.com", "base_url": "https://steward.example/"} +ANN = mail.Recipient("ann@example.com", ZoneInfo("America/New_York"), False, "ann") + + +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 test_one_alert_alone_keeps_its_usual_email(): + assert outbox.alert_words(CFG, [_alert("web01")], ANN) == mail.alert_email(CFG, _alert("web01"), ANN) + + +def test_several_alerts_are_one_email_listing_each_oldest_first(): + burst = [_alert("web01"), _alert("web02"), _alert("db01", "RESOLVED", 12.0)] + subject, text, html = outbox.alert_words(CFG, burst, ANN) + assert subject == "[Steward] 3 alerts: 2 firing, 1 resolved" + 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") + assert "Fri 9 Oct 2026 08:00 EDT" in text # the reader's zone diff --git a/tests/integration/test_alert_email_outbox.py b/tests/integration/test_alert_email_outbox.py new file mode 100644 index 0000000..17a0e2d --- /dev/null +++ b/tests/integration/test_alert_email_outbox.py @@ -0,0 +1,169 @@ +"""Integration: alert email through the outbox (#5507), against Postgres. + +Alerts that fire within 2 minutes of the first join one held email per +reader; when the window closes it is written as one message listing them, +oldest first; each alert's event records the outcome once every email +carrying it has settled, naming a reader it didn't reach. Only the SMTP +conversation is replaced. +""" +from __future__ import annotations + +import asyncio +import os +import smtplib +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +def _alert(host: str, at: datetime) -> dict: + return {"rule_name": f"Agent offline: {host}", "state": "FIRING", "metric": "seconds_since_seen", + "value": 400.0, "operator": ">", "threshold": 180.0, "resource": host, + "source_module": "host_agent", "note": None, "timestamp": at.isoformat()} + + +async def _events(app, tag: str, times: list[datetime]) -> list[str]: + from steward.alerts import service as alert_svc + from steward.core.alerts import record_metrics + from steward.models.alerts import AlertEvent + from steward.models.metrics import PluginMetric + from steward.models.users import User, UserRole + res = f"box-{tag}" + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"ob-{tag}", email=f"ob-{tag}@test", password_hash="x", + role=UserRole.admin) + s.add(user) + await record_metrics(s, [PluginMetric(source_module="host_agent", resource_name=res, + metric_name="cpu_pct", value=1.0)]) + await s.flush() + fields = await alert_svc.rule_fields( + s, name=f"hot {tag}", source_module="host_agent", resource_name=res, + metric_name="cpu_pct", operator=">", threshold=90) + rule = await alert_svc.add_rule(s, fields, created_by=user.id) + events = [AlertEvent(rule_id=rule.id, from_state="inactive", to_state="firing", + metric_value=95.0, transitioned_at=at) for at in times] + s.add_all(events) + await s.flush() + return [e.id for e in events] + + +async def _emails_for(app, event_ids: list[str]): + from sqlalchemy import select + from steward.models.outbox import EmailOutbox, email_outbox_events as link + async with app.db_sessionmaker() as s: + rows = (await s.execute( + select(EmailOutbox).where(EmailOutbox.id.in_( + select(link.c.outbox_id).where(link.c.event_id.in_(event_ids)))))).scalars().all() + links = (await s.execute(select(link.c.outbox_id, link.c.event_id) + .where(link.c.event_id.in_(event_ids)))).all() + return rows, links + + +async def _event(app, event_id): + from steward.models.alerts import AlertEvent + async with app.db_sessionmaker() as s: + return await s.get(AlertEvent, event_id) + + +def test_alerts_in_one_window_are_one_email_and_each_event_records_it(app, monkeypatch): + from steward.core import outbox + + tag = uuid.uuid4().hex[:8] + extra = f"oncall-{tag}@test" + app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test", + "recipients": [extra], "tz": "America/New_York"} + t0 = datetime(2026, 10, 9, 12, 0, tzinfo=timezone.utc) + sent = [] + + def fake_deliver(cfg, messages): + sent.extend(messages) + return [smtplib.SMTPRecipientsRefused({m["To"]: (550, b"no such user")}) + if m["To"] == extra and "web03" in m["Subject"] else None for m in messages] + + monkeypatch.setattr(outbox, "deliver_sync", fake_deliver) + + async def _go(): + ev1, ev2, ev3 = await _events(app, tag, [t0, t0 + timedelta(seconds=30), + t0 + timedelta(minutes=5)]) + assert await outbox.hold_alert(app, _alert("web01", t0), ev1, now=t0) >= 1 + await outbox.hold_alert(app, _alert("web02", t0), ev2, now=t0 + timedelta(seconds=30)) + + rows, links = await _emails_for(app, [ev1, ev2]) + mine = [r for r in rows if r.recipient == extra] + assert len(mine) == 1, "both alerts joined the one held email" + (held,) = mine + assert held.state == "held" and held.subject is None + assert held.next_attempt_at == t0 + timedelta(minutes=2) + assert {e for o, e in links if o == held.id} == {ev1, ev2} + # Each reader has exactly one email carrying the pair. + assert len({r.recipient for r in rows}) == len(rows) + + # The window closes: written, sent, and recorded on both events. + await outbox.attempt(app, [r.id for r in rows], now=t0 + timedelta(minutes=2)) + (msg,) = [m for m in sent if m["To"] == extra] + assert msg["Subject"] == "[Steward] 2 alerts: 2 firing" + body = msg.get_body(("plain",)).get_content() + assert body.index("web01") < body.index("web02") + assert "Fri 9 Oct 2026 08:00 EDT" in body # the Also-send-to zone + for ev in (ev1, ev2): + event = await _event(app, ev) + assert (event.email_sent, event.email_error) == (True, None) + + # A later alert is its own email; refused for one reader, it is + # recorded as not sent, naming who and why. + await outbox.hold_alert(app, _alert("web03", t0), ev3, now=t0 + timedelta(minutes=5)) + rows, _ = await _emails_for(app, [ev3]) + assert all(r.state == "held" for r in rows) + await outbox.attempt(app, [r.id for r in rows], now=t0 + timedelta(minutes=7)) + (refused,) = [m for m in sent if m["To"] == extra and "web03" in m["Subject"]] + assert refused["Subject"].startswith("[Steward]") # one alert: its usual email + event = await _event(app, ev3) + assert event.email_sent is False + assert event.email_error == f"{extra}: 550 no such user" + + asyncio.run(_go()) + + +def test_an_email_whose_alerts_were_deleted_fails_and_says_so(app, monkeypatch): + from sqlalchemy import delete + from steward.core import outbox + from steward.models.alerts import AlertEvent + + tag = uuid.uuid4().hex[:8] + extra = f"gone-{tag}@test" + app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test", "recipients": [extra]} + monkeypatch.setattr(outbox, "deliver_sync", lambda cfg, msgs: [None] * len(msgs)) + t0 = datetime.now(timezone.utc) + + async def _go(): + (ev,) = await _events(app, tag, [t0]) + await outbox.hold_alert(app, _alert("web09", t0), ev, now=t0) + rows, _ = await _emails_for(app, [ev]) + (row,) = [r for r in rows if r.recipient == extra] + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(delete(AlertEvent).where(AlertEvent.id == ev)) + states = await outbox.attempt(app, [row.id]) + return states[row.id], row.id + + state, rid = asyncio.run(_go()) + assert state == "failed" + + async def _reason(): + from steward.models.outbox import EmailOutbox + async with app.db_sessionmaker() as s: + return (await s.get(EmailOutbox, rid)).last_error + + assert asyncio.run(_reason()) == "The alerts this email carried were deleted" diff --git a/tests/integration/test_email_outbox.py b/tests/integration/test_email_outbox.py index 9e08c44..649bea2 100644 --- a/tests/integration/test_email_outbox.py +++ b/tests/integration/test_email_outbox.py @@ -72,7 +72,7 @@ def test_the_sender_settles_each_row_on_what_the_server_said(app, monkeypatch): by = {k: rows[to[k]] for k in to} assert by["ok"].state == "sent" and by["ok"].sent_at is not None and by["ok"].attempts == 1 assert by["refused"].state == "failed" - assert by["refused"].last_error == f"{to['refused']}: 550 no such user" + assert by["refused"].last_error == "550 no such user" later = by["later"] assert later.state == "pending" and later.attempts == 1 and later.last_error == "451 greylisted" assert later.next_attempt_at > later.first_attempt_at