feat(host_agent): a dropped ingest says which step of the metric write it stalled in (#5149)

The first instrumented drop on dev read "dropped after 10.0s … (completed:
parse=0.1s)": the whole stall is inside record_metrics. It now marks its
three steps (flush of the rows, the rule and state reads, the per-sample
evaluation) on the caller's phases, and the drop line carries the request's
samples and metric row count.

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

plugins/host_agent/routes.py

@@ -250,8 +250,8 @@ async def ingest():
 
             # Through the alert pipeline, so rules on host metrics can fire.
             from steward.core.alerts import record_metrics
-            await record_metrics(session, metric_rows)
-            phases.mark("metrics")
+            phases.size = f"samples={accepted} metric_rows={len(metric_rows)}"
+            await record_metrics(session, metric_rows, phases)
 
             # Hand host-scoped container data to the docker plugin if it's enabled
             # (opportunistic synergy via the capability registry — no hard import,
@@ -331,6 +331,7 @@ class _Phases:
     def __init__(self):
         self.start = self.last = time.monotonic()
         self.took: dict[str, float] = {}
+        self.size = ""
 
     def mark(self, name: str) -> None:
         now = time.monotonic()
@@ -339,9 +340,9 @@ class _Phases:
 
     def cancelled(self, who: str) -> None:
         current_app.logger.warning(
-            "host_agent ingest: dropped after %.1fs for token=%s… (completed: %s)",
+            "host_agent ingest: dropped after %.1fs for token=%s… (completed: %s) %s",
             time.monotonic() - self.start, who,
-            " ".join(f"{k}={v:.1f}s" for k, v in self.took.items()) or "nothing")
+            " ".join(f"{k}={v:.1f}s" for k, v in self.took.items()) or "nothing", self.size)
 
     def report(self, host: str, samples: int, metrics: int, containers: int, log_lines: int) -> None:
         total = time.monotonic() - self.start

steward/app.py

@@ -367,7 +367,7 @@ def create_app(
 def _register_core_tasks(app: Quart) -> None:
     from .core.scheduler import ScheduledTask
 
-    poll_interval = app.config.get("MONITORS_POLL_INTERVAL", 60)
+    from .monitors.scheduler import DISPATCH_TICK_SECONDS as MONITOR_DISPATCH_TICK
     cleanup_interval = 3600  # hourly
 
     async def run_monitor_checks():
@@ -418,10 +418,12 @@ def _register_core_tasks(app: Quart) -> None:
             interval_seconds=60,
             run_on_startup=False,
         ),
