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/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

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