diff --git a/.github/workflows/native-supervisor-test.yml b/.github/workflows/native-supervisor-test.yml new file mode 100644 index 00000000..ba0a08d5 --- /dev/null +++ b/.github/workflows/native-supervisor-test.yml @@ -0,0 +1,24 @@ +name: Native Supervisor Tests +on: + push: + branches: [main] + paths: [deploy/catchup/**, .github/workflows/native-supervisor-test.yml] + pull_request: + paths: [deploy/catchup/**, .github/workflows/native-supervisor-test.yml] + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + admission: + runs-on: ubuntu-24.04 + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + - name: Install disposable etcd fixture dependency + run: sudo apt-get update && sudo apt-get install -y etcd-server + - name: Verify supervisor identity, CAS, backoff and join behavior + env: + ETCD_BIN: /usr/bin/etcd + run: python3 -m unittest discover -s deploy/catchup -p 'test_native_supervisor.py' -v diff --git a/deploy/catchup/native-publisher.example.json b/deploy/catchup/native-publisher.example.json new file mode 100644 index 00000000..d73093dc --- /dev/null +++ b/deploy/catchup/native-publisher.example.json @@ -0,0 +1,28 @@ +{ + "apiVersion": "v1", + "kind": "Pod", + "metadata": {"name": "native-publisher-example"}, + "spec": { + "restartPolicy": "Never", + "automountServiceAccountToken": false, + "terminationGracePeriodSeconds": 1800, + "containers": [{ + "name": "publisher", + "image": "YOUR_REGISTRY/lance-context@sha256:REPLACE_WITH_TESTED_IMAGE_DIGEST", + "command": ["python3", "-u", "/supervisor/native_supervisor.py"], + "envFrom": [{"secretRef": {"name": "lance-native-publisher-config"}}], + "env": [ + {"name": "POD_UID", "valueFrom": {"fieldRef": {"apiVersion": "v1", "fieldPath": "metadata.uid"}}}, + {"name": "HOSTNAME", "valueFrom": {"fieldRef": {"apiVersion": "v1", "fieldPath": "metadata.name"}}} + ], + "resources": { + "requests": {"cpu": "2", "memory": "16Gi"}, + "limits": {"cpu": "2", "memory": "16Gi"} + }, + "readinessProbe": {"exec": {"command": ["test", "-f", "/tmp/native-supervisor-ready"]}}, + "volumeMounts": [{"name": "supervisor", "mountPath": "/supervisor", "readOnly": true}], + "securityContext": {"allowPrivilegeEscalation": false} + }], + "volumes": [{"name": "supervisor", "configMap": {"name": "native-supervisor-REPLACE_WITH_CONTENT_HASH"}}] + } +} diff --git a/deploy/catchup/native-publisher.md b/deploy/catchup/native-publisher.md new file mode 100644 index 00000000..5c13a0c0 --- /dev/null +++ b/deploy/catchup/native-publisher.md @@ -0,0 +1,76 @@ +# Persistent native publisher admission + +`native_supervisor.py` tracks the existing external persistent-publisher +supervisor alongside the native catch-up deployment assets. It adds Pod UID +admission fencing. It does not replace or enable the master's native Kubernetes +Job controller described in `docs/master-catchup.md`, and does not create, delete, +restart, or take ownership of any Pod or table. + +A Kubernetes Pod name can be reused. Previously a replacement under the same +name could accept an operator record intended for its predecessor. Before every +attempt, this supervisor now requires all of: + +- operator `phase` is `native_parallel_active`; +- `native_job` and the exact persistent `catchup-active/` owner match + `CATCHUP_JOB_NAME`; +- `native_pod` matches the Pod name; +- `native_pod_uid` matches `POD_UID`, supplied by the downward API. + +A missing local UID fails startup before readiness. A missing or mismatched +authorized UID leaves the supervisor in standby without touching the attempt or +failure ledger. The attempt CAS still compares the **entire raw operator value**, +the owner, and the old attempt. A concurrent lifecycle change after the UID check +therefore rejects admission. Attempts include the admitted `pod_uid` for diagnosis. + +## Deployment contract + +Use `native-publisher.example.json` as the standalone Pod shape. Mount an immutable, +content-addressed ConfigMap containing this script, and pin a tested image with +Python 3 and `/usr/local/bin/lance-context-master`. Do not feed this standalone +Pod manifest into `CATCHUP_POD_TEMPLATE`, which accepts the native Job controller's +different PodSpec contract. + +**Require one container and `restartPolicy: Never`, and verify both on the live +Pod before activation.** The Pod UID then identifies one supervisor incarnation. +An `Always` or `OnFailure` Pod can restart its container under the same UID and is +not supported by this protocol. The example uses the downward API for both UID +and name; do not supply a literal UID or reuse another Pod's value. The supervisor +has no Kubernetes credential and cannot independently inspect its restart policy. + +The reviewed environment/Secret must configure the target, persistent job, +storage, etcd prefix/endpoints, worker endpoints, historical shard identities, +owned target, append settings and byte budgets for the native binary. In +particular `MERGE_OWNED_TARGETS` must include `CATCHUP_TARGET`. Size memory limits +above the supervisor's existing 12 GiB admission watermark; that watermark is not +a bound on one admitted operation. Keep native byte reservations and Lance memory +limits in force. Configure `CATCHUP_SLICE_SECS` as a soft admission slice and allow +already admitted native work to join during graceful termination. + +Create each new Pod in standby, read its actual Ready UID/container identity, then +CAS the operator to authorize that UID while preserving the persistent owner and +all attempt/failure records. A same-name foreign Pod created between observation +and activation cannot use the stale UID, even if the activation CAS succeeds. +Do not activate a replacement until the predecessor has qualified terminal/join +evidence and native execution recovery has completed. Pod disappearance, an API +timeout, or an expired lease alone is not that evidence. + +The wrapper retains the existing backoff and child-join behavior: SIGTERM stops +new admissions and waits for the native child; unknown outcomes do not clear +ownership or bypass the native execution fence. The native executable remains +responsible for claims, cancellation, memory reservations, storage barriers and +watermarks. This script has no general automatic recreation loop. Migration to +the native Job controller remains separate work. + +## Tests + +```sh +ETCD_BIN=/path/to/etcd python3 -m unittest discover \ + -s deploy/catchup -p 'test_native_supervisor.py' -v +``` + +Tests run real supervisor processes and a disposable loopback-only etcd. The +native child is a recording stub: they prove stale UID rejection, full-operator +CAS races, subsequent admission checks, preserved backoff and joined shutdown, +not Lance payload correctness or Kubernetes lifecycle behavior. Validate the +changed supervisor separately with real Kubernetes downward API identities and +the native binary against isolated storage before production migration. diff --git a/deploy/catchup/native_supervisor.py b/deploy/catchup/native_supervisor.py new file mode 100644 index 00000000..6dda1477 --- /dev/null +++ b/deploy/catchup/native_supervisor.py @@ -0,0 +1,246 @@ +"""Admission supervisor for externally managed persistent native publishers. + +Every admission is bound to the actual Pod UID, not just its reusable name. +Data work and execution fencing remain in the native binary. This supervisor +never clears ownership, replays an uncertain attempt, or kills a native child. +Deploy only in a one-container Pod with restartPolicy Never; see native-publisher.md. +""" + +import base64 +import json +import os +import pathlib +import re +import signal +import subprocess +import threading +import time +import urllib.request +import uuid + +NATIVE_BINARY = "/usr/local/bin/lance-context-master" +MEMORY_CURRENT = pathlib.Path("/sys/fs/cgroup/memory.current") +READY_FILE = pathlib.Path("/tmp/native-supervisor-ready") + + +def authorized(operator, job, pod_name, pod_uid): + """Reject missing identity and stale authorization for a reused Pod name.""" + return bool( + pod_uid + and operator.get("phase") == "native_parallel_active" + and operator.get("native_job") == job + and operator.get("native_pod") == pod_name + and operator.get("native_pod_uid") == pod_uid + ) + + +def main(): + UID = os.environ.get("POD_UID", "").strip() + if not UID: + raise SystemExit( + "POD_UID from metadata.uid is required; refusing native admission" + ) + T = os.environ["CATCHUP_TARGET"] + J = os.environ["CATCHUP_JOB_NAME"] + P = os.environ["ETCD_PREFIX"].rstrip("/") + H = T.encode().hex() + E = os.environ["ETCD_ENDPOINTS"].rstrip("/") + "/v3/" + stop = threading.Event() + proc = None + enc = lambda s: base64.b64encode(s.encode()).decode() + + def log(event, **kw): + print( + json.dumps( + {"at": time.time(), "event": event, "target": T, "job": J, **kw} + ), + flush=True, + ) + + def read(suffix): + r = urllib.request.Request( + E + "kv/range", + data=json.dumps({"key": enc(P + "/" + suffix)}).encode(), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(r, timeout=5) as f: + k = json.load(f).get("kvs", []) + return base64.b64decode(k[0]["value"]).decode() if k else None + + def cas_attempt(old, new, operator): + if not authorized(json.loads(operator), J, os.environ["HOSTNAME"], UID): + return None + key = P + "/merge-rollout-attempts/" + H + compares = [ + { + "key": enc(P + "/catchup-active/" + H), + "target": "VALUE", + "result": "EQUAL", + "value": enc(J), + }, + { + "key": enc(P + "/merge-rollout-operator/" + H), + "target": "VALUE", + "result": "EQUAL", + "value": enc(operator), + }, + ] + compares.append( + {"key": enc(key), "target": "VERSION", "result": "EQUAL", "version": "0"} + if old is None + else { + "key": enc(key), + "target": "VALUE", + "result": "EQUAL", + "value": enc(old), + } + ) + value = json.dumps(new, separators=(",", ":")) + r = urllib.request.Request( + E + "kv/txn", + data=json.dumps( + { + "compare": compares, + "success": [ + {"request_put": {"key": enc(key), "value": enc(value)}} + ], + } + ).encode(), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(r, timeout=5) as f: + out = json.load(f) + return value if out.get("succeeded") else None + + def halt(*args): + stop.set() + log("admission_stopping_waiting_for_native_exit") + + signal.signal(signal.SIGTERM, halt) + signal.signal(signal.SIGINT, halt) + READY_FILE.touch() + log("standby") + while not stop.is_set(): + try: + operator = read("merge-rollout-operator/" + H) + op = json.loads(operator or "{}") + if ( + not authorized(op, J, os.environ["HOSTNAME"], UID) + or read("catchup-active/" + H) != J + ): + stop.wait(10) + continue + # The admission supervisor does not bypass native persistent failure backoff. + failure = json.loads( + read("merge-failures/" + H + "/" + b"master:catchup".hex()) or "{}" + ) + old_attempt = read("merge-rollout-attempts/" + H) + previous = json.loads(old_attempt or "{}") + if previous and previous.get("job") != J: + raise RuntimeError("attempt ledger belongs to a different job") + delay = max( + 0, + max(failure.get("next_retry_ms", 0), previous.get("next_retry_ms", 0)) + / 1000 + - time.time(), + ) + if delay > 0: + log("persistent_backoff", retry_after_seconds=round(delay)) + stop.wait(min(delay, 60)) + continue + if int(MEMORY_CURRENT.read_text()) > 12 * 2**30: + log("memory_admission_delayed") + stop.wait(15) + continue + # The native executable claims only J's target, reconciles old execution, + # enforces shared byte reservations and idle progress, then commits watermarks. + attempt = { + "job": J, + "pod_uid": UID, + "id": str(uuid.uuid4()), + "started_at": time.time(), + "consecutive_failures": previous.get("consecutive_failures", 0) + 1, + "state": "admitted", + } + attempt["next_retry_ms"] = int( + ( + time.time() + + min(3600, 60 * 2 ** min(attempt["consecutive_failures"], 6)) + ) + * 1000 + ) + attempt["needs_attention"] = attempt["consecutive_failures"] >= 3 + admitted = cas_attempt(old_attempt, attempt, operator) + if admitted is None: + stop.wait(10) + continue + started = time.monotonic() + reclaimed = 0 + pass_errors = 0 + proc = subprocess.Popen( + [NATIVE_BINARY], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + ) + log("native_pass_started", pid=proc.pid) + for line in proc.stdout: + print(line.rstrip(), flush=True) + clean = re.sub(r"\x1b\[[0-9;]*m", "", line) + if "parallel rollout append completed" in clean: + n = re.search(r"\breclaimed=(\d+)", clean) + if n: + reclaimed += int(n.group(1)) + n = re.search(r"\bfailed=(\d+)", clean) + if n: + pass_errors += int(n.group(1)) + code = proc.wait() + seconds = time.monotonic() - started + proc = None + terminal = dict( + attempt, + state="succeeded" if code == 0 else "failed", + exit_code=code, + finished_at=time.time(), + reclaimed=reclaimed, + seconds=seconds, + ) + if code == 0: + terminal.update( + consecutive_failures=0, + needs_attention=False, + next_retry_ms=int((time.time() + (2 if reclaimed else 30)) * 1000), + ) + else: + terminal["next_retry_ms"] = int( + ( + time.time() + + min(3600, 60 * 2 ** min(attempt["consecutive_failures"], 6)) + ) + * 1000 + ) + if cas_attempt(admitted, terminal, operator) is None: + log("attempt_record_changed_ownership_untouched") + log( + "native_pass_finished", + exit_code=code, + reclaimed=reclaimed, + seconds=seconds, + generations_per_second=reclaimed / seconds, + failed_shards=pass_errors, + ) + # Let ordinary flushes proceed; zero-work passes must not hammer etcd/catalogs. + stop.wait(2 if code == 0 and reclaimed > 0 else 30) + except Exception as e: # noqa: BLE001 - preserve native child join on any supervisor error + log("supervisor_error_ownership_untouched", error=repr(e)) + if proc is not None: + # Do not launch a second child after a controller error. Wait for native exit. + proc.wait() + proc = None + stop.wait(30) + log("supervisor_stopped") + + +if __name__ == "__main__": + main() diff --git a/deploy/catchup/test_native_supervisor.py b/deploy/catchup/test_native_supervisor.py new file mode 100644 index 00000000..2b0afb4f --- /dev/null +++ b/deploy/catchup/test_native_supervisor.py @@ -0,0 +1,315 @@ +"""Process/real-etcd admission regressions; only disposable loopback fixtures. + +ETCD_BIN must name an etcd binary. Native work is a recording, blocking stub; +these tests prove supervisor admission/CAS/join behavior, not Lance throughput. +""" + +import base64 +import http.server +import json +import os +import shutil +import socket +import subprocess +import sys +import tempfile +import threading +import time +import unittest +import urllib.request +import uuid +from pathlib import Path + + +def enc(value): + return base64.b64encode(value.encode()).decode() + + +def free_port(): + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class SupervisorTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + binary = os.environ.get("ETCD_BIN") or shutil.which("etcd") + if not binary: + raise unittest.SkipTest("set ETCD_BIN to run the real-etcd regressions") + cls.directory = tempfile.TemporaryDirectory(prefix="native-supervisor-") + cls.endpoint = "http://127.0.0.1:" + str(free_port()) + peer = "http://127.0.0.1:" + str(free_port()) + cls.etcd_log = cls.enterClassContext( + open(Path(cls.directory.name) / "etcd.log", "w") + ) + cls.etcd = subprocess.Popen( + [ + binary, + "--name", + "test", + "--data-dir", + cls.directory.name + "/data", + "--listen-client-urls", + cls.endpoint, + "--advertise-client-urls", + cls.endpoint, + "--listen-peer-urls", + peer, + "--initial-advertise-peer-urls", + peer, + "--initial-cluster", + "test=" + peer, + ], + stdout=cls.etcd_log, + stderr=cls.etcd_log, + ) + try: + for _ in range(100): + if cls.etcd.poll() is not None: + raise RuntimeError("disposable etcd exited") + try: + cls.rpc("kv/range", {"key": enc("/health")}) + return + except OSError: + time.sleep(0.1) + raise RuntimeError("disposable etcd startup timed out") + except BaseException: + cls.tearDownClass() + raise + + @classmethod + def tearDownClass(cls): + cls.etcd.terminate() + cls.etcd.wait(timeout=10) + cls.etcd_log.close() + cls.directory.cleanup() + + @classmethod + def rpc(cls, path, body): + request = urllib.request.Request( + cls.endpoint + "/v3/" + path, + data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(request, timeout=3) as response: + return json.load(response) + + def setUp(self): + self.folder = Path(self.directory.name) / uuid.uuid4().hex + self.folder.mkdir() + self.prefix = "/tests/native-supervisor/" + self.folder.name + self.keys = { + name: self.prefix + "/" + suffix + for name, suffix in { + "owner": "catchup-active/" + b"table".hex(), + "operator": "merge-rollout-operator/" + b"table".hex(), + "attempt": "merge-rollout-attempts/" + b"table".hex(), + "failure": "merge-failures/" + + b"table".hex() + + "/" + + b"master:catchup".hex(), + }.items() + } + self.process = None + self.proxy = None + self.logs = self.enterContext(open(self.folder / "supervisor.log", "w")) + self.stub = self.folder / "native" + self.stub.write_text( + "#!" + sys.executable + "\n" + "import os,pathlib,time\n" + 'p=pathlib.Path(os.environ["FIXTURE_DIR"])\n' + 'with (p/"started").open("a") as f:f.write(str(os.getpid())+"\\n")\n' + "deadline=time.monotonic()+10\n" + 'while not (p/"release").exists() and time.monotonic()