feat(host_agent): say where a slow ingest spends its time, and how far a dropped one got (#5149)

Every host-agent ingest in the service logs ends at ~10 s with no status:
the agent gives up at its urlopen timeout (agent.py) before Steward answers.
Nothing says which part of ingest is slow, so this measures it:

- each phase (parse, metrics with alert evaluation, docker persist,
  commit) is timed, and a request past 2 s logs one line with the phases,
  sample count, metric rows, containers and log lines;
- a request the server drops when the agent hangs up (CancelledError) logs
  how long it ran and which phases completed, then re-raises.

The fix follows once the logs say which phase it is.

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

plugins/host_agent/routes.py

@@ -1,7 +1,9 @@
 # plugins/host_agent/routes.py
 from __future__ import annotations
 
+import asyncio
 import hashlib
+import time
 from datetime import datetime, timezone
 from typing import Any
 
@@ -188,122 +190,170 @@ async def ingest():
         return _error(400, "malformed_payload", "'samples' must be a non-empty list")
 
     token_hash = _hash_token(raw)
-    async with current_app.db_sessionmaker() as session:
+    phases = _Phases()
-        reg = (await session.execute(
+    try:
-            select(HostAgentRegistration).where(
+        async with current_app.db_sessionmaker() as session:
-                HostAgentRegistration.token_hash == token_hash)
+            reg = (await session.execute(
-        )).scalar_one_or_none()
+                select(HostAgentRegistration).where(
-        if reg is None:
+                    HostAgentRegistration.token_hash == token_hash)
-            return _error(401, "invalid_token")
+            )).scalar_one_or_none()
-        host = (await session.execute(
+            if reg is None:
-            select(Host).where(Host.id == reg.host_id)
+                return _error(401, "invalid_token")
-        )).scalar_one_or_none()
+            host = (await session.execute(
-        if host is None:
+                select(Host).where(Host.id == reg.host_id)
-            return _error(401, "invalid_token")
+            )).scalar_one_or_none()
+            if host is None:
+                return _error(401, "invalid_token")
 
-        accepted = 0
+            accepted = 0
-        metric_rows: list[PluginMetric] = []
+            metric_rows: list[PluginMetric] = []
-        latest_ts: datetime | None = None
+            latest_ts: datetime | None = None
-        docker_snapshots: list[tuple[datetime, list]] = []
+            docker_snapshots: list[tuple[datetime, list]] = []
-        docker_log_batches: list[tuple[datetime, list]] = []
+            docker_log_batches: list[tuple[datetime, list]] = []
-        latest_swarm: dict | None = None
+            latest_swarm: dict | None = None
-        latest_swarm_ts: datetime | None = None
+            latest_swarm_ts: datetime | None = None
-        latest_disk: dict | None = None
+            latest_disk: dict | None = None
-        latest_disk_ts: datetime | None = None
+            latest_disk_ts: datetime | None = None
-        for sample in samples:
+            for sample in samples:
-            try:
-                recorded_at = _parse_ts(sample["ts"])
-            except (KeyError, ValueError):
-                continue
-            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))
-            # Container logs are time-series (append every line, not newest-only);
-            # each record carries its own Docker ts, recorded_at is the fallback.
-            docker_logs = sample.get("docker_logs")
-            if isinstance(docker_logs, list) and docker_logs:
-                docker_log_batches.append((recorded_at, docker_logs))
-            # Swarm is current-state, not time-series — keep only the newest
-            # sample's topology (a manager re-reports it every interval).
-            swarm = sample.get("swarm")
-            if isinstance(swarm, dict) and (
-                    latest_swarm_ts is None or recorded_at > latest_swarm_ts):
-                latest_swarm, latest_swarm_ts = swarm, recorded_at
-            # Disk usage is current-state too — keep only the newest sample's.
-            disk = sample.get("docker_disk")
-            if isinstance(disk, dict) and (
-                    latest_disk_ts is None or recorded_at > latest_disk_ts):
-                latest_disk, latest_disk_ts = disk, recorded_at
-            accepted += 1
-            if latest_ts is None or recorded_at > latest_ts:
-                latest_ts = recorded_at
-
-        if accepted == 0:
-            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
-        # whole ingest, so the metrics above still land.
-        if (docker_snapshots or latest_swarm is not None
-                or latest_disk is not None or docker_log_batches):
-            from steward.core.capabilities import has_capability, invoke_capability
-            if has_capability("docker.persist_host_samples"):
                 try:
