From 22b0be2b6c6813d333e65993f24c6390d146ae84 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 14 Oct 2026 18:42:12 -0400 Subject: [PATCH] feat(host_agent): Steward hands a host a maintenance job and follows it through the reboot (#5767) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #5767 "Steward hands a host a maintenance job and follows it to the end, through its own restarts", milestone 514. A host_maintenance_jobs row per job: queued → updating → [draining →] rebooting → [undraining →] done, or failed / timed_out. Steward's answer to an ingest post carries the step that host's agent should carry out now; the agent (1.18.0) leaves each step id for the root helper once, and the helper refuses an id it already ran, so repeats are harmless. Results come back as the #5766 maintenance events, matched by step id; the reboot completes when the host's `started` event has a boot_at after the reboot was handed out. The job reboots only when the update says one is needed (or never, if asked). A swarm node is drained first by itself if it's a manager or by a manager with the helper, and made active after; with no manager to drain it the job is refused. A minute tick times out a job past its state's deadline. A maintenance window over host_agent/ lasts the rest of the job. Everything is on the row, so a Steward restart, or the reboot of the host Steward runs on, resumes from the next post. Ingest runs it in its own savepoint so a job going wrong never costs the samples. The Agent panel leads with a running job and says what a finished one did; Manage agent has Update now (with "never reboot"), or says why a job can't start. Integration test: hand off, Steward restarts, the post-boot events close the job. Co-Authored-By: Claude Opus 5.5 --- docs/plugins/host-agent-design.md | 47 ++- plugins/host_agent/__init__.py | 4 +- plugins/host_agent/agent.py | 45 +- plugins/host_agent/maintenance.py | 384 ++++++++++++++++++ .../versions/host_agent_007_maint_jobs.py | 46 +++ plugins/host_agent/models.py | 33 ++ plugins/host_agent/routes.py | 56 ++- plugins/host_agent/scheduler.py | 22 + plugins/host_agent/templates/panel.html | 24 +- tests/core/test_host_page.py | 29 ++ .../integration/test_host_maintenance_job.py | 118 ++++++ tests/plugins/host_agent/test_maintenance.py | 154 +++++++ 12 files changed, 954 insertions(+), 8 deletions(-) create mode 100644 plugins/host_agent/maintenance.py create mode 100644 plugins/host_agent/migrations/versions/host_agent_007_maint_jobs.py create mode 100644 tests/integration/test_host_maintenance_job.py create mode 100644 tests/plugins/host_agent/test_maintenance.py diff --git a/docs/plugins/host-agent-design.md b/docs/plugins/host-agent-design.md index 1f7dfd7..d23e50f 100644 --- a/docs/plugins/host-agent-design.md +++ b/docs/plugins/host-agent-design.md @@ -174,7 +174,52 @@ host that has an older agent. So a stolen agent token can make its host update, reboot or drain, and nothing else. Steward handing a host a job and following it through the reboot is -#5767; the cadence that does it one host at a time is #5768. +#5767 (below); the cadence that does it one host at a time is #5768. + +### A maintenance job, followed through the reboot (1.18.0, #5767) + +Steward keeps one `host_maintenance_jobs` row per job and moves it on only from +what the agents post, so a restart of Steward, or a reboot of the host Steward +runs on, loses nothing. `plugins/host_agent/maintenance.py` has the states and +transitions (model-free, unit-tested) and the database side. + +``` +queued → updating → [draining →] rebooting → [undraining →] done + ↘ done (no reboot needed, or the job may never reboot) +any state → failed (a step's result said so) | timed_out (its deadline passed) +``` + +- **Hand-off.** Steward's answer to an ingest post carries + `"maintenance": [{"id", "action", ["node"]}]`, the step that host's agent + should carry out now. It is repeated on every post until a result for that + step id arrives; the agent leaves each id for the helper once + (`take_maintenance`), and the helper refuses an id it already has a result + for, so a repeat, or a re-hand after the agent restarts, never runs twice. + A step id is `-`. +- **Results** are the `maintenance` events of #5766, matched to the job by step + id, so a drain reported by a manager reaches the job of the node it drained. +- **The reboot completes** when the host's agent sends `started` with a + `boot_at` after the reboot was handed out (60 s of clock slack). An agent + restart on a host that stayed up has an old `boot_at` and changes nothing. +- **Reboot only when required**: the update's result says whether a reboot is + needed; a job made with "never reboot" stops after the update. +- **Swarm nodes** are drained first and made active after. The node is found + by hostname in the Docker plugin's swarm nodes; a manager drains itself, a + worker is drained by a manager whose agent has the helper. With no such + manager the job is refused rather than rebooting an undrained node. +- **Deadlines** (`DEADLINES`): queued 1 h, updating 90 min, draining 10 min, + rebooting 30 min, undraining 10 min. A minute tick (`host_agent_maintenance`) + times out a job past its deadline. +- **A maintenance window** scoped to the host's agent metrics + (`host_agent/`) opens when the job starts, lasts as long as the rest of + the job could take, and ends when it does. +- Ingest applies all of this in a savepoint of its own: a job going wrong is + logged and never costs the host its samples. + +On the host's Agent panel: a running job leads what the host has waiting, with +since-when; a finished one says what it did ("Done: 12 packages updated, +rebooted"). Under Manage agent, **Update now** (with "never reboot") starts a +job, or the panel says why one can't start. ### What happened to the host (1.14.0, #5756) diff --git a/plugins/host_agent/__init__.py b/plugins/host_agent/__init__.py index 56565e4..0b7ca2e 100644 --- a/plugins/host_agent/__init__.py +++ b/plugins/host_agent/__init__.py @@ -37,8 +37,8 @@ def setup(app: "Quart") -> None: def get_scheduled_tasks() -> list: - from .scheduler import make_task - return [make_task(_app)] + from .scheduler import make_maintenance_task, make_task + return [make_task(_app), make_maintenance_task(_app)] def get_blueprint(): diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index a2085c2..41d6387 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -22,7 +22,7 @@ from collections import deque from concurrent.futures import ThreadPoolExecutor, wait as wait_futures from datetime import datetime, timezone -AGENT_VERSION = "1.17.0" +AGENT_VERSION = "1.18.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 @@ -1484,6 +1484,27 @@ def request_maintenance(job_id: str, action: str, node: str | None = None, os.replace(tmp, os.path.join(requests_dir, f"{job_id}.json")) +def take_maintenance(steps, requested: set, requests_dir: str = MAINT_REQUESTS) -> list: + """Leave the helper each step Steward handed over that this agent hasn't + already left it; returns the ids left. Steward hands a step on every post + until its result arrives, so a step is taken once here, and the helper + also refuses an id it has a result for (after a reboot this set is empty).""" + taken = [] + for step in steps if isinstance(steps, list) else []: + if not isinstance(step, dict) or step.get("id") in requested: + continue + try: + request_maintenance(str(step.get("id") or ""), str(step.get("action") or ""), + step.get("node"), requests_dir=requests_dir) + except (ValueError, OSError) as exc: + _log("WARN", f"maintenance step {step.get('id')!r} not taken: {exc}") + continue + requested.add(step["id"]) + taken.append(step["id"]) + _log("INFO", f"maintenance: {step['action']} requested ({step['id']})") + return taken + + def maintenance_results(seen: set, results_dir: str = MAINT_RESULTS, now: float | None = None) -> tuple: """(events, keys): a `maintenance` event for each result whose (id, state) @@ -1774,7 +1795,15 @@ def next_backoff(current: int) -> int: return min(current * 2, BACKOFF_CAP) -def post_payload(url: str, token: str, payload: dict) -> tuple[bool, int | None]: +# Steward's answer to a post is small ({"ok", counts, maybe a maintenance +# step}); anything past this is not read. +ANSWER_MAX_BYTES = 65536 + + +def post_payload(url: str, token: str, payload: dict, + answer: dict | None = None) -> tuple[bool, int | None]: + """POST the payload. When `answer` is given, Steward's JSON reply is put + in it (#5767: it carries any maintenance step for this host).""" body = json.dumps(payload).encode("utf-8") req = urllib.request.Request( url.rstrip("/") + "/plugins/host_agent/ingest", @@ -1788,6 +1817,13 @@ def post_payload(url: str, token: str, payload: dict) -> tuple[bool, int | None] ) try: with urllib.request.urlopen(req, timeout=10) as resp: + if answer is not None: + try: + got = json.loads(resp.read(ANSWER_MAX_BYTES).decode("utf-8")) + if isinstance(got, dict): + answer.update(got) + except (ValueError, UnicodeDecodeError, OSError): + pass # the post landed; only the reply is lost return (200 <= resp.status < 300, resp.status) except urllib.error.HTTPError as e: return (False, e.code) @@ -1842,6 +1878,7 @@ def main_loop(conf_path: str) -> int: # Carries previous net/disk counters + monotonic ts for rate computation. rate_state: dict = {} maint_seen: set = set() + maint_requested: set = set() _log("INFO", f"steward-host-agent {AGENT_VERSION} starting " f"(url={cfg['url']}, interval={cfg['interval_seconds']}s)") @@ -1873,9 +1910,11 @@ def main_loop(conf_path: str) -> int: events=events + ([lost_contact_event(lost_since)] if lost_since else []) + maint_events, ) - ok, status = post_payload(cfg["url"], cfg["token"], payload) + answer: dict = {} + ok, status = post_payload(cfg["url"], cfg["token"], payload, answer) if ok: + take_maintenance(answer.get("maintenance"), maint_requested) if lost_since: _log("INFO", f"reached Steward again (unreachable since {lost_since})") events, lost_since = [], None diff --git a/plugins/host_agent/maintenance.py b/plugins/host_agent/maintenance.py new file mode 100644 index 0000000..0cd612d --- /dev/null +++ b/plugins/host_agent/maintenance.py @@ -0,0 +1,384 @@ +# plugins/host_agent/maintenance.py +"""A host's maintenance job, followed to its end through any restart (#5767). + +Steward plans and the host's agent acts: a job is one host's update and, only +when the update says one is needed, its reboot, with the host's swarm node +drained first and made active again after. Each action is one step the agent +hands to its root helper (maint.py, #5766). Everything lives on the job's row, +so Steward restarting, or the host Steward runs on rebooting, loses nothing: +the next post from the agent carries the step's result, and the `started` +event a host sends after booting is what completes a reboot. + +The transitions here are model-free (a job is anything with the attributes), +so they're unit-tested without a database; routes.ingest and the scheduler's +tick apply them. + +States: queued → updating → [draining →] rebooting → [undraining →] done, +or failed / timed_out. A job with no reboot needed goes updating → done. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +ACTIVE = ("queued", "updating", "draining", "rebooting", "undraining") +FINISHED = ("done", "failed", "timed_out") +# The helper action each handed state asks for. +STEP_ACTION = {"updating": "update", "draining": "drain", "rebooting": "reboot", + "undraining": "undrain"} +# How long a state may last before the job is given up. Queued counts from the +# job's creation (the host may be silent); the rest from their hand-off. +DEADLINES = { + "queued": timedelta(hours=1), + "updating": timedelta(minutes=90), # the helper itself stops at 60 + "draining": timedelta(minutes=10), + "rebooting": timedelta(minutes=30), + "undraining": timedelta(minutes=10), +} +# A host's clock and Steward's differ by seconds; a boot this far before the +# reboot was handed out is still taken as that reboot. +BOOT_SLACK = timedelta(seconds=60) +DETAIL_MAX = 500 + + +def step_id(job, state: str) -> str: + """The helper request id for a job's step: unique per job and step, and in + the helper's id alphabet (letters, digits, dashes).""" + return f"{job.id}-{STEP_ACTION[state]}" + + +def _enter(job, state: str, now: datetime) -> None: + job.state, job.state_at = state, now + job.step_id = step_id(job, state) if state in STEP_ACTION else None + job.step_seen = False # a new step is handed out until its result arrives + if state in FINISHED: + job.finished_at = now + + +def _finish(job, state: str, now: datetime, detail: str | None = None) -> None: + if detail: + job.detail = detail[:DETAIL_MAX] + _enter(job, state, now) + + +def start(job, now: datetime) -> None: + """The host's agent is about to be handed the job: update first.""" + if job.state == "queued": + _enter(job, "updating", now) + + +def executor(job) -> str: + """The host whose agent carries out the current step: the job's own host, + except a drain or undrain, which a swarm manager does.""" + if job.state in ("draining", "undraining") and job.drain_host_id: + return job.drain_host_id + return job.host_id + + +def handoff(job) -> dict | None: + """What to hand the executing agent now, or None. Handed on every post + until a result for the step arrives; the agent and the helper both ignore + an id they've already taken, so a repeat is harmless.""" + if job.state not in STEP_ACTION or job.step_seen: + return None + out = {"id": job.step_id, "action": STEP_ACTION[job.state]} + if job.state in ("draining", "undraining") and job.node_id: + out["node"] = job.node_id + return out + + +def _after_update(job, now: datetime) -> None: + if not (job.reboot_needed and job.reboot_policy != "never"): + _enter(job, "done", now) + elif job.node_id and job.drain_host_id: + _enter(job, "draining", now) + else: + _enter(job, "rebooting", now) + + +def apply_result(job, event: dict, now: datetime) -> bool: + """Apply a `maintenance` event for this job's current step; True when the + job changed. A result for a step the job has moved past is ignored, so a + repeated or late report does nothing.""" + if job.state not in STEP_ACTION or event.get("id") != job.step_id: + return False + state = event.get("state") + if state in ("running", "rebooting"): + if job.step_seen: + return False + job.step_seen = True + return True + if state not in ("done", "failed"): + return False + if state == "failed" or event.get("ok") is False: + what = STEP_ACTION[job.state] + detail = f"{what} failed: {event.get('detail') or 'no detail'}" + if job.state == "undraining": + detail += " (the node is still drained)" + _finish(job, "failed", now, detail) + return True + if job.state == "updating": + if isinstance(event.get("packages"), int): + job.packages = event["packages"] + job.reboot_needed = event.get("reboot_required") is True + _after_update(job, now) + elif job.state == "draining": + _enter(job, "rebooting", now) + elif job.state == "undraining": + _enter(job, "done", now) + else: # rebooting: the boot completes it, not this + return False + return True + + +def apply_boot(job, boot_at: datetime | None, now: datetime) -> bool: + """A `started` event from the job's host. A boot after the reboot was + handed out is that reboot: the host is back.""" + if job.state != "rebooting" or boot_at is None or job.state_at is None: + return False + if boot_at < job.state_at - BOOT_SLACK: + return False # the agent restarted; the host didn't + job.back_at = now + if job.node_id and job.drain_host_id: + _enter(job, "undraining", now) + else: + _enter(job, "done", now) + return True + + +def check_deadline(job, now: datetime) -> bool: + """Give up on a job whose state has lasted past its deadline.""" + limit = DEADLINES.get(job.state) + since = job.state_at or job.created_at + if limit is None or since is None or now - since <= limit: + return False + waiting = {"queued": "the host's agent didn't check in", + "updating": "the update didn't finish", + "draining": "the node wasn't drained", + "rebooting": "the host didn't come back", + "undraining": "the node wasn't made active again"}[job.state] + _finish(job, "timed_out", now, f"{waiting} within {int(limit.total_seconds() // 60)} min") + return True + + +def window_end(job, now: datetime) -> datetime: + """When the job's maintenance window should end: the longest the rest of + the job could take, or now once it has finished.""" + if job.state in FINISHED: + return now + left = sum((DEADLINES[s] for s in ACTIVE[ACTIVE.index(job.state):]), timedelta()) + return now + left + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +# ── applied to the database: ingest, the tick, and starting a job ──────────── + +async def _window(session, job, host_name: str, now: datetime) -> None: + """Keep the job's maintenance window over the host until the job ends: one + scoped to this host's agent metrics, so an alert the job causes waits.""" + from steward.models.alerts import MaintenanceWindow + if job.window_id: + w = await session.get(MaintenanceWindow, job.window_id) + if w is not None: + w.end_at = max(w.start_at, window_end(job, now)) + return + if job.created_by is None or job.state in FINISHED: + return # a window needs an owner; none for a finished job + w = MaintenanceWindow(name=f"Maintenance: {host_name}"[:128], start_at=now, + end_at=window_end(job, now), scope_source="host_agent", + scope_resource=host_name, created_by=job.created_by) + session.add(w) + await session.flush() + job.window_id = w.id + + +async def _host_name(session, host_id: str) -> str: + from steward.models.hosts import Host + host = await session.get(Host, host_id) + return host.name if host is not None else host_id + + +async def apply_events(session, host_id: str, events, now: datetime) -> int: + """Apply one post's `maintenance` results and `started` boots to the jobs + they belong to; returns how many jobs changed. A result is matched by its + step id, so a drain reported by a manager reaches the job of the node it + drained.""" + from sqlalchemy import select + from .downtime import _at + from .models import HostMaintenanceJob + if not isinstance(events, list): + return 0 + changed = set() + for event in events: + if not isinstance(event, dict): + continue + if event.get("kind") == "maintenance" and isinstance(event.get("id"), str): + job = (await session.execute(select(HostMaintenanceJob).where( + HostMaintenanceJob.step_id == event["id"], + HostMaintenanceJob.state.in_(ACTIVE)))).scalar_one_or_none() + if job is not None and apply_result(job, event, now): + changed.add(job) + elif event.get("kind") == "started": + boot_at = _at(event.get("boot_at")) + for job in (await session.execute(select(HostMaintenanceJob).where( + HostMaintenanceJob.host_id == host_id, + HostMaintenanceJob.state == "rebooting"))).scalars(): + if apply_boot(job, boot_at, now): + changed.add(job) + for job in changed: + await _window(session, job, await _host_name(session, job.host_id), now) + return len(changed) + + +async def handoffs(session, host_id: str, host_name: str, now: datetime) -> list[dict]: + """The steps this host's agent should carry out now. A queued job for the + host starts here, when its agent is known to be listening.""" + from sqlalchemy import or_, select + from .models import HostMaintenanceJob + jobs = (await session.execute(select(HostMaintenanceJob).where( + HostMaintenanceJob.state.in_(ACTIVE), + or_(HostMaintenanceJob.host_id == host_id, + HostMaintenanceJob.drain_host_id == host_id) + ).order_by(HostMaintenanceJob.created_at))).scalars().all() + out = [] + for job in jobs: + if job.state == "queued" and job.host_id == host_id: + start(job, now) + await _window(session, job, host_name, now) + if executor(job) == host_id: + step = handoff(job) + if step is not None: + out.append(step) + return out + + +async def tick(session, now: datetime) -> int: + """Give up on jobs past their deadline; returns how many.""" + from sqlalchemy import select + from .models import HostMaintenanceJob + gave_up = 0 + for job in (await session.execute(select(HostMaintenanceJob).where( + HostMaintenanceJob.state.in_(ACTIVE)))).scalars(): + if check_deadline(job, now): + gave_up += 1 + await _window(session, job, await _host_name(session, job.host_id), now) + return gave_up + + +class CannotStart(Exception): + """Why a host can't take a maintenance job now, in words for the page.""" + + +def pick_drainer(node_hostname_to_id: dict, managers: dict, host_name: str, + host_id: str, helpers: set) -> tuple: + """(node_id, drain_host_id) for a host in a swarm, (None, None) for one + that isn't. `managers` maps a manager's host id to its swarm node id; + `helpers` is the hosts whose agent has the maintenance helper. The host + drains itself when it is a manager, else another manager does.""" + node_id = node_hostname_to_id.get(host_name.lower()) + if node_id is None: + return None, None + if host_id in managers and host_id in helpers: + return node_id, host_id + for mgr in sorted(managers): + if mgr != host_id and mgr in helpers: + return node_id, mgr + raise CannotStart("it is a swarm node and no swarm manager has the maintenance " + "helper to drain it: run Update agent on a manager") + + +async def plan_job(session, host, reg, reboot_policy: str = "if_required") -> tuple: + """(node_id, drain_host_id) for a job on `host`, or CannotStart saying why + it can't have one now. The panel asks this before offering the button.""" + from sqlalchemy import select + from .models import HostAgentRegistration, HostMaintenanceJob + if reg is None or not reg.maint_helper: + raise CannotStart("its agent has no maintenance helper: run Update agent first") + if reboot_policy not in ("if_required", "never"): + raise CannotStart(f"unknown reboot policy {reboot_policy!r}") + busy = (await session.execute(select(HostMaintenanceJob.id).where( + HostMaintenanceJob.host_id == host.id, + HostMaintenanceJob.state.in_(ACTIVE)))).first() + if busy is not None: + raise CannotStart("a maintenance job is already running on it") + if reboot_policy == "never": + return None, None + nodes, managers = await _swarm(session) + helpers = set((await session.execute(select(HostAgentRegistration.host_id).where( + HostAgentRegistration.maint_helper.is_(True)))).scalars()) + return pick_drainer(nodes, managers, host.name, host.id, helpers) + + +async def create_job(session, host, reg, user_id: str | None, + reboot_policy: str = "if_required", now: datetime | None = None): + """Queue a maintenance job for `host`; it starts with the agent's next post.""" + from .models import HostMaintenanceJob + now = now or utcnow() + node_id, drain_host_id = await plan_job(session, host, reg, reboot_policy) + job = HostMaintenanceJob(host_id=host.id, state="queued", state_at=now, + reboot_policy=reboot_policy, node_id=node_id, + drain_host_id=drain_host_id, created_by=user_id, created_at=now) + session.add(job) + await session.flush() + return job + + +async def latest_job(session, host_id: str, now: datetime, days: int = 7): + """The host's running job, else its last one in `days`, for the panel.""" + from sqlalchemy import select + from .models import HostMaintenanceJob + return (await session.execute(select(HostMaintenanceJob).where( + HostMaintenanceJob.host_id == host_id, + HostMaintenanceJob.created_at >= now - timedelta(days=days), + ).order_by(HostMaintenanceJob.state.in_(ACTIVE).desc(), + HostMaintenanceJob.created_at.desc()).limit(1))).scalar_one_or_none() + + +LABELS = { + "queued": "Waiting for the agent", + "updating": "Updating", + "draining": "Draining its swarm node", + "rebooting": "Rebooting", + "undraining": "Making its swarm node active again", + "done": "Done", + "failed": "Failed", + "timed_out": "Timed out", +} + + +def summary(job) -> str: + """One line on what the job did, for the panel.""" + if job.state not in FINISHED: + return LABELS[job.state] + parts = [] + if job.packages is not None: + parts.append(f"{job.packages} package{'s' if job.packages != 1 else ''} updated") + if job.back_at is not None: + parts.append("rebooted") + elif job.reboot_needed and job.reboot_policy == "never": + parts.append("reboot needed, not allowed by this job") + elif job.state == "done" and job.reboot_needed is False: + parts.append("no reboot needed") + head = LABELS[job.state] + return f"{head}: {', '.join(parts)}" if parts else head + + +async def _swarm(session) -> tuple: + """({node hostname: node id}, {manager host id: its node id}) from what the + Docker plugin last stored; empty when it isn't installed. A manager is a + host whose agent reported the swarm's nodes.""" + try: + from plugins.docker.models import DockerSwarmNode + except ImportError: + return {}, {} + from sqlalchemy import select + from steward.models.hosts import Host + rows = (await session.execute(select(DockerSwarmNode))).scalars().all() + nodes = {(n.hostname or "").lower(): n.node_id for n in rows if n.hostname} + reporters = {n.host_id for n in rows} + names = dict((await session.execute(select(Host.id, Host.name).where( + Host.id.in_(reporters)))).all()) if reporters else {} + managers = {hid: nodes.get(name.lower()) for hid, name in names.items()} + return nodes, managers diff --git a/plugins/host_agent/migrations/versions/host_agent_007_maint_jobs.py b/plugins/host_agent/migrations/versions/host_agent_007_maint_jobs.py new file mode 100644 index 0000000..c1a6178 --- /dev/null +++ b/plugins/host_agent/migrations/versions/host_agent_007_maint_jobs.py @@ -0,0 +1,46 @@ +"""host_agent: a host's maintenance job, followed through its reboot (#5767) + +Revision ID: host_agent_007_maint_jobs +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "host_agent_007_maint_jobs" +down_revision: Union[str, None] = "host_agent_006_maint_helper" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "host_maintenance_jobs", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False), + sa.Column("state", sa.String(16), nullable=False), + sa.Column("state_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("reboot_policy", sa.String(16), nullable=False), + sa.Column("step_id", sa.String(64), nullable=True), + sa.Column("step_seen", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column("node_id", sa.String(64), nullable=True), + sa.Column("drain_host_id", sa.String(36), sa.ForeignKey("hosts.id", ondelete="SET NULL"), + nullable=True), + sa.Column("packages", sa.Integer(), nullable=True), + sa.Column("reboot_needed", sa.Boolean(), nullable=True), + sa.Column("back_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("detail", sa.Text(), nullable=True), + sa.Column("window_id", sa.String(36), + sa.ForeignKey("maintenance_windows.id", ondelete="SET NULL"), nullable=True), + sa.Column("created_by", sa.String(36), sa.ForeignKey("users.id", ondelete="SET NULL"), + nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_host_maintenance_jobs_host_id", "host_maintenance_jobs", ["host_id"]) + op.create_index("ix_host_maintenance_jobs_state", "host_maintenance_jobs", ["state"]) + op.create_index("ix_host_maintenance_jobs_step_id", "host_maintenance_jobs", ["step_id"]) + + +def downgrade() -> None: + op.drop_table("host_maintenance_jobs") diff --git a/plugins/host_agent/models.py b/plugins/host_agent/models.py index 52257dd..ef7e13b 100644 --- a/plugins/host_agent/models.py +++ b/plugins/host_agent/models.py @@ -73,3 +73,36 @@ class HostDowntime(Base): started_at = Column(DateTime(timezone=True), nullable=False) ended_at = Column(DateTime(timezone=True), nullable=True) created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) + + +class HostMaintenanceJob(Base): + """One host's update, and its reboot when the update needs one (#5767). + + Steward keeps the job here and the host's agent carries out each step + through its root helper; maintenance.py has the states and transitions. + step_id is the helper request id of the current step, and step_seen says a + result for it has arrived, so it stops being handed out. node_id and + drain_host_id are set when the host is a swarm node: the node to drain and + the manager whose agent drains it. + """ + __tablename__ = "host_maintenance_jobs" + + id = Column(String(36), primary_key=True, default=_uuid) + host_id = Column(String(36), ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False, index=True) + state = Column(String(16), nullable=False, default="queued", index=True) + state_at = Column(DateTime(timezone=True), nullable=True) + reboot_policy = Column(String(16), nullable=False, default="if_required") + step_id = Column(String(64), nullable=True, index=True) + step_seen = Column(Boolean, nullable=False, default=False) + node_id = Column(String(64), nullable=True) + drain_host_id = Column(String(36), ForeignKey("hosts.id", ondelete="SET NULL"), nullable=True) + packages = Column(Integer, nullable=True) + reboot_needed = Column(Boolean, nullable=True) + back_at = Column(DateTime(timezone=True), nullable=True) + detail = Column(Text, nullable=True) + window_id = Column(String(36), ForeignKey("maintenance_windows.id", ondelete="SET NULL"), + nullable=True) + created_by = Column(String(36), ForeignKey("users.id", ondelete="SET NULL"), nullable=True) + created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) + finished_at = Column(DateTime(timezone=True), nullable=True) diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index fcb005a..c9eae93 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -17,6 +17,7 @@ from steward.models.users import UserRole from sqlalchemy import select, or_, and_ from datetime import timedelta +from steward.core.audit import audit_request from steward.core.formatting import worst_reading from steward.core.health import ingest as ingest_tally from steward.core.metric_series import points as metric_points @@ -26,6 +27,7 @@ from steward.hosts import service as host_svc from steward.models.hosts import Host from steward.models.metrics import PluginMetric from .downtime import recent as recent_downtime, record_events +from . import maintenance from .models import HostAgentRegistration from .scheduler import arrivals # Query helpers live in a model-free module so integration tests can import them @@ -347,6 +349,7 @@ async def ingest(): token_hash = _hash_token(raw) phases = _Phases() + maint_steps: list = [] samples, dropped = newest_samples(samples, MAX_SAMPLES_PER_INGEST) if dropped: current_app.logger.warning( @@ -461,6 +464,20 @@ async def ingest(): # was a restart without a shutdown. if events: await record_events(session, host.id, events, reg.last_seen_at) + # Maintenance jobs (#5767): results and boots move a job on, then + # whatever step is due for this host's agent goes back in the answer. + # Each in its own savepoint: a job going wrong never costs the samples. + try: + async with session.begin_nested(): + if events: + await maintenance.apply_events(session, host.id, events, received_at) + if reg.maint_helper: + maint_steps = await maintenance.handoffs( + session, host.id, host.name, received_at) + except Exception: + maint_steps = [] + current_app.logger.exception( + "host_agent ingest: maintenance jobs failed for host=%s", host.name) md = payload.get("metadata") or {} changed = False @@ -518,7 +535,10 @@ async def ingest(): phases.report(host.name, accepted, len(metric_rows), sum(len(c) for _, c in docker_snapshots), sum(len(lines) for _, lines in docker_log_batches)) - return jsonify({"ok": True, "samples_accepted": accepted, "samples_dropped": dropped}), 200 + answer = {"ok": True, "samples_accepted": accepted, "samples_dropped": dropped} + if maint_steps: + answer["maintenance"] = maint_steps + return jsonify(answer), 200 # The agent gives up on a post after 10 s (agent.py); an ingest past this @@ -964,6 +984,14 @@ async def host_panel(host_id: str): target = (await db.execute(select(AnsibleTarget).where( AnsibleTarget.host_id == host_id))).scalar_one_or_none() downtime = await recent_downtime(db, host_id, datetime.now(timezone.utc)) if reg else [] + # Maintenance (#5767): the running or latest job, and whether one can start. + maint_job = await maintenance.latest_job(db, host_id, datetime.now(timezone.utc)) if reg else None + maint_blocker = None + if reg is not None: + try: + await maintenance.plan_job(db, host, reg) + except maintenance.CannotStart as exc: + maint_blocker = str(exc) ls = reg.last_seen_at if reg else None # "Deployed" is gated on the agent actually checking in (last_seen_at), NOT on @@ -976,6 +1004,8 @@ async def host_panel(host_id: str): return await render_template( "panel.html", host=host, reg=reg, reporting=reporting, stale=stale, target=target, downtime=downtime, + maint_job=maint_job, maint_blocker=maint_blocker, + maint_summary=maintenance.summary(maint_job) if maint_job else None, ansible_available=has_capability("ansible.run_playbook"), managed_key_set=bool((ansible_cfg.get("ssh_public_key") or "").strip()), ) @@ -1060,6 +1090,30 @@ async def rotate_token(host_id: str): return redirect(url_for("host_agent.fleet") + f"?new_token={raw}&host_id={host_id}") +@host_agent_bp.post("/fleet//maintain") +@require_role(UserRole.admin) +async def start_maintenance(host_id: str): + """Update this host now, rebooting it only if the update needs it (#5767). + The job starts with the agent's next post and is followed on the panel.""" + form = await request.form + policy = "never" if form.get("reboot_policy") == "never" else "if_required" + user_id = session.get("user_id") + async with current_app.db_sessionmaker() as db: + host = await db.get(Host, host_id) + if host is None: + return _error(404, "not_found") + reg = (await db.execute(select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host_id))).scalar_one_or_none() + try: + job = await maintenance.create_job(db, host, reg, user_id, policy) + except maintenance.CannotStart as exc: + return _error(409, "cannot_start", f"{host.name} can't be updated now: {exc}") + await db.commit() + await audit_request("host_agent.maintenance", "host", host_id, + {"job_id": job.id, "reboot_policy": policy}) + return redirect(f"/hosts/{host_id}") + + @host_agent_bp.post("/fleet//delete") @require_role(UserRole.admin) async def delete_registration(host_id: str): diff --git a/plugins/host_agent/scheduler.py b/plugins/host_agent/scheduler.py index c1d8dbb..40775d6 100644 --- a/plugins/host_agent/scheduler.py +++ b/plugins/host_agent/scheduler.py @@ -119,3 +119,25 @@ def make_task(app) -> ScheduledTask: interval_seconds=60, run_on_startup=False, ) + + +def make_maintenance_task(app) -> ScheduledTask: + """Gives up on maintenance jobs past their deadline (#5767). Every other + move a job makes comes from its agent's posts.""" + async def _tick(): + from . import maintenance + try: + async with app.db_sessionmaker() as session: + async with session.begin(): + gave_up = await maintenance.tick(session, datetime.now(timezone.utc)) + if gave_up: + logger.warning("host_agent: %d maintenance job(s) timed out", gave_up) + except Exception: + logger.exception("host_agent: maintenance tick failed") + + return ScheduledTask( + name="host_agent_maintenance", + coro_factory=_tick, + interval_seconds=60, + run_on_startup=False, + ) diff --git a/plugins/host_agent/templates/panel.html b/plugins/host_agent/templates/panel.html index 48d530c..b6108c4 100644 --- a/plugins/host_agent/templates/panel.html +++ b/plugins/host_agent/templates/panel.html @@ -21,8 +21,16 @@ {# What the host has waiting, from the agent's hourly check (#5765): a needed reboot first, then the updates. #} - {% if reg.updates_checked_at or reg.reboot_required %} + {% if reg.updates_checked_at or reg.reboot_required or maint_job %}
+ {# Steward's own maintenance job on this host (#5767): a running one first. #} + {% if maint_job %} + {% set running = maint_job.state not in ('done', 'failed', 'timed_out') %} +
+ Maintenance: {{ maint_summary }} + · {% if running %}since {{ maint_job.state_at|ago }}{% else %}{{ maint_job.finished_at|ago }}{% endif %}{% if maint_job.state in ('failed', 'timed_out') and maint_job.detail %} · {{ maint_job.detail|truncate(90) }}{% endif %} +
+ {% endif %} {% if reg.reboot_required %}
Reboot required{% if reg.reboot_required_since %} since {{ reg.reboot_required_since|ago }}{% endif %}{% if reg.reboot_packages %} @@ -78,6 +86,20 @@ Link an Ansible target (below) to update the agent from here.

{% endif %} + {# Update now (#5767): the agent's helper updates the host and reboots it + only if the update needs it, draining its swarm node first. #} + {% if maint_blocker %} +

Can't update from here: {{ maint_blocker }}.

+ {% else %} +
{{ csrf_field() }} + + +
+ {% endif %} {# The root helper that carries out scheduled updates and reboots (#5766). #}

{% if reg.maint_helper %}Maintenance helper installed: Steward can update and reboot this host. diff --git a/tests/core/test_host_page.py b/tests/core/test_host_page.py index 1d726e1..893b82d 100644 --- a/tests/core/test_host_page.py +++ b/tests/core/test_host_page.py @@ -279,3 +279,32 @@ async def test_manage_agent_says_whether_the_maintenance_helper_is_installed(app reg=SimpleNamespace(**{**vars(reg), "maint_helper": None}), **kw) assert "Maintenance helper installed" in installed assert "No maintenance helper yet: Update agent installs it" in missing + + +async def test_the_agent_card_follows_a_maintenance_job_and_says_why_it_cant_start_one(app): + """#5767: a running job leads what the host has waiting; Update now is + offered only when a job can start, else the reason is shown.""" + from datetime import datetime, timedelta, timezone + from plugins.host_agent.maintenance import summary + from tests._templates import render_plugin_template + now = datetime.now(timezone.utc) + host = SimpleNamespace(id="h1", name="box", address="10.0.0.5") + reg = SimpleNamespace(agent_version="1.18.0", last_seen_at=now, maint_helper=True, + updates_checked_at=now, updates_pending=4, updates_security=0, + reboot_required=False) + job = SimpleNamespace(state="rebooting", state_at=now - timedelta(minutes=2), finished_at=None, + detail=None, packages=4, back_at=None, reboot_needed=True, + reboot_policy="if_required") + kw = dict(host=host, reg=reg, reporting=True, stale=False, target=None, ansible_available=False, + managed_key_set=False, downtime=[]) + async with app.test_request_context("/plugins/host_agent/panel/h1"): + session["user_role"] = "admin" + busy = await render_plugin_template( + app, "host_agent", "panel.html", maint_job=job, maint_summary=summary(job), + maint_blocker="a maintenance job is already running on it", **kw) + free = await render_plugin_template(app, "host_agent", "panel.html", maint_job=None, + maint_blocker=None, **kw) + assert busy.index("Maintenance: Rebooting") < busy.index("4 updates waiting") + assert "Can't update from here: a maintenance job is already running on it." in busy + assert ">Update now" not in busy + assert ">Update now" in free and 'name="reboot_policy" value="never"' in free diff --git a/tests/integration/test_host_maintenance_job.py b/tests/integration/test_host_maintenance_job.py new file mode 100644 index 0000000..3fa4ead --- /dev/null +++ b/tests/integration/test_host_maintenance_job.py @@ -0,0 +1,118 @@ +"""Integration (#5767): Steward hands a host a maintenance job and follows it +to the end, through a restart of its own, from the agent's posts alone. + +Rows are seeded and read with plain SQL: importing the plugin's models again +after the app loaded the plugin raises "Table already defined" (see +test_host_agent_alerts).""" +from __future__ import annotations + +import asyncio +import hashlib +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +def _app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +async def _seed(app): + from sqlalchemy import text + from steward.models.hosts import Host + from steward.models.users import User + tag = uuid.uuid4().hex[:8] + token = f"tok-{tag}-{uuid.uuid4().hex}" + job_id = str(uuid.uuid4()) + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"mj-{tag}", email=f"mj-{tag}@test", password_hash="x") + host = Host(name=f"maint-{tag}", address="10.9.9.7") + s.add_all([user, host]) + await s.flush() + await s.execute(text( + "INSERT INTO host_agent_registrations " + "(id, host_id, token_hash, token_created_at, last_seen_at, maint_helper, " + " created_at, updated_at) " + "VALUES (:id, :host, :hash, now(), now(), true, now(), now())"), + {"id": str(uuid.uuid4()), "host": host.id, + "hash": hashlib.sha256(token.encode()).hexdigest()}) + await s.execute(text( + "INSERT INTO host_maintenance_jobs " + "(id, host_id, state, state_at, reboot_policy, step_seen, created_by, created_at) " + "VALUES (:id, :host, 'queued', now(), 'if_required', false, :user, now())"), + {"id": job_id, "host": host.id, "user": user.id}) + host_id = host.id + return host_id, token, job_id + + +async def _post(app, token, events=(), sample=True): + now = datetime.now(timezone.utc).isoformat() + resp = await app.test_client().post( + "/plugins/host_agent/ingest", + json={"samples": [{"ts": now, "cpu_pct": 2.0}] if sample else [], + "events": list(events)}, + headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 200, await resp.get_data(as_text=True) + return (await resp.get_json()).get("maintenance", []) + + +async def _job(app, job_id): + from sqlalchemy import text + async with app.db_sessionmaker() as s: + async with s.begin(): + return (await s.execute(text( + "SELECT j.state, j.packages, j.reboot_needed, j.back_at, w.end_at " + "FROM host_maintenance_jobs j LEFT JOIN maintenance_windows w ON w.id = j.window_id " + "WHERE j.id = :id"), {"id": job_id})).one() + + +def test_a_job_is_followed_through_a_steward_restart_and_the_host_reboot(): + app = _app() + + async def _first_half(): + host_id, token, job_id = await _seed(app) + # The agent's next post starts the job: update first, under a window. + assert await _post(app, token) == [{"id": f"{job_id}-update", "action": "update"}] + state, *_, window_end = await _job(app, job_id) + assert state == "updating" and window_end > datetime.now(timezone.utc) + # The helper took it: the step is no longer handed out. + running = {"kind": "maintenance", "id": f"{job_id}-update", "action": "update", + "state": "running"} + assert await _post(app, token, [running]) == [] + return token, job_id + + token, job_id = asyncio.run(_first_half()) + + # Steward restarts: a new process, nothing in memory but the database. + app = _app() + + async def _second_half(): + done = {"kind": "maintenance", "id": f"{job_id}-update", "action": "update", + "state": "done", "ok": True, "packages": 12, "reboot_required": True} + assert await _post(app, token, [done]) == [{"id": f"{job_id}-reboot", "action": "reboot"}] + assert (await _job(app, job_id))[:3] == ("rebooting", 12, True) + # The agent restarting on a host that stayed up doesn't count as the reboot. + old_boot = datetime.now(timezone.utc) - timedelta(days=3) + await _post(app, token, [{"kind": "started", "at": datetime.now(timezone.utc).isoformat(), + "boot_at": old_boot.isoformat()}]) + assert (await _job(app, job_id))[0] == "rebooting" + # After the reboot: the helper's result and the agent's boot, in one post. + boot = datetime.now(timezone.utc) + rebooted = {"kind": "maintenance", "id": f"{job_id}-reboot", "action": "reboot", + "state": "rebooting", "ok": True} + started = {"kind": "started", "at": (boot + timedelta(seconds=20)).isoformat(), + "boot_at": boot.isoformat()} + assert await _post(app, token, [rebooted, started]) == [] + state, packages, reboot_needed, back_at, window_end = await _job(app, job_id) + assert state == "done" and back_at is not None + assert window_end <= datetime.now(timezone.utc) + + asyncio.run(_second_half()) diff --git a/tests/plugins/host_agent/test_maintenance.py b/tests/plugins/host_agent/test_maintenance.py new file mode 100644 index 0000000..c45aaa5 --- /dev/null +++ b/tests/plugins/host_agent/test_maintenance.py @@ -0,0 +1,154 @@ +# tests/plugins/host_agent/test_maintenance.py +"""#5767: a host's maintenance job, step by step, from the agent's reports. + +The transitions are model-free, so a job here is a SimpleNamespace with the +row's attributes; the database side is tests/integration/test_host_maintenance_job.py. +""" +import json +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from plugins.host_agent import agent as a +from plugins.host_agent import maintenance as m + +T0 = datetime(2026, 10, 14, 23, 0, tzinfo=timezone.utc) + + +def _job(**kw): + base = dict(id="job-1", host_id="h1", state="queued", state_at=T0, created_at=T0, + reboot_policy="if_required", step_id=None, step_seen=False, node_id=None, + drain_host_id=None, packages=None, reboot_needed=None, back_at=None, + detail=None, finished_at=None) + base.update(kw) + return SimpleNamespace(**base) + + +def _result(job, state, **kw): + return {"kind": "maintenance", "id": job.step_id, "state": state, **kw} + + +def test_an_update_with_no_reboot_needed_is_done_after_the_update(): + job = _job() + m.start(job, T0) + assert m.handoff(job) == {"id": "job-1-update", "action": "update"} + assert m.apply_result(job, _result(job, "running"), T0) + assert m.handoff(job) is None # taken: not handed again + m.apply_result(job, _result(job, "done", ok=True, packages=3, reboot_required=False), T0) + assert (job.state, job.packages, job.finished_at) == ("done", 3, T0) + assert m.summary(job) == "Done: 3 packages updated, no reboot needed" + + +def test_a_swarm_node_is_drained_rebooted_and_made_active_again(): + job = _job(node_id="n" * 25, drain_host_id="mgr") + m.start(job, T0) + m.apply_result(job, _result(job, "done", ok=True, packages=12, reboot_required=True), T0) + assert job.state == "draining" and m.executor(job) == "mgr" + assert m.handoff(job) == {"id": "job-1-drain", "action": "drain", "node": "n" * 25} + m.apply_result(job, _result(job, "done", ok=True), T0) + assert job.state == "rebooting" and m.executor(job) == "h1" + assert m.handoff(job) == {"id": "job-1-reboot", "action": "reboot"} + t1 = T0 + timedelta(minutes=4) + assert m.apply_boot(job, T0 + timedelta(minutes=2), t1) + assert job.state == "undraining" and job.back_at == t1 and m.executor(job) == "mgr" + m.apply_result(job, _result(job, "done", ok=True), t1) + assert job.state == "done" + assert m.summary(job) == "Done: 12 packages updated, rebooted" + + +def test_a_job_that_may_never_reboot_stops_after_the_update(): + job = _job(reboot_policy="never") + m.start(job, T0) + m.apply_result(job, _result(job, "done", ok=True, packages=1, reboot_required=True), T0) + assert job.state == "done" + assert "reboot needed, not allowed" in m.summary(job) + + +def test_only_a_boot_after_the_reboot_was_handed_out_completes_it(): + job = _job(state="rebooting", step_id="job-1-reboot", state_at=T0) + assert not m.apply_boot(job, T0 - timedelta(days=3), T0) # the agent restarted, not the host + assert not m.apply_boot(job, None, T0) + assert job.state == "rebooting" + assert m.apply_boot(job, T0 - timedelta(seconds=30), T0) # within the clock slack + assert job.state == "done" + + +def test_a_failed_step_fails_the_job_and_says_which(): + job = _job(state="undraining", step_id="job-1-undrain", node_id="n" * 25, drain_host_id="mgr") + m.apply_result(job, _result(job, "failed", ok=False, detail="This node is not a swarm manager."), T0) + assert job.state == "failed" + assert job.detail.startswith("undrain failed: This node is not a swarm manager.") + assert "still drained" in job.detail + + +def test_a_late_or_repeated_result_does_nothing(): + job = _job() + m.start(job, T0) + stale = {"kind": "maintenance", "id": "job-1-drain", "state": "done", "ok": True} + assert not m.apply_result(job, stale, T0) + assert m.apply_result(job, _result(job, "running"), T0) + assert not m.apply_result(job, _result(job, "running"), T0) + m.apply_result(job, _result(job, "done", ok=True, reboot_required=False), T0) + assert not m.apply_result(job, {"kind": "maintenance", "id": "job-1-update", "state": "done"}, T0) + + +@pytest.mark.parametrize("state,minutes,expected", [ + ("queued", 59, False), ("queued", 61, True), + ("updating", 91, True), ("rebooting", 29, False), ("rebooting", 31, True), +]) +def test_a_job_is_given_up_past_its_state_deadline(state, minutes, expected): + job = _job(state=state, step_id=f"job-1-{m.STEP_ACTION.get(state, 'x')}") + assert m.check_deadline(job, T0 + timedelta(minutes=minutes)) is expected + if expected: + assert job.state == "timed_out" and "within" in job.detail + + +def test_the_window_lasts_the_rest_of_the_job_and_ends_with_it(): + job = _job(state="rebooting", step_id="job-1-reboot") + assert m.window_end(job, T0) == T0 + timedelta(minutes=30 + 10) + job.state = "done" + assert m.window_end(job, T0) == T0 + + +def test_a_manager_drains_itself_and_a_worker_is_drained_by_a_manager(): + nodes = {"docker02": "a" * 25, "docker03": "b" * 25} + managers = {"h2": "a" * 25} + assert m.pick_drainer(nodes, managers, "Docker02", "h2", {"h2", "h3"}) == ("a" * 25, "h2") + assert m.pick_drainer(nodes, managers, "Docker03", "h3", {"h2", "h3"}) == ("b" * 25, "h2") + assert m.pick_drainer(nodes, managers, "Util01", "h4", {"h2"}) == (None, None) + with pytest.raises(m.CannotStart, match="no swarm manager"): + m.pick_drainer(nodes, managers, "Docker03", "h3", {"h3"}) + + +# ── the agent's side ───────────────────────────────────────────────────────── + +def test_the_agent_leaves_each_handed_step_once(tmp_path): + requested: set = set() + steps = [{"id": "job-1-update", "action": "update"}, {"id": "job-1-x", "action": "shell"}, "junk"] + assert a.take_maintenance(steps, requested, requests_dir=str(tmp_path)) == ["job-1-update"] + assert json.loads((tmp_path / "job-1-update.json").read_text()) == { + "id": "job-1-update", "action": "update"} + (tmp_path / "job-1-update.json").unlink() # the helper took it + assert a.take_maintenance(steps, requested, requests_dir=str(tmp_path)) == [] + assert not (tmp_path / "job-1-update.json").exists() + assert a.take_maintenance(None, requested, requests_dir=str(tmp_path)) == [] + + +def test_the_agent_reads_stewards_answer_when_asked(monkeypatch): + class Resp: + status = 200 + + def read(self, n): + return b'{"ok": true, "maintenance": [{"id": "job-1-reboot", "action": "reboot"}]}' + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + monkeypatch.setattr(a.urllib.request, "urlopen", lambda req, timeout: Resp()) + answer: dict = {} + assert a.post_payload("https://x", "tok", {}, answer) == (True, 200) + assert answer["maintenance"] == [{"id": "job-1-reboot", "action": "reboot"}]