feat(docker): retention + hourly rollup for metrics/events with Settings windows

Bounds Docker time-series growth (the main scaling concern). New
docker_metrics_hourly table + docker_006 migration; a plugin retention module
(docker.run_retention capability) rolls raw docker_metrics older than the raw
window into hourly averages (idempotent upsert), deletes the rolled raw rows,
then prunes stale rollups + lifecycle events. Core cleanup.py drives it each
hourly run via the capability (no plugin-model import), reading the three
retention windows fresh from settings so changes apply without restart (rule 25).

Settings → "Thresholds & Retention" gains a Docker retention card (raw /
rolled-up / events windows, working defaults 7/90/30 days). Unit tests cover the
hour-aligned cutoff/bucketing helpers; integration test exercises the real
rollup-average + prune across both windows.

Milestone 77 task #941.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC

plugins/docker/__init__.py

@@ -21,12 +21,21 @@ 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
     register_capability(
         "docker.persist_host_samples", persist_host_docker,
         label="Persist host Docker samples",
         description="Store per-host container state + metrics pushed by the host agent.",
         required_role=UserRole.viewer,
     )
+    # Roll up + prune Docker time-series, driven by the core cleanup task without
+    # it importing our models. Same trusted server-side data-plane role as above.
+    register_capability(
+        "docker.run_retention", run_docker_retention,
+        label="Run Docker retention",
+        description="Roll up old docker_metrics to hourly + prune stale metrics/events.",
+        required_role=UserRole.viewer,
+    )
 
 
 def get_scheduled_tasks() -> list:

plugins/docker/migrations/versions/docker_006_metric_rollup.py

@@ -0,0 +1,45 @@
+"""Docker hourly metric rollup table
+
+Adds docker_metrics_hourly — the coarse series that retention rolls raw
+docker_metrics into before pruning them, so multi-day history stays cheap.
+One row per (host, container, hour bucket); the unique constraint is the
+conflict target for the idempotent rollup upsert. Additive create_table.
+
+Revision ID: docker_006_metric_rollup
+Revises: docker_005_swarm_placement
+Create Date: 2026-06-19
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "docker_006_metric_rollup"
+down_revision: Union[str, None] = "docker_005_swarm_placement"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "docker_metrics_hourly",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("host_id", sa.String(length=36), nullable=False),
+        sa.Column("container_name", sa.String(length=255), nullable=False),
+        sa.Column("bucket", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("cpu_pct", sa.Float(), nullable=False, server_default="0"),
+        sa.Column("mem_pct", sa.Float(), nullable=False, server_default="0"),
+        sa.Column("mem_usage_bytes", sa.BigInteger(), nullable=False, server_default="0"),
+        sa.Column("sample_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.ForeignKeyConstraint(["host_id"], ["hosts.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("host_id", "container_name", "bucket",
+                            name="uq_docker_metrics_hourly_bucket"),
+    )
+    op.create_index("ix_docker_metrics_hourly_bucket",
+                    "docker_metrics_hourly", ["bucket"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_docker_metrics_hourly_bucket",
+                  table_name="docker_metrics_hourly")
+    op.drop_table("docker_metrics_hourly")

plugins/docker/models.py

@@ -4,6 +4,7 @@ import uuid
 from datetime import datetime, timezone
 from sqlalchemy import (
     BigInteger, Boolean, DateTime, Float, ForeignKey, Index, Integer, String, Text,
+    UniqueConstraint,
 )
 from sqlalchemy.orm import Mapped, mapped_column
 from steward.models.base import Base
@@ -87,6 +88,39 @@ class DockerMetric(Base):
     )
 
 
+class DockerMetricHourly(Base):
+    """Hourly rollup of docker_metrics — avg cpu/mem per container per hour.
+
+    Raw per-sample rows (~2880/container/day at 30s) are pruned beyond a short
+    window; before deletion they're aggregated here so multi-day history stays
+    cheap to store and query. One row per (host, container, hour bucket); the
+    unique constraint lets retention upsert idempotently if it re-runs before the
+    raw rows are deleted. `bucket` is the hour-truncated sample time.
+    """
+    __tablename__ = "docker_metrics_hourly"
+
+    id: Mapped[str] = mapped_column(
+        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
+    )
+    host_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("hosts.id", ondelete="CASCADE"), nullable=False
+    )
+    container_name: Mapped[str] = mapped_column(String(255), nullable=False)
+    bucket: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    cpu_pct: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
+    mem_pct: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
+    mem_usage_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    sample_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    __table_args__ = (
+        # One bucket per container per host — the conflict target for the
+        # idempotent rollup upsert; doubles as the history-query index.
+        UniqueConstraint("host_id", "container_name", "bucket",
+                         name="uq_docker_metrics_hourly_bucket"),
+        Index("ix_docker_metrics_hourly_bucket", "bucket"),
+    )
+
+
 class DockerEvent(Base):
     """Lifecycle events derived by diffing consecutive host snapshots.
 

