Node tags, and --tag on the fleet commands

README.md

@@ -320,6 +320,22 @@ node that is never confirmed puts itself back without anybody driving there. A
 fleet-wide apply refuses to give that window up: it is the one place warden will
 not let it be left out.
 
+**And not always all of them.** A node says what it is, in its own
+configuration — `tags = ["web", "eu-west"]` — and it travels in the
+announcement it already sends, so the hub knows without anybody keeping a
+second list:
+
+```sh
+warden firewall apply --fleet --tag web --rollback 120
+warden update --fleet --tag eu-west
+warden ls --all --tag web
+```
+
+Kept on the machine rather than in groups on the hub: two places that know what
+a machine is disagree the first time one is rebuilt. A tag nothing carries is
+refused and told which ones exist — an apply that quietly touched no machines
+is worse than one that would not run.
+
 Reading the rules is what a token already allows. **Changing** them needs
 `allow_remote_firewall` set on the machine being asked, and it is off out of
 the box — a warden that will change its own firewall on request, listens beyond

src/warden/api.py

@@ -31,7 +31,7 @@
 from warden.core.here import Here
 from warden.core.rounds import Rounds
 from warden.core.store import RuleStore, Snapshots, Store
-from warden.errors import NotPermittedError, WardenError
+from warden.errors import NotPermittedError, UnknownTagError, WardenError
 from warden.firewall import catalogue, guard, link
 from warden.firewall import model as firewall
 from warden.firewall.backends import base
@@ -424,19 +424,39 @@ def forget(name: str, fleet: FleetDep) -> None:
         prefix="/v1/fleet", tags=["fleet"], dependencies=[Depends(known_caller)]
     )
 
+    def _picked(fleet: Fleet, tag: str | None) -> tuple[list[Node], bool]:
+        """The nodes a tag picks out, and whether this warden is one of them.
+
+        Nothing matching is a refusal rather than an empty answer: an apply
+        that quietly touched no machines is worse than one that would not run.
+        """
+        if tag is None:
+            return fleet.nodes(), True
+        chosen = fleet.tagged(tag)
+        mine = tag in settings.tags
+        if not chosen and not mine:
+            known = sorted(set(fleet.tags()) | set(settings.tags))
+            raise UnknownTagError(
+                f"no warden here is tagged {tag!r}"
+                + (f" - there is {', '.join(known)}" if known else " - none of them is tagged")
+            )
+        return chosen, mine
+
     @fleet_view.get("/services", summary="Everything the whole fleet holds")
     async def fleet_services(
         manager: Manager,
         fleet: FleetDep,
         project: str | None = None,
         kind: str | None = None,
+        tag: str | None = None,
     ) -> FleetServices:
+        chosen, mine = _picked(fleet, tag)
         async with aggregate.client(settings.cluster_token) as http:
             return await aggregate.gather_services(
                 http,
-                fleet.nodes(),
+                chosen,
                 here=settings.node,
-                local=manager.list(project=project, kind=kind),
+                local=manager.list(project=project, kind=kind) if mine else [],
                 project=project,
                 kind=kind,
             )
