From 07a841d91e6b43b48873c0fe4e92d5f4e6b8714d Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 19 Jul 2026 18:51:12 -0400 Subject: [PATCH] feat(docker): per-container size+age log ring rotation [M79 step 4] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bound docker_logs growth in the periodic cleanup task (same architecture as the metrics/events retention). run_docker_retention gains logs_retention_days + logs_max_bytes_per_container: it prunes lines past the age window, then keeps only the newest ~cap bytes per (host, container) via a window-function ring (exclusive-prefix sum, so the newest line always survives even if it alone exceeds the cap). Containers rotate independently. - settings DEFAULTS: docker.logs.enabled/exclude/retention_days(3)/ max_bytes_per_container(5MB) — operator preference: ~3 days / ~5 MB - cleanup.py reads the two windows fresh each run (rule 25, no restart) - integration rotate test: age prune + per-container byte cap + isolation Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CAGR73DUowdVFVvYzLXC5C --- plugins/docker/retention.py | 34 +++++++++++++++++- steward/core/cleanup.py | 6 +++- steward/core/settings.py | 8 +++++ tests/integration/test_docker.py | 61 ++++++++++++++++++++++++++++++++ 4 files changed, 107 insertions(+), 2 deletions(-) diff --git a/plugins/docker/retention.py b/plugins/docker/retention.py index 9919f92..ae3ebe0 100644 --- a/plugins/docker/retention.py +++ b/plugins/docker/retention.py @@ -38,6 +38,8 @@ async def run_docker_retention( events_days: int, metrics_raw_days: int, metrics_rollup_days: int, + logs_retention_days: int = 3, + logs_max_bytes_per_container: int = 5_000_000, now: datetime | None = None, ) -> dict: """Roll up + prune Docker time-series. Returns a counts dict for logging. @@ -47,18 +49,22 @@ async def run_docker_retention( re-run is idempotent, then delete those raw rows. 2. Prune rolled-up rows older than the rollup window. 3. Prune docker_events older than the events window. + 4. Prune docker_logs with a per-container size+age ring (m79): drop lines + older than the age window, then keep only the newest ~cap bytes per + (host, container). """ from datetime import timezone from sqlalchemy import delete, func, select from sqlalchemy.dialects.postgresql import insert as pg_insert - from .models import DockerEvent, DockerMetric, DockerMetricHourly + from .models import DockerEvent, DockerLog, DockerMetric, DockerMetricHourly if now is None: now = datetime.now(timezone.utc) rolled = rolled_rows = events_pruned = rollup_pruned = 0 + logs_age_pruned = logs_size_pruned = 0 # ── 1. Roll up raw metrics older than the raw window into hourly buckets ── raw_cutoff = _rollup_cutoff(now, metrics_raw_days) @@ -120,9 +126,35 @@ async def run_docker_retention( ) events_pruned = res.rowcount or 0 + # ── 4. Container-log ring: age cutoff, then per-container byte cap (m79) ── + logs_cutoff = now - timedelta(days=logs_retention_days) + res = await session.execute( + delete(DockerLog).where(DockerLog.ts < logs_cutoff) + ) + logs_age_pruned = res.rowcount or 0 + + # Size ring: per (host, container), sum line bytes newest-first; delete a row + # once its strictly-newer siblings already fill the cap. Using the EXCLUSIVE + # prefix (running total minus this row) means the newest row always survives, + # so a single line larger than the cap is never wiped out. + running = func.sum(func.length(DockerLog.line)).over( + partition_by=[DockerLog.host_id, DockerLog.container_name], + order_by=[DockerLog.ts.desc(), DockerLog.id.desc()], + ) + prefix_excl = (running - func.length(DockerLog.line)).label("prefix_excl") + ranked = select(DockerLog.id, prefix_excl).subquery() + over_cap = select(ranked.c.id).where( + ranked.c.prefix_excl >= logs_max_bytes_per_container) + res = await session.execute( + delete(DockerLog).where(DockerLog.id.in_(over_cap)) + ) + logs_size_pruned = res.rowcount or 0 + return { "buckets_rolled": rolled, "raw_rows_rolled": rolled_rows, "rollup_pruned": rollup_pruned, "events_pruned": events_pruned, + "logs_age_pruned": logs_age_pruned, + "logs_size_pruned": logs_size_pruned, } diff --git a/steward/core/cleanup.py b/steward/core/cleanup.py index e1f9975..28bbe76 100644 --- a/steward/core/cleanup.py +++ b/steward/core/cleanup.py @@ -76,10 +76,14 @@ async def _run_docker_retention(session, now: datetime) -> None: raw_days = int(await get_setting(session, "docker.retention.metrics_raw_days") or 7) rollup_days = int(await get_setting(session, "docker.retention.metrics_rollup_days") or 90) events_days = int(await get_setting(session, "docker.retention.events_days") or 30) + logs_days = int(await get_setting(session, "docker.logs.retention_days") or 3) + logs_cap = int( + await get_setting(session, "docker.logs.max_bytes_per_container") or 5_000_000) counts = await invoke_capability( "docker.run_retention", UserRole.viewer, session, events_days=events_days, metrics_raw_days=raw_days, - metrics_rollup_days=rollup_days, now=now, + metrics_rollup_days=rollup_days, logs_retention_days=logs_days, + logs_max_bytes_per_container=logs_cap, now=now, ) if counts and any(counts.values()): logger.info("Docker retention: %s", counts) diff --git a/steward/core/settings.py b/steward/core/settings.py index 76c6905..d41c2c3 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -84,6 +84,14 @@ DEFAULTS: dict[str, Any] = { "docker.retention.metrics_raw_days": 7, "docker.retention.metrics_rollup_days": 90, "docker.retention.events_days": 30, + # Container logs (m79): on by default for every container (operator + # preference). `exclude` names containers the server drops on ingest; the + # per-container ring bounds storage (rotate oldest past whichever of ~age or + # ~bytes hits first — a chatty container just keeps a shorter window). + "docker.logs.enabled": True, + "docker.logs.exclude": [], + "docker.logs.retention_days": 3, + "docker.logs.max_bytes_per_container": 5_000_000, # Host/plugin metrics retention (plugin_metrics): keep a short raw window at # the agent's ~30s cadence, then roll up to hourly averages kept much longer. "metrics.retention.raw_days": 7, diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 2bf9381..9703cf4 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -511,6 +511,67 @@ def test_retention_rollup_and_prune(app): assert counts["events_pruned"] == 1 and counts["rollup_pruned"] == 1 +@_NEEDS_DB +def test_retention_logs_ring(app): + """docker_logs ring (m79): lines past the age window are pruned; within it, + each container keeps only the newest ~cap bytes; containers are independent.""" + from datetime import timedelta + from sqlalchemy import text + from steward.models.hosts import Host + + run_retention = _retention_fn(app) + now = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc) + line40 = "x" * 40 # 40 chars/line; cap=100 keeps 3 lines (excl-prefix 0/40/80) + + 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="loghost2", address="10.7.7.8") + s.add(h) + await s.flush() + hid = h.id + + def ins(cn, ts, line): + return s.execute(text( + "INSERT INTO docker_logs " + "(id, host_id, container_name, ts, stream, line) " + "VALUES (:id,:h,:cn,:ts,'stdout',:line)"), + {"id": str(uuid.uuid4()), "h": hid, "cn": cn, + "ts": ts, "line": line}) + + # Past the 3-day age window → age-pruned. + await ins("old", now - timedelta(days=10), "z") + # 5 recent 'web' lines (40 bytes each = 200 > cap 100) → keep 3. + for i in range(1, 6): + await ins("web", now - timedelta(minutes=6 - i), line40) + # 2 recent 'db' lines (80 bytes ≤ cap) → both survive (isolation). + await ins("db", now - timedelta(minutes=2), line40) + await ins("db", now - timedelta(minutes=1), line40) + + async with s.begin(): + counts = await run_retention( + s, events_days=30, metrics_raw_days=7, metrics_rollup_days=90, + logs_retention_days=3, logs_max_bytes_per_container=100, now=now, + ) + + def n(cn): + return s.execute(text( + "SELECT COUNT(*) FROM docker_logs WHERE host_id=:h AND container_name=:cn"), + {"h": hid, "cn": cn}) + web = (await n("web")).scalar() + db = (await n("db")).scalar() + old = (await n("old")).scalar() + return counts, web, db, old + + counts, web, db, old = asyncio.run(_go()) + assert old == 0 # age window + assert web == 3 # newest ~100 bytes kept, oldest 2 dropped + assert db == 2 # under cap → untouched (per-container) + assert counts["logs_age_pruned"] == 1 + assert counts["logs_size_pruned"] == 2 + + def test_widget_dedup_collapses_cross_manager_duplicates(): """The same swarm task is reported by every manager (identical container_id); the dashboard widget must count it once. Older agents send no container_id,