feat(docker): per-container enrichment — health, restarts, exit code, I/O, grouping

First slice of milestone 77 (Docker monitoring depth). Surfaces real per-container
stats beyond basic state, all read-only on the existing push model.

- agent (→1.4.0): collect_docker now inspects each container (health, restart
  count, exit code, OOM) and reads net + block I/O from the stats payload; pulls
  compose project + swarm service/task/node from container labels. Per-container
  inspect+stats calls run over a small bounded ThreadPool so the ~1s-per-stats
  blocking doesn't stretch the sample on a busy host.
- schema (docker_003): additive columns on docker_containers — health, exit_code,
  oom_killed, compose_project, service_name, task_id, node_id, and BigInteger
  net/blk byte counters.
- ingest: persists the enrichment + restart_count (.get keeps older agents working).
- ui: Docker page rows now show health badge, uptime ("up 3d 4h"), restart count,
  exit code (+OOM) for stopped containers, and compose/service grouping label.
- tests: agent helpers (grouping, inspect fields, net/IO sum) + collect_docker
  assembly incl. inspect; integration asserts enrichment round-trips.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC

plugins/host_agent/agent.py

@@ -17,9 +17,10 @@ import time
 import urllib.error
 import urllib.request
 from collections import deque
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.3.0"
+AGENT_VERSION = "1.4.0"
 
 # Default path to the local Docker Engine socket. Overridable via the
 # `docker_socket` config key; collection is silently skipped if it's absent or
@@ -515,6 +516,117 @@ def _docker_ports(ports: list) -> list:
     return out
 
 