-                    # SAVEPOINT so a docker-side failure rolls back only the
+                    recorded_at = _parse_ts(sample["ts"])
-                    # docker writes, leaving the host metrics intact to commit.
+                except (KeyError, ValueError):
-                    async with session.begin_nested():
+                    continue
-                        await invoke_capability(
+                metric_rows.extend(_expand_sample_to_metrics(sample, host.name, recorded_at))
-                            "docker.persist_host_samples", UserRole.admin,
+                docker = sample.get("docker")
-                            session, host, docker_snapshots, latest_swarm,
+                if isinstance(docker, list) and docker:
-                            latest_disk, docker_log_batches,
+                    docker_snapshots.append((recorded_at, docker))
-                        )
+                # Container logs are time-series (append every line, not newest-only);
-                except Exception:
+                # each record carries its own Docker ts, recorded_at is the fallback.
-                    current_app.logger.exception(
+                docker_logs = sample.get("docker_logs")
-                        "host_agent ingest: docker persist failed for host=%s",
+                if isinstance(docker_logs, list) and docker_logs:
-                        host.name)
+                    docker_log_batches.append((recorded_at, docker_logs))
+                # Swarm is current-state, not time-series — keep only the newest
+                # sample's topology (a manager re-reports it every interval).
+                swarm = sample.get("swarm")
+                if isinstance(swarm, dict) and (
+                        latest_swarm_ts is None or recorded_at > latest_swarm_ts):
+                    latest_swarm, latest_swarm_ts = swarm, recorded_at
+                # Disk usage is current-state too — keep only the newest sample's.
+                disk = sample.get("docker_disk")
+                if isinstance(disk, dict) and (
+                        latest_disk_ts is None or recorded_at > latest_disk_ts):
+                    latest_disk, latest_disk_ts = disk, recorded_at
+                accepted += 1
+                if latest_ts is None or recorded_at > latest_ts:
+                    latest_ts = recorded_at
 
-        md = payload.get("metadata") or {}
+            if accepted == 0:
-        changed = False
+                await session.rollback()
-        if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at):
+                return _error(400, "malformed_payload", "no valid samples")
-            reg.last_seen_at = latest_ts
+            phases.mark("parse")
-            changed = True
+
-        version = payload.get("agent_version")
+            # Through the alert pipeline, so rules on host metrics can fire.
-        if version and reg.agent_version != version:
+            from steward.core.alerts import record_metrics
-            reg.agent_version = version
+            await record_metrics(session, metric_rows)
-            changed = True
+            phases.mark("metrics")
-        for field in ("kernel", "distro", "arch"):
+
-            if field in md and getattr(reg, field) != md[field]:
+            # Hand host-scoped container data to the docker plugin if it's enabled
-                setattr(reg, field, md[field])
+            # (opportunistic synergy via the capability registry — no hard import,
+            # no-op when docker is disabled). A failure here must never sink the
+            # whole ingest, so the metrics above still land.
+            if (docker_snapshots or latest_swarm is not None
+                    or latest_disk is not None or docker_log_batches):
+                from steward.core.capabilities import has_capability, invoke_capability
+                if has_capability("docker.persist_host_samples"):
+                    try:
+                        # SAVEPOINT so a docker-side failure rolls back only the
+                        # docker writes, leaving the host metrics intact to commit.
+                        async with session.begin_nested():
+                            await invoke_capability(
+                                "docker.persist_host_samples", UserRole.admin,
+                                session, host, docker_snapshots, latest_swarm,
+                                latest_disk, docker_log_batches,
+                            )
+                    except Exception:
+                        current_app.logger.exception(
+                            "host_agent ingest: docker persist failed for host=%s",
+                            host.name)
+                    phases.mark("docker")
+
+            md = payload.get("metadata") or {}
+            changed = False
+            if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at):
+                reg.last_seen_at = latest_ts
                 changed = True
-        host_ip = md.get("host_ip")
+            version = payload.get("agent_version")
-        if host_ip and reg.host_ip != host_ip:
+            if version and reg.agent_version != version:
-            reg.host_ip = host_ip
+                reg.agent_version = version
-            changed = True
+                changed = True
-        # Mirror the reported IP into Host.address only when it's blank.
+            for field in ("kernel", "distro", "arch"):
-        new_address = pick_host_address(host.address, host_ip)
+                if field in md and getattr(reg, field) != md[field]:
-        if new_address is not None:
+                    setattr(reg, field, md[field])
-            host.address = new_address
+                    changed = True
-        if changed:
+            host_ip = md.get("host_ip")
-            reg.updated_at = datetime.now(timezone.utc)
+            if host_ip and reg.host_ip != host_ip:
+                reg.host_ip = host_ip
+                changed = True
+            # Mirror the reported IP into Host.address only when it's blank.
+            new_address = pick_host_address(host.address, host_ip)
+            if new_address is not None:
+                host.address = new_address
+            if changed:
+                reg.updated_at = datetime.now(timezone.utc)
 
