feat(docker): a swarm service's metrics stay under one key across its tasks (#6718)

#6718 "A swarm container's metrics, normal and anomalies are keyed by its task
name, so they start over whenever the task moves". The operator: "the metrics
should be keys to the service name not the task name... as that's what's
consistent between updates."

CPU, memory, restart count and health now go under "swarm/<service>", so a
normal, its anomalies and an alert rule carry across a redeploy or a move
between nodes. Network rates stay "<host>/<service>", the task id gone, as they
answer who on a host is sending. Tasks of a multi-task service share the key,
so its level reads as an average across them (of 51 live services, only
authentik-worker and portainer_agent run more than one task).

History under task names moves in a background task, 5000 raw samples every
15 s, not a migration: the raw table holds about 2M such rows and the app
would wait on it at boot. Hourly rows combine where two tasks share an hour,
and rows still to review merge to one per metric.

A service's anomalies no longer reach a host's page, as it has no one host;
get_anomaly reads what happened around one from the service's own events.

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

plugins/docker/__init__.py

@@ -49,9 +49,12 @@ def setup(app: "Quart") -> None:
     from steward.core.norms import floor_setting, learn_normal
     learn_normal("docker", "cpu_pct", floor_setting(app, "container_cpu_pct"))
     learn_normal("docker", "mem_used_pct", floor_setting(app, "container_mem_pct"))
-    # A host's agent carries its containers' figures: they go silent with it (#5721).
+    # A host's agent carries its containers' figures: they go silent with it
+    # (#5721). A swarm service's aren't tied to one host.
     from steward.core.alerts import agent_fed
-    agent_fed("docker", lambda resource: resource.split("/", 1)[0])
+    from .service import SERVICE_RESOURCE
+    agent_fed("docker", lambda resource: None if resource.startswith(SERVICE_RESOURCE)
+              else resource.split("/", 1)[0])
     # An alert on a swarm figure names the services or nodes it counts (#6400).
     from steward.core.alerts import names_behind
     from .swarm_health import NAMERS
@@ -63,9 +66,10 @@ def setup(app: "Quart") -> None:
 
 
 def get_scheduled_tasks() -> list:
-    # Collection is now agent-driven (pushed to the host_agent ingest); the
-    # central socket scrape was removed. No periodic task to register.
-    return []
+    # Collection is agent-driven (pushed to the host_agent ingest). The one
+    # task here moves task-named history to its service.
+    from .rekey import make_rekey_task
+    return [make_rekey_task(_app)]
 
 
 def get_blueprint():
@@ -88,16 +92,25 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic
     return await events(db, host_id, host_name, since, until)
 
 
+async def mcp_resource_events(db, source: str, resource: str, since, until) -> list[dict]:
+    from .service import SERVICE_RESOURCE
+    if source != "docker" or not resource.startswith(SERVICE_RESOURCE):
+        return []
+    from .mcp_tools import service_events
+    return await service_events(db, resource[len(SERVICE_RESOURCE):], since, until)
+
+
 def get_nav() -> list[dict]:
     # Surfaces in the sidebar's Infrastructure group (fleet-wide container view).
     return [{"label": "Docker", "href": "/plugins/docker/"}]
 
 
 def get_alert_metrics() -> dict[str, list[str]]:
