From faecac3ec6eb675e851e1b98d3a8f0b4eca0686d Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 18 Jun 2026 21:40:57 -0400 Subject: [PATCH] feat(docker): retention + hourly rollup for metrics/events with Settings windows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC --- plugins/docker/__init__.py | 9 ++ .../versions/docker_006_metric_rollup.py | 45 ++++++ plugins/docker/models.py | 34 +++++ plugins/docker/retention.py | 128 ++++++++++++++++++ steward/core/cleanup.py | 37 ++++- steward/core/settings.py | 6 + steward/settings/routes.py | 17 +++ steward/templates/settings/_tabs.html | 2 +- steward/templates/settings/thresholds.html | 30 +++- tests/integration/test_docker.py | 96 +++++++++++++ tests/plugins/docker/test_retention.py | 45 ++++++ 11 files changed, 445 insertions(+), 4 deletions(-) create mode 100644 plugins/docker/migrations/versions/docker_006_metric_rollup.py create mode 100644 plugins/docker/retention.py create mode 100644 tests/plugins/docker/test_retention.py diff --git a/plugins/docker/__init__.py b/plugins/docker/__init__.py index 4272f51..7183961 100644 --- a/plugins/docker/__init__.py +++ b/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: diff --git a/plugins/docker/migrations/versions/docker_006_metric_rollup.py b/plugins/docker/migrations/versions/docker_006_metric_rollup.py new file mode 100644 index 0000000..782abfd --- /dev/null +++ b/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") diff --git a/plugins/docker/models.py b/plugins/docker/models.py index fb3dd63..cfbfee8 100644 --- a/plugins/docker/models.py +++ b/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. diff --git a/plugins/docker/retention.py b/plugins/docker/retention.py new file mode 100644 index 0000000..9919f92 --- /dev/null +++ b/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, + } diff --git a/steward/core/cleanup.py b/steward/core/cleanup.py index 0a9e556..563a918 100644 --- a/steward/core/cleanup.py +++ b/steward/core/cleanup.py @@ -16,9 +16,11 @@ logger = logging.getLogger(__name__) async def run_cleanup(app: "Quart") -> None: - """Delete rows older than DATA_RETENTION_DAYS from time-series tables.""" + """Delete rows older than DATA_RETENTION_DAYS from time-series tables, then + run Docker-specific rollup + retention (delegated to the docker plugin).""" retention_days: int = app.config.get("DATA_RETENTION_DAYS", 90) - cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days) + now = datetime.now(timezone.utc) + cutoff = now - timedelta(days=retention_days) async with app.db_sessionmaker() as session: async with session.begin(): @@ -32,3 +34,34 @@ async def run_cleanup(app: "Quart") -> None: ) if result.rowcount: logger.info(f"Pruned {result.rowcount} rows from {model.__tablename__}") + + await _run_docker_retention(session, now) + + +async def _run_docker_retention(session, now: datetime) -> None: + """Drive the docker plugin's rollup + prune via its capability, 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). + """ + from steward.core.capabilities import has_capability, invoke_capability + if not has_capability("docker.run_retention"): + return + from steward.core.settings import get_setting + from steward.models.users import UserRole + + # Reads + rollup/prune share one transaction — get_setting's SELECT would + # otherwise autobegin one, making a later session.begin() raise. + async with session.begin(): + 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) + 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, + ) + 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 ef7f0bc..23bece8 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -78,6 +78,12 @@ DEFAULTS: dict[str, Any] = { "thresholds.load_warn": 80, "thresholds.load_crit": 100, "thresholds.temp_warn": 70, "thresholds.temp_crit": 85, "thresholds.uptime_warn": 99.0, "thresholds.uptime_crit": 95.0, + # Docker time-series retention (rule 25 — tunable, no restart). Raw 30s + # samples are heavy, so keep a short raw window then roll up to hourly + # averages kept much longer; lifecycle events are light, keep a month. + "docker.retention.metrics_raw_days": 7, + "docker.retention.metrics_rollup_days": 90, + "docker.retention.events_days": 30, "plugins.index_url": "https://git.fabledsword.com/bvandeusen/Steward-plugins/raw/branch/main/index.yaml", # Default-enabled plugins. These are the generic, non-vendor-specific # bundled plugins (protocols/standards, not a single product) — useful on diff --git a/steward/settings/routes.py b/steward/settings/routes.py index 8b1f0f6..517d2e1 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -127,6 +127,14 @@ _THRESHOLD_FIELDS = [ ("latency_warn", "ping.threshold.good_ms", False), ("latency_crit", "ping.threshold.warn_ms", False), ] +# Docker time-series retention windows (days). Read fresh by the cleanup task, +# so a save takes effect on the next hourly run — no app.config wiring needed. +_RETENTION_FIELDS = [ + ("docker_metrics_raw_days", "docker.retention.metrics_raw_days"), + ("docker_metrics_rollup_days", "docker.retention.metrics_rollup_days"), + ("docker_events_days", "docker.retention.events_days"), +] + @settings_bp.get("/thresholds/") @require_role(UserRole.admin) @@ -151,6 +159,15 @@ async def save_thresholds(): except (TypeError, ValueError): continue await set_setting(db, key, val) + for field, key in _RETENTION_FIELDS: + raw = form.get(field, "") + if raw == "": + continue + try: + val = max(1, int(raw)) # at least a day — 0 would prune everything + except (TypeError, ValueError): + continue + await set_setting(db, key, val) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "thresholds"}) diff --git a/steward/templates/settings/_tabs.html b/steward/templates/settings/_tabs.html index 65bb7fd..97e1201 100644 --- a/steward/templates/settings/_tabs.html +++ b/steward/templates/settings/_tabs.html @@ -3,7 +3,7 @@
{% set tabs = [ ("general", "General", "/settings/general/"), - ("thresholds", "Thresholds", "/settings/thresholds/"), + ("thresholds", "Thresholds & Retention", "/settings/thresholds/"), ("notifications", "Notifications", "/settings/notifications/"), ("reports", "Reports", "/settings/reports/"), ("auth", "Auth", "/settings/auth/"), diff --git a/steward/templates/settings/thresholds.html b/steward/templates/settings/thresholds.html index 1a7509e..204b049 100644 --- a/steward/templates/settings/thresholds.html +++ b/steward/templates/settings/thresholds.html @@ -1,6 +1,6 @@ {# steward/templates/settings/thresholds.html — tunable degraded/critical cutoffs #} {% extends "base.html" %} -{% block title %}Settings — Thresholds — Steward{% endblock %} +{% block title %}Settings — Thresholds & Retention — Steward{% endblock %} {% block content %} {% set active_tab = "thresholds" %} {% include "settings/_tabs.html" %} @@ -42,6 +42,34 @@ {{ pair("Uptime / SLA", "uptime_warn", "thresholds.uptime_warn", "uptime_crit", "thresholds.uptime_crit", "%", "Lower is worse — amber below warn, red below crit.", step="0.1") }}
+{% macro days(label, field, key, hint) %} +
+ +
+ +
+ {% if hint %}
{{ hint }}
{% endif %} +
+{% endmacro %} + +
+

Docker data retention

+

+ Bounds how much Docker history is stored. Raw per-sample container metrics are + kept for the raw window, then rolled up into hourly averages kept for the + rollup window; lifecycle events are kept for the events window. Applied by the + hourly cleanup task — a change takes effect on its next run. +

+ + {{ days("Raw metrics", "docker_metrics_raw_days", "docker.retention.metrics_raw_days", + "Keep per-sample container CPU/memory points this long, then roll up to hourly.") }} + {{ days("Rolled-up metrics", "docker_metrics_rollup_days", "docker.retention.metrics_rollup_days", + "Keep the hourly-averaged series this long for multi-day history.") }} + {{ days("Lifecycle events", "docker_events_days", "docker.retention.events_days", + "Keep container start/stop/die/health history this long.") }} +
+
Takes effect immediately — no restart. diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 21accd5..8e98265 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -86,6 +86,15 @@ def _persist_fn(app): return persist_host_docker +def _retention_fn(app): + """Resolve run_docker_retention via capability (or direct import if unloaded).""" + from steward.core.capabilities import has_capability, get_capability + if has_capability("docker.run_retention"): + return get_capability("docker.run_retention").fn + from plugins.docker.retention import run_docker_retention + return run_docker_retention + + @_NEEDS_DB def test_persist_scopes_containers_by_host(app): from sqlalchemy import text @@ -219,3 +228,90 @@ def test_swarm_topology_persisted(app): assert svc[0] == "replicated" and svc[1] == 3 and svc[2] == 2 and svc[3] == "nginx" assert "n1" in svc[4] # placement JSON carries the node id assert node[0] == "manager" and node[2] == "ready" and node[3] is True + + +@_NEEDS_DB +def test_retention_rollup_and_prune(app): + """Old raw metrics roll up to hourly averages then delete; stale rollup + + events prune; recent rows survive.""" + 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) + # One old hour (10 days back) with three samples → one rolled-up bucket. + old_hour = datetime(2026, 6, 9, 9, 0, 0, tzinfo=timezone.utc) + recent = datetime(2026, 6, 19, 11, 0, 0, tzinfo=timezone.utc) # inside raw window + ancient_bucket = datetime(2026, 3, 1, 0, 0, 0, tzinfo=timezone.utc) # > rollup window + old_event = datetime(2026, 5, 1, 0, 0, 0, tzinfo=timezone.utc) # > events window + new_event = datetime(2026, 6, 18, 0, 0, 0, tzinfo=timezone.utc) # inside events window + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(text("DELETE FROM docker_metrics_hourly")) + await s.execute(text("DELETE FROM docker_metrics")) + await s.execute(text("DELETE FROM docker_events")) + h = Host(id=str(uuid.uuid4()), name="ret", address="10.7.7.7") + s.add(h) + await s.flush() + hid = h.id + # Three raw samples in the old hour: cpu 10/20/30, mem 40/50/60. + for i, (ts, cpu, mem, usage) in enumerate([ + (old_hour, 10.0, 40.0, 100), + (old_hour.replace(second=30), 20.0, 50.0, 200), + (old_hour.replace(minute=1), 30.0, 60.0, 300), + ]): + await s.execute(text( + "INSERT INTO docker_metrics " + "(id, host_id, container_name, scraped_at, cpu_pct, mem_pct, mem_usage_bytes) " + "VALUES (:id,:h,'web',:ts,:cpu,:mem,:usage)"), + {"id": str(uuid.uuid4()), "h": hid, "ts": ts, + "cpu": cpu, "mem": mem, "usage": usage}) + # A recent sample (within the 7-day raw window) — must survive raw. + await s.execute(text( + "INSERT INTO docker_metrics " + "(id, host_id, container_name, scraped_at, cpu_pct, mem_pct, mem_usage_bytes) " + "VALUES (:id,:h,'web',:ts,5.0,5.0,50)"), + {"id": str(uuid.uuid4()), "h": hid, "ts": recent}) + # A pre-existing rollup row older than the 90-day rollup window. + await s.execute(text( + "INSERT INTO docker_metrics_hourly " + "(id, host_id, container_name, bucket, cpu_pct, mem_pct, mem_usage_bytes, sample_count) " + "VALUES (:id,:h,'ancient',:b,1,1,1,1)"), + {"id": str(uuid.uuid4()), "h": hid, "b": ancient_bucket}) + # Events either side of the 30-day events window. + for ev_at, ev in [(old_event, "stop"), (new_event, "start")]: + await s.execute(text( + "INSERT INTO docker_events (id, host_id, container_name, event, at) " + "VALUES (:id,:h,'web',:ev,:at)"), + {"id": str(uuid.uuid4()), "h": hid, "ev": ev, "at": ev_at}) + + async with s.begin(): + counts = await run_retention( + s, events_days=30, metrics_raw_days=7, + metrics_rollup_days=90, now=now, + ) + + raw_left = (await s.execute(text( + "SELECT COUNT(*) FROM docker_metrics WHERE host_id=:h"), {"h": hid})).scalar() + bucket = (await s.execute(text( + "SELECT cpu_pct, mem_pct, mem_usage_bytes, sample_count " + "FROM docker_metrics_hourly WHERE host_id=:h AND container_name='web'"), + {"h": hid})).first() + ancient_left = (await s.execute(text( + "SELECT COUNT(*) FROM docker_metrics_hourly " + "WHERE host_id=:h AND container_name='ancient'"), {"h": hid})).scalar() + events_left = (await s.execute(text( + "SELECT COUNT(*) FROM docker_events WHERE host_id=:h"), {"h": hid})).scalar() + return counts, raw_left, bucket, ancient_left, events_left + + counts, raw_left, bucket, ancient_left, events_left = asyncio.run(_go()) + assert raw_left == 1 # only the recent sample survives raw + assert bucket is not None + assert bucket[0] == 20.0 and bucket[1] == 50.0 # avg cpu / mem over the 3 samples + assert bucket[2] == 200 and bucket[3] == 3 # avg usage + sample_count + assert ancient_left == 0 # stale rollup pruned + assert events_left == 1 # only the in-window event survives + assert counts["buckets_rolled"] == 1 and counts["raw_rows_rolled"] == 3 + assert counts["events_pruned"] == 1 and counts["rollup_pruned"] == 1 diff --git a/tests/plugins/docker/test_retention.py b/tests/plugins/docker/test_retention.py new file mode 100644 index 0000000..262d9c3 --- /dev/null +++ b/tests/plugins/docker/test_retention.py @@ -0,0 +1,45 @@ +"""Unit tests for the docker plugin's retention cutoff/bucketing helpers. + +_hour_floor and _rollup_cutoff are pure (no DB); retention.py imports its models +lazily inside run_docker_retention, so importing them here doesn't register ORM +tables — safe for the no-DB unit lane. The DB-backed rollup/prune itself is +covered in tests/integration/test_docker.py. +""" +from datetime import datetime, timedelta, timezone + +from plugins.docker.retention import _hour_floor, _rollup_cutoff + + +def test_hour_floor_drops_sub_hour_components(): + dt = datetime(2026, 6, 19, 14, 37, 52, 123456, tzinfo=timezone.utc) + assert _hour_floor(dt) == datetime(2026, 6, 19, 14, 0, 0, 0, tzinfo=timezone.utc) + + +def test_hour_floor_on_exact_hour_is_identity(): + dt = datetime(2026, 6, 19, 14, 0, 0, tzinfo=timezone.utc) + assert _hour_floor(dt) == dt + + +def test_rollup_cutoff_is_hour_aligned(): + now = datetime(2026, 6, 19, 14, 37, 52, tzinfo=timezone.utc) + cutoff = _rollup_cutoff(now, raw_days=7) + # 7 days back from 14:37 is 14:37 on the 12th, floored to 14:00. + assert cutoff == datetime(2026, 6, 12, 14, 0, 0, tzinfo=timezone.utc) + # No sub-hour remainder — only whole elapsed hours get rolled up. + assert cutoff.minute == 0 and cutoff.second == 0 and cutoff.microsecond == 0 + + +def test_rollup_cutoff_scales_with_raw_days(): + now = datetime(2026, 6, 19, 0, 0, 0, tzinfo=timezone.utc) + assert _rollup_cutoff(now, 1) == now - timedelta(days=1) + assert _rollup_cutoff(now, 30) == now - timedelta(days=30) + + +def test_rollup_cutoff_sample_classification(): + """A sample inside the raw window is kept; one before the aligned cutoff rolls up.""" + now = datetime(2026, 6, 19, 14, 37, 0, tzinfo=timezone.utc) + cutoff = _rollup_cutoff(now, raw_days=7) + recent = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc) # within window + old = datetime(2026, 6, 10, 9, 0, 0, tzinfo=timezone.utc) # older than cutoff + assert not (recent < cutoff) # kept raw + assert old < cutoff # rolled up