refactor(docker): container CPU and memory live in the one metric store (#5752)

#5752 "Container CPU and memory history lives in the one metric store;
docker_metrics and docker_metrics_hourly are retired". Container cpu_pct and
mem_pct already went to plugin_metrics for alerting; now mem_usage_bytes does
too, and every snapshot in an ingest (backlog included) is recorded at its
own time. The container history chart, Busiest Containers, a service's
history and the MCP container tools read plugin_metrics under resource
"<host>/<container>", rolled up and pruned by the shared metric retention.

docker_012_drop_metrics drops both tables. Their history is not copied
(lesson #5221 "A data migration's cost is set by the live table"); the
store already holds 7 days of container CPU and memory %. Docker retention
now prunes only lifecycle events, and Settings -> Thresholds has one
"Metric retention" card for every source.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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 "<host name>/<container name>", 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))