feat(docker): per-container size+age log ring rotation [M79 step 4]
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) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CAGR73DUowdVFVvYzLXC5C
This commit is contained in:
@@ -38,6 +38,8 @@ async def run_docker_retention(
|
|||||||
events_days: int,
|
events_days: int,
|
||||||
metrics_raw_days: int,
|
metrics_raw_days: int,
|
||||||
metrics_rollup_days: int,
|
metrics_rollup_days: int,
|
||||||
|
logs_retention_days: int = 3,
|
||||||
|
logs_max_bytes_per_container: int = 5_000_000,
|
||||||
now: datetime | None = None,
|
now: datetime | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Roll up + prune Docker time-series. Returns a counts dict for logging.
|
"""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.
|
re-run is idempotent, then delete those raw rows.
|
||||||
2. Prune rolled-up rows older than the rollup window.
|
2. Prune rolled-up rows older than the rollup window.
|
||||||
3. Prune docker_events older than the events 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 datetime import timezone
|
||||||
|
|
||||||
from sqlalchemy import delete, func, select
|
from sqlalchemy import delete, func, select
|
||||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
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:
|
if now is None:
|
||||||
now = datetime.now(timezone.utc)
|
now = datetime.now(timezone.utc)
|
||||||
|
|
||||||
rolled = rolled_rows = events_pruned = rollup_pruned = 0
|
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 ──
|
# ── 1. Roll up raw metrics older than the raw window into hourly buckets ──
|
||||||
raw_cutoff = _rollup_cutoff(now, metrics_raw_days)
|
raw_cutoff = _rollup_cutoff(now, metrics_raw_days)
|
||||||
@@ -120,9 +126,35 @@ async def run_docker_retention(
|
|||||||
)
|
)
|
||||||
events_pruned = res.rowcount or 0
|
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 {
|
return {
|
||||||
"buckets_rolled": rolled,
|
"buckets_rolled": rolled,
|
||||||
"raw_rows_rolled": rolled_rows,
|
"raw_rows_rolled": rolled_rows,
|
||||||
"rollup_pruned": rollup_pruned,
|
"rollup_pruned": rollup_pruned,
|
||||||
"events_pruned": events_pruned,
|
"events_pruned": events_pruned,
|
||||||
|
"logs_age_pruned": logs_age_pruned,
|
||||||
|
"logs_size_pruned": logs_size_pruned,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
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)
|
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)
|
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(
|
counts = await invoke_capability(
|
||||||
"docker.run_retention", UserRole.viewer, session,
|
"docker.run_retention", UserRole.viewer, session,
|
||||||
events_days=events_days, metrics_raw_days=raw_days,
|
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()):
|
if counts and any(counts.values()):
|
||||||
logger.info("Docker retention: %s", counts)
|
logger.info("Docker retention: %s", counts)
|
||||||
|
|||||||
@@ -84,6 +84,14 @@ DEFAULTS: dict[str, Any] = {
|
|||||||
"docker.retention.metrics_raw_days": 7,
|
"docker.retention.metrics_raw_days": 7,
|
||||||
"docker.retention.metrics_rollup_days": 90,
|
"docker.retention.metrics_rollup_days": 90,
|
||||||
"docker.retention.events_days": 30,
|
"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
|
# 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.
|
# the agent's ~30s cadence, then roll up to hourly averages kept much longer.
|
||||||
"metrics.retention.raw_days": 7,
|
"metrics.retention.raw_days": 7,
|
||||||
|
|||||||
@@ -511,6 +511,67 @@ def test_retention_rollup_and_prune(app):
|
|||||||
assert counts["events_pruned"] == 1 and counts["rollup_pruned"] == 1
|
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():
|
def test_widget_dedup_collapses_cross_manager_duplicates():
|
||||||
"""The same swarm task is reported by every manager (identical container_id);
|
"""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,
|
the dashboard widget must count it once. Older agents send no container_id,
|
||||||
|
|||||||
Reference in New Issue
Block a user