fix(anomalies): a burst within what a metric reaches in most hours opens nothing (#6727)

#6727 "A metric that bursts now and then opens anomalies at hours whose usual
peak reads 0". usual_peak is the p90 of a slot's hourly max, so a metric that
bursts in under about 1 hour in 10 reads 0 there however often it bursts across
the day. #6723 "An anomaly opens on a burst of minutes inside the metric's usual
peak, so most rows aren't outside the norm" therefore still opened on those.

usual_anywhere is the median, across the hours of the day, of each hour's
highest peak. Height is now judged against the greater of it and the slot's
usual peak. Bursts at any hour lift the median; a job at a time of its own
raises a few hours and leaves it low, so the same level at another time of day
is still an anomaly, as System 182 rules. A held level still opens.

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

docs/architecture.md

@@ -105,7 +105,7 @@ A two-minute spike raises the peak and leaves the p95 low; an hour spent at that
 
 A metric a plugin registers with `learn_normal()` gets a normal per hour of the day, then per hour of the week once three weeks of history exist (`metric_norms`, relearned hourly by `metric_norms`; `steward/core/norms.py`).
 
-Every minute, `metric_anomalies` averages each learned key's last 15 minutes and compares it with the normal for that hour. A level above the usual hold ×1.25, and above the metric's floor, has departed. It opens an anomaly once it is also past the usual peak for that hour, or held: the median of the last hour is over the same line. A burst of a few minutes within the usual peak lifts the 15-minute average and opens nothing (`steward/core/anomalies.py`):
+Every minute, `metric_anomalies` averages each learned key's last 15 minutes and compares it with the normal for that hour. A level above the usual hold ×1.25, and above the metric's floor, has departed. It opens an anomaly once it is also higher than the metric usually goes, or held: the median of the last hour is over the same line. How high it usually goes is the greater of the usual peak for that hour and how high it has gone in most hours of the day, so bursts that land at any hour are told from a job with a time of its own. A burst of a few minutes within that lifts the 15-minute average and opens nothing (`steward/core/anomalies.py`):
 - `metric_anomalies` holds one row per item to review. A repeat while the row still waits joins it as another occurrence (`metric_anomaly_spans`).
 - A row is `to_review` until someone marks it `normal` or `real` (Alerts → Anomalies, or MCP `review_anomaly`). An ended row nobody reviews within `anomalies.review_days` (14) becomes `lapsed`.
 - Learning skips every hour an occurrence covers unless its anomaly was marked `normal`, so a new level is never learned without review.
@@ -118,7 +118,7 @@ What learns a normal, and the floor each departure must clear (Settings → Thre
 | Host CPU | `host_agent` / `cpu_pct` | 10 points |
 | Host memory | `host_agent` / `mem_used_pct` | 5 points |
 | Host load per core | `host_agent` / `load_pct` (load_1m ÷ cores × 100, recorded at ingest) | 10 points |
-| Container CPU | `docker` / `cpu_pct` (100% is one core) | 10 points of one core |
+| Container CPU | `docker` / `cpu_pct` (100% is one core; a swarm service's under `swarm/<service>`, shared by its tasks) | 10 points of one core |
 | Container memory | `docker` / `mem_used_pct` (its processes' own memory, file cache left out, of its limit or of host RAM when it has none) | 5 points |
 | Monitor response time | `<type>` / `response_ms` | 100 ms |
 | Router 4xx share, p95 | `traefik` / `error_rate_4xx_pct`, `latency_p95_ms` | 2 points, 50 ms |

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

steward/alerts/service.py

@@ -112,7 +112,8 @@ async def resource_url_for(db, source: str, resource: str) -> str | None:
     """Where a (source, resource) of the metric store is shown, or None.
 
     host_agent resources are the host's name, or "<host>:<mount|nic|core>";
-    docker's are "<host>/<container>"; a monitor's is its name under its type.
+    docker's are "<host>/<container>" or "swarm/<service>"; a monitor's is its
+    name under its type.
     """
     from steward.models.hosts import Host
     if source == "host_agent":
@@ -125,6 +126,8 @@ async def resource_url_for(db, source: str, resource: str) -> str | None:
             select(Monitor.id).where(Monitor.type == source, Monitor.name == resource).limit(1)
         )).scalar_one_or_none()
         return f"/monitors/{monitor_id}" if monitor_id else None
+    if source == "docker" and resource.startswith("swarm/"):
+        return f"/plugins/docker/service/{resource.split('/', 1)[1]}"
     if source == "docker" and "/" in resource:
         host_name, container = resource.split("/", 1)
         host_id = (await db.execute(

steward/core/alerts.py

@@ -65,9 +65,10 @@ def agent_heartbeat(source_module: str, metric_name: str,
     _HEARTBEAT = Heartbeat(source_module, metric_name, offline_after)
 
 
-def agent_fed(source_module: str, host_of: Callable[[str], str]) -> None:
+def agent_fed(source_module: str, host_of: Callable[[str], str | None]) -> None:
     """A source whose samples a host's agent carries; `host_of(resource)`
-    names the host ("nas:/" → "nas", "nas/web" → "nas")."""
+    names the host ("nas:/" → "nas", "nas/web" → "nas"), or None for a
+    resource no one host carries."""
     _AGENT_FED[source_module] = host_of
 
 

steward/core/anomalies.py

@@ -4,10 +4,10 @@ Every minute the last WINDOW of each learned key is averaged and compared
 with the normal for the slot it falls in (steward.core.norms). A level above
 the slot's usual hold by DEPART times, and by at least the metric's floor, has
 departed. That opens an occurrence once it is also outside what the normal
-records in height or in time: higher than the slot's usual peak, or held, with
-most of the last HELD above the line. A burst of a few minutes within the
-usual peak lifts the average for a while and opens nothing. The occurrence ends
-when the level comes back under the line.
+records in height or in time: higher than the metric usually goes (Norm.
+usual_height), or held, with most of the last HELD above the line. A burst of
+a few minutes within that lifts the average for a while and opens nothing. The
+occurrence ends when the level comes back under the line.
 
 What the operator ruled (System 182 "Metric Norms and Anomalies"):
 - A departure is a warning, marked as a metric anomaly, and stays "to review"
@@ -32,7 +32,7 @@ from steward.timeline.record import anomaly_changed
 
 WINDOW = timedelta(minutes=15)   # the level judged is this window's average
 DEPART = 1.25                    # times the slot's usual hold that counts as a departure
-HELD = timedelta(hours=1)        # a level within the usual peak opens once the median of this is out
+HELD = timedelta(hours=1)        # a level within its usual height opens once the median of this is out
 # An occurrence that comes back within one window of ending continues the
 # same occurrence: a level wobbling around the line is one event, not ten.
 REJOIN = WINDOW
@@ -169,7 +169,7 @@ async def detect(session, now: datetime | None = None) -> dict:
     out_now = {key for key, level in levels.items()
                if key in norms and departs(level, norms[key], LEARNED[(key[0], key[2])].least_departure)}
     held = await held_levels(session, [key for key in out_now
-                                       if key not in open_now and levels[key] <= norms[key].usual_peak], now)
+                                       if key not in open_now and levels[key] <= norms[key].usual_height], now)
     for key, level in levels.items():
         norm = norms.get(key)
         out = key in out_now
@@ -182,7 +182,7 @@ async def detect(session, now: datetime | None = None) -> dict:
                 _close(anomaly, span, now)
                 changed.append(anomaly)
                 closed += 1
-        elif out and level > norm.usual_peak:
+        elif out and level > norm.usual_height:
             changed.append(await _occur(session, key, level, norm, now, WINDOW))
             opened += 1
         elif out and departs(held.get(key, 0.0), norm, LEARNED[(key[0], key[2])].least_departure):

steward/core/norms.py

@@ -11,7 +11,12 @@ What a slot's normal is:
   (plugin_metrics_hourly.value_p95, #5738), at its 90th percentile across
   past days. A two-minute spike doesn't raise it; an hour spent high does.
 - usual_peak: the same for each hour's max, so a page can say how high the
-  slot usually goes.
+  slot usually goes. Most hours of a metric that bursts now and then see no
+  burst, so this can read 0 for a slot such a metric bursts in as much as any.
+- usual_anywhere: how high the metric has gone in most hours of the day, the
+  median across the 24 hours of each one's highest peak. Bursts that land at
+  any hour raise it. A job at a time of its own raises a few hours and leaves
+  the median alone, so the same level at another time still stands out.
 
 The slots:
 - An hour of the day (24 slots) once a key has DAY_MIN_DAYS of history.
@@ -116,6 +121,13 @@ class Norm:
     usual_hold: float
     usual_peak: float
     days_seen: int
+    usual_anywhere: float = 0.0
+
+    @property
+    def usual_height(self) -> float:
+        """The highest this metric usually goes at this time: a level past it
+        is news however briefly it lasted."""
+        return max(self.usual_peak, self.usual_anywhere)
 
 
 def slot_of(at: datetime, kind: str) -> int:
@@ -171,6 +183,12 @@ kinds AS (
     SELECT s, r, m, days, CASE WHEN first <= :week_ready THEN 'week' ELSE 'day' END AS kind
     FROM keyed WHERE days >= CAST(:day_min_days AS int)
 ),
+anywhere AS (
+    SELECT s, r, m, percentile_cont(0.5) WITHIN GROUP (ORDER BY highest) AS peak
+    FROM (SELECT s, r, m, max(peak) AS highest FROM hours
+          GROUP BY s, r, m, EXTRACT(HOUR FROM b AT TIME ZONE 'UTC')) by_hour
+    GROUP BY s, r, m
+),
 slotted AS (
     SELECT h.s, h.r, h.m, k.kind, k.days, h.hold, h.peak,
            CASE k.kind
@@ -183,13 +201,15 @@ slotted AS (
     WHERE k.kind = 'week' OR h.b >= :day_since
 )
 INSERT INTO metric_norms (source_module, resource_name, metric_name, slot, kind,
-                          usual_hold, usual_peak, days_seen, updated_at)
-SELECT s, r, m, (slot + n.off + slots) % slots, kind,
-       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY hold),
-       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY peak),
-       days, :now
-FROM slotted CROSS JOIN generate_series(-CAST(:neighbours AS int), CAST(:neighbours AS int)) AS n(off)
-GROUP BY s, r, m, (slot + n.off + slots) % slots, kind, days
+                          usual_hold, usual_peak, usual_anywhere, days_seen, updated_at)
+SELECT l.s, l.r, l.m, (l.slot + n.off + l.slots) % l.slots, l.kind,
+       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY l.hold),
+       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY l.peak),
+       a.peak, l.days, :now
+FROM slotted l
+JOIN anywhere a ON a.s = l.s AND a.r = l.r AND a.m = l.m
+CROSS JOIN generate_series(-CAST(:neighbours AS int), CAST(:neighbours AS int)) AS n(off)
+GROUP BY l.s, l.r, l.m, (l.slot + n.off + l.slots) % l.slots, l.kind, l.days, a.peak
 """).bindparams(bindparam("keys", type_=ARRAY(String)),
                bindparam("starts", type_=ARRAY(DateTime(timezone=True))),
                bindparam("parts", type_=ARRAY(Boolean)))
@@ -235,7 +255,7 @@ async def norms_at(session, keys: list[tuple[str, str, str]],
                | ((MetricNorm.kind == "week") & (MetricNorm.slot == week)))
     )).scalars().all()
     return {(n.source_module, n.resource_name, n.metric_name):
-            Norm(n.kind, n.slot, n.usual_hold, n.usual_peak, n.days_seen) for n in rows}
+            Norm(n.kind, n.slot, n.usual_hold, n.usual_peak, n.days_seen, n.usual_anywhere) for n in rows}
 
 
 async def run_norms(app) -> None:

steward/mcp/tools/anomalies.py

@@ -53,7 +53,7 @@ async def list_anomalies(review: str = "to_review", limit: int = 50) -> dict:
     """Metric anomalies: a learned metric whose 15-minute level went well
     above what is normal for it at that hour (its usual hold ×1.25, and by
     more than the metric's floor), and either past how high it usually goes
-    (`usual_peak`) or held there for most of an hour.
+    (see get_metric_normal) or held there for most of an hour.
 
     review: "to_review" (default) — waiting for someone to mark them; "normal"
     — accepted, their hours are learned; "real" — a real problem, not learned;
@@ -291,8 +291,10 @@ async def get_metric_normal(source_module: str, resource: str, metric: str) -> d
     90th percentile across past days), `usual_peak` (how high they usually
     went), `days_seen`, and `flag_above`: the 15-minute average that has
     departed there (usual hold ×1.25, and at least `floor` above it). A
-    departure opens an anomaly once it passes `usual_peak`, or when the median
-    of the last hour is over `flag_above` too.
+    departure opens an anomaly once it passes both `usual_peak` and
+    `usual_anywhere` (how high the metric has gone in most hours of the day,
+    one figure for the whole key), or when the median of the last hour is over
+    `flag_above` too.
     `floor` is the smallest departure worth calling an anomaly, and
     `floor_setting` the Settings → Thresholds floor it comes from. Hours an
     anomaly covers are learned only once someone marks it normal.
@@ -319,6 +321,7 @@ async def get_metric_normal(source_module: str, resource: str, metric: str) -> d
     out = {
         "source_module": source_module, "resource": resource, "metric": metric,
         "learned": bool(slots), "kind": kind,
+        "usual_anywhere": rnd(rows[0].usual_anywhere, 3) if rows else None,
         "floor": rnd(floor, 3), "floor_setting": learned.floor_name,
         "only_while": ({"metric": learned.gate[0],
                         "at_least": rnd(learned.gate[1]() if callable(learned.gate[1]) else learned.gate[1], 3),

steward/migrations/versions/0050_norm_usual_anywhere.py

@@ -0,0 +1,23 @@
+"""Metric normals: how high a metric has gone in most hours of the day.
+
+Revision ID: 0050_norm_usual_anywhere
+Revises: 0049_alert_on_a_rise
+Create Date: 2026-10-30
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0050_norm_usual_anywhere"
+down_revision: Union[str, None] = "0049_alert_on_a_rise"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("metric_norms", sa.Column("usual_anywhere", sa.Float, nullable=False, server_default="0"))
+
+
+def downgrade() -> None:
+    op.drop_column("metric_norms", "usual_anywhere")

steward/models/norms.py

@@ -24,6 +24,8 @@ class MetricNorm(Base):
     # peaked: the 90th percentile, across past days, of each hour's p95 and max.
     usual_hold: Mapped[float] = mapped_column(Float, nullable=False)
     usual_peak: Mapped[float] = mapped_column(Float, nullable=False)
+    # How high the metric has gone in most hours of the day, whatever the slot.
+    usual_anywhere: Mapped[float] = mapped_column(Float, nullable=False, default=0.0, server_default="0")
     days_seen: Mapped[int] = mapped_column(Integer, nullable=False)
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
 

steward/timeline/record.py

@@ -73,12 +73,12 @@ _MONITOR_SOURCES = ("icmp", "tcp", "dns", "http")
 async def hosts_of_rule(session, source: str, resource: str) -> list[str]:
     """The host an alert rule's resource is on, by name, when it has one: an
     agent series ("<host>" or "<host>:<part>"), a container ("<host>/<name>")
-    or a monitor tied to a host."""
+    or a monitor tied to a host. A swarm service ("swarm/<service>") has none."""
     from steward.core.alerts import host_of_resource
     if source == "host_agent":
         return [host_of_resource(source, resource) or resource.split(":", 1)[0]]
     if source == "docker":
-        return [resource.split("/", 1)[0]]
+        return [] if resource.startswith("swarm/") else [resource.split("/", 1)[0]]
     if source in _MONITOR_SOURCES:
         from steward.models.hosts import Host
         from steward.models.monitors import Monitor

tests/integration/test_anomalies.py

@@ -41,11 +41,13 @@ def src():
     norms.LEARNED.pop((name, METRIC), None)
 
 
-def _norms(src: str, res: str, hold: float = 10.0, peak: float = 20.0):
-    """A learned day normal for every hour: holds `hold`, peaks at `peak`."""
+def _norms(src: str, res: str, hold: float = 10.0, peak: float = 20.0, anywhere: float = 0.0):
+    """A learned day normal for every hour: holds `hold`, peaks at `peak`, and
+    reaches `anywhere` in most hours of the day."""
     from steward.models.norms import MetricNorm
     return [MetricNorm(source_module=src, resource_name=res, metric_name=METRIC, slot=h,
-                       kind="day", usual_hold=hold, usual_peak=peak, days_seen=7, updated_at=NOW)
+                       kind="day", usual_hold=hold, usual_peak=peak, usual_anywhere=anywhere,
+                       days_seen=7, updated_at=NOW)
             for h in range(24)]
 
 
@@ -104,7 +106,8 @@ def test_a_level_within_its_normal_is_no_anomaly(app, src):
     assert rows == []
 
 
-def _hour_of(app, src: str, res: str, *, level: float, for_minutes: int, then: float = 10.0):
+def _hour_of(app, src: str, res: str, *, level: float, for_minutes: int, then: float = 10.0,
+             anywhere: float = 0.0):
     """A detection pass after an hour reading `then`, with its last
     `for_minutes` at `level`, against the normal that holds 10 and peaks at 20."""
     from sqlalchemy import select
@@ -115,7 +118,7 @@ def _hour_of(app, src: str, res: str, *, level: float, for_minutes: int, then: f
     async def _go():
         async with app.db_sessionmaker() as s:
             async with s.begin():
-                s.add_all(_norms(src, res))
+                s.add_all(_norms(src, res, anywhere=anywhere))
                 s.add_all([PluginMetric(source_module=src, resource_name=res, metric_name=METRIC,
                                         value=level if m < for_minutes else then,
                                         recorded_at=NOW - timedelta(minutes=m)) for m in range(60)])
@@ -137,6 +140,15 @@ def test_a_burst_past_its_usual_peak_is_an_anomaly(app, src):
     assert a.started_at == NOW - timedelta(minutes=15)
 
 
+def test_a_burst_past_its_slots_peak_but_within_what_most_hours_reach_is_no_anomaly(app, src):
+    assert _hour_of(app, src, _res("anywhere"), level=30.0, for_minutes=15, anywhere=40.0) == []
+
+
+def test_a_level_within_what_most_hours_reach_is_an_anomaly_once_held(app, src):
+    (a,) = _hour_of(app, src, _res("anyheld"), level=30.0, for_minutes=45, anywhere=40.0)
+    assert a.peak_level == 30.0
+
+
 def test_a_level_within_its_usual_peak_is_an_anomaly_once_held(app, src):
     # The system running at its peaked rate: never past the peak, but not a burst.
     (a,) = _hour_of(app, src, _res("held"), level=18.0, for_minutes=45)
@@ -374,6 +386,42 @@ def test_a_metric_learns_only_from_its_since(app, src):
     assert held_back is None
 
 
+def test_bursts_at_any_hour_are_told_from_a_job_with_a_time_of_its_own(app, src):
+    """Fourteen days of a metric at 0. One burst to 50 in each hour of the day
+    leaves every slot's usual peak at 0 and is usual anywhere. The same burst
+    every night at 02:00 is usual at 02:00, and still stands out at 14:00."""
+    from steward.core.norms import norms_at, rebuild_norms
+    from steward.models.metrics import PluginMetricHourly
+    sporadic, nightly = _res("sporadic"), _res("nightly")
+    midnight = NOW.replace(hour=0)
+
+    def hours(res, burst):
+        return [PluginMetricHourly(source_module=src, resource_name=res, metric_name=METRIC,
+                                   bucket=midnight - timedelta(days=d) + timedelta(hours=h),
+                                   value_avg=0.0, value_p95=0.0, sample_count=60,
+                                   value_max=50.0 if burst(d, h) else 0.0)
+                for d in range(1, 15) for h in range(24)]
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add_all(hours(sporadic, lambda d, h: h % 14 == d - 1))
+                s.add_all(hours(nightly, lambda d, h: h == 2))
+            async with s.begin():
+                await rebuild_norms(s, NOW)
+            at = {}
+            for hour in (2, 14):
+                at[hour] = await norms_at(s, [(src, sporadic, METRIC), (src, nightly, METRIC)],
+                                          NOW.replace(hour=hour))
+            return at
+
+    at = asyncio.run(_go())
+    assert at[14][(src, sporadic, METRIC)].usual_peak == 0.0
+    assert at[14][(src, sporadic, METRIC)].usual_height == 50.0
+    assert at[2][(src, nightly, METRIC)].usual_height == 50.0
+    assert at[14][(src, nightly, METRIC)].usual_height == 0.0
+
+
 def test_a_metric_learned_without_its_parts_ignores_them(app, src):
     """A host's CPU learns a normal and each of its cores doesn't (#5742):
     with parts=False, "<host>:core0" is neither learned nor judged, and the

tests/integration/test_docker_rekey.py

@@ -0,0 +1,153 @@
+"""Integration: a swarm service's metrics stay under one key across its tasks,
+and history kept under a task's own name moves onto that key in the background.
+Requires STEWARD_DATABASE_URL."""
+from __future__ import annotations
+
+import asyncio
+import importlib
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+NOW = datetime(2026, 10, 12, 10, 0, tzinfo=timezone.utc)
+TASK = "a" * 25
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+def _docker(sub):
+    """The app's loaded docker plugin module (see test_docker_recent_usage)."""
+    from steward.core.plugin_manager import get_loaded_plugin_modules
+    loaded = get_loaded_plugin_modules().get("docker")
+    base = loaded.__name__ if loaded is not None else "plugins.docker"
+    return importlib.import_module(f"{base}.{sub}")
+
+
+def _hour(res, metric, bucket, *, avg, top, count=60):
+    from steward.models.metrics import PluginMetricHourly
+    return PluginMetricHourly(source_module="docker", resource_name=res, metric_name=metric, bucket=bucket,
+                              value_avg=avg, value_max=top, value_p95=top, sample_count=count)
+
+
+def _anomaly(res, metric, at, *, peak):
+    from steward.models.norms import MetricAnomaly
+    return MetricAnomaly(source_module="docker", resource_name=res, metric_name=metric, started_at=at,
+                         last_started_at=at, ended_at=at + timedelta(minutes=30), occurrences=1,
+                         peak_level=peak, peak_at=at, usual_hold=1.0, usual_peak=2.0)
+
+
+def test_a_replica_or_a_global_task_keys_by_its_service_and_a_plain_container_by_its_host(app):
+    service = _docker("service")
+    assert service.container_resource("nas", f"stack_web.1.{TASK}") == "swarm/stack_web"
+    assert service.container_resource("nas", f"stack_agent.{'b' * 25}.{TASK}") == "swarm/stack_agent"
+    assert service.container_resource("nas", "plex") == "nas/plex"
+    assert service.container_resource("nas", "my.app.1") == "nas/my.app.1"
+    assert service.traffic_resource("nas", f"stack_web.1.{TASK}") == "nas/stack_web"
+    assert service.traffic_resource("nas", "plex") == "nas/plex"
+
+
+def test_a_tasks_history_moves_onto_its_service_a_batch_at_a_time(app, monkeypatch):
+    from sqlalchemy import select
+    from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest
+    from steward.models.norms import MetricAnomaly, MetricAnomalySpan
+    rekey = _docker("rekey")
+    monkeypatch.setattr(rekey, "BATCH", 2)
+
+    name = f"it{uuid.uuid4().hex[:8]}_web"
+    old, new, traffic = f"nas/{name}.1.{TASK}", f"swarm/{name}", "net_rx_bps"
+    shared, alone = NOW - timedelta(hours=3), NOW - timedelta(hours=2)
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add_all([PluginMetric(source_module="docker", resource_name=old, metric_name="cpu_pct",
+                                        value=float(m), recorded_at=NOW - timedelta(minutes=m)) for m in range(3)])
+                s.add(PluginMetric(source_module="docker", resource_name=old, metric_name=traffic,
+                                   value=5.0, recorded_at=NOW))
+                s.add_all([_hour(old, "cpu_pct", shared, avg=10.0, top=30.0),
+                           _hour(new, "cpu_pct", shared, avg=20.0, top=25.0),
+                           _hour(old, "cpu_pct", alone, avg=7.0, top=8.0)])
+                s.add(PluginMetricLatest(source_module="docker", resource_name=old, metric_name="cpu_pct",
+                                         value=1.0, recorded_at=NOW))
+                older = _anomaly(old, "cpu_pct", NOW - timedelta(days=2), peak=90.0)
+                newer = _anomaly(new, "cpu_pct", NOW - timedelta(days=1), peak=40.0)
+                s.add_all([older, newer])
+                await s.flush()
+                s.add(MetricAnomalySpan(anomaly_id=older.id, started_at=older.started_at,
+                                        ended_at=older.ended_at, peak_level=90.0))
+            async with s.begin():
+                found = await rekey.old_keys(s)
+            async with s.begin():
+                first = await rekey.move_key(s, old)
+            async with s.begin():
+                second = await rekey.move_key(s, old)
+
+            raw = (await s.execute(select(PluginMetric.resource_name, PluginMetric.metric_name).where(
+                PluginMetric.resource_name.in_([old, new])))).all()
+            hours = {(h.resource_name, h.bucket): h for h in (await s.execute(select(PluginMetricHourly).where(
+                PluginMetricHourly.resource_name.in_([old, new])))).scalars()}
+            latest = (await s.execute(select(PluginMetricLatest).where(
+                PluginMetricLatest.resource_name == old))).scalars().all()
+            rows = (await s.execute(select(MetricAnomaly).where(
+                MetricAnomaly.resource_name.in_([old, new])))).scalars().all()
+            spans = (await s.execute(select(MetricAnomalySpan).where(
+                MetricAnomalySpan.anomaly_id.in_([r.id for r in rows])))).scalars().all()
+            left = await rekey.old_keys(s)
+            return found, first, second, raw, hours, latest, rows, spans, left
+
+    found, first, second, raw, hours, latest, rows, spans, left = asyncio.run(_go())
+
+    assert old in found and old not in left
+    assert (first, second) == (False, True)
+    assert sorted(raw) == sorted([(new, "cpu_pct")] * 3 + [(old, traffic)])
+    assert set(hours) == {(new, shared), (new, alone)}
+    combined = hours[(new, shared)]
+    assert combined.value_avg == 15.0 and combined.value_max == 30.0 and combined.sample_count == 120
+    assert hours[(new, alone)].value_avg == 7.0
+    assert latest == []
+    (row,) = rows
+    assert row.resource_name == new and row.occurrences == 2 and row.peak_level == 90.0
+    assert row.started_at == NOW - timedelta(days=2)
+    assert [sp.anomaly_id for sp in spans] == [row.id]
+
+
+def test_an_alert_rule_on_a_task_follows_it_to_its_service(app):
+    from sqlalchemy import select
+    from steward.models.alerts import AlertOperator, AlertRule
+    from steward.models.users import User, UserRole
+    rekey = _docker("rekey")
+
+    tag = uuid.uuid4().hex[:8]
+    old, new = f"nas/it{tag}_db.1.{TASK}", f"swarm/it{tag}_db"
+    uid = str(uuid.uuid4())
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add(User(id=uid, username=f"u{tag}", email=f"{tag}@example.com",
+                           password_hash="x", role=UserRole.viewer, is_active=True))
+                await s.flush()
+                s.add(AlertRule(name=f"busy {tag}", source_module="docker", resource_name=old,
+                                metric_name="cpu_pct", operator=AlertOperator(">"), threshold=90,
+                                created_by=uid))
+            async with s.begin():
+                found = await rekey.old_keys(s)
+            async with s.begin():
+                done = await rekey.move_key(s, old)
+            names = (await s.execute(select(AlertRule.resource_name).where(
+                AlertRule.name == f"busy {tag}"))).scalars().all()
+            return found, done, names
+
+    found, done, names = asyncio.run(_go())
+    assert old in found and done
+    assert names == [new]

tests/integration/test_docker_service_page.py

@@ -59,7 +59,7 @@ def test_a_service_page_reads_every_task(app):
 
     def sample(cname, mins, cpu, mem):
         when = now - timedelta(minutes=mins)
-        return [PluginMetric(source_module="docker", resource_name=f"dock-{tag}/{cname}",
+        return [PluginMetric(source_module="docker", resource_name=f"swarm/{cname.split('.')[0]}",
                              metric_name=metric, value=value, recorded_at=when)
                 for metric, value in (("cpu_pct", cpu), ("mem_used_bytes", mem))]