feat(docker): the Docker page leads with a verdict, lists services and containers as filterable tables, and each service has its own page with logs across its tasks (#5298–#5302)

- #5298 "Docker overview: verdict line; only the problem tiles that aren't zero"
- #5299 "Docker overview: swarm services as a filterable, sortable table with a Stack column"
- #5300 "Docker overview: failures listed once, each with a link to the failed task's logs"
- #5301 "Docker overview: standalone containers in one table with a Host column"
- #5302 "Service page: replicas, image, task history, resource history and logs across all its tasks"

overview.py holds the rows, filters and sort as plain functions. The filter
bar sits outside the polled fragment and the URL keeps its state. A service's
page reads every task container Steward still holds anything for, the live
ones and removed ones known by their events, so a failed task's logs are
reachable after Docker has removed it.

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

plugins/docker/service.py

@@ -73,7 +73,8 @@ def failures_24h(events, swarm_services, host_name: dict) -> list[dict]:
     (an agent older than 1.8.0) falls back to its tasks' deaths.
 
     Each row: {kind: "container"|"service", name, host_id, host, count, at,
-    detail}.
+    detail, task_id}; task_id is the failed swarm task, so its logs can be
+    found (None for a container, or a service without task records).
     """
     by_prefix = {s["name"] + ".": s for s in swarm_services}
     rows: dict[tuple, dict] = {}
@@ -97,7 +98,9 @@ def failures_24h(events, swarm_services, host_name: dict) -> list[dict]:
         if row is None:
             rows[key] = {"kind": kind, "name": name, "host_id": e.host_id,
                          "host": host_name.get(e.host_id, e.host_id), "count": 1, "at": e.at,
-                         "detail": "OOM-killed" if e.event == "oom" else e.detail}
+                         "detail": "OOM-killed" if e.event == "oom" else e.detail,
+                         "task_id": (task_of(svc["name"], e.container_name) or (None, None))[1]
+                         if svc else None}
         else:
             row["count"] += 1
     from .swarm_view import task_time
@@ -107,7 +110,8 @@ def failures_24h(events, swarm_services, host_name: dict) -> list[dict]:
             rows[("service", s["name"])] = {
                 "kind": "service", "name": s["name"], "host_id": None, "host": err.get("node"),
                 "count": s["failures_24h"], "at": task_time(err) if err else None,
-                "detail": err.get("error") or err.get("message") or err.get("state")}
+                "detail": err.get("error") or err.get("message") or err.get("state"),
+                "task_id": err.get("task_id")}
     epoch = datetime.min.replace(tzinfo=timezone.utc)
     return sorted(rows.values(), key=lambda r: r["at"] or epoch, reverse=True)
 
@@ -470,6 +474,92 @@ async def retained_names(db, want: str, *, exact: bool, host_id: str | None = No
     return rows[:limit]
 
 
+# A swarm task's container is named "<service>.<slot>.<task id>": slot is the
+# replica number, or the node id for a global service; task ids are 25 chars.
+_TASK_SUFFIX = re.compile(r"^(?P<slot>[A-Za-z0-9]+)\.(?P<task>[a-z0-9]{25})$")
+
+
+def task_of(service: str, container_name: str) -> tuple[str, str] | None:
+    """(slot, task id) when `container_name` is one of `service`'s task
+    containers, else None — so "arr" does not claim "arr_radarr.1.…"."""
+    if not container_name.startswith(service + "."):
+        return None
+    m = _TASK_SUFFIX.match(container_name[len(service) + 1:])
+    return (m["slot"], m["task"]) if m else None
+
+
+async def service_containers(db, service: str) -> list[tuple[str, str]]:
+    """(host_id, container_name) of every task container of a swarm service
+    that Steward still holds anything for: the live ones, and the ones Docker
+    removed that left events (each task container has a start event). The
+    page's logs and history read across all of them (#5302)."""
+    pattern = escape_like(service + ".") + "%"
+    found: set[tuple[str, str]] = set()
+    for model, col in ((DockerContainer, DockerContainer.name),
+                       (DockerEvent, DockerEvent.container_name)):
+        rows = (await db.execute(
+            select(model.host_id, col).where(col.like(pattern, escape="\\")).distinct())).all()
+        found.update((h, n) for h, n in rows if task_of(service, n))
+    return sorted(found)
+
+
+async def query_service_logs(db, containers, stream: str, query: str, limit: int,
+                             task: str | None = None) -> list:
+    """The newest retained lines across a service's task containers, newest
+    first, each with its container name so the page can say which task wrote
+    it. `task` keeps one task's lines (a failure's "Logs →")."""
+    from sqlalchemy import tuple_
+    pairs = [(h, n) for h, n in containers if task is None or n.endswith("." + task)]
+    if not pairs:
+        return []
+    stmt = (select(DockerLog.ts, DockerLog.stream, DockerLog.line, DockerLog.container_name)
+            .where(tuple_(DockerLog.host_id, DockerLog.container_name).in_(pairs)))
+    if stream in ("stdout", "stderr"):
+        stmt = stmt.where(DockerLog.stream == stream)
+    if query:
+        stmt = stmt.where(DockerLog.line.ilike(f"%{escape_like(query)}%", escape="\\"))
+    stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit)
+    return (await db.execute(stmt)).all()
+
+
+def sum_series(samples, since: datetime, until: datetime, points: int = 60) -> list:
+    """[(bucket start, cpu %, memory bytes)] summed across containers: each
+    container's samples are averaged within a bucket, then the containers are
+    added, so two replicas read as their combined load. `samples` are
+    (container_name, at, cpu_pct, mem_bytes); buckets with no sample are left
+    out rather than read as zero."""
+    width = max((until - since).total_seconds() / points, 60.0)
+    acc: dict[int, dict[str, list]] = {}
+    for name, at, cpu, mem in samples:
+        i = int((at - since).total_seconds() // width)
+        if i < 0:
+            continue
+        per = acc.setdefault(i, {}).setdefault(name, [0.0, 0, 0])
+        per[0] += cpu or 0.0
+        per[1] += mem or 0
+        per[2] += 1
+    out = []
+    for i in sorted(acc):
+        cpu = sum(c / n for c, _, n in acc[i].values())
+        mem = sum(m / n for _, m, n in acc[i].values())
+        out.append((since + timedelta(seconds=i * width), cpu, int(mem)))
+    return out
+
+
+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:
+        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()
+    return sum_series(rows, since, datetime.now(timezone.utc))
+
+
 async def unhealthy_since(db, keys) -> dict[tuple[str, str], datetime]:
     """{(host_id, container_name): when it last turned unhealthy} for the given
     containers, from their health_change events; absent when none is retained."""