feat(swarm): services short, nodes down and managers lost are alertable (#5788)

On 2026-10-14 nine swarm services sat at 0/1 and nothing fired. From each
manager's swarm report the docker plugin now records three figures on
resource "swarm": swarm_services_short, swarm_nodes_down and
swarm_managers_unreachable. Drained nodes are left out of the last two,
since a drain is deliberate. The figures are recorded at most once every
30 s whichever manager reported, and are offered to alert rules.

Agent 1.19.0 sends each manager's ManagerStatus.Reachability. Older agents
fall back to the node's status.

A maintenance job on a swarm node also keeps a window over docker/swarm,
so its drain and reboot hold those alerts until the job ends.

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

tests/integration/test_docker.py

@@ -575,17 +575,26 @@ def test_swarm_topology_persisted(app):
                 s.add(h)
                 await s.flush()
                 # No containers in this sample — swarm must still persist.
+                from steward.core.plugin_manager import loaded_plugin_submodule
+                loaded_plugin_submodule("docker", "swarm_health")._last_recorded = None
+                since = datetime.now(timezone.utc)
                 await persist(s, h, [], swarm)
                 hid = h.id
+            # The swarm-wide figures alert rules watch (#5788): web is 2 of 3.
+            figures = dict((await s.execute(text(
+                "SELECT metric_name, value FROM plugin_metrics WHERE source_module = 'docker' "
+                "AND resource_name = 'swarm' AND recorded_at >= :t"), {"t": since})).all())
             svc = (await s.execute(text(
                 "SELECT mode, desired, running, image, placement_json, tasks_json "
                 "FROM docker_swarm_services WHERE host_id = :h"), {"h": hid})).first()
             node = (await s.execute(text(
                 "SELECT role, availability, status, leader "
                 "FROM docker_swarm_nodes WHERE host_id = :h"), {"h": hid})).first()
-            return svc, node
+            return svc, node, figures
 
-    svc, node = asyncio.run(_go())
+    svc, node, figures = asyncio.run(_go())
+    assert figures == {"swarm_services_short": 1.0, "swarm_nodes_down": 0.0,
+                       "swarm_managers_unreachable": 0.0}
     assert svc[0] == "replicated" and svc[1] == 3 and svc[2] == 2 and svc[3] == "nginx"
     assert "n1" in svc[4]  # placement JSON carries the node id
     (task,) = json.loads(svc[5])  # docker_010: the manager's task records survive

tests/integration/test_host_maintenance_job.py

@@ -116,3 +116,39 @@ def test_a_job_is_followed_through_a_steward_restart_and_the_host_reboot():
         assert window_end <= datetime.now(timezone.utc)
 
     asyncio.run(_second_half())
+
+
+def test_a_swarm_node_s_job_also_holds_the_swarm_wide_alerts():
+    """Draining and rebooting a node moves the swarm figures; that is the job,
+    not news (#5788). The window over them follows the job's own, and ends
+    with it."""
+    from types import SimpleNamespace
+
+    from sqlalchemy import text
+    from steward.core.plugin_manager import loaded_plugin_submodule
+    from steward.models.users import User
+    app = _app()
+    maintenance = loaded_plugin_submodule("host_agent", "maintenance")
+    tag = uuid.uuid4().hex[:8]
+    host_name = f"swarm-node-{tag}"
+
+    async def _go():
+        now = datetime.now(timezone.utc)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                user = User(username=f"sw-{tag}", email=f"sw-{tag}@test", password_hash="x")
+                s.add(user)
+                await s.flush()
+                job = SimpleNamespace(window_id=None, created_by=user.id, state="draining",
+                                      node_id="node-1", created_at=now - timedelta(seconds=1))
+                await maintenance._window(s, job, host_name, now)
+                await maintenance._window(s, job, host_name, now + timedelta(minutes=1))   # the next tick
+                job.state = "done"
+                await maintenance._window(s, job, host_name, now + timedelta(minutes=2))   # finished
+            return (await s.execute(text(
+                "SELECT scope_source, scope_resource, end_at FROM maintenance_windows "
+                "WHERE name LIKE :n ORDER BY scope_source"), {"n": f"Maintenance: {host_name}%"})).all(), now
+
+    rows, now = asyncio.run(_go())
+    assert [(r[0], r[1]) for r in rows] == [("docker", "swarm"), ("host_agent", host_name)]
+    assert all(r[2] == now + timedelta(minutes=2) for r in rows)     # ended with the job, not before