feat(cluster): record a Docker host's role on the host and keep it across runs

plugins/filter/cluster_guard.py

@@ -1,4 +1,4 @@
-"""Filters the github_runner_cluster role uses before it restarts a node (moving its container between networks, resizing its Colima VM): whether the restart would be safe for the cluster, whether the fleet-regrade Lease that serialises such steps may be taken and how it is given back, and whether the Docker host has an address on the LAN."""
+"""Filters the github_runner_cluster role uses before it restarts a node (moving its container between networks, resizing its Colima VM): whether the restart would be safe for the cluster, whether the fleet-regrade Lease that serialises such steps may be taken and how it is given back, whether the role controller holds it for a voter swap, during which no run of the role may change a node, and whether the Docker host has an address on the LAN."""
 
 from __future__ import annotations
 
@@ -41,6 +41,12 @@ def _etcd_healthy(result: Any) -> bool:
     return result.get("rc") == 0 and str(result.get("stdout", "")).strip() == "ok"
 
 
+def _reason(result: Mapping[str, Any]) -> str:
+    """The last line of what a failed command or task said, or its exit status when it said nothing."""
+    lines = [line.strip() for line in str(result.get("stderr", "") or result.get("msg", "") or "").splitlines() if line.strip()]
+    return lines[-1] if lines else f"exit status {result.get('rc')}"
+
+
 def cluster_restart_blockers(nodes: Any, pods: Any, node_name: Any, servers: Any, etcd_results: Any) -> list[str]:
     """Say why restarting (or recreating) this host's node now would be unsafe, or return an empty list when it would not be.
 
@@ -133,8 +139,7 @@ def cluster_guard_state(results: Any) -> dict[str, Any]:
             reasons.append(f"{host} did not answer within github_runner_cluster_cluster_read_timeout")
             continue
         if result.get("rc") != 0:
-            lines = [line.strip() for line in str(result.get("stderr", "") or message).splitlines() if line.strip()]
-            reasons.append(f"{host} could not be read ({lines[-1] if lines else 'exit status ' + str(result.get('rc'))})")
+            reasons.append(f"{host} could not be read ({_reason(result)})")
             continue
         try:
             state = json.loads(str(result.get("stdout", "")))
@@ -181,6 +186,14 @@ def _epoch(stamp: str) -> float:
     return datetime.strptime(re.sub(r"\.[0-9]+Z$", "Z", stamp), "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc).timestamp()
 
 
+def _lease_expires(spec: Mapping[str, Any]) -> float | None:
+    """When a Lease's holder's time runs out, as epoch seconds: its renew time (its acquire time when it was never renewed) plus its duration, the rule lease_state applies in files/regrade/fleet-regrade-core.sh; None when it records neither time."""
+    renewed = spec.get("renewTime") or spec.get("acquireTime")
+    if not renewed:
+        return None
+    return _epoch(str(renewed)) + int(spec.get("leaseDurationSeconds") or 0)
+
+
 def _lease_read(read: Any) -> dict[str, Any] | None | str:
     """The Lease a `kubectl get lease --output json` result holds, None when the API server answered that there is none, or the reason the read failed."""
     if not isinstance(read, Mapping):
@@ -219,10 +232,10 @@ def cluster_lease_claim(read: Any, holder: Any, now: Any, duration: Any, name: A
         return {"state": "absent", "holder": "", "verb": "create", "manifest": manifest, "problem": ""}
     spec = current.get("spec") or {}
     current_holder = str(spec.get("holderIdentity") or "")
-    renewed = spec.get("renewTime") or spec.get("acquireTime")
+    expires = _lease_expires(spec)
     if not current_holder:
         state = "free"
-    elif not renewed or _epoch(str(renewed)) + int(spec.get("leaseDurationSeconds") or 0) <= float(now):
+    elif expires is None or expires <= float(now):
         state = "expired"
     elif current_holder == holder:
         state = "mine"
@@ -245,6 +258,53 @@ def cluster_lease_release(read: Any, holder: Any) -> dict[str, Any]:
     return {"verb": "replace", "manifest": released}
 
 
+def cluster_role_lease_guard(reads: Any, holder: Any, swap_id_annotation: Any) -> dict[str, str]:
+    """Say whether the role controller holds the regrade Lease, which it holds for the whole of a voter swap: while it does, a run of the role must change no node, so that it never undoes or races the swap.
+
+    :param reads: the loop results of reading the Lease through each server of the inventory (tasks/role_lease_guard.yml), each with its host as ``item`` and ``stdout`` either "stopped" or "running", the reading host's clock as epoch seconds, then the Lease as ``kubectl get lease -o json`` prints it (nothing when there is none). :param holder: the holderIdentity the role controller takes the Lease as. :param swap_id_annotation: the annotation on the Lease that names the swap. :returns: ``state``, one of ``held`` (by the controller, and unexpired), ``free`` (no server runs, or the Lease is absent, has run out, is unheld or held by anything else, such as a regrading step) and ``unknown`` (no server answered and one could not be read), with ``message``, a sentence saying why. :raises AnsibleFilterError: if a server answered with something that is not a Lease.
+
+    One answer decides, since every server's API reads the same Lease. Its expiry (_lease_expires, as cluster_lease_claim judges it) is compared with the clock of the host that read it, the clock the controller renews by.
+    """
+    if not isinstance(reads, Sequence) or isinstance(reads, str):
+        raise AnsibleFilterError("The reads must be a list of loop results.")
+    unread: list[str] = []
+    for result in reads:
+        if not isinstance(result, Mapping):
+            raise AnsibleFilterError(f"{result!r} is not a loop result.")
+        host = str(result.get("item", "?"))
+        if result.get("unreachable") or result.get("rc") != 0:
+            unread.append(f"{host} could not be read ({_reason(result)})")
+            continue
+        state, _, rest = str(result.get("stdout", "")).partition("\n")
+        if state.strip() == "stopped":
+            continue
+        clock, _, text = rest.partition("\n")
+        if state.strip() != "running" or not clock.strip().isdigit():
+            raise AnsibleFilterError(f"{host} gave an answer that is not a read of the Lease: {result.get('stdout')!r}")
+        if not text.strip():
+            return {"state": "free", "message": f"there is no Lease (read through {host})"}
+        try:
+            lease = json.loads(text)
+        except ValueError as exc:
+            raise AnsibleFilterError(f"{host} gave a Lease that is not JSON: {exc}") from exc
+        if not isinstance(lease, dict):
+            raise AnsibleFilterError(f"{host} gave a Lease that is not an object: {text!r}")
+        metadata = lease.get("metadata") or {}
+        spec = lease.get("spec") or {}
+        name = f"{metadata.get('namespace', '?')}/{metadata.get('name', '?')}"
+        expires = _lease_expires(spec)
+        if str(spec.get("holderIdentity") or "") != str(holder) or expires is None:
+            return {"state": "free", "message": f"the Lease {name} is not held by {holder} (read through {host})"}
+        if expires <= int(clock):
+            return {"state": "free", "message": f"the Lease {name} that {holder} held has run out (read through {host})"}
+        swap = str((metadata.get("annotations") or {}).get(str(swap_id_annotation)) or "")
+        until = datetime.fromtimestamp(expires, tz=timezone.utc).isoformat(timespec="seconds")
+        return {"state": "held", "message": f"the role controller ({holder}) holds the Lease {name} for the voter swap {swap or '(unnamed)'}, read through {host}, until {until} unless it renews it first"}
+    if unread:
+        return {"state": "unknown", "message": "whether the role controller holds the Lease cannot be told: " + "; ".join(unread)}
+    return {"state": "free", "message": "no server of the inventory runs, so there is no Lease"}
+
+
 class FilterModule:
     """Registers the cluster guard filters with Ansible."""
 
@@ -256,4 +316,5 @@ def filters(self) -> dict[str, Any]:
             "addresses_in_network": addresses_in_network,
             "cluster_lease_claim": cluster_lease_claim,
             "cluster_lease_release": cluster_lease_release,
+            "cluster_role_lease_guard": cluster_role_lease_guard,
         }

plugins/filter/cluster_native.py

@@ -1,4 +1,4 @@
-"""Filters the github_runner_cluster role's native node uses: which k3s release asset a host's architecture takes, the release and package versions the node's pins name, which tailnet the host's Tailscale client is logged in to, which role the host runs and whether it bootstraps a cluster, and whether the cluster's other voters are Ready enough for a server to start, restart or join."""
+"""Filters the github_runner_cluster role's native node uses: which k3s release asset a host's architecture takes, the release and package versions the node's pins name, which tailnet the host's Tailscale client is logged in to, whether a server bootstraps a cluster, and whether the cluster's other voters are Ready enough for a server to start, restart or join."""
 
 from __future__ import annotations
 