-    """What ingest.py records per container (resource "<host>/<container>"):
-    restart_count is cumulative (alert on crash-looping); is_healthy is 1.0 /
-    0.0 from the container's HEALTHCHECK, only for containers that define one.
-    net_rx_bps / net_tx_bps are the bytes a second it receives and sends
+    """What ingest.py records per container (resource "swarm/<service>" for a
+    swarm task, else "<host>/<container>"): restart_count is cumulative (alert
+    on crash-looping); is_healthy is 1.0 / 0.0 from the container's HEALTHCHECK,
+    only for containers that define one. net_rx_bps / net_tx_bps are the bytes
+    a second it receives and sends, under "<host>/<service or container>"
     (#6115): "a container flooding the uplink" is net_tx_bps over a level.
     Per host (resource "<host>"), what Docker keeps on its disk (#5304).
     Swarm-wide (resource "swarm"), from any manager's report: services short

plugins/docker/ingest.py

@@ -404,6 +404,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
     """
     from steward.core.alerts import MetricBatch
     from .models import DockerContainer, DockerEvent
+    from .service import container_resource, traffic_resource
 
     async def mark(name: str) -> None:
         if phases is not None:
@@ -432,9 +433,11 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
     latest_at, latest_containers = ordered[-1]
 
     # Time series for every snapshot (running containers with a CPU read), in
-    # the one metric store (#5752): resource "<host>/<container>", so
-    # containers of the same name on different hosts don't collide. Recorded
-    # with the alert batch below, at each snapshot's own time.
+    # the one metric store: resource "swarm/<service>" for a swarm task, so its
+    # history survives the task being replaced or moved, else
+    # "<host>/<container>", so containers of the same name on different hosts
+    # don't collide. Recorded with the alert batch below, at each snapshot's
+    # own time.
     # Snapshot of stored per-container state BEFORE the upsert overwrites it —
     # the network counters each rate counts from, and the lifecycle-event diff.
     # Skip event derivation on the host's first snapshot (empty state) so we
@@ -448,7 +451,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
         for c in containers:
             if c.get("status") != "running" or c.get("cpu_pct") is None or not c.get("name"):
                 continue
-            resource = f"{host.name}/{c['name']}"
+            resource = container_resource(host.name, c["name"])
             for metric in ("cpu_pct", "mem_used_pct", "mem_used_bytes", "mem_cache_bytes"):
                 if c.get(metric) is not None:
                     batch.add(source_module="docker", resource_name=resource,
@@ -457,7 +460,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
     # can be put down to the container behind it (#6115).
     before = {r.name: (r.scraped_at, r.container_id, r.net_rx_bytes, r.net_tx_bytes) for r in old_rows}
     for recorded_at, name, rx_bps, tx_bps in net_rates(before, ordered):
-        resource = f"{host.name}/{name}"
+        resource = traffic_resource(host.name, name)
         batch.add(source_module="docker", resource_name=resource,
                   metric_name="net_rx_bps", value=round(rx_bps, 1), recorded_at=recorded_at)
         batch.add(source_module="docker", resource_name=resource,
@@ -524,7 +527,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")
 
-        resource = f"{host.name}/{name}"
+        resource = container_resource(host.name, name)
         # Restart count is alertable regardless of state (crash-looping matters
         # most while the container is down/restarting).
         batch.add(

plugins/docker/mcp_tools.py

@@ -772,6 +772,21 @@ async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dic
                   subject=f"{host_name}/{e.container_name}") for e in rows]
 
 
+async def service_events(db, service: str, since, until) -> list[dict]:
+    """What happened to a swarm service's containers on every host in a window
+    (get_anomaly): starts with their image, deaths, OOM kills, health changes.
+    A service has no one host to read events by."""
+    from steward.mcp.tools._events import event
+    from steward.models.hosts import Host
+    rows = (await db.execute(
+        select(DockerEvent, Host.name).join(Host, Host.id == DockerEvent.host_id)
+        .where(DockerEvent.container_name.startswith(f"{service}.", autoescape=True),
+               DockerEvent.at >= since, DockerEvent.at <= until)
+        .order_by(DockerEvent.at).limit(500))).all()
+    return [event(e.at, "container", f"{e.container_name} {e.event}" + (f" ({e.detail})" if e.detail else ""),
+                  subject=f"{host}/{e.container_name}") for e, host in rows]
+
+
 def register_tools(register) -> None:
     for fn in (list_containers, get_container, list_swarm_services,
                search_container_logs, get_docker_disk_usage):

plugins/docker/rekey.py

@@ -0,0 +1,179 @@
+# plugins/docker/rekey.py
+"""Moves history kept under a swarm task's own name, "<host>/<service>.<slot>.
+<task>", onto its service's key, "swarm/<service>": samples, hourly rollups,
+anomalies and the alert rules and maintenance windows that name it.
+
+The raw table holds millions of rows, too many to move in a migration that the
+app has to wait for. A small task moves a batch at a time in the background
+instead, one key after another, and does nothing once none are left. Network
+rates stay where they are: they are kept per host, and age out with retention.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING
+
+from sqlalchemy import delete, select, text, update
+
+from steward.core.scheduler import ScheduledTask
+from steward.models.alerts import AlertRule, MaintenanceWindow
+from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest
+from steward.models.norms import MetricAnomaly, MetricAnomalySpan
+from steward.models.tickets import TicketLink
+from steward.models.timeline import TimelineEntry
+from steward.timeline.record import anomaly_changed
+
+from .service import container_resource, task_service
+
+if TYPE_CHECKING:
+    from quart import Quart
+
+logger = logging.getLogger(__name__)
+
+BATCH = 5000            # raw samples moved in one go
+EVERY_S = 15
+TRAFFIC = ("net_rx_bps", "net_tx_bps")
+_TASK_KEY = r"^[^/]+/.+\.([0-9]+|[a-z0-9]{25})\.[a-z0-9]{25}$"
+_NOT_TRAFFIC = "metric_name NOT IN ('net_rx_bps', 'net_tx_bps')"
+
+# Hours the service has no row for change name; hours it has are combined, as
+# two tasks of a service can share an hour.
+_RENAME_HOURS = text(f"""
+UPDATE plugin_metrics_hourly o SET resource_name = :new
+WHERE o.source_module = 'docker' AND o.resource_name = :old AND o.{_NOT_TRAFFIC}
+  AND NOT EXISTS (SELECT 1 FROM plugin_metrics_hourly n
+                  WHERE n.source_module = 'docker' AND n.resource_name = :new
+                    AND n.metric_name = o.metric_name AND n.bucket = o.bucket)
+""")
+_COMBINE_HOURS = text("""
+UPDATE plugin_metrics_hourly n SET
+    value_avg = COALESCE((n.value_avg * n.sample_count + o.value_avg * o.sample_count)
+                         / NULLIF(n.sample_count + o.sample_count, 0), n.value_avg),
+    value_max = GREATEST(n.value_max, o.value_max),
+    value_p95 = GREATEST(n.value_p95, o.value_p95),
+    sample_count = n.sample_count + o.sample_count
+FROM plugin_metrics_hourly o
+WHERE o.source_module = 'docker' AND o.resource_name = :old
+  AND n.source_module = 'docker' AND n.resource_name = :new
+  AND n.metric_name = o.metric_name AND n.bucket = o.bucket
+""")
+_DROP_HOURS = text(f"""
+DELETE FROM plugin_metrics_hourly
+WHERE source_module = 'docker' AND resource_name = :old AND {_NOT_TRAFFIC}
+""")
+
+
+def service_key(old: str) -> str:
+    """The key a task-named resource belongs under now."""
+    return container_resource("", old.split("/", 1)[1])
+
+
+async def old_keys(session) -> list[str]:
+    """Every task-named docker resource still holding something to move. The
+    latest-value table stands for the raw samples, as a key stays in it until
+    its samples have aged out."""
+    found: set[str] = set()
+    for model in (PluginMetricLatest, PluginMetricHourly):
+        found.update((await session.execute(
+            select(model.resource_name).distinct().where(
+                model.source_module == "docker", model.metric_name.notin_(TRAFFIC),
+                model.resource_name.op("~")(_TASK_KEY)))).scalars())
+    for model in (MetricAnomaly, AlertRule):
+        found.update((await session.execute(
+            select(model.resource_name).distinct().where(
+                model.source_module == "docker", model.resource_name.op("~")(_TASK_KEY)))).scalars())
+    found.update((await session.execute(
+        select(MaintenanceWindow.scope_resource).distinct().where(
+            MaintenanceWindow.scope_source == "docker",
+            MaintenanceWindow.scope_resource.op("~")(_TASK_KEY)))).scalars())
+    return sorted(k for k in found if task_service(k.split("/", 1)[1]))
+
+
+async def _merge_anomalies(session, new: str) -> None:
+    """One row per metric still to review, as for any key: the rows waiting on
+    the service become the newest of them, with all their occurrences."""
+    rows = (await session.execute(
+        select(MetricAnomaly).where(
+            MetricAnomaly.source_module == "docker", MetricAnomaly.resource_name == new,
+            MetricAnomaly.review == "to_review")
+        .order_by(MetricAnomaly.last_started_at.desc()))).scalars().all()
+    keep: dict[str, MetricAnomaly] = {}
+    for row in rows:
+        into = keep.setdefault(row.metric_name, row)
+        if into is row:
+            continue
+        await session.execute(update(MetricAnomalySpan).where(
+            MetricAnomalySpan.anomaly_id == row.id).values(anomaly_id=into.id))
+        into.occurrences += row.occurrences
+        into.started_at = min(into.started_at, row.started_at)
+        if row.ended_at is None:
+            into.ended_at = None
+        if row.peak_level > into.peak_level:
+            into.peak_level, into.peak_at = row.peak_level, row.peak_at
+            into.usual_hold, into.usual_peak = row.usual_hold, row.usual_peak
+        linked = (await session.execute(select(TicketLink.ticket_id).where(
+            TicketLink.ref_type == "metric_anomaly", TicketLink.ref_id == into.id))).scalars().all()
+        await session.execute(update(TicketLink).where(
+            TicketLink.ref_type == "metric_anomaly", TicketLink.ref_id == row.id,
+            TicketLink.ticket_id.notin_(linked)).values(ref_id=into.id))
+        await session.execute(delete(TicketLink).where(
+            TicketLink.ref_type == "metric_anomaly", TicketLink.ref_id == row.id))
+        await session.execute(delete(TimelineEntry).where(
+            TimelineEntry.ref_type == "metric_anomaly", TimelineEntry.ref_id == row.id))
+        await session.delete(row)
+
+
+async def move_key(session, old: str) -> bool:
+    """Move the next batch of `old`, and the rest of what it holds once its raw
+    samples are done. Returns whether it is finished. Runs inside the caller's
+    transaction."""
+    new = service_key(old)
+    batch = select(PluginMetric.id).where(
+        PluginMetric.source_module == "docker", PluginMetric.resource_name == old,
+        PluginMetric.metric_name.notin_(TRAFFIC)).limit(BATCH)
+    moved = await session.execute(
+        update(PluginMetric).where(PluginMetric.id.in_(batch)).values(resource_name=new)
+        .execution_options(synchronize_session=False))
+    if moved.rowcount >= BATCH:
+        return False
+    keys = {"old": old, "new": new}
+    for statement in (_RENAME_HOURS, _COMBINE_HOURS, _DROP_HOURS):
+        await session.execute(statement, keys)
+    await session.execute(delete(PluginMetricLatest).where(
+        PluginMetricLatest.source_module == "docker", PluginMetricLatest.resource_name == old,
+        PluginMetricLatest.metric_name.notin_(TRAFFIC)))
+    await session.execute(update(AlertRule).where(
+        AlertRule.source_module == "docker", AlertRule.resource_name == old).values(resource_name=new))
+    await session.execute(update(MaintenanceWindow).where(
+        MaintenanceWindow.scope_source == "docker", MaintenanceWindow.scope_resource == old)
+        .values(scope_resource=new))
+    await session.execute(update(MetricAnomaly).where(
+        MetricAnomaly.source_module == "docker", MetricAnomaly.resource_name == old)
+        .values(resource_name=new).execution_options(synchronize_session=False))
+    session.expire_all()
+    await _merge_anomalies(session, new)
+    await session.flush()
+    for row in (await session.execute(select(MetricAnomaly).where(
+            MetricAnomaly.source_module == "docker", MetricAnomaly.resource_name == new))).scalars().all():
+        await anomaly_changed(session, row)
+    return True
+
+
+def make_rekey_task(app: "Quart") -> ScheduledTask:
+    remaining: list[str] | None = None
+    busy = asyncio.Lock()
+
+    async def rekey() -> None:
+        nonlocal remaining
+        if remaining == [] or busy.locked():
+            return
+        async with busy, app.db_sessionmaker() as session:
+            async with session.begin():
+                if remaining is None:
+                    remaining = await old_keys(session)
+                    logger.info("Swarm history under task names: %d keys to move", len(remaining))
+                if remaining and await move_key(session, remaining[-1]):
+                    remaining.pop()
+
+    return ScheduledTask(name="docker_rekey", coro_factory=rekey, interval_seconds=EVERY_S)

plugins/docker/service.py

@@ -227,7 +227,8 @@ async def container_norms(db, keys, now: datetime | None = None) -> dict:
                          now or datetime.now(timezone.utc))
     out: dict = {}
     for (_s, resource, metric), norm in got.items():
-        out.setdefault(resources[resource], {})[metric] = norm
+        for key in resources[resource]:
+            out.setdefault(key, {})[metric] = norm
     return out
 
 
@@ -318,18 +319,43 @@ async def query_logs(db, host_id: str, name: str, stream: str, query: str,
 CONTAINER_SERIES = ("cpu_pct", "mem_used_pct", "mem_used_bytes", "mem_cache_bytes")
 
 
-# A container's series live in the one metric store (#5752): source "docker",
-# resource "<host name>/<container name>", recorded at ingest.
+# A swarm task's container: "<service>.<slot or node id>.<task id>".
+_TASK = re.compile(r"^(?P<service>.+?)\.(?:\d+|[a-z0-9]{25})\.[a-z0-9]{25}$")
+SERVICE_RESOURCE = "swarm/"
+
+
+def task_service(name: str) -> str | None:
+    """The swarm service a container is a task of, from its name, or None."""
+    m = _TASK.match(name or "")
+    return m["service"] if m else None
+
+
+# A container's series live in the one metric store: source "docker", recorded
+# at ingest. A swarm service's are under "swarm/<service>", which stays the same
+# when a task is replaced or moves host, and which all its tasks share. Any
+# other container's are under "<host name>/<container name>".
 def container_resource(host_name: str, name: str) -> str:
-    return f"{host_name}/{name}"
+    service = task_service(name)
+    return f"{SERVICE_RESOURCE}{service}" if service else 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."""
+def traffic_resource(host_name: str, name: str) -> str:
+    """Where a container's network rates are kept: "<host name>/<service or
+    container>", as they say who on that host is carrying its traffic."""
+    return f"{host_name}/{task_service(name) or name}"
+
+
+async def container_resources(db, keys) -> dict[str, list[tuple[str, str]]]:
+    """{resource: [(host_id, container_name), ...]} for (host_id,
+    container_name) keys whose host still exists. A swarm service's tasks share
+    one resource."""
     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}
+    out: dict[str, list[tuple[str, str]]] = {}
+    for h, n in keys:
+        if h in hosts:
+            out.setdefault(container_resource(hosts[h].name, n), []).append((h, n))
+    return out
 
 
 async def container_points(db, host_id: str, name: str, metric: str, since,
@@ -376,7 +402,7 @@ async def average_cpu(db, keys, since: datetime) -> dict[tuple[str, str], float]
                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}
+    return {key: float(v) for r, v in rows if v is not None for key in resources[r]}
 
 
 # Busiest Containers bars (#5214): the recent past in BARS equal slices.
@@ -402,9 +428,10 @@ async def recent_usage(db, keys, since: datetime, width: int) -> dict:
     out: dict = {}
     for resource, at, metric, value in await _samples(
             db, resources, ("cpu_pct", "mem_used_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)
+        for key in resources[resource]:
+            cpu, mem = out.setdefault(key, {}).get(at, (None, None))
+            out[key][at] = (v, mem) if metric == "cpu_pct" else (cpu, v)
     return out
 
 

plugins/host_agent/senders.py

@@ -2,7 +2,7 @@
 ranked against the host's own total over a window.
 
 A host's net_tx_bps (net_rx_bps) is what its physical interfaces carried; each
-container's comes from Docker ingest (resource "<host>/<container>") and each
+container's comes from Docker ingest (resource "<host>/<service or container>") and each
 NFS mount's from the agent (resource "<host>:<mount>"). All three are read in
 the same time buckets, so the busiest bucket of the host's total can be put
 down to who was sending in it. What no container or mount accounts for is the