plugins/docker/retention.py

@@ -0,0 +1,128 @@
+# plugins/docker/retention.py
+"""Bound Docker time-series growth: roll up old metrics, prune old rows.
+
+Published as the "docker.run_retention" capability (see __init__.setup) so the
+core cleanup task can drive it WITHOUT importing the docker models (same
+opportunistic-coupling pattern as docker.persist_host_samples). Runs inside the
+caller's open transaction; never opens or commits its own.
+
+The scaling concern is docker_metrics: ~2880 rows/container/day at a 30s sample.
+We keep raw samples for a short window, then aggregate everything older into
+hourly averages (docker_metrics_hourly) and delete the raw rows — so multi-day
+history stays cheap to store and query. docker_events is light but unbounded
+without a cutoff, so it gets a (longer) window too.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+
+def _hour_floor(dt: datetime) -> datetime:
+    """Truncate a datetime down to the start of its hour (drops min/sec/µs)."""
+    return dt.replace(minute=0, second=0, microsecond=0)
+
+
+def _rollup_cutoff(now: datetime, raw_days: int) -> datetime:
+    """Hour-aligned boundary below which raw metrics get rolled up + deleted.
+
+    Aligning to the hour means we only ever roll up *whole* elapsed hours — a
+    bucket is never split across the keep/roll boundary, so re-running can't
+    produce a partial-then-complete duplicate for the same hour.
+    """
+    return _hour_floor(now - timedelta(days=raw_days))
+
+
+async def run_docker_retention(
+    session,
+    *,
+    events_days: int,
+    metrics_raw_days: int,
+    metrics_rollup_days: int,
+    now: datetime | None = None,
+) -> dict:
+    """Roll up + prune Docker time-series. Returns a counts dict for logging.
+
+    1. Aggregate docker_metrics older than the (hour-aligned) raw window into
+       docker_metrics_hourly (avg cpu/mem per container per hour), upserting so a
+       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.
+    """
+    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
+
+    if now is None:
+        now = datetime.now(timezone.utc)
+
+    rolled = rolled_rows = events_pruned = rollup_pruned = 0
+
+    # ── 1. Roll up raw metrics older than the raw window into hourly buckets ──
+    raw_cutoff = _rollup_cutoff(now, metrics_raw_days)
+    hour = func.date_trunc("hour", DockerMetric.scraped_at)
+    agg = (
+        select(
+            DockerMetric.host_id,
+            DockerMetric.container_name,
+            hour.label("bucket"),
+            func.avg(DockerMetric.cpu_pct).label("cpu_pct"),
+            func.avg(DockerMetric.mem_pct).label("mem_pct"),
+            func.avg(DockerMetric.mem_usage_bytes).label("mem_usage_bytes"),
+            func.count().label("sample_count"),
+        )
+        .where(DockerMetric.scraped_at < raw_cutoff)
+        .group_by(DockerMetric.host_id, DockerMetric.container_name, hour)
+    )
+    for r in (await session.execute(agg)).all():
+        stmt = (
+            pg_insert(DockerMetricHourly)
+            .values(
+                host_id=r.host_id,
+                container_name=r.container_name,
+                bucket=r.bucket,
+                cpu_pct=float(r.cpu_pct or 0.0),
+                mem_pct=float(r.mem_pct or 0.0),
+                mem_usage_bytes=int(r.mem_usage_bytes or 0),
+                sample_count=int(r.sample_count or 0),
+            )
+            .on_conflict_do_update(
+                constraint="uq_docker_metrics_hourly_bucket",
+                set_={
+                    "cpu_pct": float(r.cpu_pct or 0.0),
+                    "mem_pct": float(r.mem_pct or 0.0),
+                    "mem_usage_bytes": int(r.mem_usage_bytes or 0),
+                    "sample_count": int(r.sample_count or 0),
+                },
+            )
+        )
+        await session.execute(stmt)
+        rolled += 1
+        rolled_rows += int(r.sample_count or 0)
+    if rolled:
+        await session.execute(
+            delete(DockerMetric).where(DockerMetric.scraped_at < raw_cutoff)
+        )
+
+    # ── 2. Prune rolled-up rows beyond the rollup window ──
+    rollup_cutoff = now - timedelta(days=metrics_rollup_days)
+    res = await session.execute(
+        delete(DockerMetricHourly).where(DockerMetricHourly.bucket < rollup_cutoff)
+    )
+    rollup_pruned = res.rowcount or 0
+
+    # ── 3. Prune lifecycle events beyond the events window ──
+    events_cutoff = now - timedelta(days=events_days)
+    res = await session.execute(
+        delete(DockerEvent).where(DockerEvent.at < events_cutoff)
+    )
+    events_pruned = res.rowcount or 0
+
+    return {
+        "buckets_rolled": rolled,
+        "raw_rows_rolled": rolled_rows,
+        "rollup_pruned": rollup_pruned,
+        "events_pruned": events_pruned,
+    }