From 2f8f88d8d988fd1e5b9806860952ada21a167be2 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 11 Nov 2026 00:18:39 -0500 Subject: [PATCH 1/5] feat(rollouts): verify canary restore provenance and pinned modules Signed-off-by: Baylem --- deploy/rollout/canary.py | 52 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 deploy/rollout/canary.py diff --git a/deploy/rollout/canary.py b/deploy/rollout/canary.py new file mode 100644 index 000000000..89235adcc --- /dev/null +++ b/deploy/rollout/canary.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""Validate the installed bundle and the provenance of a restored rollout canary.""" + +import re + + +class Halt(RuntimeError): + """The fleet must stay at its current revision.""" + + +def require(ok, reason): + if not ok: + raise Halt(reason) + + +def validate_module(module, digest): + spec, status = module["spec"], module.get("status", {}) + require(re.fullmatch(r"sha256:[0-9a-f]{64}", digest), "pin an OCI module digest") + require(status.get("phase") == "Ready" and spec.get("version") + and spec["version"] == status.get("appliedVersion") + and status.get("observedGeneration") == module["metadata"]["generation"], + "the pinned module version has not finished reconciling") + require(digest == spec.get("digest") == status.get("appliedDigest") == status.get("verifiedDigest") + and status.get("verifyPolicy") in ("keyed", "keyless"), + "the installed module digest lacks a matching signature verification") + + +def validate_restore(backup, restore, source, clone): + """Require a successful all-data restore into a different server of this fleet.""" + source_uid, clone_uid = source["metadata"]["uid"], clone["metadata"]["uid"] + namespace = source["metadata"]["namespace"] + require(source_uid != clone_uid and all(obj["metadata"]["namespace"] == namespace + for obj in (clone, backup, restore)), "the source and canary must be distinct in one namespace") + status = backup.get("status", {}) + manifest = status.get("captureManifest") or {} + require(status.get("phase") == "Succeeded" and status.get("snapshotID") + and manifest.get("serverUID") == source_uid and manifest.get("formatVersion") == 2 + and manifest.get("volumePolicy") == "all-data" + and manifest.get("quiesceOutcome") == "already-stopped", "a stopped all-data backup is required") + volumes = manifest.get("volumes", []) + require(any(v.get("included") and v.get("classification") == "data" for v in volumes) + and all(v.get("included") or v.get("classification") == "rebuildable" for v in volumes), + "the capture omits persistent data or unclassified content") + rs = restore.get("status", {}) + plan = rs.get("recoveryPlan") or {} + require(rs.get("phase") == "Succeeded" and rs.get("snapshotID") == status["snapshotID"] + and plan.get("backupUID") == backup["metadata"]["uid"] and plan.get("targetUID") == clone_uid + and restore["spec"]["serverRef"]["name"] == clone["metadata"]["name"] + and restore["spec"]["backupRef"]["name"] == backup["metadata"]["name"] + and manifest.get("templateRevision") and plan.get("templateRevision") == manifest["templateRevision"], + "the canary has no matching completed restore") + return {v["claimName"]: v["claimUID"] for v in volumes if v.get("included")} From b2068126f4254126ca9422dde08395bb957387a7 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 11 Nov 2026 00:42:18 -0500 Subject: [PATCH 2/5] feat(rollouts): gate fleet adoption on restored joins and rollback rehearsals Signed-off-by: Baylem --- .github/workflows/rollout-canary.yaml | 21 ++ deploy/rollout/canary.py | 52 ---- docs/template-rollouts.md | 79 ++++- hack/rollout/canary.py | 421 ++++++++++++++++++++++++++ hack/rollout/test_canary.py | 188 ++++++++++++ operator/specs.md | 5 + 6 files changed, 712 insertions(+), 54 deletions(-) create mode 100644 .github/workflows/rollout-canary.yaml delete mode 100644 deploy/rollout/canary.py create mode 100644 hack/rollout/canary.py create mode 100644 hack/rollout/test_canary.py diff --git a/.github/workflows/rollout-canary.yaml b/.github/workflows/rollout-canary.yaml new file mode 100644 index 000000000..a8f9a27e8 --- /dev/null +++ b/.github/workflows/rollout-canary.yaml @@ -0,0 +1,21 @@ +name: Rollout canary qualification +on: + pull_request: + paths: + - "hack/rollout/**" + - ".github/workflows/rollout-canary.yaml" + - "test/e2e/cmd/join-once/**" + - "operator/internal/controller/restore_recovery.go" + - "operator/internal/controller/gameserver_template_revision.go" +permissions: + contents: read +jobs: + admission-and-recovery: + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - name: Exercise fleet halt, claim identity and rollback decisions + run: python3 -m unittest discover -s hack/rollout -p 'test_*.py' -v diff --git a/deploy/rollout/canary.py b/deploy/rollout/canary.py deleted file mode 100644 index 89235adcc..000000000 --- a/deploy/rollout/canary.py +++ /dev/null @@ -1,52 +0,0 @@ -#!/usr/bin/env python3 -"""Validate the installed bundle and the provenance of a restored rollout canary.""" - -import re - - -class Halt(RuntimeError): - """The fleet must stay at its current revision.""" - - -def require(ok, reason): - if not ok: - raise Halt(reason) - - -def validate_module(module, digest): - spec, status = module["spec"], module.get("status", {}) - require(re.fullmatch(r"sha256:[0-9a-f]{64}", digest), "pin an OCI module digest") - require(status.get("phase") == "Ready" and spec.get("version") - and spec["version"] == status.get("appliedVersion") - and status.get("observedGeneration") == module["metadata"]["generation"], - "the pinned module version has not finished reconciling") - require(digest == spec.get("digest") == status.get("appliedDigest") == status.get("verifiedDigest") - and status.get("verifyPolicy") in ("keyed", "keyless"), - "the installed module digest lacks a matching signature verification") - - -def validate_restore(backup, restore, source, clone): - """Require a successful all-data restore into a different server of this fleet.""" - source_uid, clone_uid = source["metadata"]["uid"], clone["metadata"]["uid"] - namespace = source["metadata"]["namespace"] - require(source_uid != clone_uid and all(obj["metadata"]["namespace"] == namespace - for obj in (clone, backup, restore)), "the source and canary must be distinct in one namespace") - status = backup.get("status", {}) - manifest = status.get("captureManifest") or {} - require(status.get("phase") == "Succeeded" and status.get("snapshotID") - and manifest.get("serverUID") == source_uid and manifest.get("formatVersion") == 2 - and manifest.get("volumePolicy") == "all-data" - and manifest.get("quiesceOutcome") == "already-stopped", "a stopped all-data backup is required") - volumes = manifest.get("volumes", []) - require(any(v.get("included") and v.get("classification") == "data" for v in volumes) - and all(v.get("included") or v.get("classification") == "rebuildable" for v in volumes), - "the capture omits persistent data or unclassified content") - rs = restore.get("status", {}) - plan = rs.get("recoveryPlan") or {} - require(rs.get("phase") == "Succeeded" and rs.get("snapshotID") == status["snapshotID"] - and plan.get("backupUID") == backup["metadata"]["uid"] and plan.get("targetUID") == clone_uid - and restore["spec"]["serverRef"]["name"] == clone["metadata"]["name"] - and restore["spec"]["backupRef"]["name"] == backup["metadata"]["name"] - and manifest.get("templateRevision") and plan.get("templateRevision") == manifest["templateRevision"], - "the canary has no matching completed restore") - return {v["claimName"]: v["claimUID"] for v in volumes if v.get("included")} diff --git a/docs/template-rollouts.md b/docs/template-rollouts.md index 5851a4cf5..7b97ddf81 100644 --- a/docs/template-rollouts.md +++ b/docs/template-rollouts.md @@ -62,8 +62,7 @@ kubectl patch gameserver my-server -n gameplane-games --type=merge \ ``` An old request never approves a newer digest. Requests can be made one server at -a time or by an administrator's batch process. There is no built-in canary, -maintenance-window or fleet concurrency scheduler. Once admitted, an update +a time or through the restored-world fleet procedure below. Once admitted, an update finishes using that exact snapshot even if another catalog update arrives or the policy changes during shutdown. @@ -131,3 +130,79 @@ content as reconstruction prerequisites. CSI restores copy the source server's effective template at restore planning time, not necessarily its template at backup time. Select a compatible Module version and `OnRequest` policy before recovery if automatic adoption would be unsafe. + +## Restored-world fleet procedure + +Use a dedicated Module installation for one fleet. Put **every** server +referencing this cluster-scoped template under `OnRequest` before publishing +a new bundle. Keep both bundle and game image in the registry. The admin +runner needs Python 3, kubectl, cosign, a trusted *game image* public key and +a join client for the game. Build the Minecraft Java/Terraria client from +the platform checkout with `go build -o join-once ./test/e2e/cmd/join-once`. +For an authenticated or proprietary game, supply a trusted entitled client +that implements the same CLI and returns `VERDICTPASSJOINEDevidence`. +The bundled Minecraft client expects offline-mode test servers. + +Create a stopped, all-data Restic backup of a representative server. Create +and start a **separate** GameServer with the same template and version, restore +that backup into it, and wait for the Restore to succeed. Give each included +data or extra volume a persistent game marker. For excluded rebuildable mod +files, verify the reconstructed content or its expected absence on the target. +The runner compares the PVC UIDs in the Restore plan with the live clone, +records all owned claims around every adoption and compares marker and game +save command output exactly. Use a trusted plan with argument arrays that +check the actual data in each listed claim. A server's checks must account for +every retained claim. Use the server name in each real claim name. + +An example plan for three Minecraft servers, each with only its primary PVC: + +```json +{ + "namespace": "games", "module": "survival-fleet", "source": "survival-a", + "backup": "pre-upgrade", "restore": "clone-from-pre-upgrade", + "rollback": "restore-required", "game": "minecraft-java", "port": 25565, + "baseline": {"revision": "<64-hex-template-revision>", "bundle": "sha256:<64-hex>", "image": "registry.example/game@sha256:<64-hex>"}, + "candidate": {"revision": "<64-hex-template-revision>", "bundle": "sha256:<64-hex>", "image": "registry.example/game@sha256:<64-hex>"}, + "canary": {"name": "survival-clone", "checks": [{"claim": "survival-clone-data", "argv": ["cat", "/data/fleet-marker"], "expected": "world-a"}]}, + "targets": [ + {"name": "survival-a", "checks": [{"claim": "survival-a-data", "argv": ["cat", "/data/fleet-marker"], "expected": "world-a"}]}, + {"name": "survival-b", "checks": [{"claim": "survival-b-data", "argv": ["cat", "/data/fleet-marker"], "expected": "world-b"}]} + ], + "save": {"argv": ["rcon-cli", "save-all flush"], "expected": "Saving the game (this may take a moment!)Saved the game"} +} +``` + +Use the exact positive acknowledgment for your qualified game version and +locale. The runner follows the server order in the target list. + +```sh +opts=(fleet.json --context games-production --image-key keys/game.pub --join ./join-once --timeout 900) +python3 hack/rollout/canary.py preflight "${opts[@]}" +# Install the signed, digest-pinned candidate Module through GitOps; await Ready. +python3 hack/rollout/canary.py canary "${opts[@]}" +# Reinstall the signed, digest-pinned baseline Module; await Ready. +python3 hack/rollout/canary.py rollback "${opts[@]}" +# Restore the exact signed candidate Module; await Ready. +python3 hack/rollout/canary.py promote "${opts[@]}" +``` + +Preflight verifies the installed baseline and full inventory across namespaces. +The canary must boot, join and save on its restored old data, adopt the exact +candidate revision, then join, validate markers and save again. A compatible +rollback must boot the baseline on the migrated save. For `restore-required`, +the runner requests a suspended baseline and waits for its pod to disappear +before restoring the original snapshot and starting it. The +production targets stay on the baseline throughout this rehearsal. Promotion +requalifies the candidate on the clone and admits one fleet member at a time. +Each member must retain its PVCs and pass the same join, marker and save checks. + +Progress, active target, PVC UIDs, image, revision and individual receipts +live in a namespace ConfigMap named `gp-rollout-`. +Resource-version updates serialize runs independently of the GameServer owner. +After a failure, inspect the journal and target before retrying the same phase +with the same plan. After a runner crash, verify that process has exited before +supplying its exact recorded token with `--recover-owner`. Archive the completed +journal before deleting it to start a different plan. Restrict this runner to +trusted release operators: it reads cluster-wide GameServers and module state, +reads namespaced pods/workloads/PVCs, patches named GameServers, creates clone +Restores and journal ConfigMaps, and uses namespaced pod exec and port-forward. diff --git a/hack/rollout/canary.py b/hack/rollout/canary.py new file mode 100644 index 000000000..59eadceb4 --- /dev/null +++ b/hack/rollout/canary.py @@ -0,0 +1,421 @@ +#!/usr/bin/env python3 +"""Admit one GameServer at a time after a restored canary and rollback rehearsal.""" + +import argparse +import hashlib +import json +import re +import subprocess +import tempfile +import time +import uuid +from pathlib import Path + + +class Halt(RuntimeError): + """The fleet must stay at its current revision.""" + + +def require(ok, reason): + if not ok: + raise Halt(reason) + + +def validate_module(module, digest): + spec, status = module["spec"], module.get("status", {}) + require(re.fullmatch(r"sha256:[0-9a-f]{64}", digest), "pin an OCI module digest") + require(status.get("phase") == "Ready" and spec.get("version") + and spec["version"] == status.get("appliedVersion") + and status.get("observedGeneration") == module["metadata"]["generation"], + "the pinned module version has not finished reconciling") + require(digest == spec.get("digest") == status.get("appliedDigest") == status.get("verifiedDigest") + and status.get("verifyPolicy") in ("keyed", "keyless"), + "the installed module digest lacks a matching signature verification") + + +def validate_restore(backup, restore, source, clone): + """Require a successful all-data restore into a different server of this fleet.""" + source_uid, clone_uid = source["metadata"]["uid"], clone["metadata"]["uid"] + namespace = source["metadata"]["namespace"] + require(source_uid != clone_uid and all(obj["metadata"]["namespace"] == namespace + for obj in (clone, backup, restore)), "the source and canary must be distinct in one namespace") + status = backup.get("status", {}) + manifest = status.get("captureManifest") or {} + require(status.get("phase") == "Succeeded" and status.get("snapshotID") + and manifest.get("serverUID") == source_uid and manifest.get("formatVersion") == 2 + and manifest.get("volumePolicy") == "all-data" + and manifest.get("quiesceOutcome") == "already-stopped", "a stopped all-data backup is required") + volumes = manifest.get("volumes", []) + require(any(v.get("included") and v.get("classification") == "data" for v in volumes) + and all(v.get("included") or v.get("classification") == "rebuildable" for v in volumes), + "the capture omits persistent data or unclassified content") + rs = restore.get("status", {}) + plan = rs.get("recoveryPlan") or {} + require(rs.get("phase") == "Succeeded" and rs.get("snapshotID") == status["snapshotID"] + and plan.get("backupUID") == backup["metadata"]["uid"] and plan.get("targetUID") == clone_uid + and restore["spec"]["serverRef"]["name"] == clone["metadata"]["name"] + and restore["spec"]["backupRef"]["name"] == backup["metadata"]["name"] + and manifest.get("templateRevision") and plan.get("templateRevision") == manifest["templateRevision"], + "the canary has no matching completed restore") + require(backup["spec"]["serverRef"]["name"] == source["metadata"]["name"] + and all(v.get("claimUID") for v in volumes if v["included"]) + and plan.get("manifestDigest"), "missing backup or volume provenance") + target_claims = {v["claimName"]: v["claimUID"] for v in plan.get("volumes", []) if v["included"]} + require(target_claims and all(target_claims.values()), "missing restored claim identities") + return {"backupUID": plan["backupUID"], "restoreUID": restore["metadata"]["uid"], + "snapshotID": status["snapshotID"], "manifestDigest": plan["manifestDigest"], + "coverage": volumes, "restoredClaims": target_claims} + + +def command(args, *, stdin=None, timeout=60): + result = subprocess.run(args, input=stdin, capture_output=True, text=True, + timeout=timeout, check=False) + if result.returncode: + raise Halt(f"{args[0]} exited {result.returncode}: {result.stderr.strip()[:500]}") + return result.stdout.strip() + + +def owned(obj, kind, uid): + return any(owner.get("controller") and owner.get("kind") == kind and owner.get("uid") == uid + for owner in obj["metadata"].get("ownerReferences", [])) + + +class Cluster: + def __init__(self, plan, args): + self.plan, self.args = plan, args + self.k = ["kubectl", "--context", args.context, "-n", plan["namespace"]] + self.timeout = args.timeout + + def get(self, resource, name): + return json.loads(command([*self.k, "get", resource, name, "-o", "json"])) + + def patch(self, gs, fields): + ops = [{"op": "test", "path": "/metadata/resourceVersion", "value": gs["metadata"]["resourceVersion"]}] + ops += [{"op": "add", "path": path, "value": value} for path, value in fields.items()] + command([*self.k, "patch", "gameserver", gs["metadata"]["name"], "--type=json", + "-p", json.dumps(ops)]) + + def pins(self, lane): + pins = self.plan[lane] + require(re.fullmatch(r"[0-9a-f]{64}", pins["revision"]), "invalid template revision") + require(re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", pins["image"]), "pin a game image digest") + module = self.get("modules.gameplane.local", self.plan["module"]) + validate_module(module, pins["bundle"]) + template = self.get("gametemplate", self.plan["module"]) + require(module["status"].get("appliedTemplate") == template["metadata"]["name"] + and owned(template, "Module", module["metadata"]["uid"]), + "the installed template is not owned by the verified module") + command(["cosign", "verify", "--key", self.args.image_key, pins["image"]], timeout=120) + return template["spec"] + + def server(self, name): + obj = self.get("gameserver", name) + require(obj["spec"]["templateRef"]["name"] == self.plan["module"] + and obj["spec"].get("templateRollout", {}).get("policy") == "OnRequest", + f"{name} must use this module and OnRequest") + return obj + + def inventory(self): + all_servers = json.loads(command([*self.k[:3], "get", "gameservers", "-A", "-o", "json"]))["items"] + fleet = [gs for gs in all_servers if gs["spec"]["templateRef"]["name"] == self.plan["module"]] + require(all(gs["metadata"]["namespace"] == self.plan["namespace"] for gs in fleet), + "a server outside the rollout namespace shares this cluster-scoped template") + return {gs["metadata"]["name"]: gs["metadata"]["uid"] for gs in fleet} + + def claims(self, server): + all_claims = json.loads(command([*self.k, "get", "pvc", "-o", "json"]))["items"] + result = {p["metadata"]["name"]: p["metadata"]["uid"] for p in all_claims + if owned(p, "GameServer", server["metadata"]["uid"])} + require(server["metadata"]["name"] + "-data" in result, "primary owned PVC missing") + return result + + def wait_server(self, name, uid, revision, suspended=False): + deadline = time.monotonic() + self.timeout + while time.monotonic() < deadline: + gs = self.server(name) + require(gs["metadata"]["uid"] == uid, f"{name} was replaced") + rev = gs.get("status", {}).get("templateRevision", {}) + if rev.get("effective") == revision and rev.get("desired") == revision: + if suspended and gs["spec"].get("suspend") and not rev.get("running"): + pod = command([*self.k, "get", "pod", name+"-0", "--ignore-not-found", "-o", "name"]) + if not pod: + return gs + if not suspended and rev.get("running") == revision: + return gs + time.sleep(2) + raise Halt(f"{name} did not reach revision {revision} within {self.timeout}s") + + def pod(self, gs, revision, image): + name = gs["metadata"]["name"] + sts = self.get("statefulset", name) + pod = self.get("pod", name+"-0") + require(owned(sts, "GameServer", gs["metadata"]["uid"]) + and owned(pod, "StatefulSet", sts["metadata"]["uid"]) + and pod["metadata"].get("annotations", {}).get("gameplane.local/template-revision") == revision + and any(c["name"] == "game" and c["image"] == image for c in pod["spec"]["containers"]) + and any(c["name"] == "game" and c.get("ready") and c.get("imageID") + for c in pod["status"].get("containerStatuses", [])), "runtime identity or readiness mismatch") + return pod + + def join(self, pod): + # kubectl binds an ephemeral local port and holds this exact pod as the + # destination. The fleet's public address could point at a different game. + with tempfile.TemporaryFile(mode="w+") as log: + proc = subprocess.Popen([*self.k, "port-forward", "--address=127.0.0.1", + "pod/"+pod, ":"+str(self.plan["port"])], stdout=log, stderr=log) + try: + deadline = time.monotonic() + min(30, self.timeout) + while time.monotonic() < deadline: + log.seek(0) + match = re.search(r"Forwarding from 127\.0\.0\.1:(\d+)", log.read()) + if match: + output = command([self.args.join, "-game", self.plan["game"], "-addr", + "127.0.0.1:"+match[1], "-deadline", "30s"]) + parts = output.split("\t", 3) + require(parts[:3] == ["VERDICT", "PASS", "JOINED"] and len(parts) == 4 + and parts[3].strip(), "game client did not produce JOINED evidence") + return parts[3] + require(proc.poll() is None, "port-forward exited before client join") + time.sleep(.15) + raise Halt("port-forward did not publish a local port") + finally: + proc.terminate() + proc.wait(timeout=10) + + def probe(self, target, gs, pins): + command([*self.k, "wait", "--for=condition=Ready", "pod/"+gs["metadata"]["name"]+"-0", + f"--timeout={self.timeout}s"], timeout=self.timeout+10) + first = self.pod(gs, pins["revision"], pins["image"]) + pod = first["metadata"]["name"] + evidence = self.join(pod) + checks = target["checks"] + claims = self.claims(gs) + require({c["claim"] for c in checks} == set(claims), "supply a marker check for every retained PVC") + markers = {} + for check in [*checks, self.plan["save"]]: + argv, expected = check["argv"], check["expected"] + require(isinstance(argv, list) and all(isinstance(a, str) for a in argv) and argv + and isinstance(expected, str) and expected.strip(), "invalid marker/save assertion") + result = command([*self.k, "exec", "pod/"+pod, "-c", "game", "--", *argv]) + require(result == expected, f"{pod} failed its {check.get('claim', 'save')} assertion") + if "claim" in check: + markers[check["claim"]] = expected + last = self.pod(self.server(gs["metadata"]["name"]), pins["revision"], pins["image"]) + require(first["metadata"]["uid"] == last["metadata"]["uid"] and claims == self.claims(gs), + "pod or PVC identity changed during the probe") + return {"serverUID": gs["metadata"]["uid"], "podUID": last["metadata"]["uid"], "pvcs": claims, + "image": pins["image"], "revision": pins["revision"], "join": evidence, + "markers": markers, "save": self.plan["save"]["expected"]} + + def restore(self, clone, backup, name): + resource = {"apiVersion": "gameplane.local/v1alpha1", "kind": "Restore", + "metadata": {"name": name, "namespace": self.plan["namespace"]}, + "spec": {"backupRef": {"name": backup}, "serverRef": {"name": clone}}} + existing = command([*self.k, "get", "restores.gameplane.local", name, "--ignore-not-found", "-o", "name"]) + if not existing: + command([*self.k, "create", "-f", "-"], stdin=json.dumps(resource)) + deadline = time.monotonic()+self.timeout + while time.monotonic() < deadline: + restore = self.get("restores.gameplane.local", name) + require(restore["spec"] == resource["spec"], "restore name already belongs to a different request") + phase = restore.get("status", {}).get("phase") + require(phase != "Failed", "restoring baseline data failed") + if phase == "Succeeded": + return restore + time.sleep(2) + raise Halt("clone restore exceeded deadline") + + +class Journal: + """CAS replacement serializes operators without owning GameServer lifecycle.""" + + def __init__(self, cluster, digest, recover): + self.k = cluster.k + self.name = "gp-rollout-" + hashlib.sha256(cluster.plan["module"].encode()).hexdigest()[:24] + self.owner = str(uuid.uuid4()) + raw = command([*self.k, "get", "configmap", self.name, "--ignore-not-found", "-o", "json"]) + if raw: + self.cm = json.loads(raw) + self.data = json.loads(self.cm["data"]["progress"]) + require(self.data["plan"] == digest, "another rollout plan occupies this journal") + require(not self.data.get("owner") or self.data["owner"] == recover, + "journal is held by another invocation; inspect its owner before recovery") + else: + self.cm = {"apiVersion": "v1", "kind": "ConfigMap", + "metadata": {"name": self.name, "namespace": cluster.plan["namespace"]}} + self.data = {"plan": digest, "receipts": {}} + self.data["owner"] = self.owner + self.write(create=not raw) + + def write(self, create=False): + self.cm["data"] = {"progress": json.dumps(self.data, separators=(",", ":"))} + self.cm = json.loads(command([*self.k, "create" if create else "replace", "-f", "-", "-o", "json"], + stdin=json.dumps(self.cm))) + + def release(self, error=""): + self.data.update(owner="", error=error[:500]) + self.write() + + +def selected_image(template, server): + spec = server["spec"] + if spec.get("image"): + return spec["image"] + versions = template.get("versions", []) + if not versions: + return template["image"] + if spec.get("version"): + selected = next((v for v in versions if v["id"] == spec["version"]), None) + require(selected, "the server's selected version is missing from the template") + else: + selected = next((v for v in versions if v.get("default")), versions[0]) + return selected["image"] + + +def admit(cluster, journal, target, lane, label, provenance=None): + p, data = cluster.plan, journal.data + name, pins = target["name"], p[lane] + template = cluster.pins(lane) + gs = cluster.server(name) + rev = gs.get("status", {}).get("templateRevision", {}) + require(rev.get("desired") == pins["revision"] and selected_image(template, gs) == pins["image"], + f"{name}: desired revision or resolved game image differs from the signed plan") + prior = p["baseline" if lane == "candidate" else "candidate"]["revision"] + fresh = data.get("active", {}).get("label") != label + if fresh: + require(not data.get("active"), "another target is unfinished") + require(not gs["spec"].get("suspend") and rev.get("effective") == rev.get("running") == prior + and gs["spec"]["templateRollout"].get("requestedRevision") != pins["revision"], + f"{name} is not running the expected previous revision") + data["active"] = {"label": label, "name": name, "serverUID": gs["metadata"]["uid"], + "pvcs": cluster.claims(gs), "from": prior, "to": pins["revision"]} + journal.write() + active = data["active"] + require(active["serverUID"] == gs["metadata"]["uid"] and active["pvcs"] == cluster.claims(gs), + f"{name}: server or claim replaced during admission") + restore_required = lane == "baseline" and p["rollback"] == "restore-required" + patch = {"/spec/templateRollout/requestedRevision": pins["revision"]} + if restore_required: + patch["/spec/suspend"] = True + journal.write() # Check the current resourceVersion before each admission. + cluster.patch(gs, patch) + if restore_required: + gs = cluster.wait_server(name, active["serverUID"], pins["revision"], suspended=True) + restore = cluster.restore(name, p["backup"], "gp-rollback-"+data["plan"][:20]) + source, backup = cluster.server(p["source"]), cluster.get("backup", p["backup"]) + restored = validate_restore(backup, restore, source, gs) + require(restored["snapshotID"] == provenance["snapshotID"], "rollback restored another snapshot") + journal.write() + cluster.patch(cluster.server(name), {"/spec/suspend": False}) + active["rollbackRestoreUID"] = restored["restoreUID"] + gs = cluster.wait_server(name, active["serverUID"], pins["revision"]) + receipt = cluster.probe(target, gs, pins) + require(receipt["pvcs"] == active["pvcs"], "workload changed its PVC set during rollout") + data["receipts"][label] = {**active, **receipt} + del data["active"] + journal.write() + + +def run_phase(cluster, journal, phase): + p, data = cluster.plan, journal.data + receipts, canary = data["receipts"], p["canary"] + backup, restore = cluster.get("backup", p["backup"]), cluster.get("restores.gameplane.local", p["restore"]) + source, clone = cluster.server(p["source"]), cluster.server(canary["name"]) + inventory = cluster.inventory() + require(set(inventory) == {p["source"], canary["name"], *(t["name"] for t in p["targets"])}, + "the dedicated module has an unexpected or omitted server in this namespace") + for name in inventory: + cluster.server(name) + if phase == "preflight": + require(not receipts and not data.get("active"), "preflight cannot replace an ongoing rehearsal") + template = cluster.pins("baseline") + for name in inventory: + gs = cluster.server(name) + rev = gs.get("status", {}).get("templateRevision", {}) + require(rev.get("effective") == rev.get("running") == rev.get("desired") == p["baseline"]["revision"] + and selected_image(template, gs) == p["baseline"]["image"] + and gs["spec"]["templateRollout"].get("requestedRevision") != p["candidate"]["revision"] + and not gs["spec"].get("suspend"), f"{name}: expected an unapproved candidate and running baseline") + data["inventory"] = inventory + else: + require(inventory == data.get("inventory"), "preflight missing or fleet identities changed") + provenance = validate_restore(backup, restore, source, clone) + require(backup["status"]["captureManifest"]["templateRevision"] == p["baseline"]["revision"], + "backup was made with another baseline revision") + live_claims = cluster.claims(clone) + require(all(live_claims.get(name) == uid for name, uid in provenance["restoredClaims"].items()), + "restored canary claims were replaced") + require(not data.get("origin") or provenance == data["origin"], "canary restore provenance changed") + data["origin"] = provenance + if phase == "preflight": + journal.write() + return + if phase == "canary": + if "candidate-canary" not in receipts: + # Check the actual restored save under its previous engine before + # admitting the first potentially format-changing candidate. + if "restored-baseline" not in receipts: + receipts["restored-baseline"] = cluster.probe(canary, clone, p["baseline"]) + journal.write() + admit(cluster, journal, canary, "candidate", "candidate-canary") + elif phase == "rollback": + require("candidate-canary" in receipts, "qualify the candidate on the restored canary first") + if "rollback-canary" not in receipts: + admit(cluster, journal, canary, "baseline", "rollback-canary", provenance) + else: + require("rollback-canary" in receipts, "complete the rollback rehearsal before promotion") + for label in ("restored-baseline", "candidate-canary", "rollback-canary"): + require(receipts[label]["serverUID"] == clone["metadata"]["uid"] + and receipts[label]["pvcs"] == cluster.claims(clone), "rehearsed canary was replaced") + if "requalified-canary" not in receipts: + admit(cluster, journal, canary, "candidate", "requalified-canary") + for target in p["targets"]: + label = "fleet/"+target["name"] + if label not in receipts: + admit(cluster, journal, target, "candidate", label) + data["complete"] = True + journal.write() + + +def validate_plan(plan): + target_names = [item["name"] for item in plan["targets"]] + names = [plan["canary"]["name"], *target_names] + require(target_names and len(names) == len(set(names)) and plan["source"] != names[0], + "select a distinct canary and unique fleet targets") + require(plan["rollback"] in ("compatible", "restore-required") + and 0 < plan["port"] <= 65535, "invalid rollback mode or game port") + require(plan["baseline"]["revision"] != plan["candidate"]["revision"], "baseline and candidate coincide") + for name in [*names, plan["source"], plan["namespace"], plan["module"], plan["backup"], plan["restore"]]: + require(re.fullmatch("[a-z0-9](?:[-.a-z0-9]*[a-z0-9])?", name), "invalid Kubernetes resource name") + + +def main(): + cli = argparse.ArgumentParser(description=__doc__) + cli.add_argument("phase", choices=["preflight", "canary", "rollback", "promote"]) + cli.add_argument("plan", type=Path) + cli.add_argument("--context", required=True) + cli.add_argument("--image-key", required=True, help="trusted local cosign public key for game images") + cli.add_argument("--join", required=True, help="join-once or a trusted game-specific adapter with its contract") + cli.add_argument("--timeout", type=int, default=600) + cli.add_argument("--recover-owner", help="exact journal owner after the previous process has exited") + args = cli.parse_args() + plan = json.loads(args.plan.read_text()) + validate_plan(plan) + require(0 < args.timeout <= 3600, "choose a timeout between 1s and 1h") + digest = hashlib.sha256(json.dumps(plan, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + cluster = Cluster(plan, args) + journal = Journal(cluster, digest, args.recover_owner) + try: + run_phase(cluster, journal, args.phase) + except Exception as exc: + journal.release(str(exc)) + raise + journal.release() + print(json.dumps(journal.data, indent=2)) + + +if __name__ == "__main__": + try: + main() + except (Halt, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc: + raise SystemExit(str(exc)) from exc diff --git a/hack/rollout/test_canary.py b/hack/rollout/test_canary.py new file mode 100644 index 000000000..f918a678d --- /dev/null +++ b/hack/rollout/test_canary.py @@ -0,0 +1,188 @@ +"""Exercise rollout decisions independently of cluster scheduling and registry access.""" +import unittest +from copy import deepcopy +from types import SimpleNamespace +from unittest.mock import patch + +from canary import Cluster, Halt, Journal, run_phase, selected_image, validate_module, validate_restore + + +def metadata(name): + return {"name": name, "namespace": "games", "uid": name+"-uid", "generation": 1} + + +class Fleet: + def __init__(self, mode): + self.plan = {"source": "source", "canary": {"name": "clone"}, "targets": [{"name": n} for n in ("source", "other")], + "module": "game", "backup": "bk", "restore": "restore", "rollback": mode, + "baseline": {"revision": "a"*64, "image": "old"}, "candidate": {"revision": "b"*64, "image": "new"}} + self.names = ["source", "clone", "other"] + self.running = {n: self.plan["baseline"]["revision"] for n in self.names} + self.suspend = {n: False for n in self.names} + self.world = {n: 1 for n in self.names} + self.uid = {n: n+"-uid" for n in self.names} + self.lane = "baseline" + self.fail_join = "" + self.admissions = [] + self.restores = [] + self.volume_suffix = "" + + def server(self, name): + return {"metadata": {**metadata(name), "uid": self.uid[name]}, + "spec": {"suspend": self.suspend[name], "templateRef": {"name": "game"}, + "templateRollout": {"policy": "OnRequest"}}, + "status": {"templateRevision": {"effective": self.running[name], "running": self.running[name], + "desired": self.plan[self.lane]["revision"]}}} + + def claims(self, gs): + name = gs["metadata"]["name"] + return {name+"-"+v: name+"-"+v+"-uid"+self.volume_suffix for v in ("data", "extra", "mods")} + + def inventory(self): + return dict(self.uid) + + def get(self, kind, name): + def coverage(server): + return [{"logicalName": v, "claimName": server+"-"+v, "claimUID": server+"-"+v+"-uid", + "included": v != "mods", "classification": "rebuildable" if v == "mods" else "data"} + for v in ("data", "extra", "mods")] + if kind == "backup": + return {"metadata": metadata(name), "spec": {"serverRef": {"name": "source"}}, + "status": {"phase": "Succeeded", "snapshotID": "snapshot", "captureManifest": { + "formatVersion": 2, "volumePolicy": "all-data", "serverUID": "source-uid", + "quiesceOutcome": "already-stopped", "templateRevision": "a"*64, "volumes": coverage("source")}}} + return {"metadata": metadata(name), "spec": {"serverRef": {"name": "clone"}, "backupRef": {"name": "bk"}}, + "status": {"phase": "Succeeded", "snapshotID": "snapshot", "recoveryPlan": {"backupUID": "bk-uid", + "targetUID": "clone-uid", "templateRevision": "a"*64, "manifestDigest": "manifest", + "volumes": coverage("clone")}}} + + def pins(self, lane): + if lane != self.lane: + raise Halt("wrong installed signed module") + return {"image": self.plan[lane]["image"]} + + def patch(self, gs, fields): + name = gs["metadata"]["name"] + if "/spec/suspend" in fields: + self.suspend[name] = fields["/spec/suspend"] + if "/spec/templateRollout/requestedRevision" in fields: + self.running[name] = fields["/spec/templateRollout/requestedRevision"] + self.admissions.append((name, self.lane)) + if self.lane == "candidate" and self.plan["rollback"] == "restore-required": + self.world[name] = 2 + + def wait_server(self, name, uid, revision, suspended=False): + assert self.uid[name] == uid and self.running[name] == revision and self.suspend[name] == suspended + return self.server(name) + + def probe(self, target, gs, pins): + name = target["name"] + if name == self.fail_join: + raise Halt("registration reached PARTIAL only") + if pins == self.plan["baseline"] and self.world[name] != 1: + raise Halt("older engine cannot load format 2") + return {"serverUID": self.uid[name], "pvcs": self.claims(gs), "join": "JOINED", "save": "ack"} + + def restore(self, name, backup, restore_name): + assert self.suspend[name] + self.world[name] = 1 + self.restores.append(name) + return self.get("restore", restore_name) + + +class Tests(unittest.TestCase): + def setup_fleet(self, mode="compatible"): + fleet = Fleet(mode) + journal = SimpleNamespace(data={"plan": "1234567890", "receipts": {}}, write=lambda: None) + run_phase(fleet, journal, "preflight") + fleet.lane = "candidate" + return fleet, journal + + def test_bad_canary_never_admits_a_fleet_member(self): + fleet, journal = self.setup_fleet() + fleet.fail_join = "clone" + with self.assertRaises(Halt): + run_phase(fleet, journal, "canary") + with self.assertRaises(Halt): + run_phase(fleet, journal, "promote") + self.assertEqual(fleet.admissions, []) + fleet.fail_join = "" + run_phase(fleet, journal, "canary") + self.assertEqual(fleet.admissions, [("clone", "candidate")]) + + def test_bad_image_after_admission_halts_before_fleet(self): + fleet, journal = self.setup_fleet() + original_probe = fleet.probe + def check(target, gs, pins): + if pins == fleet.plan["candidate"]: + raise Halt("candidate image cannot boot") + return original_probe(target, gs, pins) + fleet.probe = check + with self.assertRaisesRegex(Halt, "cannot boot"): + run_phase(fleet, journal, "canary") + self.assertEqual(fleet.admissions, [("clone", "candidate")]) + self.assertNotIn("candidate-canary", journal.data["receipts"]) + + def test_both_rollback_classes_and_per_target_halt(self): + for mode in ("compatible", "restore-required"): + with self.subTest(mode=mode): + fleet, journal = self.setup_fleet(mode) + run_phase(fleet, journal, "canary") + fleet.lane = "baseline" + run_phase(fleet, journal, "rollback") + self.assertEqual(fleet.restores, ["clone"] if mode == "restore-required" else []) + fleet.lane = "candidate" + fleet.fail_join = "source" + with self.assertRaises(Halt): + run_phase(fleet, journal, "promote") + self.assertNotIn(("other", "candidate"), fleet.admissions) + fleet.fail_join = "" + run_phase(fleet, journal, "promote") + self.assertTrue(journal.data["complete"]) + self.assertEqual(journal.data["receipts"]["fleet/other"]["from"], "a"*64) + + def test_changed_or_omitted_claims_refuse_admission(self): + fleet, journal = self.setup_fleet() + fleet.volume_suffix = "-replaced" + with self.assertRaisesRegex(Halt, "claims were replaced"): + run_phase(fleet, journal, "canary") + backup, restore = fleet.get("backup", "bk"), fleet.get("restore", "restore") + backup["status"]["captureManifest"]["volumes"][-1]["classification"] = "unclassified" + with self.assertRaises(Halt): + validate_restore(backup, restore, fleet.server("source"), fleet.server("clone")) + + def test_signed_digest_and_version_fail_closed(self): + digest = "sha256:"+"a"*64 + module = {"metadata": metadata("module"), "spec": {"version": "1.0.0", "digest": digest}, + "status": {"phase": "Ready", "appliedVersion": "1.0.0", "observedGeneration": 1, + "verifyPolicy": "keyed", "appliedDigest": digest, "verifiedDigest": digest}} + validate_module(module, digest) + for field in ("verifiedDigest", "appliedDigest", "verifyPolicy", "observedGeneration"): + changed = deepcopy(module) + del changed["status"][field] + with self.assertRaises(Halt): + validate_module(changed, digest) + with self.assertRaises(Halt): + selected_image({"versions": [{"id": "a", "image": "x"}]}, {"spec": {"version": "unknown"}}) + + def test_held_journal_cannot_be_taken_implicitly(self): + cm = '{"data":{"progress":"{\\"owner\\":\\"existing\\",\\"plan\\":\\"abc\\"}"}}' + cluster = SimpleNamespace(k=["kubectl"], plan={"module": "game"}) + with patch("canary.command", return_value=cm): + with self.assertRaisesRegex(Halt, "held"): + Journal(cluster, "abc", None) + + def test_save_and_join_are_tied_to_same_owned_instance(self): + cluster = object.__new__(Cluster) + cluster.k, cluster.timeout = ["kubectl"], 60 + cluster.plan = {"save": {"argv": ["save"], "expected": "ok"}} + gs = {"metadata": metadata("clone")} + target = {"name": "clone", "checks": [{"claim": "clone-data", "argv": ["read"], "expected": "ok"}]} + cluster.claims = lambda _: {"clone-data": "uid"} + cluster.join = lambda _: "login success" + cluster.server = lambda _: gs + pods = iter([{"metadata": metadata("first")}, {"metadata": metadata("replacement")}]) + cluster.pod = lambda *_: next(pods) + with patch("canary.command", return_value="ok"): + with self.assertRaisesRegex(Halt, "identity changed"): + cluster.probe(target, gs, {"revision": "a"*64, "image": "image"}) diff --git a/operator/specs.md b/operator/specs.md index a13bd5527..3ab077dbf 100644 --- a/operator/specs.md +++ b/operator/specs.md @@ -6,6 +6,11 @@ ## Purpose +The release-operator fleet runner in `hack/rollout/canary.py` uses the existing +OnRequest API to admit one named server at a time. A separate ConfigMap records +restore provenance, rollback rehearsal, client join and data checks, without +taking a server's lifecycle owner. See [the staged rollout procedure](../docs/template-rollouts.md#restored-world-fleet-procedure). + Backup manifests record `verified-freeze` only after a successful pod-bound pause. Checkpoint-only and unsupported live copies receive weak consistency conditions. Historical unqualified hook success remains readable with From 46e4a8e68295a93c23a2b0bec642b2b0dc2f29d5 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 11 Nov 2026 00:53:06 -0500 Subject: [PATCH 3/5] test(rollouts): rehearse failed images and format rollback on real restored claims Signed-off-by: Baylem --- hack/rollout/canary.py | 10 +- hack/rollout/test_canary.py | 3 +- test/e2e/buckets.sh | 1 + test/e2e/restored_rollout_e2e_test.go | 24 ++++ test/e2e/rollout_rehearsal.py | 186 ++++++++++++++++++++++++++ test/e2e/suite-components.json | 1 + 6 files changed, 222 insertions(+), 3 deletions(-) create mode 100644 test/e2e/restored_rollout_e2e_test.go create mode 100644 test/e2e/rollout_rehearsal.py diff --git a/hack/rollout/canary.py b/hack/rollout/canary.py index 59eadceb4..3d9361298 100644 --- a/hack/rollout/canary.py +++ b/hack/rollout/canary.py @@ -187,6 +187,11 @@ def probe(self, target, gs, pins): f"--timeout={self.timeout}s"], timeout=self.timeout+10) first = self.pod(gs, pins["revision"], pins["image"]) pod = first["metadata"]["name"] + def runtime(obj): + game = next(c for c in obj["status"]["containerStatuses"] if c["name"] == "game") + require(game.get("containerID") and game.get("imageID"), "missing runtime container identity") + return (obj["metadata"]["uid"], game["containerID"], game["imageID"], game.get("restartCount", 0)) + initial = runtime(first) evidence = self.join(pod) checks = target["checks"] claims = self.claims(gs) @@ -201,9 +206,10 @@ def probe(self, target, gs, pins): if "claim" in check: markers[check["claim"]] = expected last = self.pod(self.server(gs["metadata"]["name"]), pins["revision"], pins["image"]) - require(first["metadata"]["uid"] == last["metadata"]["uid"] and claims == self.claims(gs), + require(initial == runtime(last) and claims == self.claims(gs), "pod or PVC identity changed during the probe") - return {"serverUID": gs["metadata"]["uid"], "podUID": last["metadata"]["uid"], "pvcs": claims, + return {"serverUID": gs["metadata"]["uid"], "podUID": initial[0], "containerID": initial[1], + "runtimeImageID": initial[2], "restartCount": initial[3], "pvcs": claims, "image": pins["image"], "revision": pins["revision"], "join": evidence, "markers": markers, "save": self.plan["save"]["expected"]} diff --git a/hack/rollout/test_canary.py b/hack/rollout/test_canary.py index f918a678d..a44e18340 100644 --- a/hack/rollout/test_canary.py +++ b/hack/rollout/test_canary.py @@ -181,7 +181,8 @@ def test_save_and_join_are_tied_to_same_owned_instance(self): cluster.claims = lambda _: {"clone-data": "uid"} cluster.join = lambda _: "login success" cluster.server = lambda _: gs - pods = iter([{"metadata": metadata("first")}, {"metadata": metadata("replacement")}]) + status = {"containerStatuses": [{"name": "game", "imageID": "image", "containerID": "container"}]} + pods = iter([{"metadata": metadata("first"), "status": status}, {"metadata": metadata("replacement"), "status": status}]) cluster.pod = lambda *_: next(pods) with patch("canary.command", return_value="ok"): with self.assertRaisesRegex(Halt, "identity changed"): diff --git a/test/e2e/buckets.sh b/test/e2e/buckets.sh index 0a145dbe6..4c1b13e39 100755 --- a/test/e2e/buckets.sh +++ b/test/e2e/buckets.sh @@ -64,6 +64,7 @@ TestGameServer_ExplicitStorageClassOverridesDefault TestGameServer_NonexistentStorageClassSurfacesError TestGameServer_VersionSwitch TestGameServer_TemplateOnRequestRetainsData +TestGameServer_RestoredRolloutRehearsal TestGameServer_WipeReportsFailureOnPermissionDenied TestGameServer_CascadingDelete TestGameServer_TunnelCredentialRefusalSurfaced diff --git a/test/e2e/restored_rollout_e2e_test.go b/test/e2e/restored_rollout_e2e_test.go new file mode 100644 index 000000000..0a36ef6f6 --- /dev/null +++ b/test/e2e/restored_rollout_e2e_test.go @@ -0,0 +1,24 @@ +//go:build e2e + +package e2e + +import ( + "fmt" + "os/exec" + "testing" + "time" +) + +// Exercise rollout, storage and failure containment with real operator, kubelet +// and Restic workers. The Minecraft bot independently qualifies the game join. +func TestGameServer_RestoredRolloutRehearsal(t *testing.T) { + t.Parallel() + ensureResticRepo(t) + name := fmt.Sprintf("rollout-%x", time.Now().UnixNano()) + cmd := exec.CommandContext(t.Context(), "python3", "rollout_rehearsal.py", envInstance.Context, name) + out, err := cmd.CombinedOutput() + t.Logf("restored rollout receipts:\n%s", out) + if err != nil { + t.Fatalf("restored rollout qualification: %v", err) + } +} diff --git a/test/e2e/rollout_rehearsal.py b/test/e2e/rollout_rehearsal.py new file mode 100644 index 000000000..f38a0d9f8 --- /dev/null +++ b/test/e2e/rollout_rehearsal.py @@ -0,0 +1,186 @@ +"""Kind-only exercise of durable rollout and real PVC/Restic behavior. + +The HTTP fixture models on-disk format compatibility. Real game joins and +registry signature verification have dedicated Minecraft and module E2Es. +""" +import hashlib +import json +import sys +import time +from pathlib import Path +from types import SimpleNamespace + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "hack/rollout")) +from canary import Cluster, Halt, Journal, command, require, run_phase + + +class Fixture(Cluster): + def pins(self, lane): + template = self.get("gametemplate", self.plan["module"])["spec"] + require(self.plan[lane]["image"] == template["image"], "unexpected test fixture image") + return template + + def join(self, pod): + value = command([*self.k, "exec", pod, "-c", "game", "--", "wget", "-qO-", + "http://127.0.0.1:25565/marker"]) + require(value == "world-a", "fixture cannot serve the restored world") + return "fixture HTTP read of restored world" + + def create(self, kind, name, spec): + self.resources.append((kind.lower()+"s.gameplane.local", name)) + command([*self.k, "create", "-f", "-"], stdin=json.dumps({ + "apiVersion": "gameplane.local/v1alpha1", "kind": kind, + "metadata": {"name": name}, "spec": spec})) + + def phase(self, kind, name): + deadline = time.monotonic()+self.timeout + while time.monotonic() < deadline: + obj = self.get(kind, name) + phase = obj.get("status", {}).get("phase") + require(phase != "Failed", f"{kind}/{name}: {obj.get('status')}") + if phase == "Succeeded": + return obj + time.sleep(2) + raise Halt(f"{kind}/{name} timed out") + + def installed(self, spec): + name = self.plan["module"] + # The fixture exclusively controls spec. Avoid racing status writes, + # and replace the whole spec so rollback removes candidate-only fields. + patch = [{"op": "replace", "path": "/spec", "value": spec}] + command([*self.k, "patch", "gametemplate", name, "--type=json", "-p", json.dumps(patch)]) + deadline = time.monotonic()+self.timeout + while time.monotonic() < deadline: + # Template-controller status is QA-independent. Read the actual hash + # the GameServer reconciler computed from this installation. + gs = self.server(self.plan["source"]) + current = gs.get("status", {}).get("templateRevision", {}).get("desired") + clone = self.server(self.plan["canary"]["name"]) + clone_desired = clone.get("status", {}).get("templateRevision", {}).get("desired") + if current and current != self.previous_desired and current == clone_desired: + self.previous_desired = current + return {"revision": current, "image": spec["image"]} + time.sleep(2) + raise Halt("server did not observe the new installed template") + + def baseline_ready(self, name, revision): + gs = self.server(name) + self.wait_server(name, gs["metadata"]["uid"], revision) + command([*self.k, "wait", "--for=condition=Ready", "pod/"+name+"-0", "--timeout=180s"], timeout=190) + + def read(self, name, *cmd): + return command([*self.k, "exec", name+"-0", "-c", "game", "--", *cmd]) + +def target(cluster, name): + pod = cluster.get("pod", name+"-0") + volumes = {v["name"]: v["persistentVolumeClaim"]["claimName"] for v in pod["spec"]["volumes"] + if "persistentVolumeClaim" in v} + game = next(c for c in pod["spec"]["containers"] if c["name"] == "game") + checks = [] + for mount in game["volumeMounts"]: + if mount["name"] not in volumes: + continue + path = mount["mountPath"] + if path == "/cache": + # The source deliberately contains cache data. The clone holds a + # different value throughout restore because this root is excluded. + expected = "source-cache" if name == cluster.plan["source"] else "clone-cache" + else: + expected = {"/data": "world-a", "/extra": "extra-a"}[path] + checks.append({"claim": volumes[mount["name"]], "argv": ["cat", path+"/marker"], "expected": expected}) + require(len(checks) == 3, "missing fixture claims") + return {"name": name, "checks": checks} + + +def run(context, prefix): + require(context.startswith("kind-") and prefix.startswith("rollout-"), "disposable kind fixture required") + source, clone = prefix+"-source", prefix+"-clone" + plan = {"namespace": "gameplane-games", "module": prefix, "source": source, "backup": prefix+"-backup", + "restore": prefix+"-restore", "game": "format-fixture", "port": 25565, + "save": {"argv": ["sh", "-ec", "sync; printf saved"], "expected": "saved"}} + cluster = Fixture(plan, SimpleNamespace(context=context, timeout=240)) + cluster.resources, cluster.previous_desired = [], None + serve = "exec httpd -f -p 25565 -h /data" + base = {"displayName": prefix, "game": "rollout-fixture", "version": "1", "image": "busybox:1.36", + "command": ["sh", "-ec", "test ! -e /data/format2; "+serve], + "ports": [{"name": "game", "containerPort": 25565, "protocol": "TCP", "advertise": True}], + "security": {"runAsUser": 65532, "runAsGroup": 65532, "fsGroup": 65532}, + "storage": {"size": "1Gi", "mountPath": "/data", "extra": [ + {"name": "world", "mountPath": "/extra", "size": "1Gi", "backupRole": "data"}, + {"name": "cache", "mountPath": "/cache", "size": "1Gi", "backupRole": "rebuildable"}]}} + try: + cluster.create("GameTemplate", prefix, base) + for name in (source, clone): + cluster.create("GameServer", name, {"templateRef": {"name": prefix}, "templateRollout": {"policy": "OnRequest"}}) + deadline = time.monotonic()+cluster.timeout + while time.monotonic() < deadline: + rev = cluster.server(source).get("status", {}).get("templateRevision", {}).get("running") + if rev: + break + time.sleep(2) + require(rev, "missing initial running revision") + cluster.previous_desired = rev + plan["baseline"] = {"revision": rev, "image": base["image"]} + for name in (source, clone): + cluster.baseline_ready(name, rev) + cluster.read(name, "sh", "-ec", "echo world-a > /data/marker; echo extra-a > /extra/marker; " + +"echo "+("source-cache" if name == source else "clone-cache")+" > /cache/marker; sync") + cluster.create("Backup", plan["backup"], {"serverRef": {"name": source}, + "repoRef": {"name": "e2e-restic-creds", "key": "repo"}, + "powerPolicy": "stop-copy-restart", "volumePolicy": "all-data", "quiesce": False}) + cluster.phase("backup", plan["backup"]) + # Source and target must demonstrably differ before restoring. + cluster.read(clone, "sh", "-ec", "echo corrupt > /data/marker; echo corrupt > /extra/marker; sync") + cluster.create("Restore", plan["restore"], {"backupRef": {"name": plan["backup"]}, "serverRef": {"name": clone}}) + cluster.phase("restores.gameplane.local", plan["restore"]) + for name in (source, clone): + cluster.baseline_ready(name, rev) + plan["canary"], plan["targets"] = target(cluster, clone), [target(cluster, source)] + held_pod = cluster.get("pod", source+"-0")["metadata"]["uid"] + for scenario in ("compatible", "restore-required", "bad-image"): + candidate = {**base, "version": scenario} + if scenario == "restore-required": + candidate["command"] = ["sh", "-ec", "echo migrated > /data/format2; "+serve] + if scenario == "bad-image": + candidate["image"] = "127.0.0.1:1/no-image@sha256:"+"0"*64 + plan["candidate"] = cluster.installed(candidate) + require(cluster.installed(base) == plan["baseline"], "baseline content changed") + plan["rollback"] = "compatible" if scenario == "bad-image" else scenario + digest = hashlib.sha256(json.dumps(plan, sort_keys=True).encode()).hexdigest() + journal = Journal(cluster, digest, None) + cluster.resources.append(("configmap", journal.name)) + run_phase(cluster, journal, "preflight") + require(cluster.installed(candidate) == plan["candidate"], "candidate content changed") + if scenario == "bad-image": + cluster.timeout = 90 + try: + run_phase(cluster, journal, "canary") + except Halt: + require(journal.data.get("active", {}).get("name") == clone + and "candidate-canary" not in journal.data["receipts"], "failed at the wrong stage") + else: + raise Halt("unpullable canary passed") + statuses = cluster.get("pod", clone+"-0")["status"].get("containerStatuses", []) + require(any(s["name"] == "game" and s.get("state", {}).get("waiting", {}).get("reason") + in ("ErrImagePull", "ImagePullBackOff") for s in statuses), "expected actual image pull failure") + else: + run_phase(cluster, journal, "canary") + if scenario == "restore-required": + require(cluster.read(clone, "cat", "/data/format2") == "migrated", "fixture did not migrate") + require(cluster.installed(base) == plan["baseline"], "rollback selected another baseline") + run_phase(cluster, journal, "rollback") + require("rollback-canary" in journal.data["receipts"], "rollback receipt missing") + if scenario == "restore-required": + cluster.read(clone, "test", "!", "-e", "/data/format2") + cluster.resources.append(("restores.gameplane.local", "gp-rollback-"+digest[:20])) + require(held_pod == cluster.get("pod", source+"-0")["metadata"]["uid"], "canary rolled a fleet member") + print(json.dumps({"scenario": scenario, "observations": journal.data}, sort_keys=True), flush=True) + journal.release() + command([*cluster.k, "delete", "configmap", journal.name]) + finally: + for resource, name in reversed(cluster.resources): + command([*cluster.k, "delete", resource, name, "--ignore-not-found", "--wait=false"]) + + +if __name__ == "__main__": + run(*sys.argv[1:]) diff --git a/test/e2e/suite-components.json b/test/e2e/suite-components.json index d2f3e6746..456e499ac 100644 --- a/test/e2e/suite-components.json +++ b/test/e2e/suite-components.json @@ -16,6 +16,7 @@ "test/e2e/suite-components.json" ], "suites": [ + { "tests": "^TestGameServer_RestoredRolloutRehearsal$", "components": ["operator", "agent"], "paths": ["hack/rollout/", "test/e2e/rollout_rehearsal.py"] }, { "tests": "^TestColdWakeFleet_ConcurrentJoins$", "components": ["operator", "sentinel", "gameproto", "agent"], "paths": ["test/e2e/wake_burst.py", "test/e2e/cmd/join-once/", "test/e2e/testdata/wake-burst/"] }, { "tests": "^TestGameServer_HostportWakeStableEndpoint$", "components": ["operator", "api", "agent", "sentinel", "gameproto"] }, { "tests": "^(TestPanel_OfflineRecovery|TestAPIClient_DoContextResponseBody|TestAPIClient_ExtractIntField)$", "components": ["operator", "api", "agent"], From f6035d9f6e41c7db335cb47e986d1fc7a1bc033c Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 11 Nov 2026 01:05:52 -0500 Subject: [PATCH 4/5] fix(rollouts): recheck resumed canaries and bind writes to Gameplane resources Signed-off-by: Baylem --- hack/rollout/canary.py | 18 +++++++++++++++--- test/e2e/rollout_rehearsal.py | 2 +- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/hack/rollout/canary.py b/hack/rollout/canary.py index 3d9361298..bd852dc15 100644 --- a/hack/rollout/canary.py +++ b/hack/rollout/canary.py @@ -87,12 +87,14 @@ def __init__(self, plan, args): self.timeout = args.timeout def get(self, resource, name): + resource = {"backup": "backups.gameplane.local", "gametemplate": "gametemplates.gameplane.local", + "gameserver": "gameservers.gameplane.local"}.get(resource, resource) return json.loads(command([*self.k, "get", resource, name, "-o", "json"])) def patch(self, gs, fields): ops = [{"op": "test", "path": "/metadata/resourceVersion", "value": gs["metadata"]["resourceVersion"]}] ops += [{"op": "add", "path": path, "value": value} for path, value in fields.items()] - command([*self.k, "patch", "gameserver", gs["metadata"]["name"], "--type=json", + command([*self.k, "patch", "gameservers.gameplane.local", gs["metadata"]["name"], "--type=json", "-p", json.dumps(ops)]) def pins(self, lane): @@ -116,7 +118,7 @@ def server(self, name): return obj def inventory(self): - all_servers = json.loads(command([*self.k[:3], "get", "gameservers", "-A", "-o", "json"]))["items"] + all_servers = json.loads(command([*self.k[:3], "get", "gameservers.gameplane.local", "-A", "-o", "json"]))["items"] fleet = [gs for gs in all_servers if gs["spec"]["templateRef"]["name"] == self.plan["module"]] require(all(gs["metadata"]["namespace"] == self.plan["namespace"] for gs in fleet), "a server outside the rollout namespace shares this cluster-scoped template") @@ -180,7 +182,11 @@ def join(self, pod): raise Halt("port-forward did not publish a local port") finally: proc.terminate() - proc.wait(timeout=10) + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=10) def probe(self, target, gs, pins): command([*self.k, "wait", "--for=condition=Ready", "pod/"+gs["metadata"]["name"]+"-0", @@ -375,6 +381,12 @@ def run_phase(cluster, journal, phase): and receipts[label]["pvcs"] == cluster.claims(clone), "rehearsed canary was replaced") if "requalified-canary" not in receipts: admit(cluster, journal, canary, "candidate", "requalified-canary") + else: + # A paused fleet may resume days later. Recheck the live canary + # before advancing to the next target. + cluster.pins("candidate") + current = cluster.probe(canary, cluster.server(canary["name"]), p["candidate"]) + require(current["pvcs"] == receipts["requalified-canary"]["pvcs"], "canary claims changed") for target in p["targets"]: label = "fleet/"+target["name"] if label not in receipts: diff --git a/test/e2e/rollout_rehearsal.py b/test/e2e/rollout_rehearsal.py index f38a0d9f8..7b0f212ff 100644 --- a/test/e2e/rollout_rehearsal.py +++ b/test/e2e/rollout_rehearsal.py @@ -48,7 +48,7 @@ def installed(self, spec): # The fixture exclusively controls spec. Avoid racing status writes, # and replace the whole spec so rollback removes candidate-only fields. patch = [{"op": "replace", "path": "/spec", "value": spec}] - command([*self.k, "patch", "gametemplate", name, "--type=json", "-p", json.dumps(patch)]) + command([*self.k, "patch", "gametemplates.gameplane.local", name, "--type=json", "-p", json.dumps(patch)]) deadline = time.monotonic()+self.timeout while time.monotonic() < deadline: # Template-controller status is QA-independent. Read the actual hash From 37ba78e04f41f8ac84b7ba10c0a626bb50c2c2b8 Mon Sep 17 00:00:00 2001 From: Baylem Date: Wed, 11 Nov 2026 02:47:35 -0500 Subject: [PATCH 5/5] fix(rollout): qualify UDP pod joins and reject incompatible clients Signed-off-by: Baylem --- docs/template-rollouts.md | 5 +++ hack/rollout/canary.py | 32 ++++++++++---- hack/rollout/test_canary.py | 13 ++++++ test/e2e/restored_rollout_e2e_test.go | 8 +++- test/e2e/rollout_rehearsal.py | 60 ++++++++++++++++++++------- 5 files changed, 94 insertions(+), 24 deletions(-) diff --git a/docs/template-rollouts.md b/docs/template-rollouts.md index 7b97ddf81..2abf84f4f 100644 --- a/docs/template-rollouts.md +++ b/docs/template-rollouts.md @@ -142,6 +142,11 @@ the platform checkout with `go build -o join-once ./test/e2e/cmd/join-once`. For an authenticated or proprietary game, supply a trusted entitled client that implements the same CLI and returns `VERDICTPASSJOINEDevidence`. The bundled Minecraft client expects offline-mode test servers. +The plan's `port` and `protocol` must name a game-container port. TCP (the +default) uses a port-forward to that specific Pod. For a UDP game set +`"protocol": "UDP"` and run its qualified join client on a host or in a Pod +that can route to the game's Pod IP. Permit this runner's traffic to the +declared UDP port in the selected workload's ingress policy. Create a stopped, all-data Restic backup of a representative server. Create and start a **separate** GameServer with the same template and version, restore diff --git a/hack/rollout/canary.py b/hack/rollout/canary.py index bd852dc15..11e57c0eb 100644 --- a/hack/rollout/canary.py +++ b/hack/rollout/canary.py @@ -3,6 +3,7 @@ import argparse import hashlib +import ipaddress import json import re import subprocess @@ -159,24 +160,38 @@ def pod(self, gs, revision, image): for c in pod["status"].get("containerStatuses", [])), "runtime identity or readiness mismatch") return pod + def client_join(self, address): + result = subprocess.run([self.args.join, "-game", self.plan["game"], "-addr", address, + "-deadline", "30s"], capture_output=True, text=True, timeout=45) + parts = result.stdout.strip().split("\t", 3) + verdict = " ".join(parts[:3])[:120] + require(result.returncode == 0 and parts[:3] == ["VERDICT", "PASS", "JOINED"] + and len(parts) == 4 and parts[3].strip(), f"game client did not join ({verdict})") + return parts[3] + def join(self, pod): + protocol = self.plan.get("protocol", "TCP") + port = self.plan["port"] + game = next(c for c in pod["spec"]["containers"] if c["name"] == "game") + require(any(p["containerPort"] == port and p.get("protocol", "TCP") == protocol + for p in game.get("ports", [])), "planned game port and protocol absent from game container") + if protocol == "UDP": + # kubectl has no UDP port-forward. An in-cluster or routed runner + # addresses this exact canary, including on IPv6-only clusters. + ip = ipaddress.ip_address(pod["status"].get("podIP", "")) + return self.client_join(f"[{ip}]:{port}" if ip.version == 6 else f"{ip}:{port}") # kubectl binds an ephemeral local port and holds this exact pod as the # destination. The fleet's public address could point at a different game. with tempfile.TemporaryFile(mode="w+") as log: proc = subprocess.Popen([*self.k, "port-forward", "--address=127.0.0.1", - "pod/"+pod, ":"+str(self.plan["port"])], stdout=log, stderr=log) + "pod/"+pod["metadata"]["name"], ":"+str(port)], stdout=log, stderr=log) try: deadline = time.monotonic() + min(30, self.timeout) while time.monotonic() < deadline: log.seek(0) match = re.search(r"Forwarding from 127\.0\.0\.1:(\d+)", log.read()) if match: - output = command([self.args.join, "-game", self.plan["game"], "-addr", - "127.0.0.1:"+match[1], "-deadline", "30s"]) - parts = output.split("\t", 3) - require(parts[:3] == ["VERDICT", "PASS", "JOINED"] and len(parts) == 4 - and parts[3].strip(), "game client did not produce JOINED evidence") - return parts[3] + return self.client_join("127.0.0.1:"+match[1]) require(proc.poll() is None, "port-forward exited before client join") time.sleep(.15) raise Halt("port-forward did not publish a local port") @@ -198,7 +213,7 @@ def runtime(obj): require(game.get("containerID") and game.get("imageID"), "missing runtime container identity") return (obj["metadata"]["uid"], game["containerID"], game["imageID"], game.get("restartCount", 0)) initial = runtime(first) - evidence = self.join(pod) + evidence = self.join(first) checks = target["checks"] claims = self.claims(gs) require({c["claim"] for c in checks} == set(claims), "supply a marker check for every retained PVC") @@ -402,6 +417,7 @@ def validate_plan(plan): "select a distinct canary and unique fleet targets") require(plan["rollback"] in ("compatible", "restore-required") and 0 < plan["port"] <= 65535, "invalid rollback mode or game port") + require(plan.get("protocol", "TCP") in ("TCP", "UDP"), "unsupported game transport") require(plan["baseline"]["revision"] != plan["candidate"]["revision"], "baseline and candidate coincide") for name in [*names, plan["source"], plan["namespace"], plan["module"], plan["backup"], plan["restore"]]: require(re.fullmatch("[a-z0-9](?:[-.a-z0-9]*[a-z0-9])?", name), "invalid Kubernetes resource name") diff --git a/hack/rollout/test_canary.py b/hack/rollout/test_canary.py index a44e18340..76f59df03 100644 --- a/hack/rollout/test_canary.py +++ b/hack/rollout/test_canary.py @@ -172,6 +172,19 @@ def test_held_journal_cannot_be_taken_implicitly(self): with self.assertRaisesRegex(Halt, "held"): Journal(cluster, "abc", None) + def test_udp_joins_only_the_declared_port_on_the_qualified_pod(self): + cluster = object.__new__(Cluster) + cluster.plan = {"protocol": "UDP", "port": 8303} + cluster.client_join = lambda address: address + for ip, endpoint in (("10.0.3.19", "10.0.3.19:8303"), ("fd00::19", "[fd00::19]:8303")): + port = {"containerPort": 8303, "protocol": "UDP"} + pod = {"status": {"podIP": ip}, + "spec": {"containers": [{"name": "game", "ports": [port]}]}} + self.assertEqual(cluster.join(pod), endpoint) + port["protocol"] = "TCP" + with self.assertRaisesRegex(Halt, "protocol absent"): + cluster.join(pod) + def test_save_and_join_are_tied_to_same_owned_instance(self): cluster = object.__new__(Cluster) cluster.k, cluster.timeout = ["kubectl"], 60 diff --git a/test/e2e/restored_rollout_e2e_test.go b/test/e2e/restored_rollout_e2e_test.go index 0a36ef6f6..12dbd80fa 100644 --- a/test/e2e/restored_rollout_e2e_test.go +++ b/test/e2e/restored_rollout_e2e_test.go @@ -5,6 +5,7 @@ package e2e import ( "fmt" "os/exec" + "path/filepath" "testing" "time" ) @@ -15,7 +16,12 @@ func TestGameServer_RestoredRolloutRehearsal(t *testing.T) { t.Parallel() ensureResticRepo(t) name := fmt.Sprintf("rollout-%x", time.Now().UnixNano()) - cmd := exec.CommandContext(t.Context(), "python3", "rollout_rehearsal.py", envInstance.Context, name) + join := filepath.Join(t.TempDir(), "join-once") + build := exec.CommandContext(t.Context(), "go", "build", "-o", join, "./cmd/join-once") + if out, err := build.CombinedOutput(); err != nil { + t.Fatalf("build protocol client: %v\n%s", err, out) + } + cmd := exec.CommandContext(t.Context(), "python3", "rollout_rehearsal.py", envInstance.Context, name, join) out, err := cmd.CombinedOutput() t.Logf("restored rollout receipts:\n%s", out) if err != nil { diff --git a/test/e2e/rollout_rehearsal.py b/test/e2e/rollout_rehearsal.py index 7b0f212ff..ccef8d4cc 100644 --- a/test/e2e/rollout_rehearsal.py +++ b/test/e2e/rollout_rehearsal.py @@ -21,7 +21,9 @@ def pins(self, lane): return template def join(self, pod): - value = command([*self.k, "exec", pod, "-c", "game", "--", "wget", "-qO-", + if pod["metadata"]["annotations"]["gameplane.local/template-revision"] == self.reject_join_revision: + return super().join(pod) + value = command([*self.k, "exec", pod["metadata"]["name"], "-c", "game", "--", "wget", "-qO-", "http://127.0.0.1:25565/marker"]) require(value == "world-a", "fixture cannot serve the restored world") return "fixture HTTP read of restored world" @@ -81,25 +83,27 @@ def target(cluster, name): if mount["name"] not in volumes: continue path = mount["mountPath"] - if path == "/cache": - # The source deliberately contains cache data. The clone holds a - # different value throughout restore because this root is excluded. - expected = "source-cache" if name == cluster.plan["source"] else "clone-cache" + if path == "/mods": + # The rebuildable mod package differs on the two servers. Restic + # must preserve the clone's installed file across both restores. + expected = "source-package" if name == cluster.plan["source"] else "clone-package" + file = path+"/sample.pk3" else: expected = {"/data": "world-a", "/extra": "extra-a"}[path] - checks.append({"claim": volumes[mount["name"]], "argv": ["cat", path+"/marker"], "expected": expected}) + file = path+"/marker" + checks.append({"claim": volumes[mount["name"]], "argv": ["cat", file], "expected": expected}) require(len(checks) == 3, "missing fixture claims") return {"name": name, "checks": checks} -def run(context, prefix): +def run(context, prefix, join): require(context.startswith("kind-") and prefix.startswith("rollout-"), "disposable kind fixture required") source, clone = prefix+"-source", prefix+"-clone" plan = {"namespace": "gameplane-games", "module": prefix, "source": source, "backup": prefix+"-backup", - "restore": prefix+"-restore", "game": "format-fixture", "port": 25565, + "restore": prefix+"-restore", "game": "minecraft-java", "port": 25565, "save": {"argv": ["sh", "-ec", "sync; printf saved"], "expected": "saved"}} - cluster = Fixture(plan, SimpleNamespace(context=context, timeout=240)) - cluster.resources, cluster.previous_desired = [], None + cluster = Fixture(plan, SimpleNamespace(context=context, timeout=240, join=join)) + cluster.resources, cluster.previous_desired, cluster.reject_join_revision = [], None, None serve = "exec httpd -f -p 25565 -h /data" base = {"displayName": prefix, "game": "rollout-fixture", "version": "1", "image": "busybox:1.36", "command": ["sh", "-ec", "test ! -e /data/format2; "+serve], @@ -107,7 +111,7 @@ def run(context, prefix): "security": {"runAsUser": 65532, "runAsGroup": 65532, "fsGroup": 65532}, "storage": {"size": "1Gi", "mountPath": "/data", "extra": [ {"name": "world", "mountPath": "/extra", "size": "1Gi", "backupRole": "data"}, - {"name": "cache", "mountPath": "/cache", "size": "1Gi", "backupRole": "rebuildable"}]}} + {"name": "packages", "mountPath": "/mods", "size": "1Gi", "backupRole": "rebuildable"}]}} try: cluster.create("GameTemplate", prefix, base) for name in (source, clone): @@ -124,7 +128,7 @@ def run(context, prefix): for name in (source, clone): cluster.baseline_ready(name, rev) cluster.read(name, "sh", "-ec", "echo world-a > /data/marker; echo extra-a > /extra/marker; " - +"echo "+("source-cache" if name == source else "clone-cache")+" > /cache/marker; sync") + +"echo "+("source-package" if name == source else "clone-package")+" > /mods/sample.pk3; sync") cluster.create("Backup", plan["backup"], {"serverRef": {"name": source}, "repoRef": {"name": "e2e-restic-creds", "key": "repo"}, "powerPolicy": "stop-copy-restart", "volumePolicy": "all-data", "quiesce": False}) @@ -137,7 +141,7 @@ def run(context, prefix): cluster.baseline_ready(name, rev) plan["canary"], plan["targets"] = target(cluster, clone), [target(cluster, source)] held_pod = cluster.get("pod", source+"-0")["metadata"]["uid"] - for scenario in ("compatible", "restore-required", "bad-image"): + for scenario in ("compatible", "restore-required", "bad-join", "bad-image"): candidate = {**base, "version": scenario} if scenario == "restore-required": candidate["command"] = ["sh", "-ec", "echo migrated > /data/format2; "+serve] @@ -145,13 +149,39 @@ def run(context, prefix): candidate["image"] = "127.0.0.1:1/no-image@sha256:"+"0"*64 plan["candidate"] = cluster.installed(candidate) require(cluster.installed(base) == plan["baseline"], "baseline content changed") - plan["rollback"] = "compatible" if scenario == "bad-image" else scenario + plan["rollback"] = "restore-required" if scenario == "restore-required" else "compatible" digest = hashlib.sha256(json.dumps(plan, sort_keys=True).encode()).hexdigest() journal = Journal(cluster, digest, None) cluster.resources.append(("configmap", journal.name)) run_phase(cluster, journal, "preflight") require(cluster.installed(candidate) == plan["candidate"], "candidate content changed") - if scenario == "bad-image": + if scenario == "bad-join": + # The candidate runs and serves the marker over TCP, but a real + # Minecraft client cannot log in to this HTTP-only listener. + cluster.reject_join_revision = plan["candidate"]["revision"] + try: + run_phase(cluster, journal, "canary") + except Halt as exc: + require("game client did not join (VERDICT FAIL_" in str(exc), str(exc)) + require(journal.data.get("active", {}).get("name") == clone + and "candidate-canary" not in journal.data["receipts"], "failed at the wrong stage") + else: + raise Halt("running candidate admitted despite a failed wire login") + require(cluster.read(clone, "wget", "-qO-", "http://127.0.0.1:25565/marker") == "world-a", + "candidate was not serving during the failed join") + try: + run_phase(cluster, journal, "promote") + except Halt as exc: + require("complete the rollback rehearsal" in str(exc), str(exc)) + else: + raise Halt("fleet admitted after failed wire login") + # Reset only the disposable clone for the independent bad-image + # case. This remains outside the halted production rollout. + cluster.reject_join_revision = None + require(cluster.installed(base) == plan["baseline"], "manual reset selected another baseline") + cluster.patch(cluster.server(clone), {"/spec/templateRollout/requestedRevision": rev}) + cluster.baseline_ready(clone, rev) + elif scenario == "bad-image": cluster.timeout = 90 try: run_phase(cluster, journal, "canary")