+        # Ticks often and only starts checks: each monitor decides whether it
+        # is due from its own interval (steward/monitors/scheduler.py).
         ScheduledTask(
             name="monitor_check",
             coro_factory=run_monitor_checks,
-            interval_seconds=poll_interval,
+            interval_seconds=MONITOR_DISPATCH_TICK,
             run_on_startup=True,
         ),
         ScheduledTask(

steward/core/alerts.py

@@ -141,7 +141,7 @@ def _after_commit(session: AsyncSession, fn, *args) -> None:
     info[_AFTER_COMMIT].append((fn, args))
 
 
-async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> None:
+async def record_metrics(session: AsyncSession, rows: list[PluginMetric], phases=None) -> 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
@@ -155,7 +155,11 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
     happened. A maintenance window holds back notifications and actions, not
     evaluation: the state still moves, and the event says it was muted.
     Notifications and actions start only once the transaction commits.
+
+    `phases`, when given, is marked after each step (flush, rules, evaluate)
+    so a slow caller can say which one its time went to.
     """
+    mark = phases.mark if phases is not None else (lambda _name: None)
     if not rows:
         return
     now = datetime.now(timezone.utc)
@@ -164,6 +168,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
             row.recorded_at = now
     session.add_all(rows)
     await session.flush()
+    mark("flush")
 
     by_key: dict[tuple[str, str, str], list[PluginMetric]] = {}
     for row in rows:
@@ -180,6 +185,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
     )).scalars().all()
     rules = [r for r in rules if (r.source_module, r.resource_name, r.metric_name) in by_key]
     if not rules:
+        mark("rules")
         return
 
     windows = (await session.execute(
@@ -189,6 +195,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
     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()}
+    mark("rules")
 
     for rule in rules:
         alert_state = states.get(rule.id)
@@ -214,6 +221,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
                 # Admin-configured Ansible action runs only on transition-to-firing.
                 if notif_type == "firing" and rule.ansible_action:
                     _after_commit(session, _run_ansible_action, rule, sample.value, event_id)
+    mark("evaluate")
 
 
 # A series is silent once it has gone this long without a sample, or three of

steward/monitors/scheduler.py

@@ -1,52 +1,92 @@
-"""Single scheduled task that runs every due Monitor of any type."""
+"""Dispatch every due Monitor of any type, each check in its own task.
+
+The dispatcher ticks every DISPATCH_TICK_SECONDS. Each tick it reads the
+enabled monitors and their intervals fresh (a changed interval applies at the
+next tick), stamps the due ones with the time they were due to start, and
+starts each check on its own. A slow check (a 10 s timeout) holds up nobody
+else, and a monitor still being checked is not started twice.
+"""
 from __future__ import annotations
 import asyncio
 import logging
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 
 from steward.core.alerts import MetricBatch
 from steward.models.monitors import Monitor, MonitorResult
-from .runner import run_monitor
+from .runner import _BLANK, run_monitor
 
 logger = logging.getLogger(__name__)
 
+# How often the dispatcher looks for due monitors. A monitor runs within one
+# tick of its interval, so this is the cadence's worst-case lateness.
+DISPATCH_TICK_SECONDS = 5
+# A check past this is recorded down rather than left holding its monitor:
+# far beyond any probe's own timeout, so it only ever catches a hang.
+CHECK_DEADLINE_SECONDS = 120
+
+
+def is_due(monitor, now: datetime, global_interval: int) -> bool:
+    """Whether a monitor's interval has elapsed since its last scheduled start.
+
+    Half a tick of slack, so a check stamped at t isn't missed at t+interval
+    by a few milliseconds of tick jitter and pushed back a whole tick."""
+    if monitor.last_checked_at is None:
+        return True
+    interval = monitor.check_interval_seconds or global_interval
+    elapsed = (now - monitor.last_checked_at).total_seconds()
+    return elapsed >= interval - DISPATCH_TICK_SECONDS / 2
+
+
+def _state(app) -> tuple[set, set]:
+    """(monitor ids being checked, their tasks) for this app."""
+    if not hasattr(app, "_monitor_checks"):
+        app._monitor_checks = (set(), set())
+    return app._monitor_checks
+
 
 async def run_due_monitors(app) -> None:
-    """Check every enabled Monitor whose interval has elapsed, persist results.
+    """Start a check for every enabled Monitor that is due and not already
+    being checked. Returns once they are started, not finished.
 
     Per-monitor `check_interval_seconds` overrides the global
-    MONITORS_POLL_INTERVAL (0 = use global), mirroring how the scheduler fires
-    on the global cadence but each monitor decides if it is actually due.
-    """
+    MONITORS_POLL_INTERVAL (0 = use global)."""
     global_interval = int(app.config.get("MONITORS_POLL_INTERVAL", 60))
     now = datetime.now(timezone.utc)
+    in_flight, tasks = _state(app)
 
     async with app.db_sessionmaker() as session:
         monitors = list((await session.execute(
             select(Monitor).where(Monitor.enabled.is_(True))
         )).scalars())
+        due = [m for m in monitors if m.id not in in_flight and is_due(m, now, global_interval)]
+        if not due:
+            return
+        # Stamp the scheduled start before checking: the next run is due one
+        # interval from now, however long this check takes.
+        await session.execute(
+            update(Monitor).where(Monitor.id.in_([m.id for m in due])).values(last_checked_at=now))
+        await session.commit()
 
-    due: list[Monitor] = []
-    for m in monitors:
-        interval = m.check_interval_seconds or global_interval
-        if m.last_checked_at is None or (now - m.last_checked_at).total_seconds() >= interval:
-            due.append(m)
-    if not due:
-        return
+    for m in due:
+        in_flight.add(m.id)
+        task = asyncio.create_task(_check_and_record(app, m))
+        tasks.add(task)
+        task.add_done_callback(tasks.discard)
 
-    pairs = await asyncio.gather(
-        *[run_monitor(m) for m in due], return_exceptions=True
-    )
-    check_time = datetime.now(timezone.utc)
 
-    async with app.db_sessionmaker() as session:
-        async with session.begin():
-            for monitor, res in zip(due, pairs):
-                if isinstance(res, Exception):
-                    logger.error("Monitor %r check errored: %s", monitor.name, res)
-                    continue
+async def _check_and_record(app, monitor: Monitor) -> None:
+    """Run one monitor's check and store its result and metrics."""
+    in_flight, _ = _state(app)
+    try:
+        try:
+            res = await asyncio.wait_for(run_monitor(monitor), CHECK_DEADLINE_SECONDS)
+        except asyncio.TimeoutError:
+            res = dict(_BLANK, error_msg=f"Check did not finish within {CHECK_DEADLINE_SECONDS} s")
+        check_time = datetime.now(timezone.utc)
+        async with app.db_sessionmaker() as session:
+            async with session.begin():
                 session.add(MonitorResult(
                     monitor_id=monitor.id,
                     checked_at=check_time,
@@ -58,10 +98,6 @@ async def run_due_monitors(app) -> None:
                     tls_expires_at=res["tls_expires_at"],
                     error_msg=res["error_msg"],
                 ))
-                db_monitor = await session.get(Monitor, monitor.id)
-                if db_monitor:
-                    db_monitor.last_checked_at = check_time
-
                 # Alert pipeline — keyed by monitor type (source) + name.
                 batch = MetricBatch()
                 batch.add(
@@ -76,3 +112,7 @@ async def run_due_monitors(app) -> None:
                         value=res["response_ms"],
                     )
                 await batch.record(session)
+    except Exception:
+        logger.exception("Monitor %r check errored", monitor.name)
+    finally:
+        in_flight.discard(monitor.id)

tests/core/test_monitor_schedule.py

@@ -0,0 +1,154 @@
+"""Monitors run on their own interval, not the dispatcher's: due is measured
+from the scheduled start, each check runs on its own, one still running is not
+started twice, and a check that never finishes is recorded down (#5151)."""
+import asyncio
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace as NS
+
+import pytest
+
+from steward.monitors import scheduler as sched
+from steward.monitors.scheduler import DISPATCH_TICK_SECONDS, is_due
+
+NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)
+
+
+def _m(mid="m1", interval=0, checked_ago=None, name=None):
+    return NS(id=mid, name=name or mid, type="http", enabled=True, check_interval_seconds=interval,
+              last_checked_at=None if checked_ago is None else NOW - timedelta(seconds=checked_ago))
+
+
+@pytest.mark.parametrize("monitor, due", [
+    (_m(), True),                                   # never checked
+    (_m(checked_ago=60), True),                     # exactly one interval (global 60)
+    (_m(checked_ago=60 - DISPATCH_TICK_SECONDS / 2), True),  # within half a tick: not pushed a tick late
+    (_m(checked_ago=55), False),
+    (_m(interval=30, checked_ago=30), True),        # its own interval, not the global one
+    (_m(interval=30, checked_ago=20), False),
+    (_m(interval=300, checked_ago=120), False),
+    (_m(checked_ago=86400), True),                  # long overdue: due, and runs once
+])
+def test_is_due(monitor, due):
+    assert is_due(monitor, NOW, 60) is due
+
+
+class _Store:
+    def __init__(self, monitors):
+        self.monitors, self.stamped, self.added = monitors, [], []
+
+
+class _Session:
+    def __init__(self, store):
+        self.store = store
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    def begin(self):
+        return self
+
+    async def execute(self, stmt):
+        if getattr(stmt, "is_update", False):
+            # Like the real stamp: every monitor in this fake was due.
+            self.store.stamped.append(stmt)
+            for m in self.store.monitors:
+                m.last_checked_at = datetime.now(timezone.utc)
+            return NS()
+        return NS(scalars=lambda: iter(self.store.monitors))
+
+    async def commit(self):
+        pass
+
+    def add(self, obj):
+        self.store.added.append(obj)
+
+
+class _Batch:
+    def add(self, **kw):
+        pass
+
+    async def record(self, session):
+        pass
+
+
+def _app(store):
+    return NS(config={"MONITORS_POLL_INTERVAL": 60}, db_sessionmaker=lambda: _Session(store))
+
+
+async def _settle(app):
+    _, tasks = sched._state(app)
+    if tasks:
+        await asyncio.gather(*list(tasks))
+
+
+async def test_a_slow_check_holds_up_nobody_and_is_not_started_twice(monkeypatch):
+    release = asyncio.Event()
+    started = []
+
+    async def fake_run(m):
+        started.append(m.id)
+        if m.id == "slow":
+            await release.wait()
+        return dict(sched._BLANK, is_up=True)
+
+    monkeypatch.setattr(sched, "run_monitor", fake_run)
+    monkeypatch.setattr(sched, "MetricBatch", _Batch)
+    store = _Store([_m("slow"), _m("fast")])
+    app = _app(store)
+
+    await sched.run_due_monitors(app)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert sorted(started) == ["fast", "slow"]
+    assert [r.monitor_id for r in store.added] == ["fast"]   # recorded while slow still runs
+
+    await sched.run_due_monitors(app)                          # next tick: slow is in flight
+    await asyncio.sleep(0)
+    assert started.count("slow") == 1
+
+    release.set()
+    await _settle(app)
+    assert sorted(r.monitor_id for r in store.added) == ["fast", "slow"]
+    assert sched._state(app)[0] == set()
+    assert len(store.stamped) == 1   # stamped once, when both were due
+
+
+async def test_a_check_that_raises_frees_its_monitor(monkeypatch):
+    async def boom(m):
+        raise RuntimeError("probe broke")
+
+    monkeypatch.setattr(sched, "run_monitor", boom)
+    monkeypatch.setattr(sched, "MetricBatch", _Batch)
+    store = _Store([_m("m1")])
+    app = _app(store)
+    await sched.run_due_monitors(app)
+    await _settle(app)
+    assert store.added == [] and sched._state(app)[0] == set()
+
+
+async def test_a_check_that_never_finishes_is_recorded_down(monkeypatch):
+    async def hang(m):
+        await asyncio.sleep(3600)
+
+    monkeypatch.setattr(sched, "run_monitor", hang)
+    monkeypatch.setattr(sched, "MetricBatch", _Batch)
+    monkeypatch.setattr(sched, "CHECK_DEADLINE_SECONDS", 0.01)
+    store = _Store([_m("m1")])
+    app = _app(store)
+    await sched.run_due_monitors(app)
+    await _settle(app)
+    [res] = store.added
+    assert res.is_up is False and "did not finish" in res.error_msg
+    assert sched._state(app)[0] == set()
+
+
+async def test_nothing_due_stamps_nothing(monkeypatch):
+    m = _m("m1")
+    m.last_checked_at = datetime.now(timezone.utc) - timedelta(seconds=10)
+    store = _Store([m])
+    app = _app(store)
+    await sched.run_due_monitors(app)
+    assert store.stamped == [] and store.added == []

