From 17b7dbecbdb758613bab330047c9e948fc6bb722 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:37:04 +0100 Subject: [PATCH] fix(migrate): adopt pre-worker earners so they stay visible behind the proxy The owner migrator only took containers labelled deployed-by=worker, but CashPilot also counts deployed-by=direct as its own. On a host with such earners the fence network was left unlabelled and the earners dropped out of the worker's view. They now count as ours, and each is recreated with the worker label (with its volumes moved when they need it). --- app/owner_migrate.py | 38 +++++++++++++++++++++++++++++-------- tests/test_owner_migrate.py | 33 ++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 8 deletions(-) diff --git a/app/owner_migrate.py b/app/owner_migrate.py index c8a6c6c9..bd40104b 100644 --- a/app/owner_migrate.py +++ b/app/owner_migrate.py @@ -34,6 +34,7 @@ from pathlib import Path from typing import Any +from app.constants import CASHPILOT_DEPLOYED from app.socket_policy import OWNER, OWNER_LABEL logger = logging.getLogger("cashpilot.migrate") @@ -106,6 +107,14 @@ def owned(labels: dict[str, str] | None) -> bool: return (labels or {}).get(OWNER_LABEL) == OWNER +def ours(labels: dict[str, str] | None) -> bool: + """A container CashPilot deployed: by the worker, or by the deploy path + before workers ("direct"), whose earners still run on old installs. The + proxy shows the worker neither kind until it carries the worker label, so + the migration moves both and recreates each "direct" one with it.""" + return (labels or {}).get(OWNER_LABEL) in CASHPILOT_DEPLOYED + + def backup_name(volume: str) -> str: return volume + BACKUP_SUFFIX @@ -185,6 +194,9 @@ def _remove_helper(self) -> None: with contextlib.suppress(NotFound): self.client.containers.get(HELPER_NAME).remove(force=True) + def _ours(self) -> list[Any]: + return [c for c in self.client.containers.list(all=True, filters={"label": OWNER_LABEL}) if ours(c.labels)] + # -- the run ------------------------------------------------------------ def run(self, compose_volumes: list[tuple[str, str]]) -> int: @@ -289,7 +301,7 @@ def migrate_networks(self) -> None: names = {r["name"] for r in self.state.all("networks") if not r.get("done")} for net in self.client.networks.list(filters={"label": FENCE_LABEL}): names.add((net.attrs or {}).get("Name") or net.name) - for c in self.client.containers.list(all=True, filters={"label": f"{OWNER_LABEL}={OWNER}"}): + for c in self._ours(): names.update(((c.attrs or {}).get("NetworkSettings") or {}).get("Networks") or {}) for name in sorted(names - BUILTIN_NETWORKS): try: @@ -317,7 +329,7 @@ def migrate_network(self, name: str) -> None: attrs = net.attrs or {} members = [] for c in self._on_network(name): - if not owned(c.labels): + if not ours(c.labels): raise LeftAlone( f"{c.name} is on it and was not deployed by the worker, so the network was left as it " f"was, and the worker cannot put services on it. To move it, disconnect that container " @@ -350,7 +362,7 @@ def migrate_network(self, name: str) -> None: for c in self._on_network(name): if c.name in known: continue - if not owned(c.labels): + if not ours(c.labels): # Leave it as it is, with the members an earlier run took off back on. self._reconnect(net, record) self.state.drop("networks", name) @@ -465,10 +477,12 @@ def _run_plan_logged(self, plan: dict[str, Any]) -> None: self.fail(f"{', '.join(c['Name'].lstrip('/') for c in plan['containers'])}: {exc}") def _new_plans(self) -> list[dict[str, Any]]: - """Groups of worker containers that share unlabelled volumes, one plan each.""" + """Groups of worker containers that share unlabelled volumes, one plan each. + A "direct" container gets a plan even with no such volume: recreating + it is the only way to give it the worker label.""" from docker.errors import NotFound - workers = self.client.containers.list(all=True, filters={"label": f"{OWNER_LABEL}={OWNER}"}) + workers = self._ours() # A stored plan that did not finish already holds these, stopped by it; # a new plan would record them as stopped and bring them back stopped. held = {c["Id"] for p in self.state.all("plans") if not p.get("done") for c in p["containers"]} @@ -488,6 +502,7 @@ def _new_plans(self) -> list[dict[str, Any]]: continue users.setdefault(mount["Name"], set()).add(c.id) attrs[c.id] = c.attrs + legacy = [c for c in workers if c.id not in held and not owned(c.labels)] # Join containers that share a volume: they stop and come back together. groups: list[tuple[set[str], set[str]]] = [] for vol, cids in users.items(): @@ -497,13 +512,17 @@ def _new_plans(self) -> list[dict[str, Any]]: merged[0].update(group[0]) merged[1].update(group[1]) groups.append(merged) + for c in legacy: + if not any(c.id in g[0] for g in groups): + attrs[c.id] = c.attrs + groups.append(({c.id}, set())) groups = self._with_joiners(groups, attrs, held) plans = [] for cids, vols in groups: strangers = sorted( attrs[cid]["Name"].lstrip("/") for cid in cids - if not owned((attrs[cid].get("Config") or {}).get("Labels")) + if not ours((attrs[cid].get("Config") or {}).get("Labels")) ) if strangers: self.warn( @@ -603,7 +622,10 @@ def _current(self, info: dict[str, Any]) -> Any | None: def run_plan(self, plan: dict[str, Any]) -> None: names = [c["Name"].lstrip("/") for c in plan["containers"]] - logger.info("Moving %s onto labelled volumes: %s", ", ".join(names), ", ".join(plan["volumes"])) + if plan["volumes"]: + logger.info("Moving %s onto labelled volumes: %s", ", ".join(names), ", ".join(plan["volumes"])) + else: + logger.info("Recreating %s with %s=%s, as the worker deploys it", ", ".join(names), OWNER_LABEL, OWNER) # A container in another's namespace goes first and comes back last. in_plan = {c["Id"]: c["Name"].lstrip("/") for c in plan["containers"]} joined = [c for c in plan["containers"] if _namespace_of(c) in in_plan] @@ -653,7 +675,7 @@ def run_plan(self, plan: dict[str, Any]) -> None: self._left_for_the_worker(info) plan["done"] = True self.state.put("plans", plan["key"], plan) - logger.info("%s back as before, on labelled volumes.", ", ".join(names)) + logger.info("%s back as before, labelled for the worker.", ", ".join(names)) @staticmethod def _left_for_the_worker(info: dict[str, Any]) -> None: diff --git a/tests/test_owner_migrate.py b/tests/test_owner_migrate.py index 8f72894b..812f2b45 100644 --- a/tests/test_owner_migrate.py +++ b/tests/test_owner_migrate.py @@ -872,3 +872,36 @@ def test_a_proxied_earner_with_its_own_volume_moves_in_the_same_plan_as_its_tunn new_earner = d.containers.get("cashpilot-honeygain-p1") assert new_earner.attrs["HostConfig"]["NetworkMode"] == f"container:{new_tunnel.id}" assert d.data["honeygain-data"] == {"device": "id"} + + +def test_earners_the_pre_worker_deploy_path_made_are_adopted_with_the_fence(install, tmp_path): + """deployed-by=direct counts as CashPilot's (constants.CASHPILOT_DEPLOYED). + Left alone, the fence would stay unlabelled and the earners would vanish + from the worker behind the proxy.""" + d, myst, _ = install + direct = {"cashpilot.managed": "true", "cashpilot.service": "honeygain", om.OWNER_LABEL: "direct"} + old = d.add(Container(d, "cashpilot-honeygain", direct, networks=("cashpilot-isolated",))) + d.volume("earnapp-data", {"uuid": "sdk-node-1"}) + old_vol = d.add( + Container( + d, + "cashpilot-earnapp", + {**direct, "cashpilot.service": "earnapp"}, + [("earnapp-data", "/etc/earnapp")], + networks=("cashpilot-isolated",), + running=False, + ) + ) + m = FakeMigrator(d, om.State(tmp_path)) + assert m.run([]) == 0 and m.warnings == 0 + assert om.owned(d.nets["cashpilot-isolated"].attrs["Labels"]) + hg = d.containers.get("cashpilot-honeygain") + assert hg.id != old.id and hg.status == "running" + assert hg.labels == {**direct, **OWNER, om.RECREATED_LABEL: old.id} + assert hg.networks() == ["cashpilot-isolated"] + ea = d.containers.get("cashpilot-earnapp") + assert ea.id != old_vol.id and ea.status == "exited" and om.owned(ea.labels) + assert om.owned(d.vols["earnapp-data"].attrs["Labels"]) and d.data["earnapp-data"] == {"uuid": "sdk-node-1"} + again = FakeMigrator(d, om.State(tmp_path)) + assert again.run([]) == 0 and again.copies == [] + assert d.containers.get("cashpilot-honeygain").id == hg.id