From 326a9cd254a6386f740840e0d4d69cce9d01551f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 20:58:46 -0400 Subject: [PATCH] fix(docker): a container that crashes between two samples leaves a die event and its last log lines (#5049) Agent 1.11.0 sends finished_at (State.FinishedAt) and reads a stopped container's logs once per stop: since its cursor if it was being read, or its last 200 lines when first seen already crashed (non-zero exit, stopped in the last 10 min). Cursors and finals are forgotten once a container is no longer listed. Ingest gives a container first seen already exited non-zero a die event, and dates die/stop events by finished_at when the agent sends it. Co-Authored-By: Claude Opus 5.5 --- plugins/docker/ingest.py | 15 +++- plugins/host_agent/agent.py | 77 +++++++++++++------ tests/plugins/docker/test_ingest_events.py | 16 ++++ tests/plugins/host_agent/test_agent_docker.py | 69 +++++++++++++++++ 4 files changed, 151 insertions(+), 26 deletions(-) diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 5e97ef4..d493f21 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -40,6 +40,8 @@ def _derive_events(old_state: dict, new_containers: list) -> list: start — a container transitions into running (or a genuinely new container appears already running) + die — also for a new container first seen already exited + non-zero: a task that crashed between two samples (#5049) restart — Docker restarted it (its restart count went up); said instead of a start, since that is what happened stop — running → not-running, or a running container is removed @@ -64,10 +66,12 @@ def _derive_events(old_state: dict, new_containers: list) -> list: old = old_state.get(name) if old is None: - # Newly observed container — only a start is meaningful (we have no - # prior state to diff a death against). + # Newly observed container: a start, or, when it is already down + # with a non-zero exit, the death that happened between samples. if new_running: events.append((name, "start", c.get("image") or None)) + elif new_status in _STOPPED_STATES and new_exit not in (None, 0): + events.append((name, "die", f"exit {new_exit}")) continue old_running = (old.get("status") or "").lower() == "running" @@ -365,10 +369,15 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, for r in old_rows } if old_state: + # A stop is dated when Docker says it happened (agent ≥ 1.11), not when + # the sample that noticed it was taken. + finished = {c.get("name"): _parse_started_at(c.get("finished_at")) + for c in latest_containers} for name, event, detail in _derive_events(old_state, latest_containers): + at = finished.get(name) if event in ("die", "stop") else None session.add(DockerEvent( host_id=host.id, container_name=name, - event=event, detail=detail, at=latest_at, + event=event, detail=detail, at=at or latest_at, )) await mark("events") diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 1922523..493e14c 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -20,7 +20,7 @@ from collections import deque from concurrent.futures import ThreadPoolExecutor, wait as wait_futures from datetime import datetime, timezone -AGENT_VERSION = "1.10.0" +AGENT_VERSION = "1.11.0" # Default path to the local Docker Engine socket. Overridable via the # `docker_socket` config key; collection is silently skipped if it's absent or @@ -456,6 +456,8 @@ DOCKER_DF_INTERVAL_SECONDS = 600.0 # kept per container in the agent's rate-state. DOCKER_LOG_BATCH_MAX_BYTES = 262144 # 256 KiB of log text per push, all containers DOCKER_LOG_FIRST_TAIL = 50 # lines seeded on a container's first interval +DOCKER_LOG_FINAL_TAIL = 200 # lines read once from a container found already crashed +DOCKER_LOG_FINAL_WINDOW_SECONDS = 600 # ... when it stopped this recently def _dechunk(body: bytes) -> bytes: @@ -599,12 +601,12 @@ def _docker_inspect_fields(insp: dict): return health, restart_count, exit_code, oom_killed -def _docker_started_at(insp: dict): +def _docker_started_at(insp: dict, key: str = "StartedAt"): """State.StartedAt as ISO-8601, or None for a container never started (Docker's zero time) or an inspect that failed. When the container last started, not when it was created: a restarted container's uptime counts - from its restart.""" - raw = (((insp or {}).get("State") or {}).get("StartedAt") or "") + from its restart. key="FinishedAt" reads when it last stopped.""" + raw = (((insp or {}).get("State") or {}).get(key) or "") if not raw or raw.startswith("0001-"): return None ts = _parse_log_ts(raw) @@ -665,6 +667,9 @@ def _collect_one_container(socket_path: str, c: dict) -> dict: pass started_at = _docker_started_at(insp) + # When it last stopped (a running container's is its previous stop, or + # None): dates a die event, and tells the log collector a stop is new. + finished_at = None if state == "running" else _docker_started_at(insp, "FinishedAt") record = { "name": name, "container_id": cid[:12], @@ -676,6 +681,7 @@ def _collect_one_container(socket_path: str, c: dict) -> dict: "mem_pct": mem_pct, "ports": _docker_ports(c.get("Ports", [])), "started_at": started_at, + "finished_at": finished_at, "health": health, "restart_count": restart_count, "exit_code": exit_code, @@ -872,39 +878,64 @@ def _parse_container_logs(raw: bytes) -> list: def collect_docker_logs(socket_path: str, containers: list, state: dict, exclude=None, - max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES) -> list: - """New container-log lines since the last interval, for running containers. + max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES, + now: float | None = None) -> list: + """New container-log lines since the last interval. Keeps a per-container since-cursor in `state["docker_log_cursors"]` (name → unix ts) so each interval fetches only what's new; a container's - first interval seeds from a short tail. The whole batch is capped at - `max_bytes` of log text — once exceeded a truncation marker is appended and - collection stops (logs must never balloon a push). `exclude` container names - are skipped entirely (a per-host bandwidth opt-out). Best-effort: a - per-container transport error just contributes nothing this interval. + first interval seeds from a short tail. A running container is read every + interval. A stopped one is read once per stop (`finished_at`, remembered in + `state["docker_log_finals"]`), so the lines it wrote on its way down reach + Steward (#5049): since its cursor if it was being read, or, for one first + seen already stopped with a non-zero exit in the last + DOCKER_LOG_FINAL_WINDOW_SECONDS, its last DOCKER_LOG_FINAL_TAIL lines. That + is how a task that crashes between two samples is explained by its own + output. Both maps forget a container once it is no longer listed. + + The whole batch is capped at `max_bytes` of log text — once exceeded a + truncation marker is appended and collection stops (logs must never balloon + a push). `exclude` container names are skipped entirely (a per-host + bandwidth opt-out). Best-effort: a per-container transport error just + contributes nothing this interval. Returns [{"container", "stream", "ts", "line"}, …]. """ exclude_set = set(exclude or ()) cursors = state.setdefault("docker_log_cursors", {}) - running = {c.get("name") for c in (containers or []) - if c.get("name") and c.get("status") == "running"} - # Forget cursors for containers no longer running so state can't grow without - # bound over the agent's lifetime. - for gone in [k for k in cursors if k not in running]: - del cursors[gone] + finals = state.setdefault("docker_log_finals", {}) + now = time.time() if now is None else now + listed = {c.get("name"): c for c in (containers or []) if c.get("name")} + # Forget containers no longer listed so state can't grow without bound + # over the agent's lifetime. + for book in (cursors, finals): + for gone in [k for k in book if k not in listed]: + del book[gone] out: list = [] total = 0 truncated = False - for c in (containers or []): - name = c.get("name") - if not name or name not in running or name in exclude_set: + for name, c in listed.items(): + if name in exclude_set: continue prev = cursors.get(name) base = f"/containers/{name}/logs?stdout=1&stderr=1×tamps=1" - # since accepts a fractional Unix ts; we still dedup the boundary line. - path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None - else f"{base}&since={prev:.9f}") + if c.get("status") == "running": + # since accepts a fractional Unix ts; we still dedup the boundary line. + path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None + else f"{base}&since={prev:.9f}") + else: + stopped = c.get("finished_at") + if not stopped or finals.get(name) == stopped: + continue + if prev is not None: + path = f"{base}&since={prev:.9f}" + else: + at = _parse_log_ts(stopped) + if (c.get("exit_code") in (None, 0) or at is None + or now - at.timestamp() > DOCKER_LOG_FINAL_WINDOW_SECONDS): + continue + path = f"{base}&tail={DOCKER_LOG_FINAL_TAIL}" + finals[name] = stopped try: raw = _docker_request_raw(socket_path, path) except (OSError, ValueError): diff --git a/tests/plugins/docker/test_ingest_events.py b/tests/plugins/docker/test_ingest_events.py index 6921e37..dd5b036 100644 --- a/tests/plugins/docker/test_ingest_events.py +++ b/tests/plugins/docker/test_ingest_events.py @@ -43,6 +43,22 @@ def test_die_on_nonzero_exit(): assert events == [("job", "die", "exit 137")] +@pytest.mark.parametrize("status,exit_code,events", [ + ("exited", 1, [("codex_db.1.x", "die", "exit 1")]), # crashed between samples + ("dead", 137, [("codex_db.1.x", "die", "exit 137")]), + ("exited", 0, []), # a one-shot that finished + ("exited", None, []), # inspect failed: no claim + ("created", 1, []), # never ran +]) +def test_new_container_first_seen_down(status, exit_code, events): + # #5049: a task that starts and exits between two samples is first seen + # already down; a non-zero exit is the death the samples missed. + old = {"other": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}} + got = _derive_events(old, [_c("other", "running"), + _c("codex_db.1.x", status, exit_code=exit_code)]) + assert got == events + + def test_oom_on_flip_to_true(): old = {"hog": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}} # OOM-killed → also no longer running with a non-zero exit, so die + oom both fire. diff --git a/tests/plugins/host_agent/test_agent_docker.py b/tests/plugins/host_agent/test_agent_docker.py index 7115a9e..6bbe07d 100644 --- a/tests/plugins/host_agent/test_agent_docker.py +++ b/tests/plugins/host_agent/test_agent_docker.py @@ -634,6 +634,75 @@ def test_collect_docker_logs_forgets_gone_container_cursors(monkeypatch): assert "db" in state["docker_log_cursors"] +def _final_raw(seen): + def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT): + seen.append(path) + return (_frame(1, b"2026-10-06T12:00:00.000000000Z booting\n") + + _frame(2, b"2026-10-06T12:00:01.000000000Z fatal: no config\n")) + return fake_raw + + +def test_collect_docker_logs_reads_a_stopped_container_once(monkeypatch): + # #5049: a container that was being read and has stopped is read one last + # time since its cursor, then not again until it stops again. + seen: list = [] + monkeypatch.setattr(a, "_docker_request_raw", _final_raw(seen)) + state = {"docker_log_cursors": {"web": 1791288000.5}} # 2026-10-06T12:00:00.5Z + down = [{"name": "web", "status": "exited", "exit_code": 1, + "finished_at": "2026-10-06T12:00:02+00:00"}] + out = a.collect_docker_logs("/sock", down, state) + assert [r["line"] for r in out] == ["fatal: no config"] # since the cursor + assert "since=" in seen[0] + assert a.collect_docker_logs("/sock", down, state) == [] # same stop: not again + assert len(seen) == 1 + again = [{**down[0], "finished_at": "2026-10-06T12:05:00+00:00"}] + a.collect_docker_logs("/sock", again, state) # a new stop is read + assert len(seen) == 2 + + +@pytest.mark.parametrize("exit_code,stopped_ago,read", [ + (1, 30, True), # crashed between two samples: its last lines + (0, 30, False), # finished cleanly: nothing to explain + (None, 30, False), # inspect failed: no claim + (1, 3600, False), # an old crash already on the host when the agent came up +]) +def test_collect_docker_logs_tails_a_container_first_seen_crashed(monkeypatch, exit_code, + stopped_ago, read): + seen: list = [] + monkeypatch.setattr(a, "_docker_request_raw", _final_raw(seen)) + stopped = 1791288000.0 + down = [{"name": "codex_db.1.x", "status": "exited", "exit_code": exit_code, + "finished_at": "2026-10-06T12:00:00+00:00"}] + out = a.collect_docker_logs("/sock", down, {}, now=stopped + stopped_ago) + assert bool(out) is read + if read: + assert f"tail={a.DOCKER_LOG_FINAL_TAIL}" in seen[0] + assert out[-1]["line"] == "fatal: no config" + + +def test_collect_docker_logs_forgets_finals_of_unlisted_containers(monkeypatch): + seen: list = [] + monkeypatch.setattr(a, "_docker_request_raw", _final_raw(seen)) + state: dict = {} + down = [{"name": "job", "status": "exited", "exit_code": 2, + "finished_at": "2026-10-06T12:00:00+00:00"}] + a.collect_docker_logs("/sock", down, state, now=1791288010.0) + assert "job" in state["docker_log_finals"] + a.collect_docker_logs("/sock", [], state, now=1791288020.0) + assert state["docker_log_finals"] == {} and state["docker_log_cursors"] == {} + + +def test_finished_at_only_for_a_stopped_container(monkeypatch): + insp = {"State": {"StartedAt": "2026-10-06T11:00:00Z", "FinishedAt": "2026-10-06T12:00:00Z", + "ExitCode": 1}} + monkeypatch.setattr(a, "_docker_request", lambda s, p, timeout=a.DOCKER_API_TIMEOUT: insp) + rec = a._collect_one_container("/sock", {"Id": "abc", "Names": ["/job"], "State": "exited"}) + assert rec["finished_at"].startswith("2026-10-06T12:00:00") + insp["State"]["FinishedAt"] = "0001-01-01T00:00:00Z" # never stopped + rec = a._collect_one_container("/sock", {"Id": "abc", "Names": ["/job"], "State": "exited"}) + assert rec["finished_at"] is None + + def test_build_sample_includes_logs_when_present(monkeypatch): monkeypatch.setattr(a, "collect_docker", lambda _s: ("ok", None, [{"name": "web", "status": "running"}]))