tests/integration/test_alert_states.py

@@ -114,3 +114,51 @@ def test_acknowledge_holds_clear_after_and_reset(app):
         assert (await events())[-1] == ("firing", "resolved", "rule disabled")
 
     asyncio.run(_go())
+
+
+def test_record_metrics_marks_each_step(app):
+    """A caller timing an ingest can tell the row write from the rule reads
+    from the evaluation (#5149)."""
+    from steward.alerts import service as alert_svc
+    from steward.core.alerts import record_metrics
+    from steward.models.metrics import PluginMetric
+    from steward.models.users import User, UserRole
+
+    class _Marks:
+        def __init__(self):
+            self.names = []
+
+        def mark(self, name):
+            self.names.append(name)
+
+    def rows(res):
+        return [PluginMetric(source_module="host_agent", resource_name=res,
+                             metric_name="cpu_pct", value=5.0)]
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        watched, unwatched = f"box-{tag}", f"other-{tag}"
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                user = User(username=f"mk-{tag}", email=f"mk-{tag}@test", password_hash="x",
+                            role=UserRole.admin)
+                s.add(user)
+                s.add_all(rows(watched))
+                await s.flush()
+                fields = await alert_svc.rule_fields(
+                    s, name=f"hot {tag}", source_module="host_agent", resource_name=watched,
+                    metric_name="cpu_pct", operator=">", threshold=90)
+                await alert_svc.add_rule(s, fields, created_by=user.id)
+
+        seen = {}
+        for res in (watched, unwatched):
+            marks = _Marks()
+            async with app.db_sessionmaker() as s:
+                async with s.begin():
+                    await record_metrics(s, rows(res), marks)
+            seen[res] = marks.names
+        return seen[watched], seen[unwatched]
+
+    watched, unwatched = asyncio.run(_go())
+    assert watched == ["flush", "rules", "evaluate"]
+    assert unwatched == ["flush", "rules"]   # no rule watches it: nothing to evaluate

tests/plugins/host_agent/test_ingest_timing.py

@@ -51,7 +51,8 @@ async def test_a_dropped_ingest_says_how_far_it_got(app, monkeypatch, caplog):
         phases = routes._Phases()
         clock.t += 0.2
         phases.mark("parse")
+        phases.size = "samples=21 metric_rows=840"
         clock.t += 9.9
         with caplog.at_level(logging.WARNING):
             phases.cancelled("abcd1234")
-    assert "dropped after 10.1s for token=abcd1234… (completed: parse=0.2s)" in caplog.text
+    assert "dropped after 10.1s for token=abcd1234… (completed: parse=0.2s) samples=21 metric_rows=840" in caplog.text