diff --git a/backend/fluksio/alembic/versions/b9e4d2a7c1f6_alert_sent.py b/backend/fluksio/alembic/versions/b9e4d2a7c1f6_alert_sent.py new file mode 100644 index 0000000..367bd58 --- /dev/null +++ b/backend/fluksio/alembic/versions/b9e4d2a7c1f6_alert_sent.py @@ -0,0 +1,35 @@ +"""alert_sent + +What the alert manager has sent, so a cooldown holds across a restart. + +Revision ID: b9e4d2a7c1f6 +Revises: a3c7e5d91f42 +Create Date: 2026-10-28 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "b9e4d2a7c1f6" +down_revision = "a3c7e5d91f42" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "alert_sent", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("key", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column("ts", sa.Float(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_alert_sent_ts"), "alert_sent", ["ts"], unique=False) + + +def downgrade(): + op.drop_index(op.f("ix_alert_sent_ts"), table_name="alert_sent") + op.drop_table("alert_sent") diff --git a/backend/fluksio/flow/alert_ledger.py b/backend/fluksio/flow/alert_ledger.py new file mode 100644 index 0000000..160c3d5 --- /dev/null +++ b/backend/fluksio/flow/alert_ledger.py @@ -0,0 +1,34 @@ +"""The alerts the manager has sent, in the engine's database. + +Cooldowns used to live in memory, so a node that stayed down was announced +again by every restart. One row per delivered alert is enough to rebuild both +the cooldowns and the ceiling per hour: the newest row of a key is when it was +last sent, and the rows of the last hour are what the ceiling counts. Rows older +than the retention are deleted as new ones come in; a cooldown longer than that +is not supported. + +Blocking: the manager calls it through ``asyncio.to_thread``. +""" + +from __future__ import annotations + +from sqlalchemy import delete +from sqlmodel import Session, col, select + +from fluksio.core.db import engine +from fluksio.models import AlertSent + + +class AlertLedger: + def load(self, since: float) -> list[tuple[str, float]]: + """Every alert sent at or after *since*, as ``(key, ts)``.""" + with Session(engine) as session: + rows = session.exec(select(AlertSent).where(AlertSent.ts >= since)).all() + return [(row.key, row.ts) for row in rows] + + def add(self, key: str, ts: float, keep_since: float) -> None: + """Record one alert, and drop what is older than *keep_since*.""" + with Session(engine) as session: + session.add(AlertSent(key=key, ts=ts)) + session.exec(delete(AlertSent).where(col(AlertSent.ts) < keep_since)) + session.commit() diff --git a/backend/fluksio/flow/alerts.py b/backend/fluksio/flow/alerts.py index 36f6635..9eeac46 100644 --- a/backend/fluksio/flow/alerts.py +++ b/backend/fluksio/flow/alerts.py @@ -17,7 +17,7 @@ import logging import time from collections import deque from collections.abc import Callable -from typing import Any, Literal +from typing import Any, Literal, Protocol import httpx from pydantic import BaseModel, Field @@ -47,6 +47,14 @@ FLAP_THRESHOLD = 3 # However bad it gets, this is the most anyone is told per hour. RATE_LIMIT = 10 RATE_WINDOW_S = 3600.0 +# A connection has to stay down this long before anyone is told. A broker that +# restarts with the stack, or a device that reconnects, is back before it +# matters. The same figure as `controller.HEALTH_FLOOR`, which the screens wait +# out; it is not imported, because the controller imports this module. +HEALTH_GRACE_S = 60.0 +# How far back a restart remembers what was sent. A cooldown longer than this +# does not survive one. +LEDGER_RETAIN_S = 7 * 24 * 3600.0 class Channel(BaseModel): @@ -163,6 +171,14 @@ def dedup_key(event: dict[str, Any]) -> str: return f"{event.get('type')}:{event.get('node') or event.get('flow') or ''}" +class Ledger(Protocol): + """Where sent alerts are kept across a restart (``alert_ledger.AlertLedger``).""" + + def load(self, since: float) -> list[tuple[str, float]]: ... + + def add(self, key: str, ts: float, keep_since: float) -> None: ... + + class AlertManager: """Watches the event bus and tells someone when it matters.""" @@ -170,11 +186,17 @@ class AlertManager: self, events: EventBus, config: AlertsConfig | None = None, - now: Any = time.monotonic, + now: Any = time.time, + ledger: Ledger | None = None, + grace_s: float = HEALTH_GRACE_S, ) -> None: self._events = events self.config = config or AlertsConfig() + # Wall time, because what was sent is remembered across a restart and + # a monotonic clock means nothing to the next process. self._now = now + self._ledger = ledger + self.grace_s = grace_s #: How a "dashboard" channel puts its alert into the graph. Bound after #: construction, because the controller that publishes does not exist #: yet when the manager is built. @@ -184,6 +206,8 @@ class AlertManager: self._health_flips: dict[str, deque[float]] = {} self._flapping: dict[str, float] = {} self._recent: deque[float] = deque() + #: Connections that went down and are waiting out the grace. + self._down: dict[str, asyncio.Task[None]] = {} # ------------------------------------------------------------------------- # The loop @@ -191,13 +215,41 @@ class AlertManager: async def run(self) -> None: """Consume the bus until cancelled.""" - async with self._events.subscribe() as queue: - while True: - event = await queue.get() - try: - await self.handle(event) - except Exception: - logger.exception("Alerting failed for %s", event.get("type")) + try: + async with self._events.subscribe() as queue: + await self.restore() + while True: + event = await queue.get() + try: + await self.handle(event) + except Exception: + logger.exception("Alerting failed for %s", event.get("type")) + finally: + for task in self._down.values(): + task.cancel() + self._down.clear() + + async def restore(self) -> None: + """Take what was sent before this process from the ledger. + + Without it every restart starts with nothing on cooldown, and a node + that stays down is announced again each time. + """ + if self._ledger is None: + return + now = self._now() + try: + rows = await asyncio.to_thread(self._ledger.load, now - LEDGER_RETAIN_S) + except Exception: + logger.exception("Could not read the alert ledger; cooldowns start empty") + return + for key, ts in sorted(rows, key=lambda row: row[1]): + # A clock that was set back would otherwise hold a key silent until + # it caught up. + ts = min(ts, now) + self._last_sent[key] = ts + if ts > now - RATE_WINDOW_S: + self._recent.append(ts) async def handle(self, event: dict[str, Any]) -> None: if not self.config.enabled: @@ -205,9 +257,37 @@ class AlertManager: kind = str(event.get("type") or "") if kind not in ALERTING_EVENTS: return + node = str(event.get("node") or "") + if kind == "node_health" and event.get("health") != "down": + # Back before anyone was told, so there is nothing to tell. + self._cancel_down(node) if self._flaps(event): return + if kind == "node_health" and event.get("health") == "down" and self.grace_s > 0: + self._cancel_down(node) + self._down[node] = asyncio.create_task(self._down_after_grace(node, event)) + return + await self._deliver(event) + def _cancel_down(self, node: str) -> None: + task = self._down.pop(node, None) + if task is not None: + task.cancel() + + async def _down_after_grace(self, node: str, event: dict[str, Any]) -> None: + """Tell someone a connection is down, unless it came back first.""" + await asyncio.sleep(self.grace_s) + if self._down.get(node) is asyncio.current_task(): + del self._down[node] + if not self.config.enabled: + return + try: + await self._deliver(event) + except Exception: + logger.exception("Alerting failed for %s", event.get("type")) + + async def _deliver(self, event: dict[str, Any]) -> None: + kind = str(event.get("type") or "") alert = describe(event) if alert is None: return @@ -218,10 +298,14 @@ class AlertManager: key = dedup_key(event) cooldown = min(r.cooldown_s for r in rules) - if not self._due(key, cooldown): + if not self._cooled_down(key, cooldown): return if not self._within_rate_limit(): + # Held back, not spent: the cooldown only starts once it is sent. + self._suppressed[key] = self._suppressed.get(key, 0) + 1 return + now = self._stamp(key) + await self._remember(key, now) held = self._suppressed.pop(key, 0) if held: @@ -238,13 +322,11 @@ class AlertManager: # Deciding whether to speak # ------------------------------------------------------------------------- - def _due(self, key: str, cooldown: float) -> bool: - now = self._now() + def _cooled_down(self, key: str, cooldown: float) -> bool: last = self._last_sent.get(key) - if last is not None and now - last < cooldown: + if last is not None and self._now() - last < cooldown: self._suppressed[key] = self._suppressed.get(key, 0) + 1 return False - self._last_sent[key] = now return True def _within_rate_limit(self) -> bool: @@ -254,9 +336,24 @@ class AlertManager: if len(self._recent) >= RATE_LIMIT: logger.warning("Alert rate limit reached; holding back") return False - self._recent.append(now) return True + def _stamp(self, key: str) -> float: + """Count an alert as sent, before anything is awaited.""" + now: float = self._now() + self._last_sent[key] = now + self._recent.append(now) + return now + + async def _remember(self, key: str, ts: float) -> None: + """Write it to the ledger. A failure here never stops the alert.""" + if self._ledger is None: + return + try: + await asyncio.to_thread(self._ledger.add, key, ts, ts - LEDGER_RETAIN_S) + except Exception: + logger.exception("Could not record the alert in the ledger") + def _flaps(self, event: dict[str, Any]) -> bool: """Is this health event part of a connection flapping up and down? diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index dcf2f78..c9ba188 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -30,6 +30,7 @@ from fluksio.core.config import settings from fluksio.core.db import engine as db_engine from fluksio.core.db import prepare from fluksio.flow import logs, modules +from fluksio.flow.alert_ledger import AlertLedger from fluksio.flow.alerts import AlertManager from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.controller import FlowController, RebuildBusy @@ -284,7 +285,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Connectors register their node types before any flow is built with them. load_plugins() - alerts = AlertManager(event_bus, config=read_alerts_config()) + alerts = AlertManager( + event_bus, config=read_alerts_config(), ledger=AlertLedger() + ) execution = ExecutionService( queue=_work_queue(), max_workers=settings.FLOW_MAX_WORKERS, diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index f8edf75..baa6897 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -586,6 +586,27 @@ class ConnectorState(SQLModel, table=True): updated_at: datetime = Field(default_factory=get_datetime_utc, sa_type=UTCDateTime) +# ----------------------------------------------------------------------------- +# Alerts sent +# +# What the alert manager has told someone, so its cooldowns and its ceiling per +# hour hold across a restart. A node that stays down would otherwise be +# announced again by every one. +# ----------------------------------------------------------------------------- + + +class AlertSent(SQLModel, table=True): + """One alert that was delivered, by the key it deduplicates on.""" + + __tablename__ = "alert_sent" + + id: int | None = Field(default=None, primary_key=True) + #: `alerts.dedup_key`: the event type and the node or flow it is about. + key: str = Field(max_length=255) + #: Seconds since the epoch, from the manager's own clock. + ts: float = Field(index=True) + + # ----------------------------------------------------------------------------- # Worker tokens # diff --git a/backend/tests/flow/test_alerts.py b/backend/tests/flow/test_alerts.py index 4f07326..9c58eb7 100644 --- a/backend/tests/flow/test_alerts.py +++ b/backend/tests/flow/test_alerts.py @@ -9,6 +9,7 @@ import pytest from fluksio.flow.alerts import ( ALERTING_EVENTS, FLAP_THRESHOLD, + LEDGER_RETAIN_S, RATE_LIMIT, Alert, AlertManager, @@ -20,6 +21,20 @@ from fluksio.flow.alerts import ( from fluksio.flow.events import EventBus +class Ledger: + """What the database holds, without one.""" + + def __init__(self) -> None: + self.rows: list[tuple[str, float]] = [] + + def load(self, since: float) -> list[tuple[str, float]]: + return [row for row in self.rows if row[1] >= since] + + def add(self, key: str, ts: float, keep_since: float) -> None: + self.rows = [row for row in self.rows if row[1] >= keep_since] + self.rows.append((key, ts)) + + class Clock: """A hand-wound clock, so cooldowns take no real time.""" @@ -39,7 +54,9 @@ def manager(clock: Clock) -> tuple[AlertManager, list[Alert]]: channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})], rules=[Rule(events=[], channels=["phone"], cooldown_s=900)], ) - alerts = AlertManager(EventBus(), config=config, now=clock) + # No grace: these tests are about what is sent, not how long a connection + # has to stay down first. + alerts = AlertManager(EventBus(), config=config, now=clock, grace_s=0) async def capture(channel, alert): sent.append(alert) @@ -153,6 +170,174 @@ def test_a_storm_is_capped(): assert len(sent) == RATE_LIMIT +def test_a_cooldown_survives_a_restart(): + """A node that stays down was announced again by every restart.""" + clock, ledger = Clock(), Ledger() + + async def scenario() -> list[Alert]: + first, sent_before = manager(clock) + first._ledger = ledger + await first.handle(error_event()) + clock.advance(120) + + second, sent_after = manager(clock) + second._ledger = ledger + await second.restore() + await second.handle(error_event()) + assert len(sent_before) == 1 + clock.advance(1000) + await second.handle(error_event()) + return sent_after + + sent_after = asyncio.run(scenario()) + + # Silent inside the cooldown, and speaking again once it has passed. + assert len(sent_after) == 1 + + +def test_the_ceiling_per_hour_survives_a_restart(): + clock, ledger = Clock(), Ledger() + clock.now = 10_000.0 + ledger.rows = [ + (f"node_error:heating.n{i}", clock.now - 60) for i in range(RATE_LIMIT) + ] + alerts, sent = manager(clock) + alerts._ledger = ledger + + async def scenario(): + await alerts.restore() + await alerts.handle(error_event("heating.new")) + + asyncio.run(scenario()) + + assert sent == [] + + +def test_a_clock_set_back_does_not_hold_a_key_silent(): + clock, ledger = Clock(), Ledger() + clock.now = 100.0 + ledger.rows = [("node_error:heating.pump", 50_000.0)] + alerts, sent = manager(clock) + alerts._ledger = ledger + + async def scenario(): + await alerts.restore() + clock.advance(1000) + await alerts.handle(error_event()) + + asyncio.run(scenario()) + + assert len(sent) == 1 + + +def test_sent_alerts_are_written_to_the_ledger_and_old_ones_dropped(): + clock, ledger = Clock(), Ledger() + clock.now = LEDGER_RETAIN_S * 2 + ledger.rows = [("node_error:heating.old", 1.0)] + alerts, _sent = manager(clock) + alerts._ledger = ledger + + asyncio.run(alerts.handle(error_event())) + + assert ledger.rows == [("node_error:heating.pump", clock.now)] + + +def test_a_ledger_that_fails_does_not_stop_the_alert(): + clock = Clock() + alerts, sent = manager(clock) + + class Broken: + def load(self, since): + raise OSError("disk") + + def add(self, key, ts, keep_since): + raise OSError("disk") + + alerts._ledger = Broken() # type: ignore[assignment] + + async def scenario(): + await alerts.restore() + await alerts.handle(error_event()) + + asyncio.run(scenario()) + + assert len(sent) == 1 + + +def test_an_alert_held_back_by_the_ceiling_does_not_spend_its_cooldown(): + clock = Clock() + alerts, sent = manager(clock) + + async def scenario(): + for i in range(RATE_LIMIT): + await alerts.handle(error_event(f"heating.n{i}")) + await alerts.handle(error_event("heating.late")) + # An hour on, the ceiling has cleared and the same fault is still due. + clock.advance(3601) + await alerts.handle(error_event("heating.late")) + + asyncio.run(scenario()) + + assert [a.node for a in sent][-1] == "heating.late" + assert "1 more since the last alert" in sent[-1].body + + +def down(node: str = "heating.mqtt") -> dict: + return {"type": "node_health", "node": node, "health": "down", "detail": "refused"} + + +def up(node: str = "heating.mqtt") -> dict: + return {"type": "node_health", "node": node, "health": "ok"} + + +def with_grace(grace_s: float) -> tuple[AlertManager, list[Alert]]: + alerts, sent = manager(Clock()) + alerts.grace_s = grace_s + return alerts, sent + + +def test_a_connection_that_comes_back_within_the_grace_alerts_nobody(): + alerts, sent = with_grace(0.05) + + async def scenario(): + await alerts.handle(down()) + await alerts.handle(up()) + await asyncio.sleep(0.1) + + asyncio.run(scenario()) + + assert sent == [] + + +def test_a_connection_that_stays_down_is_told_after_the_grace(): + alerts, sent = with_grace(0.05) + + async def scenario(): + await alerts.handle(down()) + assert sent == [] + await asyncio.sleep(0.1) + + asyncio.run(scenario()) + + assert [a.title for a in sent] == ["heating.mqtt lost its connection"] + + +def test_a_pending_alert_is_dropped_when_the_manager_stops(): + alerts, sent = with_grace(0.05) + + async def scenario(): + await alerts.handle(down()) + runner = asyncio.create_task(alerts.run()) + await asyncio.sleep(0.01) + runner.cancel() + await asyncio.gather(runner, return_exceptions=True) + await asyncio.sleep(0.1) + + asyncio.run(scenario()) + + assert sent == [] + + def test_nothing_is_sent_without_a_rule(): clock = Clock() alerts, sent = manager(clock) diff --git a/backend/tests/test_alert_ledger.py b/backend/tests/test_alert_ledger.py new file mode 100644 index 0000000..0b6bcff --- /dev/null +++ b/backend/tests/test_alert_ledger.py @@ -0,0 +1,26 @@ +"""The alert ledger: what the manager remembers across a restart.""" + +import pytest +from sqlalchemy import delete +from sqlmodel import Session + +from fluksio.core.db import engine +from fluksio.flow.alert_ledger import AlertLedger +from fluksio.models import AlertSent + + +@pytest.fixture(autouse=True) +def _empty(db: Session): # noqa: ARG001 - the schema has to exist first + with Session(engine) as session: + session.exec(delete(AlertSent)) + session.commit() + yield + + +def test_a_sent_alert_comes_back_and_old_ones_are_pruned(): + ledger = AlertLedger() + ledger.add("node_error:house.calc", 100.0, keep_since=0.0) + ledger.add("node_health:house.mqtt", 5000.0, keep_since=1000.0) + + assert ledger.load(since=0.0) == [("node_health:house.mqtt", 5000.0)] + assert ledger.load(since=6000.0) == [] diff --git a/docs/interface/operations.md b/docs/interface/operations.md index 50767c9..e4cd7b6 100644 --- a/docs/interface/operations.md +++ b/docs/interface/operations.md @@ -83,9 +83,13 @@ matches every supported alert event, including future additions. | `queue_unavailable` | the work queue becomes unreachable | | `run_finished` | a batch run ends in error | -Repeated faults are deduplicated with a 15-minute default cooldown. -Flapping connections are suppressed, and the engine caps alert dispatches at -ten per hour. A quiet channel may be cooling down; **Test** helps distinguish +Repeated faults are deduplicated with a 15-minute default cooldown. The +engine remembers what it sent in its database, so a restart does not start +every cooldown over, and the ten-per-hour cap holds across one too. Cooldowns +longer than seven days are not kept. A `node_health` alert waits until the +connection has been down for a minute, so a broker that restarts with the +stack or a device that reconnects alerts nobody. Flapping connections are +suppressed. A quiet channel may be cooling down; **Test** helps distinguish that from a broken configuration. ## Admin: users and clients