-        if latest_ts:
+            if latest_ts:
-            skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds())
+                skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds())
-            if skew > 300:
+                if skew > 300:
-                current_app.logger.warning(
+                    current_app.logger.warning(
-                    "host_agent ingest: clock skew %.0fs for host=%s", skew, host.name)
+                        "host_agent ingest: clock skew %.0fs for host=%s", skew, host.name)
 
-        await session.commit()
+            await session.commit()
+            phases.mark("commit")
+    except asyncio.CancelledError:
+        # The agent hung up (its 10 s timeout) and the server dropped the
+        # request: say how far it got before re-raising.
+        phases.cancelled(token_hash[:8])
+        raise
 
+    phases.report(host.name, accepted, len(metric_rows),
+                  sum(len(c) for _, c in docker_snapshots),
+                  sum(len(lines) for _, lines in docker_log_batches))
     return jsonify({"ok": True, "samples_accepted": accepted}), 200
 
 
+# The agent gives up on a post after 10 s (agent.py); an ingest past this
+# says where its time went, so a slow phase shows in the logs (#5149).
+SLOW_INGEST_SECONDS = 2.0
+
+
+class _Phases:
+    """Wall-clock per phase of one ingest request."""
+
+    def __init__(self):
+        self.start = self.last = time.monotonic()
+        self.took: dict[str, float] = {}
+
+    def mark(self, name: str) -> None:
+        now = time.monotonic()
+        self.took[name] = self.took.get(name, 0.0) + (now - self.last)
+        self.last = now
+
+    def cancelled(self, who: str) -> None:
+        current_app.logger.warning(
+            "host_agent ingest: dropped after %.1fs for token=%s… (completed: %s)",
+            time.monotonic() - self.start, who,
+            " ".join(f"{k}={v:.1f}s" for k, v in self.took.items()) or "nothing")
+
+    def report(self, host: str, samples: int, metrics: int, containers: int, log_lines: int) -> None:
+        total = time.monotonic() - self.start
+        if total < SLOW_INGEST_SECONDS:
+            return
+        current_app.logger.warning(
+            "host_agent ingest: slow %.1fs for host=%s (%s) samples=%d metric_rows=%d "
+            "containers=%d log_lines=%d", total, host,
+            " ".join(f"{k}={v:.1f}s" for k, v in self.took.items()),
+            samples, metrics, containers, log_lines)
+
+
 AGENT_SOURCE_PATH = Path(__file__).parent / "agent.py"
 
 

tests/plugins/host_agent/test_ingest_timing.py

@@ -0,0 +1,57 @@
+"""A slow ingest says where its time went, and one dropped when the agent hung
+up says how far it got (#5149)."""
+import logging
+
+from plugins.host_agent import routes
+
+
+class _Clock:
+    def __init__(self):
+        self.t = 100.0
+
+    def __call__(self):
+        return self.t
+
+
+async def test_a_slow_ingest_logs_its_phases(app, monkeypatch, caplog):
+    clock = _Clock()
+    monkeypatch.setattr(routes.time, "monotonic", clock)
+    async with app.app_context():
+        phases = routes._Phases()
+        clock.t += 0.1
+        phases.mark("parse")
+        clock.t += 8.0
+        phases.mark("metrics")
+        clock.t += 0.4
+        phases.mark("commit")
+        with caplog.at_level(logging.WARNING):
+            phases.report("web-1", 12, 480, 30, 900)
+    line = caplog.text
+    assert "slow 8.5s for host=web-1" in line
+    assert "parse=0.1s metrics=8.0s commit=0.4s" in line
+    assert "samples=12 metric_rows=480 containers=30 log_lines=900" in line
+
+
+async def test_a_quick_ingest_logs_nothing(app, monkeypatch, caplog):
+    clock = _Clock()
+    monkeypatch.setattr(routes.time, "monotonic", clock)
+    async with app.app_context():
+        phases = routes._Phases()
+        clock.t += 0.3
+        phases.mark("commit")
+        with caplog.at_level(logging.WARNING):
+            phases.report("web-1", 1, 40, 0, 0)
+    assert "host_agent ingest" not in caplog.text
+
+
+async def test_a_dropped_ingest_says_how_far_it_got(app, monkeypatch, caplog):
+    clock = _Clock()
+    monkeypatch.setattr(routes.time, "monotonic", clock)
+    async with app.app_context():
+        phases = routes._Phases()
+        clock.t += 0.2
+        phases.mark("parse")
+        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