feat(docker): per-host collection via the host agent; drop central scrape

Docker collection moves off the central single-socket scrape onto the host
agent, giving Docker a real per-host dimension. The Steward host now reports
its own containers like any other host, and same-named containers on different
hosts no longer collide.

- agent: stdlib UDS Docker client (AF_UNIX HTTP/1.1, Connection: close,
  chunked-aware), collect_docker() ports the cpu%/mem math; sample["docker"]
  added best-effort (silent-skip on absent/unreadable socket). AGENT_VERSION
  1.2.0 → 1.3.0; optional docker_socket config key.
- ingest: host_agent ingest hands per-host container snapshots to the docker
  plugin via a new "docker.persist_host_samples" capability (no hard import,
  no-op when docker disabled), inside a SAVEPOINT so a docker failure never
  sinks the host metrics. Resource names are host-scoped ("<host>/<name>").
- schema: docker_containers re-keyed (host_id, name); docker_metrics gains
  host_id; docker_002 migration DROP+recreates (dev-only, rule 122).
- ui: Docker page + widgets grouped by host with host links; new per-host
  Docker panel embedded on the Hosts hub (gated on docker enabled via a new
  enabled_plugins template context). Replaces the SQLite-only strftime
  bucketing with DB-agnostic Python bucketing.
- provisioning: install/provision playbooks add steward-agent to the docker
  group (best-effort) so the agent can read the socket.
- removed central scrape: docker scheduler.py + scraper.py deleted; plugin.yaml
  socket_path/scrape_interval_seconds/include_stopped dropped (plugin 2.0.0).
- tests: agent docker collector units (math, chunked decode, silent-skip,
  sample shape, config) + integration (host-scoped schema + persistence).

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

@@ -19,7 +19,12 @@ import urllib.request
 from collections import deque
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.2.0"
+AGENT_VERSION = "1.3.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
+# unreadable, so this never needs setting on a Docker-less host (zero-config).
+DEFAULT_DOCKER_SOCKET = "/var/run/docker.sock"
 
 
 class ConfigError(Exception):
@@ -62,6 +67,7 @@ def read_config(path: str) -> dict:
         raise ConfigError(f"{path}: missing required key(s): {', '.join(missing)}")
 
     cfg.setdefault("interval_seconds", 30)
+    cfg.setdefault("docker_socket", DEFAULT_DOCKER_SOCKET)
     return cfg
 
 
@@ -386,6 +392,180 @@ def _rates(cur: dict, prev: dict, dt: float) -> dict:
     return out
 
 
