diff --git a/plugins/docker/__init__.py b/plugins/docker/__init__.py index 0aa6045..38e1c13 100644 --- a/plugins/docker/__init__.py +++ b/plugins/docker/__init__.py @@ -11,7 +11,7 @@ _app: "Quart | None" = None def setup(app: "Quart") -> None: global _app _app = app - from .models import DockerContainer, DockerMetric # noqa: F401 — register with Base + from .models import DockerContainer # noqa: F401 — register with Base # Publish the per-host persist hook so the host agent can hand off the # `docker` array from each sample without importing our models (opportunistic @@ -28,12 +28,13 @@ def setup(app: "Quart") -> None: 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. + # Prune lifecycle events, driven by the core cleanup task without it + # importing our models. Same trusted server-side data-plane role as above. + # Container CPU and memory live in the one metric store and its retention. register_capability( "docker.run_retention", run_docker_retention, label="Run Docker retention", - description="Roll up old docker_metrics to hourly + prune stale metrics/events.", + description="Prune Docker lifecycle events past their window.", required_role=UserRole.viewer, ) register_capability( diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 46bcc99..0caa230 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -306,9 +306,10 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, `snapshots` is a list of (recorded_at: datetime, containers: list[dict]) — one entry per ingested sample carrying docker data (usually one; more when - the agent flushes a backlog). Every snapshot contributes time-series - DockerMetric rows; the newest snapshot drives current container state, the - alert pipeline, and lifecycle-event derivation. `swarm` is the newest + the agent flushes a backlog). Every snapshot's running containers record + cpu_pct, mem_pct and mem_usage_bytes into the one metric store, at the + snapshot's time; the newest snapshot drives current container state and + lifecycle-event derivation. `swarm` is the newest sample's swarm object (or None off managers) — persisted when present. `disk` is the newest sample's /system/df summary (or None on Docker-less hosts) — persisted when present. `logs` is a list of (recorded_at, records) @@ -323,7 +324,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, flushed before its mark, so a slow ingest names the part (#5181). """ from steward.core.alerts import MetricBatch - from .models import DockerContainer, DockerEvent, DockerMetric + from .models import DockerContainer, DockerEvent async def mark(name: str) -> None: if phases is not None: @@ -350,18 +351,20 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, ordered = sorted(snapshots, key=lambda s: s[0]) latest_at, latest_containers = ordered[-1] - # Time-series points for every snapshot (running containers with a CPU read). + # Time series for every snapshot (running containers with a CPU read), in + # the one metric store (#5752): resource "/", so + # containers of the same name on different hosts don't collide. Recorded + # with the alert batch below, at each snapshot's own time. + batch = MetricBatch() for recorded_at, containers in ordered: for c in containers: - if c.get("status") == "running" and c.get("cpu_pct") is not None: - session.add(DockerMetric( - host_id=host.id, - container_name=c["name"], - scraped_at=recorded_at, - cpu_pct=c["cpu_pct"], - mem_pct=c.get("mem_pct") or 0.0, - mem_usage_bytes=c.get("mem_usage_bytes") or 0, - )) + if c.get("status") != "running" or c.get("cpu_pct") is None or not c.get("name"): + continue + resource = f"{host.name}/{c['name']}" + for metric in ("cpu_pct", "mem_pct", "mem_usage_bytes"): + if c.get(metric) is not None: + batch.add(source_module="docker", resource_name=resource, + metric_name=metric, value=c[metric], recorded_at=recorded_at) await mark("metrics") # Snapshot of stored per-container state BEFORE the upsert overwrites it — @@ -389,8 +392,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, )) await mark("events") - # Current state + alerts from the newest snapshot only. - batch = MetricBatch() + # Current state, and the restart and health alerts, from the newest snapshot only. for c in latest_containers: name = c.get("name") if not name: @@ -424,19 +426,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, existing.blk_read_bytes = c.get("blk_read_bytes") existing.blk_write_bytes = c.get("blk_write_bytes") - # Alert pipeline — resource is host-scoped so containers of the same name - # on different hosts don't collide in the metric/alert namespace. resource = f"{host.name}/{name}" - if c.get("status") == "running" and c.get("cpu_pct") is not None: - batch.add( - source_module="docker", - resource_name=resource, metric_name="cpu_pct", value=c["cpu_pct"], - ) - if c.get("mem_pct") is not None: - batch.add( - source_module="docker", - resource_name=resource, metric_name="mem_pct", value=c["mem_pct"], - ) # Restart count is alertable regardless of state (crash-looping matters # most while the container is down/restarting). batch.add( diff --git a/plugins/docker/migrations/env.py b/plugins/docker/migrations/env.py index 3878d67..eb0c489 100644 --- a/plugins/docker/migrations/env.py +++ b/plugins/docker/migrations/env.py @@ -15,7 +15,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent)) from steward.models.base import Base import steward.models # noqa: F401 -from plugins.docker.models import DockerContainer, DockerMetric # noqa: F401 +from plugins.docker.models import DockerContainer # noqa: F401 config = context.config if config.config_file_name is not None: diff --git a/plugins/docker/migrations/versions/docker_012_drop_metrics.py b/plugins/docker/migrations/versions/docker_012_drop_metrics.py new file mode 100644 index 0000000..22f4572 --- /dev/null +++ b/plugins/docker/migrations/versions/docker_012_drop_metrics.py @@ -0,0 +1,56 @@ +"""Container CPU and memory move to the one metric store + +Drops docker_metrics and docker_metrics_hourly (#5752). Container cpu_pct, +mem_pct and mem_usage_bytes are recorded into plugin_metrics as source +"docker", resource "/", and roll up with every other metric. +CPU and memory % were already recorded there for alerting, so their history +stays; memory in bytes starts fresh. + +Revision ID: docker_012_drop_metrics +Revises: docker_011_host_status +Create Date: 2026-10-14 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_012_drop_metrics" +down_revision: Union[str, None] = "docker_011_host_status" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute("DROP TABLE IF EXISTS docker_metrics_hourly") + op.execute("DROP TABLE IF EXISTS docker_metrics") + + +def downgrade() -> None: + op.create_table( + "docker_metrics", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False, index=True), + sa.Column("container_name", sa.String(255), nullable=False, index=True), + sa.Column("scraped_at", sa.DateTime(timezone=True), nullable=False, index=True), + sa.Column("cpu_pct", sa.Float, nullable=False), + sa.Column("mem_pct", sa.Float, nullable=False), + sa.Column("mem_usage_bytes", sa.BigInteger, nullable=False), + ) + op.create_index("ix_docker_metrics_host_container_time", "docker_metrics", + ["host_id", "container_name", "scraped_at"]) + op.create_table( + "docker_metrics_hourly", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False), + sa.Column("container_name", sa.String(255), nullable=False), + sa.Column("bucket", sa.DateTime(timezone=True), nullable=False), + sa.Column("cpu_pct", sa.Float, nullable=False), + sa.Column("mem_pct", sa.Float, nullable=False), + sa.Column("mem_usage_bytes", sa.BigInteger, nullable=False), + sa.Column("sample_count", sa.Integer, nullable=False), + 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"]) diff --git a/plugins/docker/models.py b/plugins/docker/models.py index 99330de..2b63aa5 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -3,7 +3,6 @@ from __future__ import annotations 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 @@ -61,69 +60,6 @@ class DockerContainer(Base): blk_write_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True) -class DockerMetric(Base): - """Time-series CPU/memory per container — one row per sample per running - container, scoped to the reporting host.""" - __tablename__ = "docker_metrics" - - id: Mapped[str] = mapped_column( - String(36), primary_key=True, default=time_ordered_id - ) - host_id: Mapped[str] = mapped_column( - String(36), ForeignKey("hosts.id", ondelete="CASCADE"), nullable=False, index=True - ) - container_name: Mapped[str] = mapped_column(String(255), nullable=False, index=True) - scraped_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), nullable=False, - default=lambda: datetime.now(timezone.utc), - index=True, - ) - cpu_pct: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) - mem_pct: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) - # BigInteger: per-container memory usage exceeds 2^31 bytes (~2.1 GB) routinely. - mem_usage_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) - - # Per-container history lookups filter on (host_id, container_name) then sort - # by time — a composite index keeps the rows() sparkline queries cheap. - __table_args__ = ( - Index("ix_docker_metrics_host_container_time", - "host_id", "container_name", "scraped_at"), - ) - - -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=time_ordered_id - ) - 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. @@ -167,7 +103,7 @@ class DockerLog(Base): """Container log lines pushed by the host agent — one row per line. Time-series, scoped to the reporting host (container names are only unique - within a host, same identity as docker_metrics). The agent tails each running + within a host, same identity as docker_containers). The agent tails each running container incrementally and folds new lines into its metrics push; `ts` is the line's own Docker timestamp. Bounded by a per-container size+age ring (retention), so a chatty container just keeps a shorter window rather than diff --git a/plugins/docker/retention.py b/plugins/docker/retention.py index b235bd2..e08887f 100644 --- a/plugins/docker/retention.py +++ b/plugins/docker/retention.py @@ -1,133 +1,41 @@ # plugins/docker/retention.py -"""Bound Docker time-series growth: roll up old metrics, prune old rows. +"""Bound Docker history growth: prune old lifecycle events and log lines. -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. +Published as the "docker.run_retention" and "docker.prune_logs" capabilities +(see __init__.setup) so the core cleanup task can drive them WITHOUT importing +the docker models (same opportunistic-coupling pattern as +docker.persist_host_samples). Each 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. +Container CPU and memory are not here: they live in the one metric store +(plugin_metrics) and its rollup and retention (#5752). """ 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. - - The container-log ring is prune_docker_logs, run in its own transaction. - """ + """Prune docker_events older than the events window. Returns a counts + dict for logging. 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 sqlalchemy import delete - from .models import DockerEvent, DockerMetric, DockerMetricHourly + from .models import DockerEvent 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) + delete(DockerEvent).where(DockerEvent.at < now - timedelta(days=events_days)) ) - 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, - } + return {"events_pruned": res.rowcount or 0} async def prune_docker_logs( diff --git a/plugins/docker/routes.py b/plugins/docker/routes.py index 468aaf0..a76ff59 100644 --- a/plugins/docker/routes.py +++ b/plugins/docker/routes.py @@ -26,7 +26,7 @@ from .overview import ( stack_of, ) from .models import ( - DockerContainer, DockerDiskUsage, DockerHostStatus, DockerImage, DockerMetric, + DockerContainer, DockerDiskUsage, DockerHostStatus, DockerImage, DockerSwarmNode, DockerSwarmService, ) @@ -303,11 +303,7 @@ async def widget_resources(): # fill the list with duplicates of the same container. running = dedup_by_container_id([ c for c in await docker_svc.list_containers(db) if c.status == "running"]) - average = {(h, n): v for h, n, v in (await db.execute( - select(DockerMetric.host_id, DockerMetric.container_name, func.avg(DockerMetric.cpu_pct)) - .where(DockerMetric.scraped_at >= since) - .group_by(DockerMetric.host_id, DockerMetric.container_name) - )).all()} + average = await docker_svc.average_cpu(db, {(c.host_id, c.name) for c in running}, since) def cpu_of(c): v = average.get((c.host_id, c.name)) diff --git a/plugins/docker/service.py b/plugins/docker/service.py index 59d0f18..108e88d 100644 --- a/plugins/docker/service.py +++ b/plugins/docker/service.py @@ -11,12 +11,13 @@ from datetime import datetime, timedelta, timezone from sqlalchemy import func, select -from steward.core.time_range import bucket_seconds +from steward.core import metric_series from steward.models.hosts import Host +from steward.models.metrics import PluginMetric from .dedup import dedup_by_container_id from .models import ( - DockerContainer, DockerEvent, DockerLog, DockerMetric, DockerMetricHourly, DockerSwarmService, + DockerContainer, DockerEvent, DockerLog, DockerSwarmService, ) from .swarm_view import replaced_tasks @@ -270,39 +271,65 @@ async def query_logs(db, host_id: str, name: str, stream: str, query: str, CONTAINER_SERIES = ("cpu_pct", "mem_pct", "mem_usage_bytes") +# A container's series live in the one metric store (#5752): source "docker", +# resource "/", recorded at ingest. +def container_resource(host_name: str, name: str) -> str: + return f"{host_name}/{name}" + + +async def container_resources(db, keys) -> dict[str, tuple[str, str]]: + """{resource: (host_id, container_name)} for (host_id, container_name) + keys whose host still exists.""" + keys = set(keys) + hosts = await host_map(db, {h for h, _ in keys}) + return {container_resource(hosts[h].name, n): (h, n) for h, n in keys if h in hosts} + + async def container_points(db, host_id: str, name: str, metric: str, since, target: int = 120) -> list[tuple]: - """(time, average) points of one container metric since `since`, oldest first. - - Raw samples are bucket-averaged in SQL; the part of the range older than - the oldest raw sample (already rolled up and pruned by retention) comes - from the hourly table. - """ + """(time, average) points of one container metric since `since`, oldest + first: raw samples bucket-averaged, and the hourly rollup for the part of + the range past the raw window (metric_series.points).""" if metric not in CONTAINER_SERIES: raise ValueError(f"metric must be one of {', '.join(CONTAINER_SERIES)}") - raw_col = getattr(DockerMetric, metric) - width_s = bucket_seconds(since, target) - rbucket = func.date_bin( - func.make_interval(0, 0, 0, 0, 0, 0, width_s), DockerMetric.scraped_at, func.to_timestamp(0), - ).label("bucket") - raw = (await db.execute( - select(rbucket, func.avg(raw_col)) - .where(DockerMetric.host_id == host_id, DockerMetric.container_name == name, - DockerMetric.scraped_at >= since) - .group_by(rbucket).order_by(rbucket) - )).all() - oldest_raw = raw[0][0] if raw else None - hourly_stmt = ( - select(DockerMetricHourly.bucket, getattr(DockerMetricHourly, metric)) - .where(DockerMetricHourly.host_id == host_id, - DockerMetricHourly.container_name == name, - DockerMetricHourly.bucket >= since) - .order_by(DockerMetricHourly.bucket) - ) - if oldest_raw is not None: - hourly_stmt = hourly_stmt.where(DockerMetricHourly.bucket < oldest_raw) - hourly = (await db.execute(hourly_stmt)).all() - return [(t, float(v)) for t, v in hourly] + [(t, float(v)) for t, v in raw] + host = await db.get(Host, host_id) + if host is None: + return [] + return await metric_series.points(db, "docker", container_resource(host.name, name), + metric, since, target=target) + + +async def _samples(db, resources, metrics, since: datetime, width: int | None = None): + """Rows of (resource, time, metric, value) for the given containers since + `since`: each sample, or with `width` (seconds) each bucket's average.""" + at = PluginMetric.recorded_at + if width: + at = func.date_bin(func.make_interval(0, 0, 0, 0, 0, 0, int(width)), + PluginMetric.recorded_at, func.to_timestamp(0)) + value = func.avg(PluginMetric.value) if width else PluginMetric.value + q = (select(PluginMetric.resource_name, at.label("at"), PluginMetric.metric_name, value) + .where(PluginMetric.source_module == "docker", + PluginMetric.resource_name.in_(list(resources)), + PluginMetric.metric_name.in_(list(metrics)), + PluginMetric.recorded_at >= since)) + if width: + q = q.group_by(PluginMetric.resource_name, at, PluginMetric.metric_name) + return (await db.execute(q.order_by(at))).all() + + +async def average_cpu(db, keys, since: datetime) -> dict[tuple[str, str], float]: + """{(host_id, container_name): its average cpu_pct since `since`}.""" + resources = await container_resources(db, keys) + if not resources: + return {} + rows = (await db.execute( + select(PluginMetric.resource_name, func.avg(PluginMetric.value)) + .where(PluginMetric.source_module == "docker", + PluginMetric.resource_name.in_(list(resources)), + PluginMetric.metric_name == "cpu_pct", + PluginMetric.recorded_at >= since) + .group_by(PluginMetric.resource_name))).all() + return {resources[r]: float(v) for r, v in rows if v is not None} # Busiest Containers bars (#5214): the recent past in BARS equal slices. @@ -322,25 +349,15 @@ def bar_slots(now: datetime, hours: int, bars: int = BARS) -> tuple[list[datetim async def recent_usage(db, keys, since: datetime, width: int) -> dict: """{(host_id, container_name): {bucket start: (cpu_pct, mem_usage_bytes)}} averaged per bucket since `since`, for the given containers, in one query.""" - keys = set(keys) - if not keys: + resources = await container_resources(db, keys) + if not resources: return {} - bucket = func.date_bin( - func.make_interval(0, 0, 0, 0, 0, 0, int(width)), DockerMetric.scraped_at, func.to_timestamp(0), - ).label("bucket") - rows = (await db.execute( - select(DockerMetric.host_id, DockerMetric.container_name, bucket, - func.avg(DockerMetric.cpu_pct), func.avg(DockerMetric.mem_usage_bytes)) - .where(DockerMetric.scraped_at >= since, - DockerMetric.host_id.in_({h for h, _ in keys}), - DockerMetric.container_name.in_({n for _, n in keys})) - .group_by(DockerMetric.host_id, DockerMetric.container_name, bucket) - )).all() out: dict = {} - for host_id, name, at, cpu, mem in rows: - if (host_id, name) in keys: - out.setdefault((host_id, name), {})[at] = ( - None if cpu is None else float(cpu), None if mem is None else float(mem)) + for resource, at, metric, value in await _samples( + db, resources, ("cpu_pct", "mem_usage_bytes"), since, width): + cpu, mem = out.setdefault(resources[resource], {}).get(at, (None, None)) + v = None if value is None else float(value) + out[resources[resource]][at] = (v, mem) if metric == "cpu_pct" else (cpu, v) return out @@ -523,15 +540,16 @@ def sum_series(samples, since: datetime, until: datetime, points: int = 60) -> l async def service_history(db, containers, since: datetime) -> list: """sum_series over a service's task containers' samples since `since`.""" - from sqlalchemy import tuple_ - if not containers: + resources = await container_resources(db, containers) + if not resources: return [] - rows = (await db.execute( - select(DockerMetric.container_name, DockerMetric.scraped_at, - DockerMetric.cpu_pct, DockerMetric.mem_usage_bytes) - .where(tuple_(DockerMetric.host_id, DockerMetric.container_name).in_(list(containers)), - DockerMetric.scraped_at >= since) - .order_by(DockerMetric.scraped_at))).all() + joined: dict[tuple[str, datetime], list] = {} + for resource, at, metric, value in await _samples( + db, resources, ("cpu_pct", "mem_usage_bytes"), since): + pair = joined.setdefault((resource, at), [None, None]) + pair[0 if metric == "cpu_pct" else 1] = value + rows = [(resource, at, cpu, mem) for (resource, at), (cpu, mem) in + sorted(joined.items(), key=lambda kv: kv[0][1])] return sum_series(rows, since, datetime.now(timezone.utc)) diff --git a/steward/backup/contents.py b/steward/backup/contents.py index 3e4c5a5..dcb35f9 100644 --- a/steward/backup/contents.py +++ b/steward/backup/contents.py @@ -40,7 +40,7 @@ EXCLUDED: frozenset[str] = frozenset({ # Run history; its logs live on disk, not in the database. "ansible_runs", "docker_containers", "docker_disk_usage", "docker_events", "docker_host_status", - "docker_images", "docker_logs", "docker_metrics", "docker_metrics_hourly", + "docker_images", "docker_logs", "docker_swarm_nodes", "docker_swarm_services", "traefik_access_summaries", "traefik_certs", "traefik_global_stats", "traefik_metrics", "traefik_status_codes", diff --git a/steward/core/cleanup.py b/steward/core/cleanup.py index 6ad8621..85e2199 100644 --- a/steward/core/cleanup.py +++ b/steward/core/cleanup.py @@ -23,8 +23,9 @@ logger = logging.getLogger(__name__) # Time-series columns pruned at data.retention_days (#5155). A plugin adds its # own with prune_by_age() in setup(), so the core never imports plugin models. -# Tables kept another way are not listed: plugin_metrics and the docker tables -# are rolled up (below), and the audit log has its own, longer window. +# Tables kept another way are not listed: plugin_metrics is rolled up (below), +# the docker events and logs have their own windows, and the audit log has its +# own, longer window. _AGED: list = [MonitorResult.checked_at, AnsibleRun.started_at, AlertEvent.transitioned_at, EmailOutbox.created_at] @@ -40,9 +41,7 @@ def prune_by_age(column) -> None: _METRIC_COUNTS = [("raw_rows_rolled", "plugin_metrics: raw samples rolled into hourly"), ("rollup_pruned", "plugin_metrics_hourly: past the rollup window"), ("latest_pruned", "plugin_metric_latest: stopped reporting")] -_DOCKER_COUNTS = [("raw_rows_rolled", "docker_metrics: raw samples rolled into hourly"), - ("rollup_pruned", "docker_metrics_hourly: past the rollup window"), - ("events_pruned", "docker_events: past the events window"), +_DOCKER_COUNTS = [("events_pruned", "docker_events: past the events window"), ("logs_age_pruned", "docker_logs: past the age window"), ("logs_size_pruned", "docker_logs: over a container's size cap")] @@ -142,13 +141,13 @@ async def _run_metrics_retention(session, now: datetime) -> dict: async def _run_docker_retention(session, now: datetime) -> dict | None: - """Drive the docker plugin's rollup + prune and its log ring via its + """Drive the docker plugin's event 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). 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 + prune above, and the log ring's deletes don't wait on the event prune (#5449). No-op when the docker plugin is disabled (capability absent). """ from steward.core.capabilities import has_capability, invoke_capability @@ -160,13 +159,10 @@ async def _run_docker_retention(session, now: datetime) -> dict | None: # 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, + events_days=events_days, now=now, ) async with session.begin(): logs_days = int(await get_setting(session, "docker.logs.retention_days") or 3) diff --git a/steward/core/formatting.py b/steward/core/formatting.py index 699f661..d6d32b8 100644 --- a/steward/core/formatting.py +++ b/steward/core/formatting.py @@ -311,6 +311,7 @@ METRICS: dict[str, tuple[str, str]] = { "total_clients": ("Connected clients", "devices on the network"), "cpu_pct": ("CPU", "CPU in use; for a container, 100% is one core"), "mem_pct": ("Memory (% of limit)", "memory in use, against the container's limit or the host's RAM"), + "mem_usage_bytes": ("Memory used", "a container's memory in use, in bytes"), "restart_count": ("Restarts", "times the container has restarted, in total"), "disk_total_bytes": ("Docker disk use", "everything Docker keeps on the host (resource: the host)"), "disk_images_bytes": ("Docker images", "space Docker's images take on the host"), diff --git a/steward/core/settings.py b/steward/core/settings.py index c80453b..51a8798 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -117,11 +117,8 @@ DEFAULTS: dict[str, Any] = { # Days an ended metric anomaly waits to be reviewed before it lapses: # it leaves the to-review list and is kept as not reviewed (#5740). "anomalies.review_days": 14, - # 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 lifecycle events are light; keep a month (rule 25 — tunable, no + # restart). Container CPU and memory follow metrics.retention.* (#5752). "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 diff --git a/steward/settings/routes.py b/steward/settings/routes.py index f9b795b..7ac8603 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -149,8 +149,6 @@ _THRESHOLD_FIELDS = [ # 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"), ("docker_logs_retention_days", "docker.logs.retention_days"), ("metrics_raw_days", "metrics.retention.raw_days"), diff --git a/steward/templates/settings/thresholds.html b/steward/templates/settings/thresholds.html index 6a1f0ae..38f14ed 100644 --- a/steward/templates/settings/thresholds.html +++ b/steward/templates/settings/thresholds.html @@ -104,18 +104,13 @@
-

