From b3a5bbfeffaae2ea727909308bf22be2c4afc51a Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 8 Oct 2026 22:31:00 -0400 Subject: [PATCH] perf(docker): the hourly log prune ranks only the containers over their cap (#5449) #5449 "The hourly Docker log prune ranks every log line in the table to find the few over their cap": the size ring put a running-total window over every row of docker_logs each hour, sorting the whole table on NFS-backed storage to find the handful of containers past docker.logs.max_bytes_per_container. - prune_docker_logs (capability docker.prune_logs) does the age cutoff, then a per-container byte total picks the containers at or over the cap, and only their rows are ranked and trimmed. Same keep rule as before: the newest line always survives. - run_docker_retention keeps the metrics rollup and the event prune. - Cleanup runs the log ring in its own transaction after them. Arose from #5431 "Ingest dropped for ~3 minutes after each deploy while startup retention runs". Co-Authored-By: Claude Opus 5.5 --- plugins/docker/__init__.py | 8 ++- plugins/docker/retention.py | 93 ++++++++++++++++++++------------ steward/core/cleanup.py | 24 +++++---- tests/integration/test_docker.py | 43 ++++++++++----- 4 files changed, 111 insertions(+), 57 deletions(-) diff --git a/plugins/docker/__init__.py b/plugins/docker/__init__.py index aca8d1f..0aa6045 100644 --- a/plugins/docker/__init__.py +++ b/plugins/docker/__init__.py @@ -21,7 +21,7 @@ def setup(app: "Quart") -> None: from steward.core.capabilities import register_capability from steward.models.users import UserRole from .ingest import persist_host_docker - from .retention import run_docker_retention + from .retention import prune_docker_logs, run_docker_retention register_capability( "docker.persist_host_samples", persist_host_docker, label="Persist host Docker samples", @@ -36,6 +36,12 @@ def setup(app: "Quart") -> None: description="Roll up old docker_metrics to hourly + prune stale metrics/events.", required_role=UserRole.viewer, ) + register_capability( + "docker.prune_logs", prune_docker_logs, + label="Prune Docker logs", + description="Keep each container's logs within the age window and byte cap.", + required_role=UserRole.viewer, + ) def get_scheduled_tasks() -> list: diff --git a/plugins/docker/retention.py b/plugins/docker/retention.py index ae3ebe0..b235bd2 100644 --- a/plugins/docker/retention.py +++ b/plugins/docker/retention.py @@ -38,8 +38,6 @@ 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. @@ -49,22 +47,20 @@ 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). + + The container-log ring is prune_docker_logs, run in its own transaction. """ from datetime import timezone from sqlalchemy import delete, func, select from sqlalchemy.dialects.postgresql import insert as pg_insert - from .models import DockerEvent, DockerLog, DockerMetric, DockerMetricHourly + from .models import DockerEvent, 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) @@ -126,35 +122,66 @@ 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, } + + +async def prune_docker_logs( + session, + *, + retention_days: int, + max_bytes_per_container: int, + now: datetime | None = None, +) -> dict: + """The docker_logs ring (m79): drop lines older than the age window, then + keep only the newest ~cap bytes per (host, container). Runs inside the + caller's open transaction, like run_docker_retention. + + Only containers at or over the cap are ranked (#5449). Ranking every line + in the table each hour to find the few containers over it sorted the whole + table on slow storage. + """ + from datetime import timezone + + from sqlalchemy import and_, delete, func, select + + from .models import DockerLog + + if now is None: + now = datetime.now(timezone.utc) + + res = await session.execute( + delete(DockerLog).where(DockerLog.ts < now - timedelta(days=retention_days)) + ) + age_pruned = res.rowcount or 0 + + size = func.length(DockerLog.line) + over = ( + select(DockerLog.host_id, DockerLog.container_name) + .group_by(DockerLog.host_id, DockerLog.container_name) + .having(func.sum(size) >= max_bytes_per_container) + .subquery() + ) + # Per container, sum line bytes newest-first; delete a row once its + # strictly-newer siblings already fill the cap. 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(size).over( + partition_by=[DockerLog.host_id, DockerLog.container_name], + order_by=[DockerLog.ts.desc(), DockerLog.id.desc()], + ) + ranked = ( + select(DockerLog.id, (running - size).label("prefix_excl")) + .join(over, and_(DockerLog.host_id == over.c.host_id, + DockerLog.container_name == over.c.container_name)) + .subquery() + ) + res = await session.execute( + delete(DockerLog).where(DockerLog.id.in_( + select(ranked.c.id).where(ranked.c.prefix_excl >= max_bytes_per_container))) + ) + return {"logs_age_pruned": age_pruned, "logs_size_pruned": res.rowcount or 0} diff --git a/steward/core/cleanup.py b/steward/core/cleanup.py index 5161bf1..bafde0b 100644 --- a/steward/core/cleanup.py +++ b/steward/core/cleanup.py @@ -103,12 +103,14 @@ async def _run_metrics_retention(session, now: datetime) -> None: async def _run_docker_retention(session, now: datetime) -> None: - """Drive the docker plugin's rollup + prune via its capability, if loaded. + """Drive the docker plugin's rollup + prune and its log ring via its + capabilities, if loaded. Windows are read fresh from settings each run (rule 25 — a change in the - Settings UI takes effect on the next hourly cleanup, no restart). Kept in its - own transaction so a docker-side failure can't roll back the generic prune - above. No-op when the docker plugin is disabled (capability absent). + Settings UI takes effect on the next hourly cleanup, no restart). Each runs + in its own transaction, so a docker-side failure can't roll back the generic + prune above, and the log ring's deletes don't wait on the metrics rollup + (#5449). No-op when the docker plugin is disabled (capability absent). """ from steward.core.capabilities import has_capability, invoke_capability if not has_capability("docker.run_retention"): @@ -122,14 +124,18 @@ 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, logs_retention_days=logs_days, - logs_max_bytes_per_container=logs_cap, now=now, + metrics_rollup_days=rollup_days, now=now, + ) + async with session.begin(): + 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.prune_logs", UserRole.viewer, session, + retention_days=logs_days, max_bytes_per_container=logs_cap, now=now, ) if counts and any(counts.values()): logger.info("Docker retention: %s", counts) diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index d12c879..94861e0 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -118,6 +118,15 @@ def _retention_fn(app): return run_docker_retention +def _prune_logs_fn(app): + """Resolve prune_docker_logs via capability (or direct import if unloaded).""" + from steward.core.capabilities import has_capability, get_capability + if has_capability("docker.prune_logs"): + return get_capability("docker.prune_logs").fn + from plugins.docker.retention import prune_docker_logs + return prune_docker_logs + + @_NEEDS_DB def test_persist_scopes_containers_by_host(app): from sqlalchemy import text @@ -739,12 +748,14 @@ def test_retention_rollup_and_prune(app): @_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.""" + each container keeps only the newest ~cap bytes; containers are independent, + including a same-named container on another host (#5449 ranks only the + containers over the cap, joined on host and name).""" from datetime import timedelta from sqlalchemy import text from steward.models.hosts import Host - run_retention = _retention_fn(app) + prune_logs = _prune_logs_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) @@ -753,16 +764,17 @@ def test_retention_logs_ring(app): 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) + h2 = Host(id=str(uuid.uuid4()), name="loghost3", address="10.7.7.9") + s.add_all([h, h2]) await s.flush() - hid = h.id + hid, hid2 = h.id, h2.id - def ins(cn, ts, line): + def ins(cn, ts, line, host=None): 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, + {"id": str(uuid.uuid4()), "h": host or hid, "cn": cn, "ts": ts, "line": line}) # Past the 3-day age window → age-pruned. @@ -773,26 +785,29 @@ def test_retention_logs_ring(app): # 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) + # 'web' on the other host: 2 lines, under the cap → both survive. + await ins("web", now - timedelta(minutes=2), line40, host=hid2) + await ins("web", now - timedelta(minutes=1), line40, host=hid2) 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, - ) + counts = await prune_logs( + s, retention_days=3, max_bytes_per_container=100, now=now) - def n(cn): + def n(cn, host=None): return s.execute(text( "SELECT COUNT(*) FROM docker_logs WHERE host_id=:h AND container_name=:cn"), - {"h": hid, "cn": cn}) + {"h": host or hid, "cn": cn}) web = (await n("web")).scalar() db = (await n("db")).scalar() old = (await n("old")).scalar() - return counts, web, db, old + web2 = (await n("web", hid2)).scalar() + return counts, web, db, old, web2 - counts, web, db, old = asyncio.run(_go()) + counts, web, db, old, web2 = 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 web2 == 2 # same name, other host → its own ring assert counts["logs_age_pruned"] == 1 assert counts["logs_size_pruned"] == 2