@@ -82,38 +82,12 @@ def cluster_native_tailnet_login(status: Any, tailnet_domain: Any) -> dict[str,
     return {"state": state, "backend": backend, "domain": domain}
 
 
-NODE_ROLES = ("server", "agent")
-
-
 def _reason(result: dict[str, Any]) -> str:
     """The last line of what a failed command or task said, or its exit status when it said nothing."""
     lines = [line.strip() for line in str(result.get("stderr") or result.get("msg") or "").splitlines() if line.strip()]
     return lines[-1] if lines else f"exit status {result.get('rc')}"
 
 
-def cluster_native_role(recorded: Any, agent_installed: Any, eligible: Any, inventory_role: Any, running: Any = ()) -> dict[str, str]:
-    """Decide the role a native host runs now and the role it is to run after this run.
-
-    :param recorded: what the host's role file holds (empty when it has none). :param agent_installed: whether the host has the k3s-agent unit, which a host set up before the role file existed has. :param eligible: whether the host is marked github_runner_cluster_voter_eligible. :param inventory_role: the role the topology gives the host. :param running: the roles whose unit is active now. :returns: ``current`` (``server``, ``agent``, or empty on a first install) and ``desired``. :raises AnsibleFilterError: if the role file holds anything but a role, or the inventory role or a running role is not one.
-
-    The role file is the host's own record of its role and outlives restarts, reboots and runs of the role. A switch writes the new role to it before it stops the old unit, so while exactly one unit runs, that unit's role is the one the host runs now, and a switch that stopped part-way (a member removal that did not finish) is taken up again by the next run. With neither unit running, the file says; a host from before the file existed ran only the agent, so its unit counts as the agent role. For a host marked eligible the inventory gives only the role it starts with: once the file names a role, that role is kept, since whatever moves votes between hosts changes it at run time and a run of the role must not undo that. For any other host the inventory decides.
-    """
-    text = "" if recorded is None else str(recorded).strip()
-    if text and text not in NODE_ROLES:
-        raise AnsibleFilterError(f"The role file holds {text!r}; it must be 'server' or 'agent'. Write the role the host runs into it, or remove it on a host that runs neither unit.")
-    if inventory_role not in NODE_ROLES:
-        raise AnsibleFilterError(f"{inventory_role!r} is not a node role (server or agent).")
-    active = [str(role) for role in running]
-    if any(role not in NODE_ROLES for role in active):
-        raise AnsibleFilterError(f"{active!r} are not node roles (server or agent).")
-    if len(active) == 1:
-        current = active[0]
-    else:
-        current = text or ("agent" if bool(agent_installed) else "")
-    desired = text if text and bool(eligible) else str(inventory_role)
-    return {"current": current, "desired": desired}
-
-
 def cluster_native_cluster_init(reads: Any, this_host: Any) -> dict[str, Any]:
     """Say whether a native bootstrap host with no datastore of its own may start a new cluster: only when no other server of the inventory runs one.
 
@@ -174,7 +148,6 @@ def filters(self) -> dict[str, Any]:
             "cluster_native_k3s_release": cluster_native_k3s_release,
             "cluster_native_tailscale_package_version": cluster_native_tailscale_package_version,
             "cluster_native_tailnet_login": cluster_native_tailnet_login,
-            "cluster_native_role": cluster_native_role,
             "cluster_native_cluster_init": cluster_native_cluster_init,
             "cluster_native_voter_blockers": cluster_native_voter_blockers,
         }

plugins/filter/cluster_role.py

@@ -0,0 +1,42 @@
+"""Filters the github_runner_cluster role uses to decide which role, server or agent, a host runs, whether it runs natively or under Docker Compose."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ansible.errors import AnsibleFilterError
+
+NODE_ROLES = ("server", "agent")
+
+
+def cluster_host_role(recorded: Any, legacy_agent: Any, eligible: Any, inventory_role: Any, running: Any = ()) -> dict[str, str]:
+    """Decide the role a host runs now and the role it is to run after this run.
+
+    :param recorded: what the host's role file holds (empty when it has none). :param legacy_agent: whether the host runs, or is set up to run, the agent from before the role file existed, which counts as the agent role when the file is absent and nothing runs (a native host's k3s-agent unit; never for a Docker host, whose role file is new). :param eligible: whether the host is marked github_runner_cluster_voter_eligible. :param inventory_role: the role the topology gives the host. :param running: the roles whose unit or container runs now. :returns: ``current`` (``server``, ``agent``, or empty on a first install) and ``desired``. :raises AnsibleFilterError: if the role file holds anything but a role, or the inventory role or a running role is not one.
+
+    The role file is the host's own record of its role and outlives restarts, reboots and runs of the role. A switch writes the new role to it before it stops the old unit or container, so while exactly one of them runs, its role is the one the host runs now, and a switch that stopped part-way (a member removal that did not finish) is taken up again by the next run. With neither running, the file says. For a host marked eligible the inventory gives only the role it starts with: once the file names a role, that role is kept, since whatever moves votes between hosts changes it at run time and a run of the role must not undo that. For any other host the inventory decides.
+    """
+    text = "" if recorded is None else str(recorded).strip()
+    if text and text not in NODE_ROLES:
+        raise AnsibleFilterError(f"The role file holds {text!r}; it must be 'server' or 'agent'. Write the role the host runs into it, or remove it on a host that runs neither.")
+    if inventory_role not in NODE_ROLES:
+        raise AnsibleFilterError(f"{inventory_role!r} is not a node role (server or agent).")
+    active = [str(role) for role in running]
+    if any(role not in NODE_ROLES for role in active):
+        raise AnsibleFilterError(f"{active!r} are not node roles (server or agent).")
+    if len(active) == 1:
+        current = active[0]
+    else:
+        current = text or ("agent" if bool(legacy_agent) else "")
+    desired = text if text and bool(eligible) else str(inventory_role)
+    return {"current": current, "desired": desired}
+
+
+class FilterModule:
+    """Registers the role filters with Ansible."""
+
+    def filters(self) -> dict[str, Any]:
+        """Return the filters this plugin provides."""
+        return {
+            "cluster_host_role": cluster_host_role,
+        }