From 4978d28a513f55dd4edcda4f302517f806992d08 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 20:50:05 -0400 Subject: [PATCH] fix(docker): the slow-ingest line names which part of the Docker persist took the time (#5181) persist_host_docker marks docker.status, .swarm, .disk, .logs, .metrics, .events, .containers, .alerts and .reap on the ingest's phase timer, each flushed before its mark so a part's writes are charged to it. Co-Authored-By: Claude Opus 5.5 --- plugins/docker/ingest.py | 20 +++++++++++++++++- plugins/host_agent/routes.py | 1 + tests/integration/test_docker.py | 35 ++++++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 1 deletion(-) diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index c9a0857..5e97ef4 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -289,7 +289,7 @@ async def _expire_swarm(session, host) -> None: async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, - logs=None, status=None) -> None: + logs=None, status=None, phases=None) -> None: """Upsert containers + time-series + lifecycle events + swarm for one host. `snapshots` is a list of (recorded_at: datetime, containers: list[dict]) — @@ -305,20 +305,33 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, sample's {"status", "detail", "at", "swarm_manager"} (agent ≥ 1.9.0): an "ok" with an empty snapshot clears the host's containers, and a node that says it is no longer a manager loses the swarm rows it reported as one. + + `phases`, when given, is marked after each part (docker.status, .swarm, + .disk, .logs, .metrics, .events, .containers, .alerts, .reap), each + flushed before its mark, so a slow ingest names the part (#5181). """ from steward.core.alerts import MetricBatch from .models import DockerContainer, DockerEvent, DockerMetric + async def mark(name: str) -> None: + if phases is not None: + await session.flush() + phases.mark(f"docker.{name}") + if status is not None: await _persist_status(session, host, status) if status.get("swarm_manager") is False: await _expire_swarm(session, host) + await mark("status") if swarm is not None: await _persist_swarm(session, host, swarm) + await mark("swarm") if disk is not None: await _persist_disk(session, host, disk) + await mark("disk") if logs: await _persist_logs(session, host, logs) + await mark("logs") if not snapshots: return @@ -337,6 +350,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, mem_pct=c.get("mem_pct") or 0.0, mem_usage_bytes=c.get("mem_usage_bytes") or 0, )) + await mark("metrics") # Snapshot of stored per-container state BEFORE the upsert overwrites it — # used to diff lifecycle events. Skip event derivation on the host's first @@ -356,6 +370,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, host_id=host.id, container_name=name, event=event, detail=detail, at=latest_at, )) + await mark("events") # Current state + alerts from the newest snapshot only. batch = MetricBatch() @@ -423,7 +438,9 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, value=1.0 if health == "healthy" else 0.0, ) + await mark("containers") await batch.record(session) + await mark("alerts") # Reap containers that vanished from the host's latest listing. Without this, # removed one-shot containers (CI job runners, buildkit builders, codex jobs) @@ -437,3 +454,4 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, if seen_names: reap = reap.where(DockerContainer.name.notin_(seen_names)) await session.execute(reap) + await mark("reap") diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 8efc3cf..dbdf77c 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -283,6 +283,7 @@ async def ingest(): "docker.persist_host_samples", UserRole.admin, session, host, docker_snapshots, latest_swarm, latest_disk, docker_log_batches, docker_status, + phases=phases, ) except Exception: current_app.logger.exception( diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index b8c407b..3b4fe93 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -383,6 +383,41 @@ def test_vanished_container_is_reaped(app): assert after == ["keep"] # the vanished container was reaped +@_NEEDS_DB +def test_persist_marks_each_part_for_the_slow_ingest_line(app): + """Each part of the docker persist is marked, so the slow-ingest line + names the part its time went to (#5181).""" + from steward.models.hosts import Host + + class _Marks: + def __init__(self): + self.names = [] + + def mark(self, name): + self.names.append(name) + + persist = _persist_fn(app) + at = datetime(2026, 10, 6, 23, 0, 0, tzinfo=timezone.utc) + c = {"name": "web", "status": "running", "cpu_pct": 1.0, "mem_pct": 1.0, + "restart_count": 0, "exit_code": None, "oom_killed": False, "health": None} + logs = [(at, [{"container": "web", "stream": "stdout", "ts": None, "line": "hello"}])] + marks = _Marks() + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + h = Host(id=str(uuid.uuid4()), name="marks", address="10.9.9.7") + s.add(h) + await s.flush() + await persist(s, h, [(at, [c])], None, None, logs, + {"status": "ok", "detail": None, "at": at, "swarm_manager": None}, + phases=marks) + + asyncio.run(_go()) + assert marks.names == ["docker.status", "docker.logs", "docker.metrics", "docker.events", + "docker.containers", "docker.alerts", "docker.reap"] + + @_NEEDS_DB def test_an_empty_snapshot_from_a_docker_that_answered_clears_the_host(app): """Docker answering with no containers removes the host's rows (#5152); a