feat(alerts): one evaluation path; host-agent ingest fires rules (#5060)

#5060 "One evaluation path: record_metrics batch API; host-agent ingest
evaluates rules".

- record_metrics(session, rows) writes metric rows and evaluates the rules
  that watch them, reading rules, states and maintenance windows once per
  batch. Each rule sees its key's samples oldest first, so an agent catching
  up after an outage counts consecutive checks as they happened.
- Host-agent ingest now goes through it. Before, it wrote plugin_metrics
  directly and no host_agent rule could ever fire. The catalog gains the
  per-mount disk_used_pct ("<host>:<mount>").
- MetricBatch gathers a transaction's samples; the docker ingest, traefik,
  unifi, snmp, monitor and self-monitor writers record one batch each instead
  of a rule query per sample.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/docker/ingest.py

@@ -266,7 +266,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
     log batches (m79) — appended one row per line when present, independent of
     whether this sample also carried container metrics.
     """
-    from steward.core.alerts import record_metric
+    from steward.core.alerts import MetricBatch
     from .models import DockerContainer, DockerEvent, DockerMetric
 
     if swarm is not None:
@@ -313,6 +313,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
             ))
 
     # Current state + alerts from the newest snapshot only.
+    batch = MetricBatch()
     for c in latest_containers:
         name = c.get("name")
         if not name:
@@ -350,19 +351,19 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
         # on different hosts don't collide in the metric/alert namespace.
         resource = f"{host.name}/{name}"
         if c.get("status") == "running" and c.get("cpu_pct") is not None:
-            await record_metric(
-                session=session, source_module="docker",
+            batch.add(
+                source_module="docker",
                 resource_name=resource, metric_name="cpu_pct", value=c["cpu_pct"],
             )
             if c.get("mem_pct") is not None:
-                await record_metric(
-                    session=session, source_module="docker",
+                batch.add(
+                    source_module="docker",
                     resource_name=resource, metric_name="mem_pct", value=c["mem_pct"],
                 )
         # Restart count is alertable regardless of state (crash-looping matters
         # most while the container is down/restarting).
-        await record_metric(
-            session=session, source_module="docker",
+        batch.add(
+            source_module="docker",
             resource_name=resource, metric_name="restart_count",
             value=float(c.get("restart_count", 0) or 0),
         )
@@ -371,12 +372,14 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
         # container).
         health = c.get("health")
         if health in ("healthy", "unhealthy", "starting"):
-            await record_metric(
-                session=session, source_module="docker",
+            batch.add(
+                source_module="docker",
                 resource_name=resource, metric_name="is_healthy",
                 value=1.0 if health == "healthy" else 0.0,
             )
 
+    await batch.record(session)
+
     # Reap containers that vanished from the host's latest listing. Without this,
     # removed one-shot containers (CI job runners, buildkit builders, codex jobs)
     # accumulate forever as permanent "stopped" rows and inflate every count.

plugins/host_agent/routes.py

@@ -201,6 +201,7 @@ async def ingest():
             return _error(401, "invalid_token")
 
         accepted = 0
+        metric_rows: list[PluginMetric] = []
         latest_ts: datetime | None = None
         docker_snapshots: list[tuple[datetime, list]] = []
         docker_log_batches: list[tuple[datetime, list]] = []
@@ -213,9 +214,7 @@ async def ingest():
                 recorded_at = _parse_ts(sample["ts"])
             except (KeyError, ValueError):
                 continue
-            metrics = _expand_sample_to_metrics(sample, host.name, recorded_at)
-            for m in metrics:
-                session.add(m)
+            metric_rows.extend(_expand_sample_to_metrics(sample, host.name, recorded_at))
             docker = sample.get("docker")
             if isinstance(docker, list) and docker:
                 docker_snapshots.append((recorded_at, docker))
@@ -243,6 +242,10 @@ async def ingest():
             await session.rollback()
             return _error(400, "malformed_payload", "no valid samples")
 
+        # Through the alert pipeline, so rules on host metrics can fire.
+        from steward.core.alerts import record_metrics
+        await record_metrics(session, metric_rows)
+
         # Hand host-scoped container data to the docker plugin if it's enabled
         # (opportunistic synergy via the capability registry — no hard import,
         # no-op when docker is disabled). A failure here must never sink the

plugins/snmp/scheduler.py

@@ -27,7 +27,7 @@ def make_poll_task(app: "Quart") -> ScheduledTask:
 
 async def _do_poll(app: "Quart") -> None:
     from .poller import poll_device
-    from steward.core.alerts import record_metric
+    from steward.core.alerts import MetricBatch
 
     devices: list[dict] = app.config["PLUGINS"]["snmp"].get("devices", [])
     if not devices:
@@ -35,6 +35,7 @@ async def _do_poll(app: "Quart") -> None:
 
     async with app.db_sessionmaker() as session:
         async with session.begin():
+            batch = MetricBatch()
             for device in devices:
                 if not isinstance(device, dict):
                     continue
@@ -55,8 +56,7 @@ async def _do_poll(app: "Quart") -> None:
                     continue
 
                 for label, value in readings.items():
-                    await record_metric(
-                        session=session,
+                    batch.add(
                         source_module="snmp",
                         resource_name=name,
                         metric_name=label,
@@ -67,3 +67,4 @@ async def _do_poll(app: "Quart") -> None:
                     logger.debug("SNMP polled %s (%s): %d OID(s)", name, host, len(readings))
                 else:
                     logger.debug("SNMP polled %s (%s): no readings", name, host)
+            await batch.record(session)

plugins/traefik/scheduler.py

@@ -48,7 +48,7 @@ async def _do_scrape(app: "Quart") -> None:
         extract_certs,
         fetch_metrics,
     )
-    from steward.core.alerts import record_metric
+    from steward.core.alerts import MetricBatch
 
     metrics_url: str = app.config["PLUGINS"]["traefik"]["metrics_url"]
 
@@ -71,6 +71,7 @@ async def _do_scrape(app: "Quart") -> None:
 
     async with app.db_sessionmaker() as session:
         async with session.begin():
+            batch = MetricBatch()
             # ── Per-service metrics ───────────────────────────────────────────
             for router_name, metrics in router_metrics.items():
                 row = TraefikMetric(
@@ -95,8 +96,7 @@ async def _do_scrape(app: "Quart") -> None:
                     ("latency_p99_ms", metrics["latency_p99_ms"]),
                     ("response_bytes_rate", metrics["response_bytes_rate"]),
                 ]:
-                    await record_metric(
-                        session=session,
+                    batch.add(
                         source_module="traefik",
                         resource_name=router_name,
                         metric_name=metric_name,
@@ -125,14 +125,15 @@ async def _do_scrape(app: "Quart") -> None:
 
                 # Emit cert_expiry_days for alert rules
                 days_remaining = (cert["not_after"] - scraped_at).total_seconds() / 86400.0
-                await record_metric(
-                    session=session,
+                batch.add(
                     source_module="traefik",
                     resource_name=cert["cn"] or cert["serial"],
                     metric_name="cert_expiry_days",
                     value=max(0.0, days_remaining),
                 )
 
+            await batch.record(session)
+
     logger.debug(
         "Traefik scrape: %d router(s), %d cert(s)",
         len(router_metrics),

plugins/unifi/scheduler.py

@@ -54,7 +54,7 @@ async def _do_poll(app: "Quart") -> None:
 
     from .client import UnifiClient
     from .models import UnifiClientSnapshot, UnifiDevice, UnifiWanStat
-    from steward.core.alerts import record_metric
+    from steward.core.alerts import MetricBatch
 
     cfg = app.config["PLUGINS"]["unifi"]
 
@@ -171,28 +171,27 @@ async def _do_poll(app: "Quart") -> None:
                     ))
 
             # ── Alert metrics ─────────────────────────────────────────────────
+            batch = MetricBatch()
             if wan_stat.latency_ms is not None:
-                await record_metric(
-                    session=session,
+                batch.add(
                     source_module="unifi",
                     resource_name="wan",
                     metric_name="latency_ms",
                     value=wan_stat.latency_ms,
                 )
-            await record_metric(
-                session=session,
+            batch.add(
                 source_module="unifi",
                 resource_name="wan",
                 metric_name="is_up",
                 value=1.0 if wan_stat.is_up else 0.0,
             )
-            await record_metric(
-                session=session,
+            batch.add(
                 source_module="unifi",
                 resource_name="clients",
                 metric_name="total_clients",
                 value=float(len(clients)),
             )
+            await batch.record(session)
 
     logger.debug(
         "UniFi poll: wan=%s, clients=%d, devices=%d",

steward/alerts/service.py

@@ -33,6 +33,8 @@ METRIC_CATALOG: dict[str, list[str]] = {
     "host_agent": [
         "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes",
         "disk_used_pct_worst", "load_1m", "load_5m", "load_15m", "uptime_secs",
+        # Per mount: the resource is "<host>:<mount>".
+        "disk_used_pct",
     ],
     # snmp metrics are user-defined OID labels — populated dynamically from plugin_metrics
     "snmp":    [],

steward/core/alerts.py

@@ -6,7 +6,7 @@ from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING
 
-from sqlalchemy import and_, or_, select
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from steward.models.alerts import (
@@ -57,82 +57,109 @@ async def record_metric(
     metric_name: str,
     value: float,
 ) -> None:
-    """Write a metric row and evaluate alert rules inline.
+    """One metric sample: record_metrics with a single row."""
+    await record_metrics(session, [PluginMetric(
+        source_module=source_module, resource_name=resource_name,
+        metric_name=metric_name, value=value)])
 
-    Must be called inside an active transaction:
-        async with session.begin():
-            await record_metric(session, ...)
 
-    Notification I/O is deferred via asyncio.create_task() so no network
-    calls occur while the transaction is open.
+class MetricBatch:
+    """Gathers one transaction's samples so they are recorded, and their alert
+    rules evaluated, in a single record_metrics call:
+
+        batch = MetricBatch()
+        batch.add(source_module="unifi", resource_name="wan", metric_name="is_up", value=1.0)
+        await batch.record(session)
     """
-    now = datetime.now(timezone.utc)
 
-    # Write metric row
-    metric = PluginMetric(
-        source_module=source_module,
-        resource_name=resource_name,
-        metric_name=metric_name,
-        value=value,
-        recorded_at=now,
-    )
-    session.add(metric)
+    def __init__(self) -> None:
+        self.rows: list[PluginMetric] = []
+
+    def add(self, source_module: str, resource_name: str, metric_name: str, value: float,
+            recorded_at: datetime | None = None) -> None:
+        self.rows.append(PluginMetric(source_module=source_module, resource_name=resource_name,
+                                      metric_name=metric_name, value=float(value),
+                                      recorded_at=recorded_at))
+
+    async def record(self, session: AsyncSession) -> None:
+        rows, self.rows = self.rows, []
+        await record_metrics(session, rows)
+
+
+def _in_window(w: MaintenanceWindow, source_module: str, resource_name: str) -> bool:
+    if w.scope_source is None:
+        return True
+    return w.scope_source == source_module and w.scope_resource in (None, resource_name)
+
+
+async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> None:
+    """Write metric rows and evaluate the alert rules that watch them.
+
+    Every ingest path comes through here, so a rule on any metric Steward
+    records can fire. Must be called inside an active transaction:
+        async with session.begin():
+            await record_metrics(session, rows)
+
+    Maintenance windows, rules and their states are read once per batch, not
+    once per row. Each rule sees its key's samples oldest first, so a backlog
+    (an agent catching up after an outage) counts consecutive checks as they
+    happened. Notification I/O is deferred via asyncio.create_task() so no
+    network calls occur while the transaction is open.
+    """
+    if not rows:
+        return
+    now = datetime.now(timezone.utc)
+    for row in rows:
+        if row.recorded_at is None:
+            row.recorded_at = now
+    session.add_all(rows)
     await session.flush()
 
-    # Check active maintenance windows — skip rule evaluation if suppressed
-    mw_result = await session.execute(
-        select(MaintenanceWindow).where(
-            MaintenanceWindow.start_at <= now,
-            MaintenanceWindow.end_at >= now,
-            or_(
-                MaintenanceWindow.scope_source.is_(None),
-                and_(
-                    MaintenanceWindow.scope_source == source_module,
-                    or_(
-                        MaintenanceWindow.scope_resource.is_(None),
-                        MaintenanceWindow.scope_resource == resource_name,
-                    ),
-                ),
-            ),
-        ).limit(1)
-    )
-    if mw_result.scalar_one_or_none() is not None:
-        return  # suppressed by maintenance window
+    by_key: dict[tuple[str, str, str], list[PluginMetric]] = {}
+    for row in rows:
+        by_key.setdefault((row.source_module, row.resource_name, row.metric_name), []).append(row)
+    sources = {k[0] for k in by_key}
+    metric_names = {k[2] for k in by_key}
 
-    # Load matching enabled rules
-    result = await session.execute(
+    rules = (await session.execute(
         select(AlertRule).where(
-            AlertRule.source_module == source_module,
-            AlertRule.resource_name == resource_name,
-            AlertRule.metric_name == metric_name,
+            AlertRule.source_module.in_(sources),
+            AlertRule.metric_name.in_(metric_names),
             AlertRule.enabled.is_(True),
         )
-    )
-    rules = result.scalars().all()
+    )).scalars().all()
+    rules = [r for r in rules if (r.source_module, r.resource_name, r.metric_name) in by_key]
+    if not rules:
+        return
+
+    windows = (await session.execute(
+        select(MaintenanceWindow).where(
+            MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now)
+    )).scalars().all()
+    states = {st.rule_id: st for st in (await session.execute(
+        select(AlertState).where(AlertState.rule_id.in_([r.id for r in rules]))
+    )).scalars()}
 
     for rule in rules:
-        state_result = await session.execute(
-            select(AlertState).where(AlertState.rule_id == rule.id)
-        )
-        alert_state = state_result.scalar_one_or_none()
+        if any(_in_window(w, rule.source_module, rule.resource_name) for w in windows):
+            continue  # suppressed by maintenance window
+        alert_state = states.get(rule.id)
         if alert_state is None:
-            logger.warning(f"No alert_state row for rule {rule.id}, skipping")
+            logger.warning("No alert_state row for rule %s, skipping", rule.id)
             continue
-
-        notifications = await _evaluate_rule(session, rule, alert_state, value, now)
-
-        for notif_type, event_id in notifications:
-            if _app is not None:
-                asyncio.create_task(
-                    _dispatch_notification(
-                        _app, notif_type, rule, value, event_id
-                    )
-                )
-                # Admin-configured Ansible action runs only on transition-to-firing.
-                if notif_type == "firing" and rule.ansible_action:
+        samples = sorted(by_key[(rule.source_module, rule.resource_name, rule.metric_name)],
+                         key=lambda r: r.recorded_at)
+        for sample in samples:
+            notifications = await _evaluate_rule(session, rule, alert_state, sample.value,
+                                                 sample.recorded_at)
+            for notif_type, event_id in notifications:
+                if _app is not None:
                     asyncio.create_task(
-                        _run_ansible_action(_app, rule, value, event_id)
-                    )
+                        _dispatch_notification(_app, notif_type, rule, sample.value, event_id))
+                    # Admin-configured Ansible action runs only on transition-to-firing.
+                    if notif_type == "firing" and rule.ansible_action:
+                        asyncio.create_task(
+                            _run_ansible_action(_app, rule, sample.value, event_id))
 
 
 def _is_breached(value: float, operator: AlertOperator, threshold: float) -> bool:

