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 <noreply@anthropic.com>

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}