From f802c9f153f217492bc738890dd2a906c0f7bf1a Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 18:34:54 -0400 Subject: [PATCH] feat(host_agent): a dropped ingest says which step of the metric write it stalled in (#5149) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- plugins/host_agent/routes.py | 9 ++-- steward/core/alerts.py | 10 +++- tests/integration/test_alert_states.py | 48 +++++++++++++++++++ .../plugins/host_agent/test_ingest_timing.py | 3 +- 4 files changed, 64 insertions(+), 6 deletions(-) diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 8fbbf9d..cb75548 100644 --- a/plugins/host_agent/routes.py +++ b/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 diff --git a/steward/core/alerts.py b/steward/core/alerts.py index 665da88..852be9f 100644 --- a/steward/core/alerts.py +++ b/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 diff --git a/tests/integration/test_alert_states.py b/tests/integration/test_alert_states.py index 82cbd21..c2efcdb 100644 --- a/tests/integration/test_alert_states.py +++ b/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 diff --git a/tests/plugins/host_agent/test_ingest_timing.py b/tests/plugins/host_agent/test_ingest_timing.py index 5b5c9e3..9d31844 100644 --- a/tests/plugins/host_agent/test_ingest_timing.py +++ b/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