From fc9730a2e526eafe633bdd595dfb9532098a8ab0 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 18:28:30 -0400 Subject: [PATCH] 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 --- plugins/host_agent/routes.py | 262 +++++++++++------- .../plugins/host_agent/test_ingest_timing.py | 57 ++++ 2 files changed, 213 insertions(+), 106 deletions(-) create mode 100644 tests/plugins/host_agent/test_ingest_timing.py diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 503c711..8fbbf9d 100644 --- a/plugins/host_agent/routes.py +++ b/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: - reg = (await session.execute( - select(HostAgentRegistration).where( - HostAgentRegistration.token_hash == token_hash) - )).scalar_one_or_none() - if reg is None: - return _error(401, "invalid_token") - host = (await session.execute( - select(Host).where(Host.id == reg.host_id) - )).scalar_one_or_none() - if host is None: - return _error(401, "invalid_token") + phases = _Phases() + try: + async with current_app.db_sessionmaker() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.token_hash == token_hash) + )).scalar_one_or_none() + if reg is None: + return _error(401, "invalid_token") + host = (await session.execute( + select(Host).where(Host.id == reg.host_id) + )).scalar_one_or_none() + if host is None: + 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]] = [] - latest_swarm: dict | None = None - latest_swarm_ts: datetime | None = None - latest_disk: dict | None = None - latest_disk_ts: datetime | None = None - 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"): + accepted = 0 + metric_rows: list[PluginMetric] = [] + latest_ts: datetime | None = None + docker_snapshots: list[tuple[datetime, list]] = [] + docker_log_batches: list[tuple[datetime, list]] = [] + latest_swarm: dict | None = None + latest_swarm_ts: datetime | None = None + latest_disk: dict | None = None + latest_disk_ts: datetime | None = None + for sample in 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) + 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 - 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 - version = payload.get("agent_version") - if version and reg.agent_version != version: - reg.agent_version = version - changed = True - for field in ("kernel", "distro", "arch"): - if field in md and getattr(reg, field) != md[field]: - setattr(reg, field, md[field]) + if accepted == 0: + await session.rollback() + return _error(400, "malformed_payload", "no valid samples") + phases.mark("parse") + + # 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") + + # 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 + # 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") - 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) + version = payload.get("agent_version") + if version and reg.agent_version != version: + reg.agent_version = version + changed = True + for field in ("kernel", "distro", "arch"): + if field in md and getattr(reg, field) != md[field]: + setattr(reg, field, md[field]) + changed = True + host_ip = md.get("host_ip") + 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: - skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds()) - if skew > 300: - current_app.logger.warning( - "host_agent ingest: clock skew %.0fs for host=%s", skew, host.name) + if latest_ts: + skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds()) + if skew > 300: + current_app.logger.warning( + "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" diff --git a/tests/plugins/host_agent/test_ingest_timing.py b/tests/plugins/host_agent/test_ingest_timing.py new file mode 100644 index 0000000..5b5c9e3 --- /dev/null +++ b/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