From a8de3570fe105f1df70b7b176e756ba09167d0cc Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 19 Jul 2026 18:46:31 -0400 Subject: [PATCH] feat(docker): persist pushed container logs into docker_logs [M79 step 3] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ingest side of container logs. The host_agent ingest route extracts sample["docker_logs"] as time-series batches (append every line, each carrying its own Docker ts with recorded_at as fallback) and hands them to the docker capability alongside the existing container/swarm/disk data — still under the begin_nested SAVEPOINT so a logs failure can't sink host metrics. - ingest.py: _persist_logs + pure _log_rows(batches, host_id) that shapes/filters records (drops the _steward truncation marker, malformed + lineless records, normalises unknown stream → stdout); persist_host_docker gains a logs= param - routes.py: accumulate docker_log_batches, add to the guard + capability call - unit test for _log_rows shaping/filtering; integration push→store→query test Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CAGR73DUowdVFVvYzLXC5C --- plugins/docker/ingest.py | 49 +++++++++++++++++++++- plugins/host_agent/routes.py | 11 ++++- tests/integration/test_docker.py | 38 +++++++++++++++++ tests/plugins/docker/test_ingest_events.py | 32 ++++++++++++++ 4 files changed, 126 insertions(+), 4 deletions(-) diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index b67a404..7194edc 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -92,6 +92,46 @@ def _derive_events(old_state: dict, new_containers: list) -> list: return events +def _log_rows(log_batches, host_id: str): + """Pure: flatten the agent's log batches → docker_logs row kwargs (no DB). + + `log_batches` is a list of (recorded_at, records) where each record is + {"container", "stream", "ts", "line"}. A line's own Docker timestamp wins; + recorded_at is the fallback when the agent couldn't parse one. Drops the + agent's advisory truncation marker (container "_steward" — it signals a + deferral, not a real container), malformed records, and lineless records; + normalises an unknown stream to stdout. Unit-testable in isolation. + """ + for recorded_at, records in log_batches: + if not isinstance(records, list): + continue + for rec in records: + if not isinstance(rec, dict): + continue + name = rec.get("container") + line = rec.get("line") + if not name or name == "_steward" or line is None: + continue + ts = _parse_started_at(rec.get("ts")) or recorded_at + stream = rec.get("stream") + if stream not in ("stdout", "stderr"): + stream = "stdout" + yield {"host_id": host_id, "container_name": str(name)[:255], + "ts": ts, "stream": stream, "line": str(line)} + + +async def _persist_logs(session, host, log_batches) -> None: + """Append pushed container log lines (one row per line) for this host. + + Time-series / append-only — the per-container size+age ring (retention) + bounds growth, so a chatty container just keeps a shorter window. + """ + from .models import DockerLog + + for row in _log_rows(log_batches, host.id): + session.add(DockerLog(**row)) + + async def _persist_swarm(session, host, swarm: dict) -> None: """Upsert this manager's swarm topology; drop rows no longer reported. @@ -201,7 +241,8 @@ async def _persist_disk(session, host, disk: dict) -> None: await session.execute(stale) -async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) -> None: +async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, + logs=None) -> None: """Upsert containers + time-series + lifecycle events + swarm for one host. `snapshots` is a list of (recorded_at: datetime, containers: list[dict]) — @@ -211,7 +252,9 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) - alert pipeline, and lifecycle-event derivation. `swarm` is the newest sample's swarm object (or None off managers) — persisted when present. `disk` is the newest sample's /system/df summary (or None on Docker-less - hosts) — persisted when present. + hosts) — persisted when present. `logs` is a list of (recorded_at, records) + log batches (m79) — appended one row per line when present, independent of + whether this sample also carried container metrics. """ from steward.core.alerts import record_metric from .models import DockerContainer, DockerEvent, DockerMetric @@ -220,6 +263,8 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) - await _persist_swarm(session, host, swarm) if disk is not None: await _persist_disk(session, host, disk) + if logs: + await _persist_logs(session, host, logs) if not snapshots: return diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index b63ae1a..f74b7f8 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -202,6 +202,7 @@ async def ingest(): accepted = 0 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 @@ -217,6 +218,11 @@ async def ingest(): 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") @@ -240,7 +246,8 @@ async def ingest(): # (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: + 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: @@ -250,7 +257,7 @@ async def ingest(): await invoke_capability( "docker.persist_host_samples", UserRole.admin, session, host, docker_snapshots, latest_swarm, - latest_disk, + latest_disk, docker_log_batches, ) except Exception: current_app.logger.exception( diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 0210cd0..2bf9381 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -172,6 +172,44 @@ def test_persist_scopes_containers_by_host(app): assert enrich == ("healthy", 2, "web", 1000) # enrichment round-trips +@_NEEDS_DB +def test_persist_logs_stores_lines(app): + """Pushed container logs land in docker_logs, host-scoped, one row per line, + ordered by their own ts; the agent's _steward truncation marker is dropped.""" + from sqlalchemy import text + from steward.models.hosts import Host + + persist = _persist_fn(app) + now = datetime.now(timezone.utc) + log_batches = [(now, [ + {"container": "web", "stream": "stdout", + "ts": "2026-01-01T00:00:01+00:00", "line": "started"}, + {"container": "web", "stream": "stderr", + "ts": "2026-01-01T00:00:02+00:00", "line": "warn: x"}, + {"container": "_steward", "stream": "stderr", "ts": None, "line": "truncated"}, + ])] + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(text("DELETE FROM docker_logs")) + h = Host(id=str(uuid.uuid4()), name="loghost", address="10.0.0.9") + s.add(h) + await s.flush() + hid = h.id + # snapshots empty; logs passed as the 6th positional arg. + await persist(s, h, [], None, None, log_batches) + rows = (await s.execute(text( + "SELECT container_name, stream, line FROM docker_logs " + "WHERE host_id = :h ORDER BY ts"), {"h": hid})).all() + return [tuple(r) for r in rows] + + assert asyncio.run(_go()) == [ + ("web", "stdout", "started"), + ("web", "stderr", "warn: x"), + ] + + @_NEEDS_DB def test_large_memory_values_persist_as_bigint(app): """A container using >2^31 bytes of RAM must persist. Regression: mem_usage_bytes diff --git a/tests/plugins/docker/test_ingest_events.py b/tests/plugins/docker/test_ingest_events.py index b5b50e8..72324c1 100644 --- a/tests/plugins/docker/test_ingest_events.py +++ b/tests/plugins/docker/test_ingest_events.py @@ -65,3 +65,35 @@ def test_no_event_when_unchanged(): old = {"web": {"status": "running", "health": "healthy", "oom_killed": False, "exit_code": None}} events = _derive_events(old, [_c("web", "running", health="healthy")]) assert events == [] + + +# ── container-log row shaping (m79, pure) ───────────────────────────────────── + +def test_log_rows_shapes_and_filters(): + from datetime import datetime, timezone + from plugins.docker.ingest import _log_rows + + rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + batches = [(rec_at, [ + {"container": "web", "stream": "stdout", + "ts": "2026-01-01T00:00:01+00:00", "line": "hello"}, + {"container": "web", "stream": "stderr", "ts": None, "line": "no-ts"}, + {"container": "_steward", "stream": "stderr", + "ts": None, "line": "truncated"}, # advisory marker → dropped + {"container": "web", "stream": "weird", "ts": None, "line": "bad-stream"}, + {"container": "", "line": "x"}, # no name → dropped + {"container": "web", "ts": None, "line": None}, # no line → dropped + "not-a-dict", # malformed → dropped + ])] + rows = list(_log_rows(batches, "host-1")) + assert [r["line"] for r in rows] == ["hello", "no-ts", "bad-stream"] + assert rows[0]["host_id"] == "host-1" and rows[0]["container_name"] == "web" + assert rows[1]["ts"] == rec_at # None ts → recorded_at fallback + assert rows[2]["stream"] == "stdout" # unknown stream normalised + + +def test_log_rows_skips_non_list_records(): + from datetime import datetime, timezone + from plugins.docker.ingest import _log_rows + rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + assert list(_log_rows([(rec_at, None)], "h")) == []