+def _docker_grouping(labels: dict) -> dict:
+    """Pull compose project + swarm service/task/node out of container labels.
+
+    These are set by Docker itself (compose / swarm), so reading them off the
+    container costs nothing extra and lets Steward group + place containers
+    without a manager-node query.
+    """
+    labels = labels or {}
+    return {
+        "compose_project": labels.get("com.docker.compose.project"),
+        "service_name": labels.get("com.docker.swarm.service.name"),
+        "task_id": labels.get("com.docker.swarm.task.id"),
+        "node_id": labels.get("com.docker.swarm.node.id"),
+    }
+
+
+def _docker_inspect_fields(insp: dict):
+    """Return (health, restart_count, exit_code, oom_killed) from an inspect.
+
+    health is None for containers without a HEALTHCHECK (no State.Health).
+    """
+    state = (insp or {}).get("State") or {}
+    health = (state.get("Health") or {}).get("Status")  # healthy|unhealthy|starting
+    restart_count = insp.get("RestartCount", 0) if isinstance(insp, dict) else 0
+    exit_code = state.get("ExitCode")
+    oom_killed = bool(state.get("OOMKilled", False))
+    return health, restart_count, exit_code, oom_killed
+
+
+def _docker_net_io(stats: dict):
+    """Return cumulative (net_rx, net_tx, blk_read, blk_write) bytes from stats.
+
+    Counters are cumulative since container start (rates can be derived later);
+    block I/O comes from the cgroup io_service_bytes list, absent on some setups.
+    """
+    net_rx = net_tx = blk_read = blk_write = 0
+    for iface in (stats.get("networks") or {}).values():
+        net_rx += iface.get("rx_bytes", 0) or 0
+        net_tx += iface.get("tx_bytes", 0) or 0
+    for e in ((stats.get("blkio_stats") or {}).get("io_service_bytes_recursive") or []):
+        op = (e.get("op") or "").lower()
+        if op == "read":
+            blk_read += e.get("value", 0) or 0
+        elif op == "write":
+            blk_write += e.get("value", 0) or 0
+    return net_rx, net_tx, blk_read, blk_write
+
+
+def _collect_one_container(socket_path: str, c: dict) -> dict:
+    """Build one container's enriched record (runs in a worker thread).
+
+    Does a per-container inspect (health/restart/exit/oom — only available there)
+    plus a stats read for running containers (cpu/mem/net/io). All best-effort:
+    a failed call just leaves the affected fields at their defaults.
+    """
+    cid = c.get("Id", "") or ""
+    names = c.get("Names") or []
+    name = names[0].lstrip("/") if names else cid[:12]
+    state = c.get("State", "unknown")
+
+    # Inspect every container (incl. stopped) — exit codes + restart counts only
+    # live here, and stopped containers are exactly where exit_code matters.
+    health = exit_code = None
+    restart_count = 0
+    oom_killed = False
+    try:
+        insp = _docker_request(socket_path, f"/containers/{cid}/json")
+        health, restart_count, exit_code, oom_killed = _docker_inspect_fields(insp)
+    except (OSError, ValueError):
+        pass
+
+    cpu_pct = mem_usage = mem_limit = mem_pct = None
+    net_rx = net_tx = blk_read = blk_write = None
+    if state == "running":
+        try:
+            stats = _docker_request(socket_path, f"/containers/{cid}/stats?stream=false")
+            cpu_pct = _docker_cpu_pct(stats)
+            mem_usage, mem_limit, mem_pct = _docker_mem(stats)
+            net_rx, net_tx, blk_read, blk_write = _docker_net_io(stats)
+        except (OSError, ValueError):
+            pass
+
+    created = c.get("Created")
+    started_at = (
+        datetime.fromtimestamp(created, tz=timezone.utc).isoformat()
+        if isinstance(created, (int, float)) else None
+    )
+    record = {
+        "name": name,
+        "container_id": cid[:12],
+        "image": c.get("Image", ""),
+        "status": state,
+        "cpu_pct": cpu_pct,
+        "mem_usage_bytes": mem_usage,
+        "mem_limit_bytes": mem_limit,
+        "mem_pct": mem_pct,
+        "ports": _docker_ports(c.get("Ports", [])),
+        "started_at": started_at,
+        "health": health,
+        "restart_count": restart_count,
+        "exit_code": exit_code,
+        "oom_killed": oom_killed,
+        "net_rx_bytes": net_rx,
+        "net_tx_bytes": net_tx,
+        "blk_read_bytes": blk_read,
+        "blk_write_bytes": blk_write,
+    }
+    record.update(_docker_grouping(c.get("Labels")))
+    return record
+
+
 def collect_docker(socket_path: str) -> list:
     """Per-container state from the local Docker socket, or [] if unavailable.
 
@@ -526,44 +638,16 @@ def collect_docker(socket_path: str) -> list:
         containers = _docker_request(socket_path, "/containers/json?all=true")
     except (OSError, ValueError):
         return []
-    if not isinstance(containers, list):
+    if not isinstance(containers, list) or not containers:
         return []
 
-    out = []
-    for c in containers:
-        cid = c.get("Id", "") or ""
-        names = c.get("Names") or []
-        name = names[0].lstrip("/") if names else cid[:12]
-        state = c.get("State", "unknown")
-
-        cpu_pct = mem_usage = mem_limit = mem_pct = None
-        if state == "running":
-            try:
-                stats = _docker_request(
-                    socket_path, f"/containers/{cid}/stats?stream=false")
-                cpu_pct = _docker_cpu_pct(stats)
-                mem_usage, mem_limit, mem_pct = _docker_mem(stats)
-            except (OSError, ValueError):
-                pass  # stats unavailable for this container — leave gauges None
-
-        created = c.get("Created")
-        started_at = (
-            datetime.fromtimestamp(created, tz=timezone.utc).isoformat()
-            if isinstance(created, (int, float)) else None
-        )
-        out.append({
-            "name": name,
-            "container_id": cid[:12],
-            "image": c.get("Image", ""),
-            "status": state,
-            "cpu_pct": cpu_pct,
-            "mem_usage_bytes": mem_usage,
-            "mem_limit_bytes": mem_limit,
-            "mem_pct": mem_pct,
-            "ports": _docker_ports(c.get("Ports", [])),
-            "started_at": started_at,
-        })
-    return out
+    # Each container needs an inspect (+ a stats read if running), and the stats
+    # call blocks ~1s while the daemon computes the cpu delta. Done serially that
+    # stretches the whole sample on a busy host, so fan the per-container work out
+    # over a small bounded thread pool (I/O-bound → threads are enough).
+    workers = min(8, len(containers))
+    with ThreadPoolExecutor(max_workers=workers) as ex:
+        return list(ex.map(lambda c: _collect_one_container(socket_path, c), containers))
 
 
 # ─── ring buffer ─────────────────────────────────────────────────────────────