@@ -885,6 +905,7 @@ async def firewall_everywhere(
         fleet: FleetDep,
         rollback: int | None = None,
         snapshot: int | None = None,
+        tag: str | None = None,
     ) -> FleetFirewallResult:
         """Every node does it to its own firewall, and answers for itself.
 
@@ -907,26 +928,29 @@ async def firewall_everywhere(
                 "to open them again",
             )
 
-        try:
-            here = FirewallResult(
-                node=settings.node,
-                url=settings.advertise_url,
-                ok=True,
-                detail=aggregate._said(
-                    what, _here(what, rules, snapshots, rollback=seconds, snapshot=snapshot)
-                ),
-            )
-        except WardenError as exc:
-            here = FirewallResult(
-                node=settings.node, url=settings.advertise_url, ok=False, detail=exc.message
-            )
+        chosen, mine = _picked(fleet, tag)
+        here = None
+        if mine:
+            try:
+                here = FirewallResult(
+                    node=settings.node,
+                    url=settings.advertise_url,
+                    ok=True,
+                    detail=aggregate._said(
+                        what, _here(what, rules, snapshots, rollback=seconds, snapshot=snapshot)
+                    ),
+                )
+            except WardenError as exc:
+                here = FirewallResult(
+                    node=settings.node, url=settings.advertise_url, ok=False, detail=exc.message
+                )
 
         params: dict[str, object] = {"rollback": seconds} if what == "apply" else {}
         if snapshot is not None:
             params["snapshot"] = snapshot
         async with aggregate.client(settings.cluster_token, timeout=30.0) as http:
             return await aggregate.firewall_fleet(
-                http, fleet.nodes(), what, params=params, here=here
+                http, chosen, what, params=params, here=here
             )
 
 
@@ -948,20 +972,23 @@ def update_self() -> dict[str, str]:
         return {"detail": updates.apply(settings)}
 
     @fleet_view.post("/update", summary="Ask every warden in the fleet to update itself")
-    async def update_everyone(fleet: FleetDep) -> FleetUpdate:
-        try:
-            here = UpdateResult(
-                node=settings.node,
-                url=settings.advertise_url,
-                ok=True,
-                detail=updates.apply(settings),
-            )
-        except WardenError as exc:
-            here = UpdateResult(
-                node=settings.node, url=settings.advertise_url, ok=False, detail=exc.message
-            )
+    async def update_everyone(fleet: FleetDep, tag: str | None = None) -> FleetUpdate:
+        chosen, mine = _picked(fleet, tag)
+        here = None
+        if mine:
+            try:
+                here = UpdateResult(
+                    node=settings.node,
+                    url=settings.advertise_url,
+                    ok=True,
+                    detail=updates.apply(settings),
+                )
+            except WardenError as exc:
+                here = UpdateResult(
+                    node=settings.node, url=settings.advertise_url, ok=False, detail=exc.message
+                )
         async with aggregate.client(settings.cluster_token, timeout=300.0) as http:
-            return await aggregate.update_fleet(http, fleet.nodes(), here=here)
+            return await aggregate.update_fleet(http, chosen, here=here)
 
     def may_change_the_firewall() -> None:
         """The switch that has to be thrown before a caller may touch the rules.

src/warden/cli/commands/firewall.py

@@ -602,15 +602,18 @@ def _across_the_fleet(
     what: str,
     *,
     rollback: int | None = None,
+    tag: str | None = None,
     as_json: bool,
 ) -> None:
     """One line per node, whichever way each of them went."""
     if what == "apply":
-        found = _asking(url, token, lambda client: client.firewall_apply_fleet(rollback=rollback))
+        found = _asking(
+            url, token, lambda client: client.firewall_apply_fleet(rollback=rollback, tag=tag)
+        )
     elif what == "confirm":
-        found = _asking(url, token, lambda client: client.firewall_confirm_fleet())
+        found = _asking(url, token, lambda client: client.firewall_confirm_fleet(tag=tag))
     else:
-        found = _asking(url, token, lambda client: client.firewall_restore_fleet())
+        found = _asking(url, token, lambda client: client.firewall_restore_fleet(tag=tag))
 
     if as_json:
         _dump(found.model_dump(mode="json"))
