diff --git a/.github/actions/validate-nvx/action.yml b/.github/actions/validate-nvx/action.yml index 364367cf7..6c9c9763c 100644 --- a/.github/actions/validate-nvx/action.yml +++ b/.github/actions/validate-nvx/action.yml @@ -10,7 +10,7 @@ runs: run: | set -euo pipefail python3 -m compileall -q scripts - python3 -m unittest scripts/test_performance.py scripts/test_nvx_tools.py scripts/test_microvm_tests.py scripts/test_development_release.py scripts/test_egress_policy.py scripts/test_control_session.py scripts/test_managed_agent.py scripts/test_guest_time.py scripts/test_time_abi.py -v + python3 -m unittest scripts/test_performance.py scripts/test_nvx_tools.py scripts/test_microvm_tests.py scripts/test_development_release.py scripts/test_egress_policy.py scripts/test_control_session.py scripts/test_state_control.py scripts/test_managed_agent.py scripts/test_guest_time.py scripts/test_time_abi.py -v python3 scripts/test_adversarial.py -v python3 .github/skills/nvx-host-connect/scripts/test_hosts.py -v python3 scripts/nvx.py --help @@ -29,7 +29,7 @@ runs: $ErrorActionPreference = "Stop" python -m compileall -q scripts if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } - python -m unittest scripts/test_performance.py scripts/test_nvx_tools.py scripts/test_microvm_tests.py scripts/test_development_release.py scripts/test_egress_policy.py scripts/test_control_session.py scripts/test_managed_agent.py scripts/test_guest_time.py scripts/test_time_abi.py -v + python -m unittest scripts/test_performance.py scripts/test_nvx_tools.py scripts/test_microvm_tests.py scripts/test_development_release.py scripts/test_egress_policy.py scripts/test_control_session.py scripts/test_state_control.py scripts/test_managed_agent.py scripts/test_guest_time.py scripts/test_time_abi.py -v if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } python scripts\test_adversarial.py -v if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } diff --git a/doc/ci.md b/doc/ci.md index bffd376f8..46940829f 100644 --- a/doc/ci.md +++ b/doc/ci.md @@ -40,8 +40,8 @@ pending checks. The test runners ask after every cold boot whose shell is on a console the harness reads, the OpenVMM console or a virtio console over TCP, before any other input, and wait for the query to exit, so the console's echo of later input cannot split its lines. Cold boots with no shell there are -only scanned: one-shot workloads, and the managed lifecycle, where init starts -the managed agent instead of a shell. A failing boot check still powers them +only scanned: one-shot workloads, and the managed lifecycle and `pause-resume` +scenarios, where init starts the managed agent instead of a shell. A failing boot check still powers them off with status 193, which fails the run. The query must exit 0 with a passing `NVX-TIME-ABI` boot line, which reports ABI version 1, generation 0, a plausible TSC rate, and the backend's LAPIC rate; a missing @@ -99,6 +99,30 @@ harness stages its check, report `/sys/kernel/rcu_stall_count` as 0 two seconds later, and show an uptime of at least 30 s, which proves that monotonic time advanced by the downtime. The captures share one downtime window, so the scenario adds about a minute per backend. +The `pause-resume` scenario covers [host pause](design/time-abi.md#host-pause). +It boots the largest requested vCPU count with the managed lifecycle and +OpenVMM's state-control endpoint (`--microvm-state-control`), which +authenticates hosts with the control console's capability. A host that does +not authenticate, or presents the wrong capability, must get no response. +While a managed workload spins on one vCPU, the harness pauses the VM for +30 s, longer than the guest's 21 s RCU stall timeout and the debug kernel's +20 s soft-lockup threshold, and then twice for 2 s. It repeats every pause and +resume request, which must change nothing the second time, queries the run +state while paused, and requires the transition count to rise by one on every +change. On Linux hosts, the OpenVMM process may use at most a tenth of a +pause, plus 0.2 s, in CPU time while paused. Afterwards the workload must +finish, having started at least 0.5 s of guest uptime before the long pause +and stopped after it, guest uptime must have advanced by the time the VM ran to +within 1 s, and `/sys/kernel/rcu_stall_count` must read 0. Host UTC kept +running, so the guest's wall clock must return to within 1.5 s of host UTC +within 150 s, through at least one wall-clock discipline step that +`/run/nvx/time/state` records, with no new violation there. A second host, +connected after the first one disconnects, must see the same broker instance +and transition count, and pauses the VM again; a third host must then find it +still paused. Finally the harness terminates OpenVMM, which must still be +running and must exit with the termination status. The scenario prints one +`NVX-PAUSE-RESUME:` line with its measurements and adds about a minute per +backend. The `time-abi-conformance` scenario boots the largest requested vCPU count and runs the guest's exhaustive CI check, `/sbin/nvx-time exhaustive`, which the boot check leaves to CI. On every online CPU it checks every leaf @@ -144,7 +168,8 @@ on the CI debug kernel (`build/vmlinux-debug`, built from `kernel/config-microvm-debug`), whose soft-lockup and hung-task detectors production kernels leave out. It selects the same-host restore scenarios `smp`, `smp-snapshot`, `restore-processors`, `restore-downtime`, and -`snapshot-tiers`. The jobs skip every other test step, because those boot the +`snapshot-tiers`, and the host pause scenario `pause-resume`. The jobs skip +every other test step, because those boot the production kernel, which the `nvx-microvm-tests-{kvm,mshv,whp}` jobs already cover: the public managed execution configuration, the Ubuntu and Azure Linux guest tests, the Ubuntu sandbox layer and live-share smoke tests, the @@ -159,7 +184,7 @@ whose guest OpenVMM stops at the restore gate on purpose. The harness refuses a kernel whose `vmlinux-debug.config` lacks the detectors, because the guest's `C11` check passes vacuously without them. To bound the cost, pull requests run the debug kernel on KVM only and `dev` pushes run it on every backend; -each job takes about five minutes on its own runner, in parallel with the +each job takes about six minutes on its own runner, in parallel with the other microVM jobs. The jobs gate the required status check, the development release, and performance persistence. The GitHub-hosted `debug-kernel` job builds the debug kernel beside the shared `artifacts` job diff --git a/doc/design/concurrency-and-trust-boundaries.md b/doc/design/concurrency-and-trust-boundaries.md index 4bef0dc5a..acd6ee7a6 100644 --- a/doc/design/concurrency-and-trust-boundaries.md +++ b/doc/design/concurrency-and-trust-boundaries.md @@ -23,7 +23,11 @@ the OpenVMM user, requires a launcher-provided capability before any host data reaches the guest, and bounds every record, queue, and credit window in both directions; malformed, stale, or out-of-order records fail closed. The capability never appears in arguments, environment variables, logs, snapshots, -or attachment identities. +or attachment identities. The state-control endpoint, which pauses and resumes +the VM without reaching the guest, admits the same local peer and requires the +same capability before it answers. It serves one host and one request at a +time; a failed or stalled authentication, a malformed request, or 60 s without +a request closes the connection without a response. Host-facing outputs are bounded as well. The optional outcome report is written after teardown to a previously absent local path and records only a schema diff --git a/doc/design/host-attachment-model.md b/doc/design/host-attachment-model.md index 6b5dace4e..b0b4c9618 100644 --- a/doc/design/host-attachment-model.md +++ b/doc/design/host-attachment-model.md @@ -14,6 +14,7 @@ Each external resource has a stable ID and a declarative reconstruction policy. | filesystem | FUSE namespace, handles, cookies, root/object identity, access mode, and denied paths, or explicit dormant state | Fresh host-directory attachment with the same canonical path, target, mode, and denied paths; a dormant slot may stay unattached or bind a new attachment | | sandbox block | Queue/device state, fixed roles, access, geometry, read-only layer identities, and scratch policy | Matching read-only layers plus a verified private paired scratch copy, or a new same-geometry scratch file | | Readiness endpoint | Nothing | Optional single-use Unix socket or named pipe supplied for one restore | +| State-control endpoint | Nothing | Optional same-user Unix socket or Windows named pipe, supplied with a live control console and authenticated with its fresh capability | Attachment resolution happens before vCPU start. A supplied attachment must reproduce the saved identity except that an eligible listener may replace only diff --git a/doc/design/machine-and-device-abi.md b/doc/design/machine-and-device-abi.md index 40fa0c15b..254744a30 100644 --- a/doc/design/machine-and-device-abi.md +++ b/doc/design/machine-and-device-abi.md @@ -299,6 +299,16 @@ The workload protocol carried inside the brokered data stream belongs to the guest agent; see [Control protocol and checkpoint handoff](sandbox-filesystem-and-agent-architecture.md#control-protocol-and-checkpoint-handoff). +A machine with a live, authenticated control console may also bind a +state-control endpoint (`--microvm-state-control listen=`), through +which the host pauses, resumes, and queries the VM without a snapshot; +[Host pause](time-abi.md#host-pause) defines guest time across a pause. The +endpoint follows the private-endpoint and peer-identity rules above, must +differ from both console endpoints, and answers a host only after it presents +the control console's capability. OpenVMM serves it outside the VM's devices, +so it keeps answering while the VM is paused. Snapshots never record it; a +restore may supply a fresh one. + ### Network The optional NIC has one RX/TX queue pair and an exact feature mask: diff --git a/doc/design/time-abi.md b/doc/design/time-abi.md index e66f69f43..3923135a1 100644 --- a/doc/design/time-abi.md +++ b/doc/design/time-abi.md @@ -37,10 +37,12 @@ Goals: rejected with a stable error code. - Restore within one backend across hosts of the same CPU generation and CPU profile, including restores after a host reboot. -- Guest monotonic time advances by the snapshot downtime. Wall clock is set +- Guest monotonic time advances by the snapshot downtime and stands still + while the host pauses the VM ([Host pause](#host-pause)). Wall clock is set on every restore and disciplined afterwards. - RCU-stall, soft-lockup, hung-task, clocksource, and TSC-warp detectors never - trip across capture and restore. A detected violation fails fast. + trip across capture, restore, and host pause. A detected violation fails + fast. - No performance regression beyond the gate in [Performance expectations and acceptance gate](#performance-expectations-and-acceptance-gate). @@ -878,6 +880,40 @@ round trip on Azure takes about 12 µs, so nested MSHV pairs at a median of about 1.5× margin over the median and fail restores at random. Downtime needs only millisecond accuracy, so a 100 µs pairing error is negligible. +## Host pause + +The host can pause a running microVM and later resume it in the same VM +process, without a snapshot, through OpenVMM's state-control endpoint +(`--microvm-state-control`; see the [appendix](#appendix-openvmm-guide)). +Where a restore advances guest time by the downtime, a host pause holds it: + +- **Monotonic time.** Once the vCPUs stop, OpenVMM takes VP 0's guest TSC and + every vCPU's LAPIC state. Resume writes that TSC to every vCPU with the + [synchronized TSC set](#backend-obligations), writes the LAPIC states back + before any vCPU runs, and then starts the VM. The TSC, the LAPIC timers, and + the VM time behind the emulated timers continue from where they stopped, so + guest `CLOCK_MONOTONIC`, `CLOCK_BOOTTIME`, and jiffies count only the time + the VM ran, and no RCU-stall, soft-lockup, or hung-task detector sees a + pause of any length. The generation counter does not change, and no restore + repair runs. +- **Rejection.** As at capture, a pause is rejected when a vCPU has an armed + periodic LAPIC timer, which a guest under this ABI arms at most before its + first tick, or a TSC-deadline timer, which it never arms (see + [Clocksource, tick, and PIT](#clocksource-tick-and-pit)). The vCPUs then + start again, and the guest sees only a brief stall. +- **Wall clock.** Host UTC keeps running, and so do the CMOS RTC's date and + time and the [time sample](#time-sample). After a resume, guest + `CLOCK_REALTIME` is behind host UTC by the paused time until the + [wall-clock discipline](#wall-clock-discipline)'s next poll, at most 64 s + of guest time later, steps it and counts a discontinuity + (`last_discontinuity=step`); a lag under 128 ms is slewed instead. The + discipline follows it as it follows any host wall-clock step, without a + violation. +- **Snapshots.** A guest snapshot request still pending when a pause stops + the vCPUs is cancelled, as when the VM stops for any other reason. A pause + at a snapshot boundary or during the post-restore gate is refused as busy + and changes nothing. + ## Restore algorithm Restore runs these steps in order. Any failure rejects the restore before a @@ -2180,8 +2216,8 @@ harness reads, and after every restore in the `smp-snapshot`, `restore-processors`, `restore-downtime`, and `snapshot-tiers` scenarios, except `snapshot-tiers`' gate-timeout check, whose guest OpenVMM stops at the restore gate on purpose. CI's other restores and its guests without a shell, -the managed lifecycle and one-shot workloads, are only scanned for violation -events; a failed check still powers the guest off with status 193. Each query +the managed lifecycle, host pause, and one-shot workloads, are only scanned +for violation events; a failed check still powers the guest off with status 193. Each query requires exit status 0 and, after a restore, a `phase=restore` line with `status=ok`, the restored `generation`, and `cpus` equal to the online CPUs; after a cold boot, a `phase=boot` line with @@ -2234,6 +2270,13 @@ watcher active, report no RCU, soft-lockup, hung-task, clocksource, or warp message. The CI debug-kernel variant runs the same-host cases with the soft-lockup and hung-task detectors enabled. +**Host pause.** CI's [`pause-resume`](../ci.md) scenario pauses a managed VM +for 30 s, longer than the 21 s RCU stall timeout, while a workload spins on +one vCPU, and then twice for 2 s, on every backend and on the debug kernel. +It checks the rules of [Host pause](#host-pause): guest uptime counts only the +time the VM ran, to within 1 s; `rcu_stall_count` stays 0, with no violation; +and the guest's wall clock returns to host UTC through a discipline step. + **Performance.** The gate above, computed by the performance agent from the CI benchmark matrix. @@ -2356,3 +2399,8 @@ The OpenVMM Guide documents the user-facing contract in The CLI reference documents `--cpu-profile` and `--x-time-abi-verify`. The test hooks stay undocumented in the Guide; this document describes them. + +`Guide/src/reference/openvmm/management/state_control_protocol.md` documents +[host pause](#host-pause): the state-control endpoint and its authentication, +the protocol, and guest time across a pause. The CLI reference documents +`--microvm-state-control`. diff --git a/doc/usage.md b/doc/usage.md index bc213926a..ab5f7ecf9 100644 --- a/doc/usage.md +++ b/doc/usage.md @@ -263,7 +263,8 @@ the Alpine-prompt-specific `console-snapshot` scenario. `--debug-kernel` boots the CI debug kernel, `build/vmlinux-debug`, whose soft-lockup and hung-task detectors the guest's time ABI watcher reports; without `--scenario`, it runs only the same-host restore scenarios `smp`, `smp-snapshot`, -`restore-processors`, `restore-downtime`, and `snapshot-tiers`. The command +`restore-processors`, `restore-downtime`, and `snapshot-tiers`, and the host +pause scenario `pause-resume`. The command requires `build/vmlinux` (with `--debug-kernel`, `build/vmlinux-debug` and its `build/vmlinux-debug.config`), the selected initramfs, and `openvmm/target/release/openvmm[.exe]`. diff --git a/openvmm b/openvmm index 07850d27a..647e635c3 160000 --- a/openvmm +++ b/openvmm @@ -1 +1 @@ -Subproject commit 07850d27af783ea8765111741763cb6d326228df +Subproject commit 647e635c308b2ca463843153116582c4192d0e1c diff --git a/scripts/nvx_tools/control_session.py b/scripts/nvx_tools/control_session.py index b0fbf2e2f..b04b6b7c3 100644 --- a/scripts/nvx_tools/control_session.py +++ b/scripts/nvx_tools/control_session.py @@ -10,7 +10,7 @@ import time from dataclasses import dataclass from pathlib import Path -from typing import Any, cast +from typing import Any, Protocol, cast from .common import ScriptError, remaining_timeout @@ -242,8 +242,26 @@ def close(self) -> None: os.close(self._fd) +class EndpointStream(Protocol): + """A connected byte stream to one of OpenVMM's ``listen=`` endpoints.""" + + def read_exact(self, length: int, deadline: float) -> bytes: ... + + def write_all(self, data: bytes) -> None: ... + + def close(self) -> None: ... + + +def connect_endpoint(endpoint: Path, timeout: float) -> EndpointStream: + """Connects to an OpenVMM endpoint: a Unix-domain socket, or on Windows a + named pipe.""" + if os.name == "nt": + return _NamedPipeStream.connect(endpoint, timeout) + return _SocketStream.connect(endpoint, timeout) + + class ControlSession: - def __init__(self, stream: _SocketStream | _NamedPipeStream) -> None: + def __init__(self, stream: EndpointStream) -> None: self._stream = stream self._instance_id = bytes(16) self._epoch = 0 @@ -259,11 +277,7 @@ def connect( ) -> ControlSession: if len(capability) != 32 or capability == bytes(32): raise ValueError("control capability must be 32 nonzero bytes") - stream = ( - _NamedPipeStream.connect(endpoint, timeout) - if os.name == "nt" - else _SocketStream.connect(endpoint, timeout) - ) + stream = connect_endpoint(endpoint, timeout) session = cls(stream) try: session._write_outer( diff --git a/scripts/nvx_tools/microvm_tests.py b/scripts/nvx_tools/microvm_tests.py index 690a020d5..4c0aee751 100644 --- a/scripts/nvx_tools/microvm_tests.py +++ b/scripts/nvx_tools/microvm_tests.py @@ -7,6 +7,7 @@ import os import queue import secrets +import signal import socket import struct import subprocess @@ -55,11 +56,19 @@ require_file, sha256_file, ) -from .control_session import ControlSession, ManagedExecRefused +from .control_session import ControlSession, ManagedExecRefused, ManagedExecResult from .egress_policy import CompiledEgressPolicy, compile_policy_file from .guests import GUEST_NAMES, GuestDescriptor, guest_descriptor from .managed_exec_tests import run_managed_exec_configuration from .openvmm_process import OpenvmmProcess, TcpConsole +from .state_control import ( + AUTHENTICATE, + PAUSE, + QUERY, + RESUME, + StateControlSession, + StateResponse, +) from .time_abi import ( ABI_VERSION, CHECK_CPU_BUDGET_US, @@ -96,6 +105,7 @@ "managed-lifecycle", "managed-exec-config", "network-snapshot", + "pause-resume", "restore-downtime", "restore-memory", "restore-processors", @@ -121,12 +131,14 @@ ) # The spec runs the same-host restore cases on the CI debug kernel, whose # soft-lockup and hung-task detectors the guest's time ABI watcher reports. +# Host pause holds guest time for longer than the soft-lockup threshold too. DEBUG_KERNEL_SCENARIOS = ( "smp", "smp-snapshot", "restore-processors", "restore-downtime", "snapshot-tiers", + "pause-resume", ) # Scenarios that run only when named, never in a default suite. The time ABI # hides TSC-deadline on every backend, so `smp` already runs on the one-shot @@ -226,6 +238,33 @@ # Gives the RCU stall detector time to report a stall that the release of # the stall suppression exposed. RESTORE_DOWNTIME_SETTLE_SECONDS = 2 +# Host pause holds guest time: the first pause outlasts the 21 s RCU stall +# timeout, as the long-downtime restore does, while a guest busy loop spans it. +PAUSE_RESUME_LONG_SECONDS = 30.0 +PAUSE_RESUME_SHORT_SECONDS = 2.0 +PAUSE_RESUME_CYCLES = 3 +PAUSE_RESUME_SPIN_SECONDS = 10 +# Spins one vCPU, without forking, for PAUSE_RESUME_SPIN_SECONDS of guest +# uptime, and prints the uptimes at which it started and stopped. +PAUSE_RESUME_SPIN_SCRIPT = ( + 'read -r now _ None: help=( "boot the CI debug kernel (build/vmlinux-debug), which enables the " "soft-lockup and hung-task detectors; selects the same-host restore " - "scenarios unless --scenario is given" + "and host pause scenarios unless --scenario is given" ), ) parser.add_argument( @@ -1078,6 +1117,443 @@ def run_managed_lifecycle( ) +def _expect_state( + response: StateResponse, step: str, status: str, state: str, transitions: int +) -> None: + if (response.status, response.state, response.transitions) != ( + status, + state, + transitions, + ): + raise RuntimeError( + f"state control {step} returned status={response.status} " + f"state={response.state} transitions={response.transitions} " + f"detail={response.detail!r}; expected status={status} " + f"state={state} transitions={transitions}" + ) + + +def _process_cpu_seconds(pid: int) -> float | None: + """Returns a Linux process's user plus system CPU time, or None elsewhere.""" + if sys.platform != "linux": + return None + stat = Path(f"/proc/{pid}/stat").read_text(encoding="ascii") + fields = stat.rsplit(")", 1)[1].split() + return (int(fields[11]) + int(fields[12])) / os.sysconf("SC_CLK_TCK") + + +def _managed_output( + session: ControlSession, arguments: tuple[str, ...], timeout: float +) -> str: + result = session.exec(arguments, timeout_ms=10_000, response_timeout=timeout) + if result.returncode != 0 or result.category != "exit" or result.stderr: + raise RuntimeError( + f"managed guest command {' '.join(arguments)} failed: " + f"returncode={result.returncode} category={result.category} " + f"stderr={result.stderr!r}" + ) + return result.stdout.decode("utf-8") + + +def _guest_uptime(session: ControlSession, timeout: float) -> tuple[float, float]: + """Returns the guest's uptime and the host's monotonic time of the read.""" + before = time.monotonic() + output = _managed_output(session, ("/bin/cat", "/proc/uptime"), timeout) + return float(output.split()[0]), (before + time.monotonic()) / 2 + + +def _guest_wall_clock_lag(session: ControlSession, timeout: float) -> float: + """Returns host UTC minus the guest's wall clock, in seconds.""" + before = time.time() + guest = int(_managed_output(session, ("/bin/date", "-u", "+%s"), timeout)) + return (before + time.time()) / 2 - guest + + +def _guest_time_state(session: ControlSession, timeout: float) -> dict[str, str]: + """Returns the guest time component's published state.""" + output = _managed_output(session, ("/bin/cat", "/run/nvx/time/state"), timeout) + return dict(line.split("=", 1) for line in output.splitlines() if "=" in line) + + +def _spin_uptimes(result: ManagedExecResult) -> tuple[float, float]: + """Returns the guest uptimes at which the busy loop started and stopped.""" + if result.returncode != 0 or result.category != "exit" or result.stderr: + raise RuntimeError(f"guest busy loop failed: {result}") + try: + start, end = (float(line) for line in result.stdout.decode().split()) + except ValueError as error: + raise RuntimeError(f"guest busy loop printed {result.stdout!r}") from error + return start, end + + +def _time_state_counter(state: dict[str, str], key: str) -> int: + try: + return int(state[key]) + except (KeyError, ValueError) as error: + raise RuntimeError(f"guest time state has no valid {key}: {state}") from error + + +def _hold_host_pause( + host: StateControlSession, + pid: int, + transitions: int, + seconds: float, + timeout: float, +) -> tuple[float, float | None]: + """Pauses the VM, holds it for ``seconds``, and resumes it, repeating each + request to check that it is idempotent. Returns how long the VM was paused + and, where the host reports it, OpenVMM's CPU time meanwhile.""" + requested = time.monotonic() + _expect_state( + host.request(PAUSE, timeout), "pause", "ok", "paused", transitions + 1 + ) + # The guest stopped while the request was outstanding. + paused_at = (requested + time.monotonic()) / 2 + _expect_state( + host.request(PAUSE, timeout), "repeated pause", "ok", "paused", transitions + 1 + ) + cpu_before = _process_cpu_seconds(pid) + time.sleep(seconds) + cpu_after = _process_cpu_seconds(pid) + _expect_state( + host.request(QUERY, timeout), + "query while paused", + "ok", + "paused", + transitions + 1, + ) + requested = time.monotonic() + _expect_state( + host.request(RESUME, timeout), "resume", "ok", "running", transitions + 2 + ) + resumed_at = (requested + time.monotonic()) / 2 + _expect_state( + host.request(RESUME, timeout), + "repeated resume", + "ok", + "running", + transitions + 2, + ) + if cpu_before is None or cpu_after is None: + return resumed_at - paused_at, None + used = cpu_after - cpu_before + if ( + used + > PAUSE_RESUME_PAUSED_CPU_SHARE * seconds + + PAUSE_RESUME_PAUSED_CPU_ALLOWANCE_SECONDS + ): + raise RuntimeError( + f"OpenVMM used {used:.2f} s of CPU time while paused for {seconds:.0f} s" + ) + return resumed_at - paused_at, used + + +def _check_host_pause( + session: ControlSession, + pid: int, + endpoint: Path, + capability: bytes, + timeout: float, +) -> dict[str, str]: + """Drives the state-control endpoint and checks the guest across pauses. + Leaves the VM paused and returns the evidence fields.""" + # Only a host that presents the control capability gets a response. + with StateControlSession.connect(endpoint, timeout) as anonymous: + anonymous.send(QUERY) + if not anonymous.closed_without_response(timeout): + raise RuntimeError( + "state control answered a host that did not authenticate" + ) + with StateControlSession.connect(endpoint, timeout) as forged: + forged.send(AUTHENTICATE, bytes(byte ^ 1 for byte in capability)) + if not forged.closed_without_response(timeout): + raise RuntimeError( + "state control answered a host with the wrong capability" + ) + + host, first = StateControlSession.authenticate(endpoint, capability, timeout) + with host: + _expect_state(first, "authentication", "ok", "running", 0) + if first.instance_id == bytes(16): + raise RuntimeError("state control reported no broker instance ID") + _expect_state(host.request(QUERY, timeout), "query", "ok", "running", 0) + _expect_state( + host.request(RESUME, timeout), "resume while running", "ok", "running", 0 + ) + + time_before = _guest_time_state(session, timeout) + uptime_start, host_start = _guest_uptime(session, timeout) + + # One vCPU spins across the long pause, so the pause stops a busy guest. + spin_results: list[ManagedExecResult] = [] + spin_errors: list[Exception] = [] + + def spin() -> None: + try: + spin_results.append( + session.exec( + ("/bin/sh", "-c", PAUSE_RESUME_SPIN_SCRIPT), + timeout_ms=600_000, + response_timeout=timeout + + PAUSE_RESUME_LONG_SECONDS + + PAUSE_RESUME_SPIN_SECONDS, + ) + ) + except Exception as error: + spin_errors.append(error) + + spinner = threading.Thread(target=spin, daemon=True) + spinner.start() + time.sleep(2) + busy_before = _process_cpu_seconds(pid) + time.sleep(1) + busy_after = _process_cpu_seconds(pid) + # The VM has run without a pause so far, so its uptime follows the host. + pause_uptime = uptime_start + time.monotonic() - host_start + paused, paused_cpu = _hold_host_pause( + host, pid, 0, PAUSE_RESUME_LONG_SECONDS, timeout + ) + spinner.join(timeout + PAUSE_RESUME_SPIN_SECONDS) + if spinner.is_alive() or spin_errors or len(spin_results) != 1: + raise RuntimeError(f"guest busy loop did not finish: {spin_errors}") + spin_start, spin_end = _spin_uptimes(spin_results[0]) + if not spin_start + PAUSE_RESUME_SPIN_LEAD_SECONDS <= pause_uptime < spin_end: + raise RuntimeError( + f"the guest busy loop ran from uptime {spin_start:.2f} s to " + f"{spin_end:.2f} s, not across the pause at {pause_uptime:.2f} s" + ) + for cycle in range(1, PAUSE_RESUME_CYCLES): + held, _ = _hold_host_pause( + host, pid, 2 * cycle, PAUSE_RESUME_SHORT_SECONDS, timeout + ) + paused += held + _managed_output(session, ("/bin/true",), timeout) + + uptime_end, host_end = _guest_uptime(session, timeout) + guest_elapsed = uptime_end - uptime_start + host_running = host_end - host_start - paused + if abs(guest_elapsed - host_running) > PAUSE_RESUME_UPTIME_TOLERANCE_SECONDS: + raise RuntimeError( + f"guest uptime advanced {guest_elapsed:.3f} s while the host ran " + f"the VM for {host_running:.3f} s and paused it for {paused:.3f} s" + ) + stalls = _managed_output( + session, ("/bin/cat", "/sys/kernel/rcu_stall_count"), timeout + ).strip() + if stalls != "0": + raise RuntimeError(f"guest reported {stalls} RCU stalls across host pauses") + + # Host UTC kept running, so the guest steps its wall clock at its next + # discipline poll. + repair_started = time.monotonic() + first_lag = lag = _guest_wall_clock_lag(session, timeout) + while abs(lag) > PAUSE_RESUME_WALL_CLOCK_TOLERANCE_SECONDS: + if ( + time.monotonic() - repair_started + > PAUSE_RESUME_WALL_CLOCK_REPAIR_SECONDS + ): + raise RuntimeError( + f"guest wall clock still differs from host UTC by {lag:.1f} s" + ) + time.sleep(2) + lag = _guest_wall_clock_lag(session, timeout) + repair = time.monotonic() - repair_started + time_after = _guest_time_state(session, timeout) + steps = _time_state_counter(time_after, "discontinuities") - ( + _time_state_counter(time_before, "discontinuities") + ) + if steps < 1 or time_after.get("last_discontinuity") != "step": + raise RuntimeError( + "guest wall clock was not repaired by a discipline step: " + f"discontinuities rose by {steps}, last_discontinuity=" + f"{time_after.get('last_discontinuity')}" + ) + violations = _time_state_counter(time_after, "violations") + if violations != _time_state_counter(time_before, "violations"): + raise RuntimeError( + f"guest time state reports {violations} violations after host pauses" + ) + + # A later host is served once the first disconnects. + second, response = StateControlSession.authenticate(endpoint, capability, timeout) + with second: + _expect_state( + response, "second authentication", "ok", "running", 2 * PAUSE_RESUME_CYCLES + ) + if response.instance_id != first.instance_id: + raise RuntimeError("state control reported a different broker instance") + _expect_state( + second.request(PAUSE, timeout), + "final pause", + "ok", + "paused", + 2 * PAUSE_RESUME_CYCLES + 1, + ) + # The pause outlives the host that made it. + third, response = StateControlSession.authenticate(endpoint, capability, timeout) + with third: + _expect_state( + response, + "authentication after the final pause", + "ok", + "paused", + 2 * PAUSE_RESUME_CYCLES + 1, + ) + if response.instance_id != first.instance_id: + raise RuntimeError("state control reported a different broker instance") + evidence = { + "busy_loop_lead_s": f"{pause_uptime - spin_start:.1f}", + "paused_s": f"{paused:.1f}", + "guest_minus_host_running_s": f"{guest_elapsed - host_running:.3f}", + "rcu_stalls": stalls, + "wall_clock_lag_after_resume_s": f"{first_lag:.1f}", + "wall_clock_repair_s": f"{repair:.1f}", + "discipline_steps": str(steps), + "last_step_ns": time_after.get("last_step_ns", "unknown"), + } + # OpenVMM's CPU time includes guest execution only where the backend + # accounts it to the vCPU threads, so the busy figure is evidence only. + if busy_before is not None and busy_after is not None: + evidence["busy_cpu_s_per_s"] = f"{busy_after - busy_before:.2f}" + if paused_cpu is not None: + evidence["paused_cpu_s"] = f"{paused_cpu:.2f}" + return evidence + + +def run_pause_resume( + executable: Path, + kernel: Path, + initrd: Path, + backend: str, + processors: int, + *, + memory_mib: int, + timeout: float, + output_dir: Path, +) -> None: + """Pauses and resumes a managed microVM through OpenVMM's state control. + + The guest must observe no elapsed monotonic time across the pauses, report + no RCU stall or time ABI violation, step its wall clock back to host UTC, + and run workloads after each resume. Hosts that do not present the control + capability get no response. The scenario ends by terminating OpenVMM while + the VM is paused. + """ + with tempfile.TemporaryDirectory(prefix="nvx-pause-resume-") as temporary: + root = Path(temporary) + control_endpoint, state_endpoint = ( + (str(root / "control.sock"), str(root / "state.sock")), + ( + f"//./pipe/openvmm-microvm-{uuid.uuid4().hex}", + f"//./pipe/openvmm-microvm-{uuid.uuid4().hex}", + ), + )[os.name == "nt"] + boot_console_address = _available_tcp_address() + capability = secrets.token_bytes(32) + command = workload_boot_command( + executable, + backend, + kernel, + initrd, + memory_mib, + "quiet loglevel=0", + processors=processors, + ) + command.extend( + ( + "--microvm-workload-identity", + "65534:65534", + "--microvm-lifecycle", + "managed", + "--virtio-console", + f"listen=tcp:{boot_console_address[0]}:{boot_console_address[1]}", + "--microvm-control-console", + f"listen={control_endpoint}", + "--microvm-control-auth-stdin", + "--microvm-state-control", + f"listen={state_endpoint}", + ) + ) + log_path = output_dir / "pause-resume.log" + guest_log_path = output_dir / "pause-resume-guest.log" + process: subprocess.Popen[bytes] | None = None + boot_console: TcpConsole | None = None + with log_path.open("wb") as log: + try: + environment = os.environ.copy() + environment["OPENVMM_LOG"] = "info" + process = subprocess.Popen( + command, + stdin=subprocess.PIPE, + stdout=log, + stderr=subprocess.STDOUT, + env=environment, + ) + record_adversarial_openvmm_pid(process.pid, environment) + if process.stdin is None: + raise RuntimeError("failed to create control capability pipe") + process.stdin.write(capability) + process.stdin.close() + # As in the managed lifecycle, init hands the boot to the + # managed agent, so the monitor only scans the guest console. + boot_console = TcpConsole.connect( + boot_console_address, timeout, monitor=TimeAbiMonitor(command) + ) + with ControlSession.connect( + Path(control_endpoint), capability, timeout + ) as session: + session.ping(timeout) + evidence = _check_host_pause( + session, process.pid, Path(state_endpoint), capability, timeout + ) + # The VM is still paused; the host stops it without the guest. + if process.poll() is not None: + raise RuntimeError( + f"OpenVMM exited with status {process.returncode} while " + f"paused; see {log_path}" + ) + process.terminate() + try: + result = process.wait(timeout=timeout) + except subprocess.TimeoutExpired as error: + raise RuntimeError( + f"OpenVMM did not exit within {timeout:g} s of being " + f"terminated while paused; see {log_path}" + ) from error + # Popen.terminate() uses TerminateProcess with exit code 1 on + # Windows, and SIGTERM, which OpenVMM does not handle, elsewhere. + terminated = 1 if os.name == "nt" else -signal.SIGTERM + if result != terminated: + raise RuntimeError( + f"OpenVMM exited with status {result} instead of being " + f"terminated while paused; see {log_path}" + ) + guest_log_path.write_bytes(boot_console.finish(check=False)) + boot_console.finish() + boot_console = None + finally: + if boot_console is not None: + guest_log_path.write_bytes(boot_console.finish(check=False)) + if process is not None and process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + print( + PAUSE_RESUME_EVIDENCE_PREFIX + + " ".join( + f"{name}={value}" + for name, value in { + "backend": backend, + "processors": str(processors), + **evidence, + }.items() + ) + ) + + def run_structured_outcome( executable: Path, kernel: Path, @@ -5291,6 +5767,22 @@ def run(args: argparse.Namespace) -> int: timeout=args.timeout, output_dir=output_dir, ) + if "pause-resume" in scenarios: + processors = max(args.processors) + print( + f"Running microVM host pause and resume ({processors} vCPU) " + f"on OpenVMM/{args.backend}" + ) + run_pause_resume( + executable, + kernel, + initrd, + args.backend, + processors, + memory_mib=args.memory_mib, + timeout=args.timeout, + output_dir=output_dir, + ) if "structured-outcome" in scenarios: print(f"Running structured microVM outcomes on OpenVMM/{args.backend}") run_structured_outcome( diff --git a/scripts/nvx_tools/state_control.py b/scripts/nvx_tools/state_control.py new file mode 100644 index 000000000..deac8e20c --- /dev/null +++ b/scripts/nvx_tools/state_control.py @@ -0,0 +1,140 @@ +"""Client for OpenVMM's authenticated microVM state-control endpoint. + +``--microvm-state-control listen=`` serves host pause, resume, and +run-state requests for a microVM that has a live authenticated control +console. The host authenticates with the control-console capability. Protocol +version 1 is documented in OpenVMM's Guide, in +``reference/openvmm/management/state_control_protocol.md``. +""" + +from __future__ import annotations + +import struct +import time +from dataclasses import dataclass +from pathlib import Path + +from .common import ScriptError +from .control_session import EndpointStream, connect_endpoint + +HEADER = struct.Struct("<4sHBBQII") +STATE_RECORD = struct.Struct(" None: + self._stream = stream + self._sequence = 0 + + @classmethod + def connect(cls, endpoint: Path, timeout: float) -> StateControlSession: + """Connects without authenticating.""" + return cls(connect_endpoint(endpoint, timeout)) + + @classmethod + def authenticate( + cls, endpoint: Path, capability: bytes, timeout: float + ) -> tuple[StateControlSession, StateResponse]: + """Connects, authenticates, and returns the session and run state.""" + if len(capability) != CAPABILITY_BYTES or capability == bytes(CAPABILITY_BYTES): + raise ValueError("control capability must be 32 nonzero bytes") + session = cls.connect(endpoint, timeout) + try: + response = session.request(AUTHENTICATE, timeout, capability) + if response.status != "ok": + raise ScriptError(f"state-control authentication failed: {response}") + return session, response + except BaseException: + session.close() + raise + + def send(self, operation: int, payload: bytes = b"") -> None: + """Sends one request without waiting for its response.""" + self._sequence += 1 + self._stream.write_all( + HEADER.pack(MAGIC, VERSION, operation, 0, self._sequence, 0, len(payload)) + + payload + ) + + def request( + self, operation: int, timeout: float, payload: bytes = b"" + ) -> StateResponse: + """Sends one request and returns its validated response.""" + self.send(operation, payload) + deadline = time.monotonic() + timeout + header: tuple[bytes, int, int, int, int, int, int] = HEADER.unpack( + self._stream.read_exact(HEADER.size, deadline) + ) + magic, version, kind, flags, sequence, status, length = header + if ( + magic != MAGIC + or version != VERSION + or kind != operation | RESPONSE_BIT + or flags != 0 + or sequence != self._sequence + or status >= len(STATUSES) + or not STATE_RECORD.size <= length <= STATE_RECORD.size + MAX_DETAIL_BYTES + ): + raise ScriptError( + "state-control endpoint returned an invalid response header: " + f"kind={kind:#x} flags={flags} sequence={sequence} " + f"status={status} length={length}" + ) + payload = self._stream.read_exact(length, deadline) + record: tuple[int, bytes, int, bytes] = STATE_RECORD.unpack_from(payload) + state, reserved, transitions, instance_id = record + if state >= len(STATES) or reserved != bytes(7): + raise ScriptError("state-control endpoint returned an invalid state record") + try: + detail = payload[STATE_RECORD.size :].decode("utf-8") + except UnicodeDecodeError as error: + raise ScriptError("state-control detail is not UTF-8") from error + return StateResponse( + STATUSES[status], STATES[state], transitions, instance_id, detail + ) + + def closed_without_response(self, timeout: float) -> bool: + """Returns whether the endpoint closes the connection before sending + anything.""" + try: + self._stream.read_exact(1, time.monotonic() + timeout) + except TimeoutError: + return False + except ConnectionError: + return True + return False + + def close(self) -> None: + self._stream.close() + + def __enter__(self) -> StateControlSession: + return self + + def __exit__(self, *_: object) -> None: + self.close() diff --git a/scripts/test_microvm_tests.py b/scripts/test_microvm_tests.py index 0ab3084ad..d545c2062 100644 --- a/scripts/test_microvm_tests.py +++ b/scripts/test_microvm_tests.py @@ -14,6 +14,7 @@ import subprocess import sys import tempfile +import threading import unittest from collections.abc import Callable from contextlib import ExitStack, redirect_stdout @@ -30,6 +31,7 @@ managed_exec_tests, microvm_tests, openvmm_process, + state_control, time_abi, ) from nvx_tools.build_constants import ( # noqa: E402 @@ -2499,16 +2501,17 @@ def test_every_tcp_console_is_monitored_and_cold_boot_shells_are_queried(self): if isinstance(node, ast.Call) and ast.unparse(node.func) == "TcpConsole.connect" ] - # The managed lifecycle, the console-snapshot capture and restore, and - # the snapshot-tier capture, restore, and gate-timeout restore. - self.assertEqual(len(connects), 6) + # The managed lifecycle, the host pause, the console-snapshot capture and + # restore, and the snapshot-tier capture, restore, and gate-timeout + # restore. + self.assertEqual(len(connects), 7) for keywords in connects: self.assertRegex( keywords.get("monitor", ""), r"^TimeAbiMonitor\(\w*command\)$" ) # Only the cold boots whose shell is on the virtio console ask for - # nvx-time status. Restores never ask, and in the managed lifecycle - # init starts the managed agent instead of a shell. + # nvx-time status. Restores never ask, and in the managed lifecycle and + # the host pause init starts the managed agent instead of a shell. self.assertEqual( [k["monitor"] for k in connects if k.get("time_abi_status") == "True"], ["TimeAbiMonitor(capture_command)"] * 2, @@ -5008,6 +5011,7 @@ def require(path: Path, _description: str) -> Path: "restore-processors": "run_restore_processors", "restore-downtime": "run_restore_downtime", "snapshot-tiers": "run_snapshot_tiers", + "pause-resume": "run_pause_resume", } self.assertEqual(tuple(dispatch), microvm_tests.DEBUG_KERNEL_SCENARIOS) with tempfile.TemporaryDirectory() as temporary: @@ -5102,6 +5106,37 @@ def require(path: Path, _description: str) -> Path: ["test-microvm", "--backend", "kvm", "--scenario", "restore-tsc-sync"] ) + def test_runner_runs_pause_resume_once_on_the_most_processors(self): + def require(path: Path, _description: str) -> Path: + return path + + with tempfile.TemporaryDirectory() as temporary: + output_dir = Path(temporary) + args = nvx.parse_args( + [ + "test-microvm", + "--backend", + "mshv", + "--scenario", + "pause-resume", + "--processors", + "2", + "4", + "--output-dir", + str(output_dir), + ] + ) + with ( + patch.object(microvm_tests, "validate_openvmm_test_backend"), + patch.object(microvm_tests, "require_file", side_effect=require), + patch.object(microvm_tests, "run_pause_resume") as run, + redirect_stdout(io.StringIO()), + ): + self.assertEqual(microvm_tests.run(args), 0) + run.assert_called_once() + self.assertEqual(run.call_args.args[3:], ("mshv", 4)) + self.assertEqual(run.call_args.kwargs["output_dir"], output_dir) + def test_runner_dispatches_console_exit_for_each_requested_cpu_count(self): def require(path: Path, _description: str) -> Path: return path @@ -5170,5 +5205,332 @@ def close(self) -> None: self.assertEqual(log_path.read_bytes(), b"complete raw output\n") +class _PauseClock: + """A host clock that sleeping advances at once. A set ``gate`` holds every + sleep until it opens, so that another thread's guest command is ordered + before the clock advances.""" + + def __init__(self) -> None: + self.now = 1_000.0 + self.gate: threading.Event | None = None + + def monotonic(self) -> float: + return self.now + + def time(self) -> float: + return 1_800_000_000.0 + self.now + + def sleep(self, seconds: float) -> None: + if self.gate is not None and not self.gate.wait(5): + raise AssertionError("the guest command the clock waits for never ran") + self.now += seconds + + +_STATE_OPERATIONS = { + state_control.AUTHENTICATE: "authenticate", + state_control.QUERY: "query", + state_control.PAUSE: "pause", + state_control.RESUME: "resume", +} + + +class _PausableVm: + """A managed VM behind OpenVMM's state control, as the scenario sees it. + + One vCPU spins whenever the VM runs; the busy loop starts ``spin_delay`` + seconds after it is requested. ``wall_clock`` is ``step`` when the guest + steps its lagging wall clock at its second read after a pause, ``never`` + when it never does, and ``follows-host`` when it never lags. + """ + + def __init__( + self, + clock: _PauseClock, + *, + holds_time: bool = True, + paused_cpu: float = 0.0, + wall_clock: str = "step", + answers_anonymous: bool = False, + spin_delay: float = 0.0, + ) -> None: + self.clock = clock + self.spin_delay = spin_delay + # The harness starts the busy loop and then sleeps. + self.spin_requested = threading.Event() + clock.gate = self.spin_requested + self.holds_time = holds_time + self.paused_cpu = paused_cpu + self.wall_clock = wall_clock + self.answers_anonymous = answers_anonymous + self.instance_id = bytes(range(1, 17)) + self.paused_since: float | None = None + self.paused_total = 0.0 + self.stepped = 0.0 + self.transitions = 0 + self.discontinuities = 0 + self.date_reads = 0 + self.requests: list[str] = [] + self.unauthenticated_payloads: list[bytes] = [] + + def response(self) -> state_control.StateResponse: + state = "running" if self.paused_since is None else "paused" + return state_control.StateResponse( + "ok", state, self.transitions, self.instance_id, "" + ) + + def request(self, operation: int) -> state_control.StateResponse: + self.requests.append(_STATE_OPERATIONS[operation]) + if operation == state_control.PAUSE and self.paused_since is None: + self.paused_since = self.clock.now + self.transitions += 1 + elif operation == state_control.RESUME and self.paused_since is not None: + self.paused_total += self.clock.now - self.paused_since + self.paused_since = None + self.transitions += 1 + return self.response() + + def _paused(self) -> float: + if self.paused_since is None: + return self.paused_total + return self.paused_total + self.clock.now - self.paused_since + + def cpu_seconds(self, _pid: int) -> float: + paused = self._paused() + return self.clock.now - paused + self.paused_cpu * paused + + def _held(self) -> float: + return self.paused_total if self.holds_time else 0.0 + + def output(self, arguments: tuple[str, ...]) -> str: + if arguments == ("/bin/cat", "/proc/uptime"): + return f"{self.clock.now - 900.0 - self._held():.2f} 0.00\n" + if arguments == ("/bin/date", "-u", "+%s"): + self.date_reads += 1 + lag = 0.0 if self.wall_clock == "follows-host" else self._held() + if ( + self.wall_clock == "step" + and self.date_reads > 1 + and self.stepped != lag + ): + self.stepped = lag + self.discontinuities += 1 + return f"{int(self.clock.time() - lag + self.stepped)}\n" + if arguments == ("/bin/cat", "/run/nvx/time/state"): + last = "step" if self.discontinuities else "none" + return ( + "version=1\n" + f"discontinuities={self.discontinuities}\n" + f"last_discontinuity={last}\n" + f"last_step_ns={int(self.stepped * 1e9)}\n" + "violations=0\n" + ) + if arguments == ("/bin/cat", "/sys/kernel/rcu_stall_count"): + return "0\n" + if arguments == ("/bin/sh", "-c", microvm_tests.PAUSE_RESUME_SPIN_SCRIPT): + start = self.clock.now - 900.0 - self._held() + self.spin_delay + self.spin_requested.set() + end = start + microvm_tests.PAUSE_RESUME_SPIN_SECONDS + return f"{start:.2f}\n{end:.2f}\n" + if arguments == ("/bin/true",): + return "" + raise AssertionError(f"unexpected guest command {arguments}") + + +class _PausableGuest: + def __init__(self, vm: _PausableVm) -> None: + self._vm = vm + self.commands: list[tuple[str, ...]] = [] + + def exec( + self, arguments: tuple[str, ...], *, timeout_ms: int, response_timeout: float + ) -> control_session.ManagedExecResult: + self.commands.append(arguments) + output = self._vm.output(arguments).encode() + return control_session.ManagedExecResult(0, "exit", output, b"") + + +def _fake_state_control(vm: _PausableVm): + class FakeStateControl: + def __init__(self, authenticated: bool) -> None: + self._authenticated = authenticated + self._closed = False + + @classmethod + def connect(cls, _endpoint: Path, _timeout: float): + vm.requests.append("connect") + return cls(False) + + @classmethod + def authenticate(cls, _endpoint: Path, _capability: bytes, _timeout: float): + vm.requests.append("authenticate") + return cls(True), vm.response() + + def send(self, operation: int, payload: bytes = b"") -> None: + vm.requests.append(f"unauthenticated {_STATE_OPERATIONS[operation]}") + vm.unauthenticated_payloads.append(payload) + + def closed_without_response(self, _timeout: float) -> bool: + return not vm.answers_anonymous + + def request( + self, operation: int, _timeout: float, _payload: bytes = b"" + ) -> state_control.StateResponse: + if not self._authenticated or self._closed: + raise AssertionError("request on an unauthenticated or closed host") + return vm.request(operation) + + def __enter__(self): + return self + + def __exit__(self, *_: object) -> None: + self._closed = True + + return FakeStateControl + + +class PauseResumeTests(unittest.TestCase): + CAPABILITY = bytes([0x5A]) * 32 + + def _check(self, vm: _PausableVm) -> tuple[dict[str, str], _PausableGuest]: + guest = _PausableGuest(vm) + with ( + patch.object(microvm_tests, "time", vm.clock), + patch.object(microvm_tests, "StateControlSession", _fake_state_control(vm)), + patch.object(microvm_tests, "_process_cpu_seconds", new=vm.cpu_seconds), + ): + evidence = microvm_tests._check_host_pause( + cast(control_session.ControlSession, guest), + 123, + Path("state-control"), + self.CAPABILITY, + 5.0, + ) + return evidence, guest + + def test_pause_holds_guest_time_and_leaves_the_vm_paused(self): + vm = _PausableVm(_PauseClock()) + evidence, guest = self._check(vm) + cycle = ["pause", "pause", "query", "resume", "resume"] + self.assertEqual( + vm.requests, + [ + "connect", + "unauthenticated query", + "connect", + "unauthenticated authenticate", + "authenticate", + "query", + "resume", + *cycle * microvm_tests.PAUSE_RESUME_CYCLES, + "authenticate", + "pause", + "authenticate", + ], + ) + self.assertEqual(vm.unauthenticated_payloads[0], b"") + forged = vm.unauthenticated_payloads[1] + self.assertEqual(len(forged), 32) + self.assertNotEqual(forged, self.CAPABILITY) + self.assertEqual(vm.transitions, 2 * microvm_tests.PAUSE_RESUME_CYCLES + 1) + self.assertIsNotNone(vm.paused_since) + self.assertIn( + ("/bin/sh", "-c", microvm_tests.PAUSE_RESUME_SPIN_SCRIPT), guest.commands + ) + self.assertEqual( + evidence, + { + "busy_loop_lead_s": "3.0", + "paused_s": "34.0", + "guest_minus_host_running_s": "0.000", + "rcu_stalls": "0", + "wall_clock_lag_after_resume_s": "34.0", + "wall_clock_repair_s": "2.0", + "discipline_steps": "1", + "last_step_ns": "34000000000", + "busy_cpu_s_per_s": "1.00", + "paused_cpu_s": "0.00", + }, + ) + + def test_failures_name_the_broken_property(self): + cases: tuple[tuple[Callable[[_PauseClock], _PausableVm], str], ...] = ( + ( + lambda clock: _PausableVm(clock, holds_time=False), + "guest uptime advanced", + ), + ( + lambda clock: _PausableVm(clock, paused_cpu=0.5), + "CPU time while paused", + ), + ( + lambda clock: _PausableVm(clock, answers_anonymous=True), + "did not authenticate", + ), + ( + lambda clock: _PausableVm(clock, spin_delay=2.6), + "not across the pause", + ), + ( + lambda clock: _PausableVm(clock, wall_clock="never"), + "still differs from host UTC", + ), + ( + lambda clock: _PausableVm(clock, wall_clock="follows-host"), + "not repaired by a discipline step", + ), + ) + for make_vm, message in cases: + with ( + self.subTest(message), + self.assertRaisesRegex(RuntimeError, message), + ): + self._check(make_vm(_PauseClock())) + + def test_unexpected_state_control_responses_are_reported(self): + response = state_control.StateResponse( + "busy", "running", 0, bytes(16), "restore gate" + ) + with self.assertRaisesRegex( + RuntimeError, + "state control pause returned status=busy state=running transitions=0 " + "detail='restore gate'; expected status=ok state=paused transitions=1", + ): + microvm_tests._expect_state(response, "pause", "ok", "paused", 1) + + def test_process_cpu_time_is_read_on_linux_only(self): + seconds = microvm_tests._process_cpu_seconds(os.getpid()) + if sys.platform == "linux": + self.assertIsNotNone(seconds) + self.assertGreaterEqual(cast(float, seconds), 0.0) + else: + self.assertIsNone(seconds) + + def test_spin_script_reports_the_uptimes_it_spun_between(self): + if sys.platform != "linux": + self.skipTest("the script uses Linux semantics") + shell = _posix_shell() + if shell is None: + self.skipTest("POSIX shell is unavailable") + seconds = f"+ {microvm_tests.PAUSE_RESUME_SPIN_SECONDS} " + self.assertIn(seconds, microvm_tests.PAUSE_RESUME_SPIN_SCRIPT) + result = subprocess.run( + [ + shell, + "-c", + microvm_tests.PAUSE_RESUME_SPIN_SCRIPT.replace(seconds, "+ 1 "), + ], + capture_output=True, + check=False, + timeout=10, + ) + start, end = microvm_tests._spin_uptimes( + control_session.ManagedExecResult( + result.returncode, "exit", result.stdout, result.stderr + ) + ) + self.assertGreaterEqual(int(end), int(start) + 1) + self.assertLess(end - start, 2) + + if __name__ == "__main__": unittest.main() diff --git a/scripts/test_state_control.py b/scripts/test_state_control.py new file mode 100644 index 000000000..397a5956b --- /dev/null +++ b/scripts/test_state_control.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +# pyright: reportPrivateUsage=false + +import socket +import sys +import time +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) +from nvx_tools import control_session, state_control # noqa: E402 +from nvx_tools.common import ScriptError # noqa: E402 + +INSTANCE = bytes(range(16)) +CAPABILITY = bytes([0x5A]) * 32 + + +def _response( + operation: int, + sequence: int, + *, + status: int = 0, + state: int = 1, + transitions: int = 0, + detail: bytes = b"", + magic: bytes = b"NVXV", + version: int = 1, + kind: int | None = None, + flags: int = 0, + reserved: bytes = bytes(7), + record_bytes: int = 32, +) -> bytes: + record = state_control.STATE_RECORD.pack(state, reserved, transitions, INSTANCE) + payload = record[:record_bytes] + detail + return ( + state_control.HEADER.pack( + magic, + version, + operation | 0x80 if kind is None else kind, + flags, + sequence, + status, + len(payload), + ) + + payload + ) + + +def _read_exact(connection: socket.socket, length: int) -> bytes: + output = bytearray() + while len(output) != length: + chunk = connection.recv(length - len(output)) + if not chunk: + raise RuntimeError("test state-control connection closed") + output.extend(chunk) + return bytes(output) + + +class StateControlSessionTests(unittest.TestCase): + def _session(self) -> tuple[state_control.StateControlSession, socket.socket]: + host, server = socket.socketpair() + server.settimeout(5) + session = state_control.StateControlSession(control_session._SocketStream(host)) + self.addCleanup(session.close) + self.addCleanup(server.close) + return session, server + + def test_requests_have_the_documented_layout(self) -> None: + self.assertEqual(state_control.HEADER.size, 24) + self.assertEqual(state_control.STATE_RECORD.size, 32) + session, server = self._session() + server.sendall(_response(state_control.AUTHENTICATE, 1)) + session.request(state_control.AUTHENTICATE, 5, CAPABILITY) + header = _read_exact(server, state_control.HEADER.size) + self.assertEqual( + header, + b"NVXV\x01\x00\x01\x00" + + (1).to_bytes(8, "little") + + bytes(4) + + (32).to_bytes(4, "little"), + ) + self.assertEqual(_read_exact(server, 32), CAPABILITY) + + server.sendall(_response(state_control.PAUSE, 2)) + session.request(state_control.PAUSE, 5) + self.assertEqual( + _read_exact(server, state_control.HEADER.size), + b"NVXV\x01\x00\x03\x00" + (2).to_bytes(8, "little") + bytes(8), + ) + + def test_responses_decode_status_state_transitions_and_detail(self) -> None: + session, server = self._session() + server.sendall( + _response( + state_control.RESUME, + 1, + status=2, + state=2, + transitions=3, + detail="restore failed \N{LATIN SMALL LETTER E WITH ACUTE}".encode(), + ) + ) + self.assertEqual( + session.request(state_control.RESUME, 5), + state_control.StateResponse( + "rejected", + "paused", + 3, + INSTANCE, + "restore failed \N{LATIN SMALL LETTER E WITH ACUTE}", + ), + ) + for sequence, (status, state, expected) in enumerate( + ( + (0, 0, ("ok", "unknown")), + (1, 4, ("busy", "busy")), + (3, 3, ("failed", "stopped")), + ), + start=2, + ): + with self.subTest(status=status, state=state): + server.sendall( + _response(state_control.QUERY, sequence, status=status, state=state) + ) + response = session.request(state_control.QUERY, 5) + self.assertEqual((response.status, response.state), expected) + + def test_malformed_responses_are_rejected(self) -> None: + cases = { + "wrong magic": _response(state_control.PAUSE, 1, magic=b"NVXW"), + "wrong version": _response(state_control.PAUSE, 1, version=2), + "nonzero flags": _response(state_control.PAUSE, 1, flags=1), + "short record": _response(state_control.PAUSE, 1, record_bytes=31), + "wrong operation": _response( + state_control.PAUSE, 1, kind=state_control.QUERY | 0x80 + ), + "wrong sequence": _response(state_control.PAUSE, 7), + "unknown status": _response(state_control.PAUSE, 1, status=4), + "unknown state": _response(state_control.PAUSE, 1, state=5), + "reserved bytes": _response( + state_control.PAUSE, 1, reserved=b"\x01" + bytes(6) + ), + "invalid detail": _response(state_control.PAUSE, 1, detail=b"\xff"), + "oversized detail": state_control.HEADER.pack( + b"NVXV", 1, state_control.PAUSE | 0x80, 0, 1, 0, 545 + ) + + bytes(545), + } + for name, response in cases.items(): + with self.subTest(name): + session, server = self._session() + server.sendall(response) + with self.assertRaises(ScriptError): + session.request(state_control.PAUSE, 5) + + def test_closed_without_response_distinguishes_close_reply_and_silence( + self, + ) -> None: + session, server = self._session() + server.sendall(b"N") + self.assertFalse(session.closed_without_response(5)) + started = time.monotonic() + self.assertFalse(session.closed_without_response(0.05)) + self.assertLess(time.monotonic() - started, 5) + server.close() + self.assertTrue(session.closed_without_response(5)) + + def test_authentication_requires_a_nonzero_capability(self) -> None: + for capability in (bytes(32), CAPABILITY[:31]): + with ( + self.subTest(length=len(capability)), + self.assertRaisesRegex(ValueError, "32 nonzero bytes"), + ): + state_control.StateControlSession.authenticate( + Path("unused"), capability, 1 + ) + + +if __name__ == "__main__": + unittest.main()