From 448258c5b4ba349fddac595a02a7b75aa513aea4 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 18 Jun 2026 20:57:18 -0400 Subject: [PATCH] feat(docker): agent manager-only swarm collector (AGENT_VERSION 1.5.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC --- plugins/host_agent/agent.py | 122 +++++++++++++++++- tests/plugins/host_agent/test_agent_docker.py | 115 +++++++++++++++++ 2 files changed, 236 insertions(+), 1 deletion(-) diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 227bc2e..d1acc08 100644 --- a/plugins/host_agent/agent.py +++ b/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 diff --git a/tests/plugins/host_agent/test_agent_docker.py b/tests/plugins/host_agent/test_agent_docker.py index e734c19..67b3654 100644 --- a/tests/plugins/host_agent/test_agent_docker.py +++ b/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):