feat(docker): agent manager-only swarm collector (AGENT_VERSION 1.5.0)

Adds collect_swarm(socket_path) to the host agent. Self-detects a Swarm
manager via /info (Swarm.ControlAvailable) — workers and non-swarm daemons
return None and never touch the manager-only endpoints (one cheap /info call,
no 503s). On a manager it queries /services, /tasks, /nodes and emits
sample["swarm"] = {services, nodes}:

  * services roll desired-vs-running replicas up from the task list (replica
    health isn't on the service object), handle replicated + global mode, strip
    the @sha256 image digest, and carry cross-node task→node placement.
  * nodes normalise role / availability / status + the manager leader flag.

build_sample omits the swarm key entirely off managers, same silent contract
as collect_docker. Unit tests cover manager detection, replica roll-up +
placement (replicated & global), node normalisation, worker silent-skip, and
build_sample wiring.

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

plugins/host_agent/agent.py

@@ -20,7 +20,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.4.0"
+AGENT_VERSION = "1.5.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
@@ -650,6 +650,117 @@ def collect_docker(socket_path: str) -> list:
         return list(ex.map(lambda c: _collect_one_container(socket_path, c), containers))
 
 
+# ─── swarm (manager-only) ────────────────────────────────────────────────────
+
+
+def _swarm_is_manager(info: dict) -> bool:
+    """True only on a reachable Swarm manager.
+
+    `/services`, `/tasks`, `/nodes` are manager-only — a worker (or a non-swarm
+    daemon) returns 503 for them. `Swarm.ControlAvailable` in `/info` is exactly
+    "this node can serve the control-plane API", so we gate on it and skip the
+    topology query everywhere else (no errors, near-zero cost on workers).
+    """
+    return bool((info or {}).get("Swarm", {}).get("ControlAvailable"))
+
+
+def _swarm_services(services: list, tasks: list) -> list:
+    """Roll desired-vs-running replicas per service up from the task list.
+
+    Replica health isn't on the service object — it's derived by counting tasks:
+    `running` = tasks currently in state running; `desired` = the service's
+    configured replica count (replicated) or the count of tasks the scheduler
+    wants running (global, which has no fixed number).
+    """
+    running_by_svc: dict = {}
+    desired_by_svc: dict = {}  # only used for global-mode services
+    placement: dict = {}       # svc_id -> {node_id: running_count}
+    for t in (tasks or []):
+        sid = t.get("ServiceID")
+        if not sid:
+            continue
+        if ((t.get("Status") or {}).get("State") or "").lower() == "running":
+            running_by_svc[sid] = running_by_svc.get(sid, 0) + 1
+            node = t.get("NodeID")
+            if node:
+                placement.setdefault(sid, {})[node] = \
+                    placement.setdefault(sid, {}).get(node, 0) + 1
+        if (t.get("DesiredState") or "").lower() == "running":
+            desired_by_svc[sid] = desired_by_svc.get(sid, 0) + 1
+
+    out = []
+    for s in (services or []):
+        sid = s.get("ID", "") or ""
+        spec = s.get("Spec") or {}
+        name = spec.get("Name") or sid[:12]
+        mode_obj = spec.get("Mode") or {}
+        if "Global" in mode_obj:
+            mode = "global"
+            desired = desired_by_svc.get(sid, 0)
+        else:
+            mode = "replicated"
+            desired = (mode_obj.get("Replicated") or {}).get("Replicas", 0) or 0
+        image = (((spec.get("TaskTemplate") or {}).get("ContainerSpec") or {})
+                 .get("Image") or "")
+        image = image.split("@", 1)[0]  # drop the @sha256:… digest suffix
+        out.append({
+            "service_name": name, "mode": mode,
+            "desired": desired, "running": running_by_svc.get(sid, 0),
+            "image": image,
+            # task→node placement of the running replicas (cross-node — a manager
+            # sees every task, even ones whose containers live on other hosts).
+            "placement": [{"node_id": n, "running": cnt}
+                          for n, cnt in sorted(placement.get(sid, {}).items())],
+        })
+    return out
+
+
+def _swarm_nodes(nodes: list) -> list:
+    """Normalise node role / availability / status (+ manager leader flag)."""
+    out = []
+    for n in (nodes or []):
+        spec = n.get("Spec") or {}
+        desc = n.get("Description") or {}
+        status = n.get("Status") or {}
+        mgr = n.get("ManagerStatus") or {}
+        out.append({
+            "node_id": n.get("ID", "") or "",
+            "hostname": desc.get("Hostname", "") or "",
+            "role": (spec.get("Role") or "worker").lower(),
+            "availability": (spec.get("Availability") or "active").lower(),
+            "status": (status.get("State") or "unknown").lower(),
+            "leader": bool(mgr.get("Leader", False)),
+        })
+    return out
+
+
+def collect_swarm(socket_path: str):
+    """Swarm topology from a manager node, or None on workers / non-swarm hosts.
+
+    Self-detects manager via `/info` (one cheap call); only then queries the
+    manager-only endpoints. Any transport failure degrades to None, same silent
+    contract as collect_docker — a non-swarm host adds nothing to the payload.
+    """
+    try:
+        info = _docker_request(socket_path, "/info")
+    except (OSError, ValueError):
+        return None
+    if not _swarm_is_manager(info):
+        return None
+    try:
+        services = _docker_request(socket_path, "/services")
+        tasks = _docker_request(socket_path, "/tasks")
+        nodes = _docker_request(socket_path, "/nodes")
+    except (OSError, ValueError):
+        return None
+    return {
+        "services": _swarm_services(
+            services if isinstance(services, list) else [],
+            tasks if isinstance(tasks, list) else []),
+        "nodes": _swarm_nodes(nodes if isinstance(nodes, list) else []),
+    }
+
+
 # ─── ring buffer ─────────────────────────────────────────────────────────────
 
 
@@ -741,6 +852,15 @@ def build_sample(mounts: list[str], state: dict,
     if docker:
         sample["docker"] = docker
 
+    # Swarm topology is only emitted by managers (collect_swarm returns None
+    # everywhere else), so the key is absent on workers / non-swarm hosts.
+    try:
+        swarm = collect_swarm(docker_socket)
+    except Exception:
+        swarm = None
+    if swarm is not None:
+        sample["swarm"] = swarm
+
     return sample
 
 

tests/plugins/host_agent/test_agent_docker.py

@@ -192,20 +192,135 @@ def test_collect_docker_skips_stats_for_stopped(monkeypatch):
     assert out[0]["health"] is None
 
 
+# ── swarm (manager-only) ──────────────────────────────────────────────────────
+
+def test_swarm_is_manager():
+    assert a._swarm_is_manager({"Swarm": {"ControlAvailable": True}}) is True
+    # Worker: in a swarm but no control plane.
+    assert a._swarm_is_manager({"Swarm": {"ControlAvailable": False}}) is False
+    # Non-swarm daemon: no Swarm block at all.
+    assert a._swarm_is_manager({}) is False
+
+
+def test_swarm_services_rolls_up_replicas():
+    services = [
+        {"ID": "svc1", "Spec": {
+            "Name": "web", "Mode": {"Replicated": {"Replicas": 3}},
+            "TaskTemplate": {"ContainerSpec": {"Image": "nginx:1@sha256:deadbeef"}}}},
+        {"ID": "svc2", "Spec": {
+            "Name": "agent", "Mode": {"Global": {}},
+            "TaskTemplate": {"ContainerSpec": {"Image": "agent:latest"}}}},
+    ]
+    tasks = [
+        # web: 2 of 3 desired replicas actually running, both on node n1
+        {"ServiceID": "svc1", "NodeID": "n1", "DesiredState": "running",
+         "Status": {"State": "running"}},
+        {"ServiceID": "svc1", "NodeID": "n1", "DesiredState": "running",
+         "Status": {"State": "running"}},
+        {"ServiceID": "svc1", "NodeID": "n2", "DesiredState": "running",
+         "Status": {"State": "failed"}},
+        # agent: global, 2 tasks desired+running across 2 nodes
+        {"ServiceID": "svc2", "NodeID": "n1", "DesiredState": "running",
+         "Status": {"State": "running"}},
+        {"ServiceID": "svc2", "NodeID": "n2", "DesiredState": "running",
+         "Status": {"State": "running"}},
+    ]
+    out = {s["service_name"]: s for s in a._swarm_services(services, tasks)}
+    assert out["web"] == {"service_name": "web", "mode": "replicated",
+                          "desired": 3, "running": 2, "image": "nginx:1",
+                          "placement": [{"node_id": "n1", "running": 2}]}
+    # Global mode has no fixed replica count → desired derived from desired tasks.
+    assert out["agent"] == {"service_name": "agent", "mode": "global",
+                            "desired": 2, "running": 2, "image": "agent:latest",
+                            "placement": [{"node_id": "n1", "running": 1},
+                                          {"node_id": "n2", "running": 1}]}
+
+
+def test_swarm_nodes_normalises_fields():
+    nodes = [
+        {"ID": "n1", "Description": {"Hostname": "mgr"},
+         "Spec": {"Role": "manager", "Availability": "active"},
+         "Status": {"State": "ready"}, "ManagerStatus": {"Leader": True}},
+        {"ID": "n2", "Description": {"Hostname": "wrk"},
+         "Spec": {"Role": "worker", "Availability": "drain"},
+         "Status": {"State": "down"}},
+    ]
+    out = {n["node_id"]: n for n in a._swarm_nodes(nodes)}
+    assert out["n1"] == {"node_id": "n1", "hostname": "mgr", "role": "manager",
+                         "availability": "active", "status": "ready", "leader": True}
+    assert out["n2"] == {"node_id": "n2", "hostname": "wrk", "role": "worker",
+                         "availability": "drain", "status": "down", "leader": False}
+
+
+def test_collect_swarm_skips_on_worker(monkeypatch):
+    # A worker reports ControlAvailable=False → None, and never hits /services.
+    def fake_request(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
+        if path == "/info":
+            return {"Swarm": {"ControlAvailable": False}}
+        raise AssertionError("manager-only endpoint queried on a worker")
+
+    monkeypatch.setattr(a, "_docker_request", fake_request)
+    assert a.collect_swarm("/var/run/docker.sock") is None
+
+
+def test_collect_swarm_assembles_on_manager(monkeypatch):
+    services = [{"ID": "svc1", "Spec": {
+        "Name": "web", "Mode": {"Replicated": {"Replicas": 1}},
+        "TaskTemplate": {"ContainerSpec": {"Image": "nginx"}}}}]
+    tasks = [{"ServiceID": "svc1", "DesiredState": "running",
+              "Status": {"State": "running"}}]
+    nodes = [{"ID": "n1", "Description": {"Hostname": "mgr"},
+              "Spec": {"Role": "manager", "Availability": "active"},
+              "Status": {"State": "ready"}, "ManagerStatus": {"Leader": True}}]
+
+    def fake_request(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
+        return {
+            "/info": {"Swarm": {"ControlAvailable": True}},
+            "/services": services, "/tasks": tasks, "/nodes": nodes,
+        }[path]
+
+    monkeypatch.setattr(a, "_docker_request", fake_request)
+    out = a.collect_swarm("/var/run/docker.sock")
+    assert out["services"][0]["service_name"] == "web"
+    assert out["services"][0]["running"] == 1 and out["services"][0]["desired"] == 1
+    assert out["nodes"][0]["hostname"] == "mgr" and out["nodes"][0]["leader"] is True
+
+
+def test_collect_swarm_silent_skip_when_no_socket():
+    assert a.collect_swarm("/nonexistent/does-not-exist.sock") is None
+
+
 # ── build_sample wiring ───────────────────────────────────────────────────────
 
 def test_build_sample_includes_docker_when_present(monkeypatch):
     monkeypatch.setattr(a, "collect_docker", lambda _s: [{"name": "web", "status": "running"}])
+    monkeypatch.setattr(a, "collect_swarm", lambda _s: None)
     sample = a.build_sample(["/"], {}, "/var/run/docker.sock")
     assert sample["docker"] == [{"name": "web", "status": "running"}]
 
 
 def test_build_sample_omits_docker_when_empty(monkeypatch):
     monkeypatch.setattr(a, "collect_docker", lambda _s: [])
+    monkeypatch.setattr(a, "collect_swarm", lambda _s: None)
     sample = a.build_sample(["/"], {}, "/var/run/docker.sock")
     assert "docker" not in sample
 
 
+def test_build_sample_includes_swarm_on_manager(monkeypatch):
+    monkeypatch.setattr(a, "collect_docker", lambda _s: [])
+    monkeypatch.setattr(a, "collect_swarm",
+                        lambda _s: {"services": [], "nodes": [{"node_id": "n1"}]})
+    sample = a.build_sample(["/"], {}, "/var/run/docker.sock")
+    assert sample["swarm"]["nodes"] == [{"node_id": "n1"}]
+
+
+def test_build_sample_omits_swarm_off_manager(monkeypatch):
+    monkeypatch.setattr(a, "collect_docker", lambda _s: [])
+    monkeypatch.setattr(a, "collect_swarm", lambda _s: None)
+    sample = a.build_sample(["/"], {}, "/var/run/docker.sock")
+    assert "swarm" not in sample
+
+
 # ── config ────────────────────────────────────────────────────────────────────
 
 def test_read_config_defaults_docker_socket(tmp_path):