+# ─── docker ──────────────────────────────────────────────────────────────────
+#
+# Per-host container collection over the local Docker socket. stdlib-only: a
+# minimal HTTP/1.1 client over an AF_UNIX socket. We send `Connection: close`
+# so the daemon closes the socket at end of response and we can read to EOF;
+# Docker still frames the body with Transfer-Encoding: chunked, so we de-chunk
+# when that header is present rather than assume Content-Length.
+
+DOCKER_API_TIMEOUT = 5.0
+
+
+def _dechunk(body: bytes) -> bytes:
+    """Decode an HTTP/1.1 chunked-transfer body into the raw payload."""
+    out = bytearray()
+    while body:
+        line, sep, rest = body.partition(b"\r\n")
+        if not sep:
+            break
+        try:
+            size = int(line.split(b";", 1)[0], 16)  # ignore chunk extensions
+        except ValueError:
+            break
+        if size == 0:
+            break
+        out += rest[:size]
+        body = rest[size + 2:]  # skip the chunk data and its trailing CRLF
+    return bytes(out)
+
+
+def _docker_request(socket_path: str, path: str, timeout: float = DOCKER_API_TIMEOUT):
+    """GET `path` from the Docker Engine API over a Unix socket; return JSON.
+
+    Raises OSError on any socket/transport problem (absent socket, permission
+    denied, non-200) and ValueError on a non-JSON body — both are caught by the
+    caller and treated as "no docker here", so collection degrades silently.
+    """
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    sock.settimeout(timeout)
+    try:
+        sock.connect(socket_path)
+        req = (
+            "GET " + path + " HTTP/1.1\r\n"
+            "Host: docker\r\n"
+            "Accept: application/json\r\n"
+            "Connection: close\r\n"
+            "\r\n"
+        )
+        sock.sendall(req.encode("ascii"))
+        chunks = []
+        while True:
+            buf = sock.recv(65536)
+            if not buf:
+                break
+            chunks.append(buf)
+    finally:
+        sock.close()
+
+    raw = b"".join(chunks)
+    head, _, body = raw.partition(b"\r\n\r\n")
+    header_text = head.decode("latin-1")
+    status_line = header_text.split("\r\n", 1)[0]
+    parts = status_line.split(None, 2)  # "HTTP/1.1 200 OK"
+    status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0
+    if status != 200:
+        raise OSError(f"docker API {path} returned {status}")
+    if "transfer-encoding: chunked" in header_text.lower():
+        body = _dechunk(body)
+    return json.loads(body.decode("utf-8"))
+
+
+def _docker_cpu_pct(stats: dict) -> float:
+    """CPU % from a Docker stats snapshot (delta of container vs system CPU).
+
+    Ported verbatim from the old central scraper's math; correct as long as both
+    cpu_stats and precpu_stats are populated (we request without `one-shot` so
+    the daemon fills precpu over one cycle).
+    """
+    try:
+        cpu = stats.get("cpu_stats", {})
+        precpu = stats.get("precpu_stats", {})
+        cpu_delta = (
+            cpu["cpu_usage"]["total_usage"] - precpu["cpu_usage"]["total_usage"]
+        )
+        sys_delta = cpu.get("system_cpu_usage", 0) - precpu.get("system_cpu_usage", 0)
+        num_cpus = cpu.get("online_cpus") or len(
+            cpu["cpu_usage"].get("percpu_usage") or [None]
+        )
+        if sys_delta <= 0 or cpu_delta < 0:
+            return 0.0
+        return round((cpu_delta / sys_delta) * num_cpus * 100.0, 2)
+    except (KeyError, TypeError, ZeroDivisionError):
+        return 0.0
+
+
+def _docker_mem(stats: dict):
+    """Return (usage_bytes, limit_bytes, mem_pct); working set excludes cache."""
+    try:
+        mem = stats.get("memory_stats", {})
+        usage = mem.get("usage", 0)
+        limit = mem.get("limit", 0)
+        # Subtract page cache for working set (cgroup v1 "cache", v2 "inactive_file").
+        cache = mem.get("stats", {}).get("cache", 0) or \
+                mem.get("stats", {}).get("inactive_file", 0)
+        actual = max(0, usage - cache)
+        pct = round(actual / limit * 100.0, 2) if limit > 0 else 0.0
+        return actual, limit, pct
+    except (KeyError, TypeError):
+        return 0, 0, 0.0
+
+
+def _docker_ports(ports: list) -> list:
+    """Normalise Docker port bindings to a compact published-ports list."""
+    out = []
+    for p in (ports or []):
+        if p.get("PublicPort"):
+            out.append({
+                "host_port": p["PublicPort"],
+                "container_port": p.get("PrivatePort"),
+                "protocol": p.get("Type", "tcp"),
+            })
+    return out
+
+
+def collect_docker(socket_path: str) -> list:
+    """Per-container state from the local Docker socket, or [] if unavailable.
+
+    Absent socket / permission denied / non-Docker endpoint all yield [] — the
+    agent silently reports no containers rather than erroring, so a Docker-less
+    host needs no configuration.
+    """
+    try:
+        containers = _docker_request(socket_path, "/containers/json?all=true")
+    except (OSError, ValueError):
+        return []
+    if not isinstance(containers, list):
+        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
+
+
 # ─── ring buffer ─────────────────────────────────────────────────────────────
 
 
@@ -408,11 +588,14 @@ class RingBuffer:
 # ─── payload ─────────────────────────────────────────────────────────────────
 
 
-def build_sample(mounts: list[str], state: dict) -> dict:
+def build_sample(mounts: list[str], state: dict,
+                 docker_socket: str = DEFAULT_DOCKER_SOCKET) -> dict:
     """Collect one full sample. Partial samples allowed if a collector fails.
 
     `state` carries the previous network/disk counters + monotonic timestamp so
     throughput rates can be derived from deltas; it is mutated in place.
+    `docker_socket` is probed best-effort — the `docker` key is omitted entirely
+    when no containers are found, so non-Docker hosts add nothing to the payload.
     """
     sample: dict = {"ts": datetime.now(timezone.utc).isoformat()}
     try:
@@ -466,6 +649,14 @@ def build_sample(mounts: list[str], state: dict) -> dict:
         for n, (rd, wr) in _rates(disk_raw, state.get("disk", {}), dt).items()
     ]
     state["net"], state["disk"], state["mono"] = net_raw, disk_raw, now_mono
+
+    try:
+        docker = collect_docker(docker_socket)
+    except Exception:
+        docker = []
+    if docker:
+        sample["docker"] = docker
+
     return sample
 
 
@@ -547,6 +738,7 @@ def main_loop(conf_path: str) -> int:
     metadata = collect_metadata()
     hostname = cfg.get("hostname") or socket.gethostname()
     mounts = cfg.get("mounts") or default_mounts()
+    docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
     buffer = RingBuffer(maxlen=20)
     backoff = 0
     # Carries previous net/disk counters + monotonic ts for rate computation.
@@ -560,13 +752,14 @@ def main_loop(conf_path: str) -> int:
             try:
                 cfg = read_config(conf_path)
                 mounts = cfg.get("mounts") or default_mounts()
+                docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
                 metadata = collect_metadata()  # refresh host_ip/distro on reload
                 _log("INFO", "config reloaded")
             except ConfigError as e:
                 _log("ERROR", f"reload failed: {e}")
             _reload_requested = False
 
-        sample = build_sample(mounts, rate_state)
+        sample = build_sample(mounts, rate_state, docker_socket)
         buffered = buffer.drain()
         payload = build_payload(
             samples=buffered + [sample],