steward/core/self_monitor.py

@@ -80,28 +80,28 @@ async def record_self_metrics(app) -> None:
         logger.debug("self_monitor: /proc/self/fd unavailable — skipping fd metrics")
         return
 
-    from .alerts import record_metric
+    from .alerts import MetricBatch
 
     soft = fd_soft_limit()
     pct = (fds / soft * 100.0) if soft else None
 
     async with app.db_sessionmaker() as session:
         async with session.begin():
-            await record_metric(
-                session=session,
+            batch = MetricBatch()
+            batch.add(
                 source_module="steward",
                 resource_name="process",
                 metric_name="open_fds",
                 value=float(fds),
             )
             if pct is not None:
-                await record_metric(
-                    session=session,
+                batch.add(
                     source_module="steward",
                     resource_name="process",
                     metric_name="open_fds_pct",
                     value=pct,
                 )
+            await batch.record(session)
 
     if pct is not None and pct >= FD_WARN_PCT:
         if not _warned:

steward/monitors/scheduler.py

@@ -6,7 +6,7 @@ from datetime import datetime, timezone
 
 from sqlalchemy import select
 
-from steward.core.alerts import record_metric
+from steward.core.alerts import MetricBatch
 from steward.models.monitors import Monitor, MonitorResult
 from .runner import run_monitor
 
@@ -63,14 +63,16 @@ async def run_due_monitors(app) -> None:
                     db_monitor.last_checked_at = check_time
 
                 # Alert pipeline — keyed by monitor type (source) + name.
-                await record_metric(
-                    session=session, source_module=monitor.type,
+                batch = MetricBatch()
+                batch.add(
+                    source_module=monitor.type,
                     resource_name=monitor.name, metric_name="is_up",
                     value=1.0 if res["is_up"] else 0.0,
                 )
                 if res["response_ms"] is not None:
-                    await record_metric(
-                        session=session, source_module=monitor.type,
+                    batch.add(
+                        source_module=monitor.type,
                         resource_name=monitor.name, metric_name="response_ms",
                         value=res["response_ms"],
                     )
+                await batch.record(session)

tests/core/test_self_monitor.py

@@ -3,7 +3,7 @@
 Exercises the fd accounting, the percentage/warning floor, and that the
 recorder degrades to a no-op where fd accounting isn't available. The DB session
 and the alert pipeline are faked — we only assert that the right metrics get
-handed to record_metric.
+handed to record_metrics.
 """
 import asyncio
 import logging
@@ -48,10 +48,10 @@ class _FakeApp:
 def _patch_recorder(monkeypatch):
     recorded: list[tuple] = []
 
-    async def fake_record_metric(session, source_module, resource_name, metric_name, value):
-        recorded.append((source_module, resource_name, metric_name, value))
+    async def fake_record_metrics(session, rows):
+        recorded.extend((r.source_module, r.resource_name, r.metric_name, r.value) for r in rows)
 
-    monkeypatch.setattr(alerts, "record_metric", fake_record_metric)
+    monkeypatch.setattr(alerts, "record_metrics", fake_record_metrics)
     return recorded
 
 

tests/integration/test_host_agent_alerts.py

@@ -0,0 +1,75 @@
+"""Integration: a host-agent ingest POST evaluates alert rules on the host's
+metrics (#5060). Before, ingest wrote rows straight to plugin_metrics and no
+host_agent rule could ever fire."""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import os
+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 test_ingest_fires_a_host_rule_from_a_backlog(app):
+    from sqlalchemy import select
+    from plugins.host_agent.models import HostAgentRegistration
+    from steward.alerts import service as alert_svc
+    from steward.models.alerts import AlertEvent, AlertState, AlertStateEnum
+    from steward.models.hosts import Host
+    from steward.models.metrics import PluginMetric
+    from steward.models.users import User, UserRole
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        hostname, token = f"agent-{tag}", f"tok-{tag}-{uuid.uuid4().hex}"
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                user = User(username=f"ha-{tag}", email=f"ha-{tag}@test", password_hash="x",
+                            role=UserRole.admin)
+                host = Host(name=hostname, address="10.9.9.9")
+                s.add_all([user, host])
+                await s.flush()
+                s.add(HostAgentRegistration(
+                    host_id=host.id, token_hash=hashlib.sha256(token.encode()).hexdigest()))
+                s.add(PluginMetric(source_module="host_agent", resource_name=hostname,
+                                   metric_name="cpu_pct", value=5.0))
+                await s.flush()
+                fields = await alert_svc.rule_fields(
+                    s, name=f"cpu {tag}", source_module="host_agent", resource_name=hostname,
+                    metric_name="cpu_pct", operator=">", threshold=90,
+                    consecutive_failures_required=2)
+                rule = await alert_svc.add_rule(s, fields, created_by=user.id)
+            rule_id = rule.id
+
+        now = datetime.now(timezone.utc)
+        # Two buffered samples in one POST, newest first: evaluated oldest
+        # first, they make the two breaching checks the rule needs.
+        samples = [{"ts": (now - timedelta(seconds=s)).isoformat(), "cpu_pct": v}
+                   for s, v in ((0, 97.0), (30, 95.0))]
+        client = app.test_client()
+        resp = await client.post("/plugins/host_agent/ingest", json={"samples": samples},
+                                 headers={"Authorization": f"Bearer {token}"})
+        assert resp.status_code == 200, await resp.get_data(as_text=True)
+
+        async with app.db_sessionmaker() as s:
+            st = (await s.execute(select(AlertState).where(
+                AlertState.rule_id == rule_id))).scalar_one()
+            events = (await s.execute(select(AlertEvent).where(
+                AlertEvent.rule_id == rule_id))).scalars().all()
+        assert st.state == AlertStateEnum.firing
+        assert [(e.from_state, e.to_state, e.metric_value) for e in events] == [
+            ("pending", "firing", 97.0)]
+
+    asyncio.run(_go())