diff --git a/.github/workflows/client.yml b/.github/workflows/client.yml index 25f8a9b..8d75a4b 100644 --- a/.github/workflows/client.yml +++ b/.github/workflows/client.yml @@ -96,6 +96,9 @@ jobs: - name: primary metadata however it's compressed (gz, EPEL 9's xz, EPEL 10's zst), by behaviour (runs the PR's own tools, so not in the gate) working-directory: . run: WEND_ALL_DECODERS=1 python3 -B tools/test_primary.py + - name: VM replays side by side (per-guest ports, the install lock, the box cache key), by behaviour with a stand-in QEMU (runs the PR's own vm.py, so not in the gate) + working-directory: . + run: python3 -B tools/test_vm_parallel.py - name: the atlas passes the client's own check working-directory: . run: client/target/x86_64-unknown-linux-musl/release/wend check diff --git a/SPEC.md b/SPEC.md index 635e3e7..9c8e053 100644 --- a/SPEC.md +++ b/SPEC.md @@ -1609,7 +1609,13 @@ The rules: runner that a pull request can trigger. Its runs go to `.wend/vm/` (schema `wend.vmrun/0`, `runner.kind = "vm"`), not to `runs/`. - QEMU user-mode networking only: nothing on the host changes. The guest reaches the mirror shim at `10.0.2.2`, and its - ssh and http are forwarded to fixed ports on `127.0.0.1`, so a host runs one VM replay at a time. + ssh is forwarded to a free port on `127.0.0.1` of its own, so a host runs several VM replays side by side. They + share the mirror shim and the installed boxes: a box is installed once per cache key (the box file, the adapter, + and the part of `tools/vm.py` the install runs), and a run that needs a box another run is installing waits for it. + Every run holds a shared lock on `/var/tmp/wend-shim/.users` for each shim port while it lives, whatever its + `$WEND_WORK` (a directory only root or the running user can write, or it's refused), and `tools/vm.py shim-stop PID` + stops the shim only when no run holds its port's lock. + Every run records the commit it ran as `head` (`commit`, and `dirty` when tracked files differ from it). - Box: an optional `[vm]` table with `ssh_port` (default 22), `skip` (roles that start their own daemons, which a running box already has) and `live` (role name to shell: the live forms of those checks; every skipped role needs one, so skipping never drops a check), `pockets` (`updates`, and `security` and/or `backports` beside it, once each: the box diff --git a/tools/sandbox.py b/tools/sandbox.py index 94aa8f6..9d92906 100644 --- a/tools/sandbox.py +++ b/tools/sandbox.py @@ -6,6 +6,7 @@ a hostname of the box's own (never the host's), daemons never start, and nothing is deleted until every mount under it is gone. """ +import fcntl import hashlib import os import subprocess @@ -257,6 +258,15 @@ def build_box(box_id, box_doc, adapter, box_toml_bytes, adapter_bytes, log=print root = f"{WORK}/boxes/{box_id.replace('/', '_')}-{key}" if os.path.exists(f"{root}/.wend-ready"): return Box(root) + os.makedirs(f"{WORK}/boxes", exist_ok=True) + with open(f"{root}.lock", "a") as lock: # runs side by side (tools/vm.py): one builds, the others wait for it + fcntl.flock(lock, fcntl.LOCK_EX) # closing the file drops it + if not os.path.exists(f"{root}/.wend-ready"): + build_unlocked(root, box_id, box_doc, adapter, log) + return Box(root) + + +def build_unlocked(root, box_id, box_doc, adapter, log): if os.path.exists(root): safe_rm(root) os.makedirs(root) @@ -269,4 +279,3 @@ def build_box(box_id, box_doc, adapter, box_toml_bytes, adapter_bytes, log=print raise RuntimeError(f"box {box_id} fails its own checks at the start: {failed}\n" + "\n".join(tail(h["checks"][c]["log"], 800) for c in failed)) open(f"{root}/.wend-ready", "w").close() - return Box(root) diff --git a/tools/test_vm_parallel.py b/tools/test_vm_parallel.py new file mode 100644 index 0000000..0d40dc6 --- /dev/null +++ b/tools/test_vm_parallel.py @@ -0,0 +1,190 @@ +"""VM replays side by side (tools/vm.py), by behaviour, the parts that don't need KVM: two guests started at once get +ssh forwards on different host ports and both run; a guest whose port another process takes first asks again for +another one; the mirror shim can't be stopped while a run uses it, whatever its WEND_WORK (shim-stop); a second run +that needs a box another run is installing waits for it; and the box cache key follows what the install runs, not the +whole file (build_source). QEMU is a stand-in that binds its hostfwd port as QEMU does and says what QEMU says when +it can't. It runs the PR's own vm.py, so the gate's schema job doesn't run it: the client job does. +python3 tools/test_vm_parallel.py""" +import os +import re +import shutil +import subprocess +import sys +import tempfile +import threading +import time + +sys.dont_write_bytecode = True +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(ROOT, "tools")) +tmp = tempfile.mkdtemp(prefix="wend-test-vm-") +os.environ["WEND_WORK"] = os.path.join(tmp, "work") # before vm reads it +import vm # noqa: E402 + +bad = 0 + + +def say(ok, what, detail=""): + global bad + print(("passed " if ok else "WRONG ") + what + ("" if ok or not detail else f"\n {detail}")) + bad += not ok + + +bin_dir = os.path.join(tmp, "bin") +os.makedirs(bin_dir) +open(os.path.join(bin_dir, "qemu-system-x86_64"), "w").write(f"""#!{sys.executable} +import re, socket, sys, time +net = sys.argv[sys.argv.index("-netdev") + 1] +port = int(re.search(r"hostfwd=tcp:127\\.0\\.0\\.1:(\\d+)-", net).group(1)) +s = socket.socket() +try: + s.bind(("127.0.0.1", port)) + s.listen() +except OSError: + sys.stderr.write(f"qemu-system-x86_64: -netdev {{net}}: Could not set up host forwarding rule 'tcp:127.0.0.1:{{port}}'\\n") + sys.exit(1) +time.sleep(60) +""") +os.chmod(os.path.join(bin_dir, "qemu-system-x86_64"), 0o755) +os.environ["PATH"] = bin_dir + os.pathsep + os.environ["PATH"] + +try: + # two guests at once: each its own port, both up, and ssh asks for that port + g1, g2 = (vm.Guest(os.path.join(tmp, f"g{i}.qcow2"), "key", 22) for i in (1, 2)) + g1.start() + g2.start() + say(g1.running() and g2.running(), "two guests run side by side") + say(g1.fwd != g2.fwd, "each guest forwards ssh from a host port of its own", f"{g1.fwd} and {g2.fwd}") + argv = g1.ssh_argv("true") + say(argv[argv.index("-p") + 1] == str(g1.fwd), "ssh goes to the guest's own forwarded port", str(argv)) + say(not re.search(r"hostfwd=tcp:127\.0\.0\.1:10022", " ".join(g1.argv())), "no fixed port is left in qemu's line") + + # a port taken between the pick and QEMU's bind: the guest asks again + real, picks = vm.free_port, [g1.fwd] + vm.free_port = lambda: picks.pop() if picks else real() + g3 = vm.Guest(os.path.join(tmp, "g3.qcow2"), "key", 22) + try: + g3.start() + say(g3.running() and g3.fwd not in (g1.fwd, g2.fwd), "a guest whose port was taken first starts on another one", + f"port {g3.fwd}") + except RuntimeError as e: + say(False, "a guest whose port was taken first starts on another one", str(e)) + finally: + vm.free_port = real + + # anything else QEMU refuses is not retried + g4 = vm.Guest(os.path.join(tmp, "g4.qcow2"), "key", 22) + open(os.path.join(bin_dir, "qemu-system-x86_64"), "w").write( + f"#!{sys.executable}\nimport sys\nsys.stderr.write('could not open disk image\\n')\nsys.exit(1)\n") + t0 = time.time() + try: + g4.start() + say(False, "qemu failing for another reason fails at once, not five times") + except RuntimeError as e: + say("could not open disk image" in str(e) and time.time() - t0 < 6, + "qemu failing for another reason fails at once, not five times", f"{time.time() - t0:.0f}s: {e}") + for g in vm.GUESTS: + g.kill() + shutil.rmtree(os.path.dirname(g.monitor_path), ignore_errors=True) + + # one run installs a box, a second waits for it, then finds it installed + lock, order = os.path.join(tmp, "work", "box.lock"), [] + + def holder(ready): + with vm.Locked(lock, "installing the box"): + ready.set() + time.sleep(1) + order.append("first done") + + ready = threading.Event() + t = threading.Thread(target=holder, args=(ready,)) + t.start() + ready.wait() + with vm.Locked(lock, "installing the box"): + order.append("second in") + t.join() + say(order == ["first done", "second in"], "a run waits while another installs the same box", str(order)) + + # the shim: shim-stop refuses while a run holds it, stops it once none does, and stops nothing else + fake = os.path.join(tmp, "checkout", "tools", "mirror.py") # a checkout's shim: vm.py and sandbox.py beside it + os.makedirs(os.path.dirname(fake)) + open(fake, "w").write("import time\ntime.sleep(60)\n") + for f in ("vm.py", "sandbox.py"): + open(os.path.join(os.path.dirname(fake), f), "w").close() + port = str(vm.free_port()) # a port no real shim uses, so its lock is this test's alone + shim = subprocess.Popen([sys.executable, fake, port, os.path.join(tmp, "cache")]) + # the run holds the port's lock under another WEND_WORK than shim-stop's: the lock follows the port + run = subprocess.Popen([sys.executable, "-B", "-c", "import sys, time; sys.path.insert(0, sys.argv[1]); import vm; " + "vm.use_shim([int(sys.argv[2])]); print('holding', flush=True); time.sleep(60)", + os.path.join(ROOT, "tools"), port], stdout=subprocess.PIPE, text=True, + env={**os.environ, "WEND_WORK": os.path.join(tmp, "other-work")}) + run.stdout.readline() + stop = lambda pid: subprocess.run([sys.executable, "-B", os.path.join(ROOT, "tools", "vm.py"), "shim-stop", str(pid)], + capture_output=True, text=True) # noqa: E731 + p = stop(shim.pid) + say(p.returncode == 3 and shim.poll() is None, "shim-stop leaves the shim running while a VM run uses it", + f"rc {p.returncode}: {p.stdout}{p.stderr}") + run.kill() + run.wait() + p = stop(shim.pid) + time.sleep(0.5) + say(p.returncode == 0 and shim.poll() is not None, "shim-stop stops the shim once no run uses it", + f"rc {p.returncode}: {p.stdout}{p.stderr}") + elsewhere = os.path.join(tmp, "srv", "mirror.py") # a mirror.py that isn't a wend checkout's + os.makedirs(os.path.dirname(elsewhere)) + shutil.copyfile(fake, elsewhere) + for what, argv in [("a process that isn't the shim", ["-c", "import time; time.sleep(60)"]), + ("another program's mirror.py", [elsewhere, port])]: + other = subprocess.Popen([sys.executable, *argv]) + p = stop(other.pid) + say(p.returncode == 2 and other.poll() is None, f"shim-stop refuses {what}", + f"rc {p.returncode}: {p.stdout}{p.stderr}") + other.kill() + shim.kill() + os.remove(os.path.join(vm.SHIM_LOCKS, f"{port}.users")) + + # a lock directory someone else could swap files in is refused, not used + real_locks = vm.SHIM_LOCKS + vm.SHIM_LOCKS = os.path.join(tmp, "open-locks") + os.mkdir(vm.SHIM_LOCKS) + os.chmod(vm.SHIM_LOCKS, 0o1777) + try: + vm.shim_lock(1) + say(False, "a world-writable lock directory is refused") + except SystemExit as e: + say("writable by nobody else" in str(e), "a world-writable lock directory is refused", str(e)) + vm.SHIM_LOCKS = real_locks + + # the cache key: what the install runs, and nothing it doesn't + src = open(os.path.join(ROOT, "tools", "vm.py")).read() + key = vm.build_source() + + def key_after(old, new): + assert src.count(old) == 1, old + p = os.path.join(tmp, "vm-changed.py") + open(p, "w").write(src.replace(old, new)) + return vm.build_source(p) + + for what, old in [("the hop walk", "def walk_vm("), ("the shelter's confinement proof", "def shelter_confine("), + ("relocate", "def relocate("), ("prune", "def prune("), ("main", "def main(")]: + say(key_after(f"\n{old}", f"\nUNUSED_BY_BUILD = 1\n\n\n{old}") == key, + f"a change beside {what} keeps the installed boxes") + say(key_after(" hops, start_net, efi_start = [], None, None\n", + " hops, start_net, efi_start = [], None, None # changed\n") == key, + "a change inside the hop walk keeps the installed boxes") + for what, old, new in [("the kickstart", 'def kickstart(box_doc, adapter, key_pub):\n', + 'def kickstart(box_doc, adapter, key_pub):\n pass\n'), + ("the install", "def install(box_id, box_doc, d):\n", + "def install(box_id, box_doc, d):\n pass\n"), + ("Guest's reboot", " def reboot(self):\n", " def reboot(self):\n pass\n"), + ("a constant the install uses", 'MAC = "52:54:00:12:34:56"', 'MAC = "52:54:00:12:34:57"'), + ("the box's checks", "def checks(g, box_doc, adapter):\n", + "def checks(g, box_doc, adapter):\n pass\n")]: + say(key_after(old, new) != key, f"a change to {what} installs the boxes again") +finally: + for g in vm.GUESTS: + g.kill() + shutil.rmtree(tmp, ignore_errors=True) + +print("ok" if not bad else f"{bad} WRONG") +sys.exit(1 if bad else 0) diff --git a/tools/vm.py b/tools/vm.py index 1c54ec0..c4b4c9a 100644 --- a/tools/vm.py +++ b/tools/vm.py @@ -11,6 +11,9 @@ python3 tools/vm.py build BOX install the box in a VM (cached by box file + adapter) python3 tools/vm.py replay ROUTE BOX walk ROUTE on a fresh copy of it, rebooting after every hop + python3 tools/vm.py shim-stop PID stop the mirror shim (tools/mirror.py) running as PID, unless a VM run is + using it (exit 3: left running); runs that start meanwhile wait, then start + a fresh one from their own checkout python3 tools/vm.py replay ROUTE BOX --from RUN HOP the same from RUN's disk after the hop before HOP (partial-pass at best) python3 tools/vm.py --extra-disk GB relocate BOX @@ -28,11 +31,20 @@ of the box first, copied into the guest before the hop that drops Python 2, and probed for confinement after every reboot from that hop on, as installed (SELinux permissive) and in an enforcing window. -Networking is QEMU user mode: nothing on the host changes. The guest reaches the mirror shim at 10.0.2.2. The guest's -ssh and http are forwarded to fixed ports on 127.0.0.1 (10022, 10080), so one host runs one VM replay at a time; a -second one fails at once ("qemu exited at once"). Disks go under $WEND_WORK/vm (default /var/tmp/wend/vm, which the -chroot harness, run as root, may own: point WEND_WORK somewhere the user can write). +Networking is QEMU user mode: nothing on the host changes. The guest reaches the mirror shim at 10.0.2.2. Each guest's +ssh is forwarded to a free port on 127.0.0.1 of its own, so one host runs several VM replays side by side (each guest +takes 2 CPUs and 3 GB); they share the mirror shim and the installed boxes. A box is installed once per cache key +(the box file, the adapter, and the part of this file the install runs: build_source), and a run that needs a box +another run is installing waits for it. A run holds a shared lock on /var/tmp/wend-shim/.users for each shim +port the adapters name, for as long as it lives, so the shim isn't stopped under it: stop the shim only with +`shim-stop`, which refuses while any run holds its port's lock. The lock follows the port, not $WEND_WORK: runs with +different work directories still share the one shim listening there. +Disks go under $WEND_WORK/vm (default /var/tmp/wend/vm, which the chroot +harness, run as root, may own: point WEND_WORK somewhere the user can write). Every run records the commit it ran +(run.json "head"), so a run on a stale head shows as one. """ +import ast +import fcntl import gzip import hashlib import http.client @@ -40,10 +52,12 @@ import json import os import re +import select import shlex import shutil import signal import socket +import stat import subprocess import sys import tempfile @@ -70,7 +84,6 @@ # The box's static address. Not 10.0.2.15, which slirp's DHCP hands out: the forwards point here, so a box that falls # back to DHCP is unreachable and fails the hop, instead of passing because DHCP happened to give the same address. ADDR = "10.0.2.20" -SSH_FWD, HTTP_FWD = 10022, 10080 # The firmware the guests boot: "bios" (SeaBIOS) or "uefi" (OVMF without Secure Boot, the host's edk2-ovmf build). # A UEFI guest's NVRAM (boot entries) is a copy of OVMF's variable store next to each disk, carried along every overlay. FIRMWARE = "bios" @@ -136,6 +149,183 @@ def log(msg): print(time.strftime("%H:%M:%S"), msg, flush=True) +def free_port(): + """A port on 127.0.0.1 nothing listens on right now, for a guest's ssh forward. Another process can take it before + QEMU binds it: Guest.start asks again when QEMU says so.""" + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +class Locked: + """An exclusive flock on `path` (created if need be) for the with block: one run installs a box, the others wait + for it and then use it. Said once when it has to wait.""" + + def __init__(self, path, what): + self.path, self.what = path, what + + def __enter__(self): + os.makedirs(os.path.dirname(self.path), exist_ok=True) + self.f = open(self.path, "a") + try: + fcntl.flock(self.f, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + log(f"another run is {self.what}; waiting for it") + fcntl.flock(self.f, fcntl.LOCK_EX) + return self + + def __exit__(self, *exc): + self.f.close() # closing the file drops the lock + + +# Where the shim locks live: one per port, the same for every run on the host whatever its $WEND_WORK, since the +# port is what runs share (the first run's shim serves them all). Read-only opens, so runs as different users share it. +SHIM_LOCKS = "/var/tmp/wend-shim" +HOLDING = [] # this run's own holds + + +def shim_lock(port): + """The open lock file for a shim port, in SHIM_LOCKS. Not /var/tmp itself: anyone can unlink and recreate a file in + a sticky world-writable directory, and a shim-stop that locked the replacement would stop a shim a run is using. + So the directory must be a real directory owned by root or by this user (or the user sudo ran as), writable by + nobody else; anything else is refused, never used.""" + try: + os.mkdir(SHIM_LOCKS) + os.chmod(SHIM_LOCKS, 0o755) # whatever the umask: runs as other users open the locks read-only + except FileExistsError: + pass + st = os.lstat(SHIM_LOCKS) + owners = {0, os.geteuid()} | ({int(os.environ["SUDO_UID"])} if os.environ.get("SUDO_UID", "").isdigit() else set()) + if not stat.S_ISDIR(st.st_mode) or st.st_uid not in owners or st.st_mode & 0o022: + raise SystemExit(f"{SHIM_LOCKS} isn't a directory owned by root or this user and writable by nobody else; " + "refusing to use the mirror shim's locks there") + fd = os.open(os.path.join(SHIM_LOCKS, f"{int(port)}.users"), os.O_RDONLY | os.O_CREAT | os.O_NOFOLLOW, 0o644) + if os.fstat(fd).st_uid == os.geteuid(): # whatever the umask, as the directory + os.fchmod(fd, 0o644) + return fd + + +def shim_ports(): + """The ports of every adapter's [mirror] shim.""" + out = set() + for f in sorted(os.listdir(os.path.join(ROOT, "adapters"))): + if f.endswith(".toml"): + a = load(f"adapters/{f}") + if "mirror" in a: + out.add(int(a["mirror"]["shim"].rsplit(":", 1)[1])) + return sorted(out) + + +def use_shim(ports=None): + """Hold a shared lock for each shim port (shim_ports) until this process exits (the kernel drops them then, + however it ends), before anything starts or reads a shim. Waits while a shim-stop holds one exclusively.""" + if not HOLDING: + for port in shim_ports() if ports is None else ports: + fd = shim_lock(port) + fcntl.flock(fd, fcntl.LOCK_SH) + HOLDING.append(fd) + + +def shim_stop(pid): + """Stop the mirror shim running as pid, only while no VM run holds the lock of the port it listens on (from its + own command line): the exclusive lock is held through the kill, so a run that starts meanwhile waits for it and + then starts its own shim. 0 stopped (or wasn't running), 3 in use and left running, 2 pid isn't a mirror shim, + 1 it didn't exit.""" + fd = None + try: + # a pidfd names this process, not its number: if it exits and the number is reused, signals through it fail + # instead of reaching the newcomer. /proc/ is read after it's open and before the first signal, while + # the pidfd says the process is still alive, so the files read are that process's. + try: + pfd = os.pidfd_open(pid) + except ProcessLookupError: + print(f"no process {pid}: nothing to stop") + return 0 + try: + try: + cmd = [a.decode(errors="replace") for a in open(f"/proc/{pid}/cmdline", "rb").read().split(b"\0")] + cwd = os.readlink(f"/proc/{pid}/cwd") + signal.pidfd_send_signal(pfd, 0) + except (FileNotFoundError, ProcessLookupError): + print(f"no process {pid}: nothing to stop") + return 0 + # a wend checkout's tools/mirror.py (a tools/ directory with vm.py and sandbox.py beside it), not any + # mirror.py + # mirror.py, then its PORT and CACHE_DIR + at = next((i for i in (1, 2) if i < len(cmd) and os.path.basename(cmd[i]) == "mirror.py"), None) + t = os.path.dirname(os.path.join(cwd, cmd[at])) if at else "" + if not (at and os.path.basename(t) == "tools" and + all(os.path.exists(os.path.join(t, f)) for f in ("vm.py", "sandbox.py")) and + at + 1 < len(cmd) and cmd[at + 1].isdigit()): + print(f"process {pid} isn't a wend checkout's tools/mirror.py: {' '.join(cmd)[:200]}") + return 2 + fd = shim_lock(cmd[at + 1]) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + print(f"the mirror shim on port {cmd[at + 1]} is in use by a VM run; left running (pid {pid})") + return 3 + for sig, wait, said in ((signal.SIGTERM, 10, "stopped"), + (signal.SIGKILL, 10, "killed (it ignored SIGTERM)")): + try: + signal.pidfd_send_signal(pfd, sig) + except ProcessLookupError: # gone already + sig = None + if sig is None or select.select([pfd], [], [], wait)[0]: # a pidfd reads ready once its process exits + print(f"{said if sig else 'stopped'} the mirror shim (pid {pid})") + return 0 + print(f"the mirror shim (pid {pid}) didn't exit after SIGKILL") + return 1 + finally: + os.close(pfd) + finally: + if fd is not None: + os.close(fd) + + +def head(): + """The commit this checkout is on and whether tracked files differ from it, or None outside a git checkout. As + root (sudo -E) in a checkout someone else owns, git refuses the repository unless told it's safe; this only reads, + and with what the checkout's own config could make git run switched off (core.fsmonitor is a command `git status` + runs, and submodules would run git in other repositories).""" + git = ["git", "-c", "safe.directory=*", "-c", "core.fsmonitor=false", "-C", ROOT] + try: + rev = subprocess.run(git + ["rev-parse", "HEAD"], capture_output=True, text=True, timeout=60) + st = subprocess.run(git + ["status", "--porcelain", "--untracked-files=no", "--ignore-submodules"], capture_output=True, text=True, + timeout=60) + except (OSError, subprocess.TimeoutExpired): + return None + if rev.returncode or st.returncode: + return None + return {"commit": rev.stdout.strip(), "dirty": bool(st.stdout.strip())} + + +def build_source(path=None): + """This file's part of the box cache key: the source of build() and of every top-level function, class and + assignment in this file that it reaches by name, through any of them. A change to what only a replay runs (the + hops, shelters, relocate, prune) keeps the installed boxes; a change to anything the install runs (the kickstart, + the preseed, Guest, the box's checks) installs them again.""" + src = open(path or os.path.abspath(__file__)).read() + defs = {} + for node in ast.parse(src).body: + if isinstance(node, (ast.FunctionDef, ast.ClassDef)): + defs.setdefault(node.name, []).append(node) + elif isinstance(node, (ast.Assign, ast.AnnAssign)): + for t in (node.targets if isinstance(node, ast.Assign) else [node.target]): + for n in ast.walk(t): + if isinstance(n, ast.Name): + defs.setdefault(n.id, []).append(node) + seen, todo = set(), ["build"] + while todo: + name = todo.pop() + if name in seen or name not in defs: + continue + seen.add(name) + todo += [n.id for node in defs[name] for n in ast.walk(node) if isinstance(n, ast.Name)] + nodes = {id(node): node for name in sorted(seen) for node in defs[name]} + return "\n".join(ast.get_source_segment(src, n) for n in sorted(nodes.values(), key=lambda n: n.lineno)).encode() + + def load(path): return tomllib.load(open(os.path.join(ROOT, path), "rb")) @@ -193,14 +383,12 @@ def hardware(adapter, release): class Guest: def __init__(self, disk, key, ssh_port, hw=("virtio", "virtio")): self.disk, self.key, self.ssh_port, self.hw = disk, key, ssh_port, hw + self.fwd = None # the host port forwarded to the guest's ssh, picked when it starts base = disk[:-len(".qcow2")] self.console, self.monitor_path, self.proc = base + ".console.log", None, None self.reboot_notes = [] # what went wrong reading boots in the last reboot(), for the run record - def start(self, extra=()): - # QEMU's monitor socket: a unix socket path has to fit in 108 bytes, which a run directory under a checkout - # doesn't, so it lives in a short directory of its own (removed on the way out, with the guest) - self.monitor_path = os.path.join(tempfile.mkdtemp(prefix="wend-vm-"), "monitor") + def argv(self, extra=()): fw = [] if FIRMWARE == "bios" else [ "-drive", f"if=pflash,format=raw,readonly=on,file={OVMF_CODE}", "-drive", f"if=pflash,format=raw,file={nvram(self.disk)}"] @@ -208,14 +396,29 @@ def start(self, extra=()): *[x.format(disk=self.disk) for x in DISKS[self.hw[0]]], *[x.format(disk=e, i=i) for i, e in enumerate(extras(self.disk), 1) for x in DISKS_EXTRA[self.hw[0]]], "-netdev", - f"user,id=n0,hostfwd=tcp:127.0.0.1:{SSH_FWD}-{ADDR}:{self.ssh_port},hostfwd=tcp:127.0.0.1:{HTTP_FWD}-{ADDR}:80", + f"user,id=n0,hostfwd=tcp:127.0.0.1:{self.fwd}-{ADDR}:{self.ssh_port}", "-device", f"{NICS[self.hw[1]]},netdev=n0,mac={MAC}", "-display", "none", "-serial", f"file:{self.console}", "-monitor", f"unix:{self.monitor_path},server,nowait", *extra] - self.proc = subprocess.Popen(argv, stdout=subprocess.DEVNULL, stderr=open(self.console + ".qemu", "a")) - GUESTS.append(self) - time.sleep(2) - if self.proc.poll() is not None: - raise RuntimeError(f"qemu exited at once: {open(self.console + '.qemu').read()[-800:]}") + return argv + + def start(self, extra=()): + for attempt in range(5): + # QEMU's monitor socket: a unix socket path has to fit in 108 bytes, which a run directory under a + # checkout doesn't, so it lives in a short directory of its own (removed on the way out, with the guest) + self.monitor_path = os.path.join(tempfile.mkdtemp(prefix="wend-vm-"), "monitor") + self.fwd = free_port() + err = self.console + ".qemu" + at = os.path.getsize(err) if os.path.exists(err) else 0 + self.proc = subprocess.Popen(self.argv(extra), stdout=subprocess.DEVNULL, stderr=open(err, "a")) + GUESTS.append(self) + time.sleep(2) + if self.proc.poll() is None: + return + said = open(err, errors="replace").read()[at:] + shutil.rmtree(os.path.dirname(self.monitor_path), ignore_errors=True) + if "host forwarding" not in said: # anything but another process taking the port first + break + raise RuntimeError(f"qemu exited at once: {said[-800:]}") def ssh(self, cmd, timeout=600, stdin=None): try: @@ -244,7 +447,7 @@ def push_tree(self, root, path, timeout=1800): return p.returncode or tar.returncode, (p.stdout + p.stderr + err.read()).decode(errors="replace") def ssh_argv(self, cmd): - return ["ssh", "-q", "-i", self.key, "-p", str(SSH_FWD), "-o", "StrictHostKeyChecking=no", + return ["ssh", "-q", "-i", self.key, "-p", str(self.fwd), "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", "-o", "ServerAliveInterval=15", "-o", "LogLevel=ERROR", # CentOS 5's OpenSSH 4.3 offers only these; later releases negotiate the modern ones @@ -537,13 +740,21 @@ def build(box_id): box_doc = load(f"boxes/{box_id}/box.toml") fam = box_doc["family"] raw = open(os.path.join(ROOT, f"boxes/{box_id}/box.toml"), "rb").read() + \ - open(os.path.join(ROOT, f"adapters/{fam}.toml"), "rb").read() + open(os.path.abspath(__file__), "rb").read() + \ + open(os.path.join(ROOT, f"adapters/{fam}.toml"), "rb").read() + build_source() + \ FIRMWARE.encode() + (json.dumps(LAYOUT, sort_keys=True).encode() if layout_tag() else b"") tag = "-".join(filter(None, [FIRMWARE, layout_tag()])) # the default layout keeps the cache key it had d = os.path.join(WORK, f"{box_id.replace('/', '_')}-{tag}-{hashlib.sha256(raw).hexdigest()[:12]}") + if not os.path.exists(os.path.join(d, "base.qcow2")): + with Locked(d + ".lock", f"installing {box_id} for this key"): + if not os.path.exists(os.path.join(d, "base.qcow2")): # it may have, while this one waited + install(box_id, box_doc, d) + return d, box_doc + + +def install(box_id, box_doc, d): + """build()'s install, into the cache directory d, under its lock.""" + fam = box_doc["family"] base = os.path.join(d, "base.qcow2") - if os.path.exists(base): - return d, box_doc inst = load(f"adapters/{fam}.toml")["releases"][box_doc["release"]]["install"] root_fs = LAYOUT["root_fs"] or box_doc.get("vm", {}).get("root_fs") if root_fs and root_fs not in inst.get("root_fs", []): @@ -642,14 +853,14 @@ def build(box_id): if failed: raise RuntimeError(f"box {box_id} fails its own checks in the VM: {sorted(failed)}\n" + "\n".join(f"{n}: {o[-600:]}" for n, o in failed.items())) + # base.qcow2 last: a run that sees it (build(), without the lock) finds its extra disks and NVRAM beside it for x, b in zip(extras(disk), extras(base)): os.rename(x, b) sandbox.host(f"chmod a-w {b}") - os.rename(disk, base) - sandbox.host(f"chmod a-w {base}") if FIRMWARE == "uefi": os.rename(nvram(disk), nvram(base)) - return d, box_doc + os.rename(disk, base) + sandbox.host(f"chmod a-w {base}") def pocket_sources(release, pockets): @@ -837,6 +1048,9 @@ def replay(route_path, box_id, resume=None): stamp = time.strftime("%Y%m%d-%H%M%S") tag = "-".join(filter(None, [FIRMWARE, layout_tag()])) out_dir = os.path.join(ROOT, ".wend", "vm", f"{stamp}-{box_id.replace('/', '_')}-{tag}-{os.getpid()}") + w.head = head() + at = w.head or {} + log(f"replaying at {at.get('commit', 'no git head')}{' (with local changes)' if at.get('dirty') else ''}") os.makedirs(out_dir) port = box_doc.get("vm", {}).get("ssh_port", 22) hw = hardware(a, route["from"]) @@ -873,7 +1087,7 @@ def walk_vm(route, route_path, box_id, box_doc, a, w, pinned, todo, resume, out_ g.start() if not g.wait_ssh(): hops.append({"to": hop, "ok": False, "stage": "the guest didn't come up before the hop"}) - g.kill() # free the forwarded ports; the overlay and console stay for a look + g.kill() # free the forwarded port; the overlay and console stay for a look break if start_net is None: start_net = net_read(g, plan) @@ -976,7 +1190,7 @@ def walk_vm(route, route_path, box_id, box_doc, a, w, pinned, todo, resume, out_ hops += prune(a, w, route["to"], box_doc, out_dir, prev, key, port, hw, start_net, plan) complete = len(hops) == len(todo) + bool(route.get("finish", {}).get("prune_obsolete")) run = {"schema": "wend.vmrun/0", "route": check.content_id(route), "box": box_id, "from": route["from"], - "to": route["to"], "started": stamp, "firmware": FIRMWARE, + "to": route["to"], "started": stamp, "head": w.head, "firmware": FIRMWARE, **({"efi_start": efi_start} if FIRMWARE == "uefi" else {}), "layout": {k: v for k, v in LAYOUT.items() if v}, "hops": hops, "result": "pass" if complete and all(h["ok"] for h in hops) and not resume else @@ -1029,7 +1243,7 @@ def relocate(box_id): g = Guest(disk, os.path.join(d, "key"), box_doc.get("vm", {}).get("ssh_port", 22), hw) g.start() stages = [] - run = {"schema": "wend.vmrelocate/0", "box": box_id, "started": stamp, "firmware": FIRMWARE, + run = {"schema": "wend.vmrelocate/0", "box": box_id, "started": stamp, "head": head(), "firmware": FIRMWARE, "layout": {k: v for k, v in LAYOUT.items() if v}, "new_disk": fills["new_disk"], "stages": stages, "runner": {"kind": "vm", "host": os.uname().nodename, "kernel": os.uname().release}} @@ -1203,6 +1417,10 @@ def main(args): LAYOUT["extra_disks"], args = LAYOUT["extra_disks"] + [int(args[1])], args[2:] else: return __doc__ + if args[:1] == ["shim-stop"] and len(args) == 2 and args[1].isdigit(): + return shim_stop(int(args[1])) + if args[:1] in (["build"], ["replay"], ["relocate"]): + use_shim() if args[:1] == ["build"] and len(args) == 2: build(args[1]) return 0 @@ -1221,6 +1439,6 @@ def main(args): try: sys.exit(main(sys.argv[1:])) finally: - for guest in GUESTS: # an exception or Ctrl-C must not leave a guest holding 3 GB and the forwarded ports + for guest in GUESTS: # an exception or Ctrl-C must not leave a guest holding 3 GB and its forwarded port guest.kill() shutil.rmtree(os.path.dirname(guest.monitor_path), ignore_errors=True) diff --git a/wend b/wend index 87fec24..3019b89 100755 --- a/wend +++ b/wend @@ -283,7 +283,8 @@ def main(argv): m = 0 if rest else subprocess.run([sys.executable, "-B", f"{ROOT}/tools/test_primary.py"]).returncode o = 0 if rest else subprocess.run([sys.executable, "-B", f"{ROOT}/tools/test_r2push.py"]).returncode h = 0 if rest else subprocess.run([sys.executable, "-B", f"{ROOT}/tools/test_gate_fetched.py"]).returncode - return a or b or u or g or n or i or k or e or r or p or f or s or d or v or m or o or h + q = 0 if rest else subprocess.run([sys.executable, "-B", f"{ROOT}/tools/test_vm_parallel.py"]).returncode + return a or b or u or g or n or i or k or e or r or p or f or s or d or v or m or o or h or q if c == "list": return cmd_list() if c == "replay":