Docker data retention

+

Docker lifecycle events

- 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. + How long container start, stop, die and health changes are kept. Container CPU + and memory are kept with the other metrics, under Metric retention below. + 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.") }}
@@ -169,16 +164,18 @@
-

Host metrics retention

+

Metric retention

- Bounds how much host-agent metric history is stored (CPU, memory, disk, network, - temps, …). Raw per-sample points are kept for the raw window, then rolled up - into hourly averages kept for the rollup window. Host charts read raw for the - recent part of a range and hourly for older data. Applied by the hourly cleanup. + Bounds how much metric history is stored: hosts (CPU, memory, disk, network, + temps, …), containers (CPU and memory), monitors and the other sources. Raw + per-sample points are kept for the raw window, then rolled up into hourly + figures (average, peak, and the level held) kept for the rollup window. Charts + read raw for the recent part of a range and hourly for older data. Applied by + the hourly cleanup.

{{ days("Raw metrics", "metrics_raw_days", "metrics.retention.raw_days", - "Keep per-sample host metrics this long, then roll up to hourly averages.") }} + "Keep per-sample metrics this long, then roll up to hourly.") }} {{ days("Rolled-up metrics", "metrics_rollup_days", "metrics.retention.rollup_days", "Keep the hourly-averaged series this long for multi-week history.") }}
diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 94861e0..251e9a5 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -36,8 +36,8 @@ def test_docker_tables_are_host_scoped(app): async def _go(): async with app.db_sessionmaker() as s: - # Both tables carry a host_id column after docker_002. - for tbl in ("docker_containers", "docker_metrics"): + # docker_containers carries a host_id column after docker_002. + for tbl in ("docker_containers",): cols = {r[0] for r in (await s.execute(text( "SELECT column_name FROM information_schema.columns " "WHERE table_name = :t"), {"t": tbl})).all()} @@ -148,7 +148,6 @@ def test_persist_scopes_containers_by_host(app): async def _go(): async with app.db_sessionmaker() as s: async with s.begin(): - await s.execute(text("DELETE FROM docker_metrics")) await s.execute(text("DELETE FROM docker_containers")) await s.execute(text( "DELETE FROM plugin_metrics WHERE source_module = 'docker'")) @@ -162,10 +161,12 @@ def test_persist_scopes_containers_by_host(app): web_rows = (await s.execute(text( "SELECT COUNT(*) FROM docker_containers WHERE name = 'web'"))).scalar() + # The series went into the one metric store, host-scoped (#5752), + # at the snapshot's time. metric_hosts = (await s.execute(text( - "SELECT COUNT(DISTINCT host_id) FROM docker_metrics " - "WHERE container_name = 'web'"))).scalar() - # Alert pipeline received host-scoped resource names. + "SELECT COUNT(DISTINCT resource_name) FROM plugin_metrics " + "WHERE source_module = 'docker' AND metric_name = 'mem_usage_bytes' " + "AND recorded_at = :at"), {"at": now})).scalar() resources = {r[0] for r in (await s.execute(text( "SELECT DISTINCT resource_name FROM plugin_metrics " "WHERE source_module = 'docker'"))).all()} @@ -295,7 +296,6 @@ def test_large_memory_values_persist_as_bigint(app): async def _go(): async with app.db_sessionmaker() as s: async with s.begin(): - await s.execute(text("DELETE FROM docker_metrics")) await s.execute(text("DELETE FROM docker_containers")) h = Host(id=str(uuid.uuid4()), name="bigmemhost", address="10.0.0.9") s.add(h) @@ -305,8 +305,9 @@ def test_large_memory_values_persist_as_bigint(app): "SELECT mem_usage_bytes, mem_limit_bytes FROM docker_containers " "WHERE name = 'bigmem'"))).first() metric = (await s.execute(text( - "SELECT mem_usage_bytes FROM docker_metrics " - "WHERE container_name = 'bigmem'"))).scalar() + "SELECT value FROM plugin_metrics WHERE source_module = 'docker' " + "AND resource_name = 'bigmemhost/bigmem' " + "AND metric_name = 'mem_usage_bytes'"))).scalar() return row, metric row, metric = asyncio.run(_go()) @@ -332,7 +333,6 @@ def test_lifecycle_events_derived_across_snapshots(app): async with app.db_sessionmaker() as s: async with s.begin(): await s.execute(text("DELETE FROM docker_events")) - await s.execute(text("DELETE FROM docker_metrics")) await s.execute(text("DELETE FROM docker_containers")) h = Host(id=str(uuid.uuid4()), name="evt", address="10.9.9.9") s.add(h) @@ -659,90 +659,43 @@ def test_disk_usage_persisted(app): @_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.""" +def test_retention_prunes_events_past_their_window(app): + """Docker retention prunes lifecycle events past the events window; the + container series roll up with the one metric store (#5752).""" 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() + counts = await run_retention(s, events_days=30, now=now) 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 + tables = {r[0] for r in (await s.execute(text( + "SELECT table_name FROM information_schema.tables " + "WHERE table_name LIKE 'docker_metrics%'"))).all()} + return counts, events_left, tables - 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 + counts, events_left, tables = asyncio.run(_go()) 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 + assert counts == {"events_pruned": 1} + assert tables == set() # docker_012 dropped both tables @_NEEDS_DB diff --git a/tests/integration/test_docker_recent_usage.py b/tests/integration/test_docker_recent_usage.py index a6b9914..b0c75b9 100644 --- a/tests/integration/test_docker_recent_usage.py +++ b/tests/integration/test_docker_recent_usage.py @@ -1,6 +1,7 @@ """Integration: Busiest Containers bars average each container's samples per -time bucket in one query (#5214). Two samples share a bucket and must merge -(lesson #5262). Requires STEWARD_DATABASE_URL.""" +time bucket in one query (#5214), read from the one metric store (#5752). Two +samples share a bucket and must merge (lesson #5262). Requires +STEWARD_DATABASE_URL.""" from __future__ import annotations import asyncio @@ -33,7 +34,7 @@ def _docker(sub): def test_recent_usage_averages_each_containers_bucket(app): from steward.models.hosts import Host - DockerMetric = _docker("models").DockerMetric + from steward.models.metrics import PluginMetric service = _docker("service") bar_slots, recent_usage = service.bar_slots, service.recent_usage @@ -43,16 +44,18 @@ def test_recent_usage_averages_each_containers_bucket(app): at = slots[-2] # a whole bucket, in the past def m(name, secs, cpu, mem): - return DockerMetric(host_id=hid, container_name=name, scraped_at=at + timedelta(seconds=secs), - cpu_pct=cpu, mem_pct=0.0, mem_usage_bytes=mem) + when = at + timedelta(seconds=secs) + return [PluginMetric(source_module="docker", resource_name=f"dock-{tag}/{name}", + metric_name=metric, value=value, recorded_at=when) + for metric, value in (("cpu_pct", cpu), ("mem_usage_bytes", mem))] async def go(): async with app.db_sessionmaker() as s: async with s.begin(): s.add(Host(id=hid, name=f"dock-{tag}", address="10.9.9.4")) async with s.begin(): - s.add_all([m(f"a-{tag}", 10, 100.0, 1000), m(f"a-{tag}", 70, 300.0, 3000), - m(f"b-{tag}", 10, 50.0, 500)]) + s.add_all(m(f"a-{tag}", 10, 100.0, 1000) + m(f"a-{tag}", 70, 300.0, 3000) + + m(f"b-{tag}", 10, 50.0, 500)) async with s.begin(): return await recent_usage(s, {(hid, f"a-{tag}")}, slots[0], width) diff --git a/tests/integration/test_docker_service_page.py b/tests/integration/test_docker_service_page.py index a96d338..7e15c2c 100644 --- a/tests/integration/test_docker_service_page.py +++ b/tests/integration/test_docker_service_page.py @@ -36,6 +36,7 @@ def _docker(sub): def test_a_service_page_reads_every_task(app): from steward.models.hosts import Host + from steward.models.metrics import PluginMetric from steward.models.users import User, UserRole m = _docker("models") @@ -57,8 +58,10 @@ def test_a_service_page_reads_every_task(app): "exit_code": 1, "at": iso(now - timedelta(minutes=10))}] def sample(cname, mins, cpu, mem): - return m.DockerMetric(host_id=hid, container_name=cname, scraped_at=now - timedelta(minutes=mins), - cpu_pct=cpu, mem_pct=0.0, mem_usage_bytes=mem) + when = now - timedelta(minutes=mins) + return [PluginMetric(source_module="docker", resource_name=f"dock-{tag}/{cname}", + metric_name=metric, value=value, recorded_at=when) + for metric, value in (("cpu_pct", cpu), ("mem_usage_bytes", mem))] async def go(): async with app.db_sessionmaker() as s: @@ -86,9 +89,9 @@ def test_a_service_page_reads_every_task(app): ts=now - timedelta(minutes=5)), m.DockerLog(host_id=hid, container_name=dead, stream="stderr", line=f"panic {tag}", ts=now - timedelta(minutes=10)), - sample(dead, 40, 200.0, 300 << 20), sample(live, 30, 40.0, 90 << 20), - sample(live, 20, 60.0, 100 << 20), sample(live, 1, 50.0, 100 << 20), ]) + s.add_all(sample(dead, 40, 200.0, 300 << 20) + sample(live, 30, 40.0, 90 << 20) + + sample(live, 20, 60.0, 100 << 20) + sample(live, 1, 50.0, 100 << 20)) client = app.test_client() async with client.session_transaction() as sess: diff --git a/tests/integration/test_mcp_plugin_tools.py b/tests/integration/test_mcp_plugin_tools.py index 38f495a..351fa17 100644 --- a/tests/integration/test_mcp_plugin_tools.py +++ b/tests/integration/test_mcp_plugin_tools.py @@ -132,12 +132,16 @@ def test_docker_and_agent_tools_over_seeded_rows(app): "VALUES (:id, :h, :n, :ts, :st, :line)"), {"id": str(uuid.uuid4()), "h": hid, "n": web, "ts": now - timedelta(minutes=5 - i), "st": stream, "line": line}) + # Container series live in the one metric store (#5752). for i, cpu in enumerate((10.0, 30.0, 50.0)): - 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, :n, :t, :c, 20.0, 1000)"), - {"id": str(uuid.uuid4()), "h": hid, "n": web, - "t": now - timedelta(minutes=30 - i * 10), "c": cpu}) + for metric, value in (("cpu_pct", cpu), ("mem_pct", 20.0), + ("mem_usage_bytes", 1000.0)): + await s.execute(text( + "INSERT INTO plugin_metrics (id, source_module, resource_name, " + "metric_name, value, recorded_at) " + "VALUES (:id, 'docker', :r, :m, :v, :t)"), + {"id": str(uuid.uuid4()), "r": f"{host_name}/{web}", "m": metric, + "v": value, "t": now - timedelta(minutes=30 - i * 10)}) await s.execute(text( "INSERT INTO docker_swarm_services (host_id, service_name, mode, desired, running, " "placement_json, tasks_json, updated_at) " diff --git a/tests/plugins/docker/test_retention.py b/tests/plugins/docker/test_retention.py deleted file mode 100644 index 262d9c3..0000000 --- a/tests/plugins/docker/test_retention.py +++ /dev/null @@ -1,45 +0,0 @@ -"""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