@@ -653,6 +656,9 @@ def firewall_apply(
     fleet: Annotated[
         bool, typer.Option("--fleet", help="Every warden in the fleet, each on its own.")
     ] = False,
+    tagged: Annotated[
+        str | None, typer.Option("--tag", help="Only the nodes carrying this tag.")
+    ] = None,
     at: Annotated[
         str | None, typer.Option("--node", help="One node in the fleet, through the hub.")
     ] = None,
@@ -670,7 +676,7 @@ def firewall_apply(
     at once. `--node build-01` asks exactly one.
     """
     if fleet:
-        _across_the_fleet(url, token, "apply", rollback=rollback, as_json=as_json)
+        _across_the_fleet(url, token, "apply", rollback=rollback, tag=tagged, as_json=as_json)
         return
     if at:
         said = _asking(url, token, lambda client: client.firewall_apply_on(at, rollback=rollback))
@@ -863,6 +869,9 @@ def firewall_confirm(
     fleet: Annotated[
         bool, typer.Option("--fleet", help="Every warden in the fleet, each on its own.")
     ] = False,
+    tagged: Annotated[
+        str | None, typer.Option("--tag", help="Only the nodes carrying this tag.")
+    ] = None,
     at: Annotated[
         str | None, typer.Option("--node", help="One node in the fleet, through the hub.")
     ] = None,
@@ -876,7 +885,7 @@ def firewall_confirm(
     back when its own window runs out.
     """
     if fleet:
-        _across_the_fleet(url, token, "confirm", as_json=as_json)
+        _across_the_fleet(url, token, "confirm", tag=tagged, as_json=as_json)
         return
     if at:
         said = _asking(url, token, lambda client: client.firewall_confirm_on(at))

src/warden/cli/commands/fleet.py

@@ -38,6 +38,9 @@ def update(
     fleet: Annotated[
         bool, typer.Option("--fleet", help="Ask every warden in the fleet to update itself.")
     ] = False,
+    tagged: Annotated[
+        str | None, typer.Option("--tag", help="Only the nodes carrying this tag.")
+    ] = None,
     yes: Annotated[bool, typer.Option("--yes", "-y", help="Do not ask first.")] = False,
     url: UrlOption = None,
     token: TokenOption = None,
@@ -56,13 +59,14 @@ def update(
                 _show_update(client.update_status(), as_json=as_json, here=url is None)
                 return
 
-            what = "every warden in the fleet" if fleet else f"the warden at {client.url}"
+            what = f"every warden tagged {tagged}" if tagged else "every warden in the fleet"
+            what = what if fleet else f"the warden at {client.url}"
             if not yes and not typer.confirm(f"Update {what}?"):
                 console.print("left alone", style=theme.BONE_DIM)
                 return
 
             if fleet:
-                _show_fleet_update(client.update_fleet(), as_json=as_json)
+                _show_fleet_update(client.update_fleet(tag=tagged), as_json=as_json)
             else:
                 detail = client.update_self()
                 _dump({"detail": detail}) if as_json else console.print(detail)
@@ -174,14 +178,15 @@ def nodes(
         return
 
     table = Table(box=None, pad_edge=False, header_style=f"bold {theme.BONE_DIM}")
-    for column in ("NODE", "URL", "POOL", "VERSION", "STATUS", "LAST SEEN"):
+    for column in ("NODE", "URL", "POOL", "VERSION", "TAGS", "STATUS", "LAST SEEN"):
         table.add_column(column)
     for node in known:
         table.add_row(
             node.name,
             Text(node.url, style=theme.BONE_DIM),
             Text(node.pool, style=theme.GLOW),
             Text(node.version, style=theme.BONE_DIM),
+            Text(", ".join(node.tags) if node.tags else "-", style=theme.BONE_DIM),
             Text(node.status, style=NODE_COLOURS[node.status]),
             Text(theme.age(node.last_seen), style=theme.BONE_DIM),
         )

src/warden/cli/commands/registry.py

@@ -219,12 +219,17 @@ def list_services(
     every: Annotated[
         bool, typer.Option("--all", help="Ask every warden in the fleet, not just this one.")
     ] = False,
+    tagged: Annotated[
+        str | None, typer.Option("--tag", help="Only the nodes carrying this tag. With --all.")
+    ] = None,
     url: UrlOption = None,
     token: TokenOption = None,
     as_json: JsonOption = False,
 ) -> None:
     """List every registered service."""
     holders = holders or stale
+    if tagged and not every:
+        raise _fail(WardenError("a tag picks out nodes, so --tag goes with --all"))
     if holders and every:
         raise _fail(
             WardenError("a holder can only be checked on the machine it runs on, so --holders "
@@ -233,7 +238,7 @@ def list_services(
     with shared._client(url, token) as client:
         try:
             fleet = (
-                client.fleet_services(project=project, kind=kind)
+                client.fleet_services(project=project, kind=kind, tag=tagged)
                 if every
                 else None
             )

src/warden/client.py

@@ -49,6 +49,12 @@
 }
 
 
+def _some(**params: object) -> dict[str, object] | None:
+    """The parameters that were actually given, or nothing at all."""
+    said = {key: value for key, value in params.items() if value is not None}
+    return said or None
+
+
 def resolve_url(url: str | None = None) -> str:
     return (url or os.environ.get("WARDEN_URL") or DEFAULT_URL).rstrip("/")
 
@@ -277,10 +283,14 @@ def forget(self, name: str) -> None:
         self._request("DELETE", f"/v1/nodes/{name}")
 
     def fleet_services(
-        self, *, project: str | None = None, kind: str | None = None
+        self, *, project: str | None = None, kind: str | None = None, tag: str | None = None
     ) -> FleetServices:
         """Everything the whole fleet holds, and the nodes that did not answer."""
-        params = {key: value for key, value in (("project", project), ("kind", kind)) if value}
+        params = {
+            key: value
+            for key, value in (("project", project), ("kind", kind), ("tag", tag))
+            if value
+        }
         return FleetServices.model_validate(
             self._request("GET", "/v1/fleet/services", params=params)
         )
@@ -470,29 +480,39 @@ def firewall_open_everywhere(
             self._request("POST", "/v1/fleet/firewall/open", json=body, timeout=60.0)
         )
 
-    def firewall_apply_fleet(self, *, rollback: int | None = None) -> FleetFirewallResult:
+    def firewall_apply_fleet(
+        self, *, rollback: int | None = None, tag: str | None = None
+    ) -> FleetFirewallResult:
         """Make every node's rules true, each with its own rollback armed.
 
         A fleet-wide apply keeps its window: pass `rollback=0` and it is
         refused. Nothing is kept until `firewall_confirm_fleet`, and a node
         that is never confirmed puts itself back on its own.
         """
-        params = {"rollback": rollback} if rollback is not None else None
+        params = _some(rollback=rollback, tag=tag)
         return FleetFirewallResult.model_validate(
             self._request("POST", "/v1/fleet/firewall/apply", params=params, timeout=60.0)
         )
 
-    def firewall_confirm_fleet(self) -> FleetFirewallResult:
+    def firewall_confirm_fleet(self, *, tag: str | None = None) -> FleetFirewallResult:
         """Keep what every node applied."""
         return FleetFirewallResult.model_validate(
-            self._request("POST", "/v1/fleet/firewall/confirm", timeout=60.0)
+            self._request(
+                "POST", "/v1/fleet/firewall/confirm", params=_some(tag=tag), timeout=60.0
+            )
         )
 
-    def firewall_restore_fleet(self, snapshot: int | None = None) -> FleetFirewallResult:
+    def firewall_restore_fleet(
+        self, snapshot: int | None = None, *, tag: str | None = None
+    ) -> FleetFirewallResult:
         """Put every node back to a snapshot, whether or not one was waiting."""
-        params = {"snapshot": snapshot} if snapshot is not None else None
         return FleetFirewallResult.model_validate(
-            self._request("POST", "/v1/fleet/firewall/restore", params=params, timeout=60.0)
+            self._request(
+                "POST",
+                "/v1/fleet/firewall/restore",
+                params=_some(snapshot=snapshot, tag=tag),
+                timeout=60.0,
+            )
         )
 
     def update_status(self) -> UpdateStatus:
@@ -503,10 +523,10 @@ def update_self(self) -> str:
         """Ask that warden to run its own update command."""
         return str(self._request("POST", "/v1/update")["detail"])
 
-    def update_fleet(self) -> FleetUpdate:
+    def update_fleet(self, *, tag: str | None = None) -> FleetUpdate:
         """Ask every warden in the fleet to update itself."""
         return FleetUpdate.model_validate(
-            self._request("POST", "/v1/fleet/update", timeout=310.0)
+            self._request("POST", "/v1/fleet/update", params=_some(tag=tag), timeout=310.0)
         )
 
     def fleet_lookup(self, node: str, name: str) -> FleetRegistration:

src/warden/core/config.py

@@ -326,6 +326,11 @@ class Settings(BaseSettings):
     upstream: str | None = None
     cluster_token: str | None = None
     node_ttl: int = Field(default=90, ge=10, le=86_400)
+    # `web`, `eu-west`, `db` - what this machine is, travelling in the
+    # announcement it already sends. On the node rather than in a list on the
+    # hub: two places that know what a machine is disagree the first time one
+    # is rebuilt, and the machine is the one that knows.
+    tags: WordSet = Field(default_factory=set)
     require_https: bool = False
 
     @classmethod