diff --git a/.github/actions/run-benchmark/action.yml b/.github/actions/run-benchmark/action.yml index 19f1fae28..7fa92ab24 100644 --- a/.github/actions/run-benchmark/action.yml +++ b/.github/actions/run-benchmark/action.yml @@ -52,7 +52,9 @@ runs: fi run_dir="data/runs/${{ inputs.platform }}/microvm-v2/1vcpu" acceptance_output="${run_dir}/acceptance.json" - rm -f "${acceptance_output}" + attempt_log="${run_dir}/acceptance-attempts.json" + rm -f "${acceptance_output}" "${attempt_log}" \ + "${run_dir}"/acceptance-attempt-*.json for attempt in 1 2; do attempt_output="${run_dir}/acceptance-attempt-${attempt}.json" python3 scripts/nvx.py benchmark \ @@ -71,7 +73,8 @@ runs: validation_status=0 python3 scripts/nvx.py performance validate-openvmm \ --platform "${{ inputs.platform }}" \ - --input "${attempt_output}" || validation_status=$? + --input "${attempt_output}" \ + --attempt-log "${attempt_log}" || validation_status=$? if [[ "${validation_status}" -eq 0 ]]; then cp "${attempt_output}" "${acceptance_output}" break @@ -220,8 +223,15 @@ runs: } $RunDir = "data/runs/${{ inputs.platform }}/microvm-v2/1vcpu" $AcceptanceOutput = "$RunDir/acceptance.json" - if (Test-Path -LiteralPath $AcceptanceOutput) { - Remove-Item -LiteralPath $AcceptanceOutput + $AttemptLog = "$RunDir/acceptance-attempts.json" + foreach ($Stale in $AcceptanceOutput, $AttemptLog) { + if (Test-Path -LiteralPath $Stale) { + Remove-Item -LiteralPath $Stale + } + } + if (Test-Path -LiteralPath $RunDir) { + Get-ChildItem -LiteralPath $RunDir -Filter "acceptance-attempt-*.json" | + Remove-Item } foreach ($Attempt in 1, 2) { $AttemptOutput = "$RunDir/acceptance-attempt-$Attempt.json" @@ -242,7 +252,8 @@ runs: if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } python scripts\nvx.py performance validate-openvmm ` --platform "${{ inputs.platform }}" ` - --input $AttemptOutput + --input $AttemptOutput ` + --attempt-log $AttemptLog $ValidationStatus = $LASTEXITCODE if ($ValidationStatus -eq 0) { Copy-Item -LiteralPath $AttemptOutput -Destination $AcceptanceOutput @@ -373,6 +384,7 @@ runs: --require-shell-snapshot \ --require-shared-suite \ --lifecycle-input "${run_dir}/acceptance.json" \ + --lifecycle-attempt-log "${run_dir}/acceptance-attempts.json" \ --summary "${GITHUB_STEP_SUMMARY}" device_run_dir="${run_dir}/device-io" device_results_dir="${device_run_dir}/results" @@ -411,6 +423,7 @@ runs: --require-shell-snapshot ` --require-shared-suite ` --lifecycle-input "$RunDir/acceptance.json" ` + --lifecycle-attempt-log "$RunDir/acceptance-attempts.json" ` --summary $env:GITHUB_STEP_SUMMARY if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } $DeviceRunDir = "$RunDir/device-io" diff --git a/.github/actions/validate-nvx/action.yml b/.github/actions/validate-nvx/action.yml index 1bccff058..835b48879 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 scripts/test_tool_versions.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_managed_agent.py scripts/test_guest_time.py scripts/test_host_telemetry.py scripts/test_time_abi.py scripts/test_tool_versions.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 scripts/test_tool_versions.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_managed_agent.py scripts/test_guest_time.py scripts/test_host_telemetry.py scripts/test_time_abi.py scripts/test_tool_versions.py -v if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } python scripts\test_adversarial.py -v if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } diff --git a/.github/skills/nvx-benchmark/SKILL.md b/.github/skills/nvx-benchmark/SKILL.md index 00e1c3132..4db56f0db 100644 --- a/.github/skills/nvx-benchmark/SKILL.md +++ b/.github/skills/nvx-benchmark/SKILL.md @@ -213,6 +213,8 @@ Verify that the retained result set contains the acceptance JSON, raw logs under `performance/`, `device-io/device-io.log`, `summary.md`, collected CSV files, and `benchmark-metadata.json` records with the expected commit, platform, and sampling counts. Require the expected artifact hashes only in the device-I/O metadata record. +Check that the `host` provenance in the acceptance JSON and metadata names the +measured machine and the scratch volume that the run used. ## Report diff --git a/doc/benchmarks.md b/doc/benchmarks.md index d3302977a..63e66e220 100644 --- a/doc/benchmarks.md +++ b/doc/benchmarks.md @@ -502,6 +502,32 @@ minor and major faults, total page faults, and, when `smaps_rollup` is available and private RSS bytes. Windows reports working set, peak working set, private commit, and page faults. +On Windows, each retained capture sample also carries `storage_telemetry`. From just before the +host requests the snapshot until the source process exits, a coordinator thread samples +cumulative host counters about every 50 ms: read and write bytes, request counts, response +times, idle time, and queue depth of the scratch, workspace, and system volumes; CPU idle, +kernel, and user time; and the system cache's dirty pages and threshold, lazy-writer and flush +pages, and modified- and mapped-page-writer pages. About every 500 ms, and in the first and last +samples, it adds the CPU time and I/O transfers of the Microsoft Defender processes +`MsMpEng.exe` and `MpDefenderCoreService.exe`. CI jobs run as Network Service, which cannot read +performance counters, so the sampler uses only interfaces that any account can query: volume +performance IOCTLs, `GetSystemTimes`, and `NtQuerySystemInformation`. Windows maintains disk +counters only on demand, and the volume IOCTL returns none until a consumer asks for them, so the +sampler keeps the disk performance WMI data block, which every account may query, open while it +samples; it changes no system setting. A failed read is recorded +in `errors` and never fails the benchmark. The raw `samples` are retained, and `phases` +summarizes `capture.snapshot_generation` and `capture.mapped_memory_flush`. Each counter group, +a volume, CPU times, cache counters, or Defender's processes, spans the samples nearest to each +phase's start and end that carry it, so a failed read does not hide what neighboring samples +recorded. The observer clock is aligned with OpenVMM's by the smallest observed-minus-reported +time of the capture's records. Each summary reports the sampled interval it spans; per-volume +throughput, IOPS, average latency, average queue length, busy time, and peak queue depth; CPU busy +time; the lazy-writer, flush, and page-writer rates; the peak dirty cache and its threshold; and +Defender CPU time and I/O. A counter group whose samples differ from the summary's, such as +Defender's sparser process counters, reports its own interval. +Linux captures do not flush a file-backed guest RAM image, take about 1 to 4 ms, and record no +storage telemetry. + `profile.phases` groups samples by `operation.phase` and stores raw `samples_ms` plus p50, nearest-rank p95, minimum, and maximum. The top-level JSON path is `snapshot_profile_matrix..`. Each memory entry contains the capture result and @@ -582,6 +608,16 @@ bimodal host-stall series into topology-wide history. On the Linux CI runners, snapshot generation takes about 1 ms on KVM and about 4 ms on MSHV, so a sub-millisecond CPU stall in two samples is enough to cross the gap limit. +Each validation also appends the attempt's file, SHA-256 digest, outcome (`valid`, +`unstable`, or `invalid`), exit status, and error to `acceptance-attempts.json`, +which the acceptance step removes, with any earlier attempt files, before measuring. +The publish step passes that log to `performance collect --lifecycle-attempt-log`. +Collection fails before it writes a CSV unless `acceptance.json` is a byte-identical +copy of exactly one valid attempt; it then prints the attempt that supplied the +published lifecycle metrics. The step summary lists every attempt with its outcome, +snapshot-generation p50, and whether it was published, and reports the storage +telemetry of each attempt, rejected ones included. + Before updating the run-scoped `benchmark--` artifact, each platform uploads its raw results and lifecycle profiles to the immutable `benchmark-diagnostics---attempt-` artifact, including @@ -592,6 +628,25 @@ platform results from an earlier workflow attempt. The `acceptance-attempt-N.jso files identify the bounded lifecycle remeasurements within one workflow attempt, not the workflow's `github.run_attempt`. +Lifecycle JSON results and workload `benchmark-metadata.json` files record host +provenance under `host`: the runner name from `RUNNER_NAME`, the machine name, +operating system, CPU model, logical processor count, and physical memory, and the +volume, file system, size, system-volume flag, disk number, and SCSI address behind +the workspace, temporary, and scratch directories. The disk number and SCSI address +tell the Azure OS disk from a data-disk LUN or the temporary disk. On a possible Azure +VM, the coordinator queries the instance metadata service once, directly rather than +through a proxy, and keeps only the VM size; each managed disk's SKU, size, caching +policy, write-accelerator and ephemeral settings, and data-disk LUN; the throughput +and IOPS limits that Ultra Disks report; and the temporary disk's size. It never +records names, resource or VM identifiers, the subscription, tags, network addresses, +or credentials, and records an unreachable service as `azure.error`. Acceptance and +diagnostic suites print the provenance before they measure, and the step summary +includes it as the benchmark host table. A device-io resume ignores `host` when it +compares controls and provenance. Runner and machine names are diagnostic only: the +regression gate pools every runner of a series into one history, and a series should +be split by runner only if evidence shows a persistent runner-specific offset larger +than the gate's tolerance. + When investigating instability, compare each attempt's `snapshot_capture..profile.raw_samples` with its `samples_ms`. For example, a slow `capture.mapped_memory_flush` with otherwise stable capture phases localizes @@ -602,6 +657,25 @@ Reproduce with the exact executable and guest artifact hashes on the same host before attributing the delay to a source change. Keep the stability thresholds and bounded remeasurement unchanged when collecting diagnostic evidence. +On Windows, the job log prints each measured capture's `capture.mapped_memory_flush` +time with the host activity during it, and the step summary tabulates the same +telemetry for every attempt: the scratch volume's write throughput, average write +latency, average queue length, and busy time; writes to the other monitored +volumes; CPU busy time; the peak dirty cache; lazy-writer throughput; and Defender +CPU time. Compare slow flushes with fast ones in the same series: + +- Disk throttling: the slow flushes write scratch at a lower, flat rate with longer + latency and queue while it stays busy, and the other columns stay quiet. Compare + that rate with the published limits of the disk SKU and size in the host table. +- Scanning: Defender CPU time or I/O rises during the slow flushes. +- Writeback: lazy writes, the dirty cache, or writes to the other volumes rise + during the slow flushes. +- Unrelated host load: CPU busy time rises while scratch stays idle for part of + the flush. + +In the JSON, the flush's own page writes appear in `memory.cache_flush_mib_per_second`, +and the raw samples show how the rates change within a flush. + Snapshot generation is bounded by the write throughput of the benchmark's temporary directory, because capture flushes guest RAM through a backing file created there. The Windows backing file stays dense so restore keeps the @@ -621,6 +695,18 @@ translation of the same directory instead of falling back to WSL's `/tmp`. Use `--scratch-dir` for manual runs whose temporary directory shares a volume with other I/O-heavy work. +Treat a change to the storage behind benchmark scratch or the runner workspace as a +performance-platform change. This includes a different VM size, disk SKU, size, +performance tier, caching policy, or bursting setting, and moving scratch between +the OS, data, and temporary disks. Because snapshot generation is bounded by the +scratch volume's write throughput, such a change shifts +`openvmm_snapshot_generation` and can shift the other lifecycle metrics. Apply it to +every pooled runner of the series at once; a change on some runners only would mix +two storage regimes in one history. In the same pull request, reset the affected +metrics by removing them from the series' history files, such as +`data/windows-whp-virtual-machine-microvm-v2-1vcpu.csv`, so that the gate restarts +their ten-point warmup instead of comparing the new storage with the old baseline. + The regression gate compares the target p50 with the median of the latest 10 p50 values on the pull request's base branch and requires all 10 matching history points. A metric regresses only when it is more than 40% diff --git a/doc/usage.md b/doc/usage.md index dc267362b..04af502f5 100644 --- a/doc/usage.md +++ b/doc/usage.md @@ -756,7 +756,7 @@ python3 scripts/nvx.py benchmark [OPTIONS] | `--timeout SECONDS` | `10` | Set the time allowed for each boot marker. | | `--teardown-mode {guest-exit,host-terminate,host-sigterm}` | `guest-exit` | Select how to stop a measured VM; `host-sigterm` is a deprecated alias. | | `--skip-build` | off | Reuse existing release binaries. | -| `--snapshot-profile` | off | Retain OpenVMM lifecycle phase samples and host counters; implied by the `snapshot-profile` suite. | +| `--snapshot-profile` | off | Retain OpenVMM lifecycle phase samples and host counters, and on Windows sample storage, CPU, writeback, and Defender telemetry during each capture; implied by the `snapshot-profile` suite. | | `--cache-state {warm,cold,both}` | `both` | Select artifact cache states for the `snapshot-profile` suite. | | `--output PATH` | none | Write the benchmark result as JSON. | | `--output-dir PATH` | none | Write canonical workload logs to a directory. | @@ -778,6 +778,8 @@ too: `auto` rejects such a catalog without falling back. The `e2e` suite uses the general memory size and measures cold start, snapshot generation, snapshot restore, guest-exit teardown, and peak RSS against the shell-ready markers. CI uses the default 128 MiB baseline. +JSON results and workload metadata record the benchmark host's provenance under +`host`, without credentials or Azure resource identifiers. See [Benchmark](benchmarks.md) for suite semantics, platform support, metric definitions, and complete examples. @@ -799,6 +801,7 @@ python3 scripts/nvx.py performance collect [--require-shared-suite] [--require-shell-snapshot-restore-512] [--lifecycle-input PATH] + [--lifecycle-attempt-log PATH] [--summary PATH] ``` @@ -810,7 +813,14 @@ restore metric from a 2-, 4-, or 8-vCPU run. result, producing the 29-metric microVM CI result. A directory whose metadata selects `device-io` is collected as five additional ABI-2, one-vCPU `ops/s` metrics; CI merges them into a 34-metric one-vCPU result. -`--summary` writes the p50 table plus lifecycle min/max/sample-count and RSS diagnostics. +`--lifecycle-attempt-log` reads the attempt log that `validate-openvmm` +records. The `--lifecycle-input` file must be a byte-identical copy of exactly +one valid logged attempt, or collection fails before it writes a CSV; the +command names that attempt. +`--summary` writes the p50 table plus lifecycle min/max/sample-count and RSS +diagnostics, the benchmark host provenance, and any snapshot-generation storage +telemetry. With `--lifecycle-attempt-log`, it also lists every attempt, marks +the published one, and reports the telemetry of rejected attempts. #### `performance validate-openvmm` @@ -818,12 +828,16 @@ metrics; CI merges them into a 34-metric one-vCPU result. python3 scripts/nvx.py performance validate-openvmm --platform PLATFORM --input PATH + [--attempt-log PATH] ``` Validates a complete 128 MiB, guest-exit OpenVMM `e2e` result without writing a CSV. Snapshot-generation instability exits with status 75 so callers can remeasure the temporary host condition selectively; other -malformed or incomplete inputs exit with status 2. +malformed or incomplete inputs exit with status 2. `--attempt-log` appends +the input's path, SHA-256 digest, outcome (`valid`, `unstable`, or +`invalid`), exit status, and error to a JSON attempt log, creating it on the +first attempt. #### `performance collect-openvmm` diff --git a/scripts/nvx_tools/benchmark.py b/scripts/nvx_tools/benchmark.py index a39be7f3e..72c97bade 100644 --- a/scripts/nvx_tools/benchmark.py +++ b/scripts/nvx_tools/benchmark.py @@ -33,7 +33,7 @@ from string import Template from typing import IO, Literal, Protocol, TextIO, TypedDict, cast -from . import common +from . import common, host_telemetry from .adversarial_oracles import ProcessTreeContainment from .build_constants import ( AlpineBuildConstants, @@ -234,6 +234,7 @@ class E2EComparison(TypedDict): class ResultDocument(TypedDict): timestamp_utc: str controls: dict[str, object] + host: dict[str, object] backends: dict[str, BenchmarkResult] snapshot_capture: dict[str, SnapshotCaptureResult] snapshot_restore: dict[str, BenchmarkResult] @@ -1226,6 +1227,75 @@ def summarize_lifecycle_profiles( } +def snapshot_profile_phase_windows( + records: Sequence[dict[str, object]], +) -> dict[str, tuple[int, int]]: + """Map each profiled phase to its start and end on the observer clock. + + OpenVMM's clock starts at its own process start, which the observer does + not see. The host observes every record after OpenVMM writes it, so the + smallest observed-minus-reported time bounds the clock offset most + tightly. + """ + offsets = [ + observed - reported + for record in records + if record.get("source") == "openvmm" + and isinstance(observed := record.get("observer_elapsed_ns"), int) + and isinstance(reported := record.get("process_elapsed_ns"), int) + ] + windows: dict[str, tuple[int, int]] = {} + for record in records: + duration = record.get("duration_ns") + reported = record.get("process_elapsed_ns") + observed = record.get("observer_elapsed_ns") + if not isinstance(duration, int): + continue + if isinstance(reported, int) and offsets: + end = reported + min(offsets) + elif isinstance(observed, int): + end = observed + else: + continue + windows[f"{record.get('operation')}.{record.get('phase')}"] = ( + end - duration, + end, + ) + return windows + + +def describe_capture_telemetry(sample: dict[str, object]) -> str | None: + """Describe the host activity during a profiled capture's RAM flush.""" + telemetry = sample.get("storage_telemetry") + records = sample.get("records") + if not isinstance(telemetry, dict) or not isinstance(records, list): + return None + telemetry = cast(dict[str, object], telemetry) + phases = telemetry.get("phases") + volumes = telemetry.get("volumes") + if not isinstance(phases, dict) or not isinstance(volumes, dict): + return None + window = cast(dict[str, object], phases).get("capture.mapped_memory_flush") + flush = next( + ( + record + for record in cast(list[dict[str, object]], records) + if record.get("operation") == "capture" + and record.get("phase") == "mapped_memory_flush" + ), + None, + ) + if not isinstance(window, dict) or flush is None: + return None + description = host_telemetry.describe_storage_window( + cast(dict[str, object], window), cast(dict[str, object], volumes) + ) + return ( + f"mapped_memory_flush {_profile_int(flush, 'duration_ns') / 1_000_000:.3f} " + f"ms: {description}" + ) + + def terminate(process: subprocess.Popen[bytes]) -> None: if process.poll() is None: process.kill() @@ -4449,6 +4519,7 @@ def write_benchmark_metadata( "host_affinity_set": args.cpus, "host_cpu_reserve": args.host_cpu_reserve, "scratch_directory": scratch_directory_control(args), + "host": benchmark_host_provenance(args), "memory_mib": { "lifecycle": args.memory_mib, "virtfs": args.virtfs_memory_mib, @@ -4494,7 +4565,7 @@ def write_benchmark_metadata( existing = json.loads(path.read_text(encoding="utf-8")) except (UnicodeError, json.JSONDecodeError) as error: raise ValueError(f"invalid resumable benchmark metadata {path}") from error - if existing != document: + if _resume_identity(existing) != _resume_identity(document): raise ValueError( f"device-io resume controls or provenance do not match {path}" ) @@ -4503,6 +4574,21 @@ def write_benchmark_metadata( return path +def _resume_identity(document: object) -> object: + """Return the metadata that a resumed device-io series must match. + + Host provenance describes the invocation that started the series and can + differ on a resume of the same host, such as after a metadata timeout. + """ + if not isinstance(document, dict): + return document + return { + key: value + for key, value in cast(dict[str, object], document).items() + if key != "host" + } + + def run_workload_benchmarks( args: argparse.Namespace, executable: Path, @@ -4771,6 +4857,7 @@ def capture_snapshot( snapshot_request_bytes = None snapshot_published_ns = None peak_bytes = 0 + storage: host_telemetry.StorageTelemetry | None = None def observe_snapshot_publication() -> None: nonlocal snapshot_published_ns @@ -4784,10 +4871,19 @@ def observe_snapshot_publication() -> None: def request_snapshot() -> None: nonlocal snapshot_requested, snapshot_started_ns nonlocal snapshot_dispatched_ns, snapshot_request_bytes, deadline + nonlocal storage payload = snapshot_request_script( processors, teardown_mode=teardown_mode, ).encode("utf-8") + if snapshot_profile and profile_sink is not None: + # Sampling starts before the request so that its first sample + # shifts no host-observed capture interval. + storage = host_telemetry.create_storage_telemetry( + capture_telemetry_volumes(snapshot_path), process_started_ns + ) + if storage is not None: + storage.start() snapshot_started_ns = time.perf_counter_ns() interaction.write_input(payload) snapshot_dispatched_ns = time.perf_counter_ns() @@ -4868,6 +4964,8 @@ def request_snapshot() -> None: monitor.finish() returncode = process.wait() source_exited_ns = time.perf_counter_ns() + if storage is not None: + storage.stop() monitor.check_exit(returncode) if snapshot_published_ns is None and snapshot_path.is_dir(): snapshot_published_ns = source_exited_ns @@ -4893,6 +4991,12 @@ def request_snapshot() -> None: source_exited_ns, snapshot_request_bytes, ) + if storage is not None: + profile_sample["storage_telemetry"] = storage.result( + snapshot_profile_phase_windows( + cast(list[dict[str, object]], profile_sample["records"]) + ) + ) if snapshot_profile and profile_sink is not None: profile_sink.append(profile_sample) request_to_publication_ms = ( @@ -4912,12 +5016,30 @@ def request_snapshot() -> None: raise RuntimeError(f"{error}\n--- OpenVMM output ---\n{tail}") from error raise finally: + if storage is not None: + storage.stop() if log_path is not None: log_path.parent.mkdir(parents=True, exist_ok=True) log_path.write_bytes(output.contents()) interaction.close() +def capture_telemetry_volumes(snapshot_path: Path) -> dict[str, Path]: + """Return the volumes whose activity can bound snapshot generation. + + OpenVMM creates the automatic guest RAM backing file beside the snapshot + destination, so the capture's RAM flush writes to the scratch volume. + """ + volumes = { + "scratch": snapshot_path.parent, + "workspace": BuildConstants.REPO_ROOT, + } + system_root = os.environ.get("SystemRoot") + if system_root: + volumes["system"] = Path(system_root) + return volumes + + def summarize_snapshot_samples( samples: list[float], request_to_publication_samples: list[float], @@ -5018,6 +5140,13 @@ def benchmark_snapshot_capture( f"peak RSS={bytes_to_mib(peak_bytes):.3f} MiB", flush=True, ) + telemetry = ( + describe_capture_telemetry(profile_samples[-1]) + if snapshot_profile + else None + ) + if telemetry is not None: + print(f" {telemetry}", flush=True) return summarize_snapshot_samples( samples, request_to_publication_samples, @@ -5885,6 +6014,7 @@ def result_document( "cache_state": args.cache_state, "scratch_directory": scratch_directory_control(args), }, + "host": benchmark_host_provenance(args), "backends": {}, "snapshot_capture": {}, "snapshot_restore": {}, @@ -5985,6 +6115,10 @@ def run_native_linux(args: argparse.Namespace) -> int: command_prefix=prefix, ) results = result_document(args, kernel, initrd, backend) + print( + "Benchmark host: " + host_telemetry.describe_host_provenance(results["host"]), + flush=True, + ) if run_guest: assert executable is not None and kernel is not None and initrd is not None @@ -6129,6 +6263,24 @@ def _resolved_scratch_directory(args: argparse.Namespace) -> Path: return Path(scratch).resolve() +def _system_temporary_directory(args: argparse.Namespace) -> Path: + temporary = getattr(args, "system_temporary_dir", None) + if temporary is None: + temporary = Path(tempfile.gettempdir()) + return Path(temporary).resolve() + + +def benchmark_host_provenance(args: argparse.Namespace) -> dict[str, object]: + """Return safe provenance for this host and the benchmark's volumes.""" + return host_telemetry.host_provenance( + { + "workspace": BuildConstants.REPO_ROOT, + "temporary": _system_temporary_directory(args), + "scratch": _resolved_scratch_directory(args), + } + ) + + def _kvm_worker_scratch_arguments(args: argparse.Namespace) -> list[str]: return [ "--scratch-dir", @@ -6338,6 +6490,7 @@ def benchmark_scratch_directory(args: argparse.Namespace) -> Generator[None]: raise ValueError(f"benchmark scratch directory does not exist: {scratch}") args.scratch_dir = scratch previous = tempfile.tempdir + args.system_temporary_dir = Path(tempfile.gettempdir()).resolve() tempfile.tempdir = str(scratch) try: yield @@ -6463,6 +6616,10 @@ def run_benchmark(args: argparse.Namespace) -> int: ) results = result_document(args, kernel, initrd) + print( + "Benchmark host: " + host_telemetry.describe_host_provenance(results["host"]), + flush=True, + ) if run_profile: assert kernel is not None and initrd is not None executable = boot_binaries["whp"] diff --git a/scripts/nvx_tools/host_telemetry.py b/scripts/nvx_tools/host_telemetry.py new file mode 100644 index 000000000..abcc8b773 --- /dev/null +++ b/scripts/nvx_tools/host_telemetry.py @@ -0,0 +1,1397 @@ +#!/usr/bin/env python3 + +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Record safe benchmark host provenance and bounded storage telemetry. + +Provenance names the runner and machine, the CPU and memory, the Azure VM +size and managed-disk configuration, and the volumes behind the benchmark's +workspace, temporary, and scratch directories. It never records credentials, +network addresses, Azure resource identifiers, names, or tags. + +Storage telemetry samples cumulative operating-system counters while a +snapshot capture runs, so that each profiled capture phase can be compared +with the disk, CPU, writeback, and Microsoft Defender activity that overlapped +it. Windows CI jobs run as Network Service, which cannot read performance +counters, so the sampler uses only interfaces that any account can query: +volume performance IOCTLs, whose on-demand counters an open disk performance +WMI data block enables, GetSystemTimes, and NtQuerySystemInformation. +""" + +from __future__ import annotations + +import copy +import ctypes +import functools +import http.client +import json +import mmap +import os +import platform +import re +import shutil +import struct +import sys +import threading +import time +import urllib.request +import uuid +from collections.abc import Callable, Mapping, Sequence +from pathlib import Path +from typing import NamedTuple, Protocol, cast + +HOST_PROVENANCE_SCHEMA_VERSION = 1 +STORAGE_TELEMETRY_SCHEMA_VERSION = 1 +STORAGE_TELEMETRY_INTERVAL_SECONDS = 0.05 +STORAGE_TELEMETRY_PROCESS_INTERVAL_SECONDS = 0.5 +STORAGE_TELEMETRY_MAX_SAMPLES = 600 +STORAGE_TELEMETRY_PHASES = ( + "capture.snapshot_generation", + "capture.mapped_memory_flush", +) +AZURE_IMDS_URL = ( + "http://169.254.169.254/metadata/instance/compute?api-version=2021-05-01" +) +AZURE_IMDS_TIMEOUT_SECONDS = 2.0 +AZURE_CHASSIS_ASSET_TAG = "7783-7084-3265-9085-8269-3286-77" +DEFENDER_PROCESS_NAMES = frozenset({"msmpeng.exe", "mpdefendercoreservice.exe"}) +MIB = 1024 * 1024 +GIB = 1024 * MIB +TICKS_PER_SECOND = 10_000_000 +"""Windows reports disk, CPU, and process times in 100 ns ticks.""" + +_WINDOWS_PROCESSOR_KEY = r"HARDWARE\DESCRIPTION\System\CentralProcessor\0" +_WINDOWS_BIOS_KEY = r"HARDWARE\DESCRIPTION\System\BIOS" +_SCSI_ADDRESS = re.compile(r"\d+:\d+:\d+:\d+") + + +def host_provenance(volumes: Mapping[str, Path]) -> dict[str, object]: + """Return safe provenance for this host and the volumes of ``volumes``. + + ``volumes`` maps a role, such as ``scratch``, to a path on the volume. + """ + return { + "schema_version": HOST_PROVENANCE_SCHEMA_VERSION, + "runner_name": os.environ.get("RUNNER_NAME") or None, + "machine_name": platform.node() or None, + "os": platform.platform(), + "cpu_model": cpu_model(), + "logical_processors": os.cpu_count(), + "memory_bytes": physical_memory_bytes(), + "azure": azure_vm_provenance(), + "volumes": {role: volume_provenance(path) for role, path in volumes.items()}, + } + + +def cpu_model() -> str | None: + """Return the host CPU's brand string as the host OS reports it.""" + try: + if sys.platform == "win32": + import winreg + + with winreg.OpenKey( + winreg.HKEY_LOCAL_MACHINE, _WINDOWS_PROCESSOR_KEY + ) as key: + value, _ = winreg.QueryValueEx(key, "ProcessorNameString") + return str(value).strip() or None + for line in Path("/proc/cpuinfo").read_text(encoding="utf-8").splitlines(): + name, separator, value = line.partition(":") + if separator and name.strip() == "model name": + return value.strip() or None + except (OSError, UnicodeError): + pass + return platform.processor() or None + + +class _MemoryStatus(ctypes.Structure): + _fields_ = [ + ("length", ctypes.c_uint32), + ("memory_load", ctypes.c_uint32), + ("total_physical", ctypes.c_uint64), + ("available_physical", ctypes.c_uint64), + ("total_page_file", ctypes.c_uint64), + ("available_page_file", ctypes.c_uint64), + ("total_virtual", ctypes.c_uint64), + ("available_virtual", ctypes.c_uint64), + ("available_extended_virtual", ctypes.c_uint64), + ] + + +def physical_memory_bytes() -> int | None: + """Return the host's installed physical memory in bytes.""" + if sys.platform == "win32": + status = _MemoryStatus() + status.length = ctypes.sizeof(status) + try: + if not _kernel32().GlobalMemoryStatusEx(ctypes.byref(status)): + return None + except OSError: + return None + return int(status.total_physical) + try: + return os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") + except (OSError, ValueError): + return None + + +def azure_vm_provenance() -> dict[str, object] | None: + """Return the Azure VM size and disks, or None off Azure. + + The instance metadata service is queried at most once per process. + """ + return copy.deepcopy(_azure_vm_provenance()) + + +@functools.cache +def _azure_vm_provenance() -> dict[str, object] | None: + if not may_be_azure_vm(): + return None + try: + return azure_compute_provenance(read_azure_compute_metadata()) + except (OSError, ValueError, http.client.HTTPException) as error: + return {"error": f"Azure instance metadata is unavailable: {error}"} + + +def may_be_azure_vm() -> bool: + """Return whether the firmware identifies a possible Azure VM. + + Linux reads Azure's chassis asset tag. Windows exposes only the Hyper-V + manufacturer and product to unprivileged accounts, so on another Hyper-V + guest the metadata query fails after its timeout and records the error. + """ + try: + if sys.platform == "win32": + import winreg + + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, _WINDOWS_BIOS_KEY) as key: + manufacturer, _ = winreg.QueryValueEx(key, "SystemManufacturer") + product, _ = winreg.QueryValueEx(key, "SystemProductName") + return (manufacturer, product) == ( + "Microsoft Corporation", + "Virtual Machine", + ) + tag = Path("/sys/class/dmi/id/chassis_asset_tag").read_text(encoding="ascii") + return tag.strip() == AZURE_CHASSIS_ASSET_TAG + except (OSError, UnicodeError): + return False + + +def read_azure_compute_metadata( + timeout: float = AZURE_IMDS_TIMEOUT_SECONDS, +) -> object: + request = urllib.request.Request(AZURE_IMDS_URL, headers={"Metadata": "true"}) + # The link-local metadata service must never be reached through a proxy. + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with opener.open(request, timeout=timeout) as response: + return json.loads(response.read(MIB).decode("utf-8")) + + +def azure_compute_provenance(compute: object) -> dict[str, object]: + """Keep only the VM size and disk performance settings of IMDS compute + metadata. Names, resource IDs, the subscription, tags, and network data + are dropped.""" + document = _string_keys(compute) + vm_size = _text(document.get("vmSize")) + if vm_size is None: + raise ValueError("compute metadata has no vmSize") + storage = _string_keys(document.get("storageProfile")) + os_disk = _string_keys(storage.get("osDisk")) + data_disks = storage.get("dataDisks") + disks = cast(list[object], data_disks) if isinstance(data_disks, list) else [] + return { + "vm_size": vm_size, + "os_disk": _azure_disk(os_disk) if os_disk else None, + "data_disks": [ + _azure_disk(disk, data_disk=True) + for disk in map(_string_keys, disks) + if disk + ], + "resource_disk_size_kib": _integer( + _string_keys(storage.get("resourceDisk")).get("size") + ), + } + + +def _azure_disk( + disk: Mapping[str, object], *, data_disk: bool = False +) -> dict[str, object]: + result: dict[str, object] = {} + if data_disk: + result["lun"] = _integer(disk.get("lun")) + result.update( + { + "storage_account_type": _text( + _string_keys(disk.get("managedDisk")).get("storageAccountType") + ), + "size_gib": _integer(disk.get("diskSizeGB")), + "caching": _text(disk.get("caching")), + "write_accelerator_enabled": _boolean(disk.get("writeAcceleratorEnabled")), + "ephemeral_option": _text( + _string_keys(disk.get("diffDiskSettings")).get("option") + ), + } + ) + if data_disk: + # IMDS populates the provisioned limits only for Ultra Disks. + result["bytes_per_second_throttle"] = _integer( + disk.get("bytesPerSecondThrottle") + ) + result["ops_per_second_throttle"] = _integer(disk.get("opsPerSecondThrottle")) + return result + + +def _string_keys(value: object) -> dict[str, object]: + if not isinstance(value, dict): + return {} + return {str(key): item for key, item in cast(dict[object, object], value).items()} + + +def _text(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _integer(value: object) -> int | None: + if isinstance(value, bool): + return None + if isinstance(value, int): + return value + if isinstance(value, str) and value.isdecimal(): + return int(value) + return None + + +def _boolean(value: object) -> bool | None: + if isinstance(value, bool): + return value + if isinstance(value, str) and value.lower() in ("true", "false"): + return value.lower() == "true" + return None + + +_VOLUME_ROLE_LABELS = ( + ("workspace", "Workspace volume"), + ("temporary", "Temporary-directory volume"), + ("scratch", "Benchmark scratch volume"), +) + + +def host_provenance_rows(host: Mapping[str, object]) -> list[tuple[str, str]]: + """Return labeled, human-readable values of recorded host provenance.""" + rows = [ + ("Runner", _text(host.get("runner_name")) or "not recorded"), + ("Machine", _text(host.get("machine_name")) or "unknown"), + ("Operating system", _text(host.get("os")) or "unknown"), + ] + cpu = _text(host.get("cpu_model")) or "unknown CPU" + processors = _integer(host.get("logical_processors")) + if processors is not None: + cpu = f"{cpu}, {processors} logical processors" + rows.append(("CPU", cpu)) + memory = _integer(host.get("memory_bytes")) + if memory is not None: + rows.append(("Memory", f"{memory / GIB:.1f} GiB")) + azure = _string_keys(host.get("azure")) + if azure: + rows.extend(_azure_rows(azure)) + volumes = _string_keys(host.get("volumes")) + for role, label in _VOLUME_ROLE_LABELS: + if role in volumes: + rows.append((label, _describe_volume(_string_keys(volumes[role])))) + return rows + + +def describe_host_provenance(host: Mapping[str, object]) -> str: + """Return recorded host provenance as one line.""" + return "; ".join(f"{label}: {value}" for label, value in host_provenance_rows(host)) + + +def _azure_rows(azure: Mapping[str, object]) -> list[tuple[str, str]]: + error = _text(azure.get("error")) + if error is not None: + return [("Azure metadata", error)] + rows = [("Azure VM size", _text(azure.get("vm_size")) or "unknown")] + os_disk = _string_keys(azure.get("os_disk")) + if os_disk: + rows.append(("Azure OS disk", _describe_azure_disk(os_disk))) + data_disks = azure.get("data_disks") + disks = cast(list[object], data_disks) if isinstance(data_disks, list) else [] + for disk in map(_string_keys, disks): + lun = _integer(disk.get("lun")) + label = "Azure data disk" if lun is None else f"Azure data disk LUN {lun}" + rows.append((label, _describe_azure_disk(disk))) + temporary = _integer(azure.get("resource_disk_size_kib")) + if temporary: + rows.append(("Azure temporary disk", f"{temporary / MIB:.1f} GiB")) + return rows + + +def _describe_azure_disk(disk: Mapping[str, object]) -> str: + parts = [_text(disk.get("storage_account_type")) or "unknown SKU"] + size = _integer(disk.get("size_gib")) + if size is not None: + parts.append(f"{size} GiB") + caching = _text(disk.get("caching")) + if caching is not None: + parts.append(f"{caching} caching") + if disk.get("write_accelerator_enabled") is True: + parts.append("write accelerator") + ephemeral = _text(disk.get("ephemeral_option")) + if ephemeral is not None: + parts.append(f"ephemeral ({ephemeral})") + throughput = _integer(disk.get("bytes_per_second_throttle")) + if throughput is not None: + parts.append(f"{throughput / MIB:.0f} MiB/s limit") + operations = _integer(disk.get("ops_per_second_throttle")) + if operations is not None: + parts.append(f"{operations} IOPS limit") + return ", ".join(parts) + + +def _describe_volume(volume: Mapping[str, object]) -> str: + error = _text(volume.get("error")) + if error is not None: + return f"unavailable ({error})" + name = _text(volume.get("volume")) or "unknown volume" + filesystem = _text(volume.get("filesystem")) + parts = [name if filesystem is None else f"{name} {filesystem}"] + size = _integer(volume.get("size_bytes")) + if size is not None: + parts.append(f"{size / GIB:.1f} GiB") + if volume.get("system_volume") is True: + parts.append("system volume") + disk = _integer(volume.get("disk_number")) + if disk is not None: + parts.append(f"disk {disk}") + device = _text(volume.get("device")) + if device is not None: + parts.append(f"device {device}") + address = _text(volume.get("scsi_address")) + if address is not None: + parts.append(f"SCSI {address}") + path = _text(volume.get("path")) + if path is not None: + parts.append(f"at {path}") + return ", ".join(parts) + + +def volume_provenance(path: Path) -> dict[str, object]: + """Describe the volume that holds ``path``.""" + try: + if sys.platform == "win32": + return _windows_volume_provenance(path) + return _linux_volume_provenance(path) + except (OSError, ValueError) as error: + return {"path": str(path), "error": str(error)} + + +def _linux_volume_provenance(path: Path) -> dict[str, object]: + if sys.platform == "win32": + raise OSError("Linux volume provenance is unavailable on Windows") + device = os.stat(path).st_dev + major, minor = os.major(device), os.minor(device) + result: dict[str, object] = {"path": str(path)} + mountinfo = Path("/proc/self/mountinfo").read_text(encoding="utf-8") + mount = linux_mount(str(path.resolve()), f"{major}:{minor}", mountinfo) + if mount is not None: + result["volume"], result["filesystem"] = mount + result["system_volume"] = mount[0] == "/" + result["size_bytes"] = shutil.disk_usage(path).total + block = Path(f"/sys/dev/block/{major}:{minor}") + if block.exists(): + resolved = block.resolve() + disk = resolved.parent if (resolved / "partition").exists() else resolved + result["device"] = resolved.name + result["disk"] = disk.name + address = (disk / "device").resolve().name + if _SCSI_ADDRESS.fullmatch(address): + result["scsi_address"] = address + return result + + +def linux_mount(target: str, device: str, mountinfo: str) -> tuple[str, str] | None: + """Return the mount point and file system of the deepest mount of + ``device`` in ``mountinfo`` that contains the resolved path ``target``.""" + deepest: tuple[str, str] | None = None + for line in mountinfo.splitlines(): + fields = line.split() + if len(fields) < 10 or fields[2] != device or "-" not in fields[6:]: + continue + mount_point = re.sub( + r"\\([0-7]{3})", lambda match: chr(int(match[1], 8)), fields[4] + ) + filesystem = fields[fields.index("-", 6) + 1] + contains = target == mount_point or target.startswith( + mount_point.rstrip("/") + "/" + ) + if contains and (deepest is None or len(mount_point) > len(deepest[0])): + deepest = (mount_point, filesystem) + return deepest + + +_FILE_SHARE_READ_WRITE = 0x3 +_OPEN_EXISTING = 3 +_INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value +_IOCTL_DISK_PERFORMANCE = 0x00070020 +_IOCTL_STORAGE_GET_DEVICE_NUMBER = 0x002D1080 +_IOCTL_SCSI_GET_ADDRESS = 0x00041018 +_SYSTEM_PERFORMANCE_INFORMATION = 2 +_SYSTEM_PROCESS_INFORMATION = 5 +_STATUS_INFO_LENGTH_MISMATCH = 0xC0000004 +_DISK_PERFORMANCE_WMI_GUID = uuid.UUID("bdd865d1-d7c1-11d0-a501-00a0c9062910") +_WMIGUID_QUERY = 0x0001 + + +class _Guid(ctypes.Structure): + _fields_ = [ + ("data1", ctypes.c_uint32), + ("data2", ctypes.c_uint16), + ("data3", ctypes.c_uint16), + ("data4", ctypes.c_uint8 * 8), + ] + + +class _DiskPerformance(ctypes.Structure): + _fields_ = [ + ("bytes_read", ctypes.c_int64), + ("bytes_written", ctypes.c_int64), + ("read_time", ctypes.c_int64), + ("write_time", ctypes.c_int64), + ("idle_time", ctypes.c_int64), + ("read_count", ctypes.c_uint32), + ("write_count", ctypes.c_uint32), + ("queue_depth", ctypes.c_uint32), + ("split_count", ctypes.c_uint32), + ("query_time", ctypes.c_int64), + ("storage_device_number", ctypes.c_uint32), + ("storage_manager_name", ctypes.c_uint16 * 8), + ] + + +class _StorageDeviceNumber(ctypes.Structure): + _fields_ = [ + ("device_type", ctypes.c_uint32), + ("device_number", ctypes.c_uint32), + ("partition_number", ctypes.c_uint32), + ] + + +class _ScsiAddress(ctypes.Structure): + _fields_ = [ + ("length", ctypes.c_uint32), + ("port_number", ctypes.c_uint8), + ("path_id", ctypes.c_uint8), + ("target_id", ctypes.c_uint8), + ("lun", ctypes.c_uint8), + ] + + +@functools.cache +def _kernel32() -> ctypes.CDLL: + if sys.platform != "win32": + raise OSError("the Windows API is unavailable") + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.CreateFileW.argtypes = ( + ctypes.c_wchar_p, + ctypes.c_uint32, + ctypes.c_uint32, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.c_uint32, + ctypes.c_void_p, + ) + kernel32.CreateFileW.restype = ctypes.c_void_p + kernel32.DeviceIoControl.argtypes = ( + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.POINTER(ctypes.c_uint32), + ctypes.c_void_p, + ) + kernel32.DeviceIoControl.restype = ctypes.c_int + kernel32.CloseHandle.argtypes = (ctypes.c_void_p,) + kernel32.CloseHandle.restype = ctypes.c_int + kernel32.GetSystemTimes.argtypes = (ctypes.POINTER(ctypes.c_uint64),) * 3 + kernel32.GetSystemTimes.restype = ctypes.c_int + kernel32.GetVolumePathNameW.argtypes = ( + ctypes.c_wchar_p, + ctypes.c_wchar_p, + ctypes.c_uint32, + ) + kernel32.GetVolumePathNameW.restype = ctypes.c_int + kernel32.GetVolumeNameForVolumeMountPointW.argtypes = ( + ctypes.c_wchar_p, + ctypes.c_wchar_p, + ctypes.c_uint32, + ) + kernel32.GetVolumeNameForVolumeMountPointW.restype = ctypes.c_int + kernel32.GetVolumeInformationW.argtypes = ( + ctypes.c_wchar_p, + ctypes.c_wchar_p, + ctypes.c_uint32, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_wchar_p, + ctypes.c_uint32, + ) + kernel32.GetVolumeInformationW.restype = ctypes.c_int + kernel32.GlobalMemoryStatusEx.argtypes = (ctypes.c_void_p,) + kernel32.GlobalMemoryStatusEx.restype = ctypes.c_int + return kernel32 + + +@functools.cache +def _ntdll() -> ctypes.CDLL: + if sys.platform != "win32": + raise OSError("the Windows API is unavailable") + ntdll = ctypes.WinDLL("ntdll") + ntdll.NtQuerySystemInformation.argtypes = ( + ctypes.c_uint32, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.POINTER(ctypes.c_uint32), + ) + ntdll.NtQuerySystemInformation.restype = ctypes.c_int32 + return ntdll + + +@functools.cache +def _advapi32() -> ctypes.CDLL: + if sys.platform != "win32": + raise OSError("the Windows API is unavailable") + advapi32 = ctypes.WinDLL("advapi32") + advapi32.WmiOpenBlock.argtypes = ( + ctypes.POINTER(_Guid), + ctypes.c_uint32, + ctypes.POINTER(ctypes.c_void_p), + ) + advapi32.WmiOpenBlock.restype = ctypes.c_uint32 + advapi32.WmiCloseBlock.argtypes = (ctypes.c_void_p,) + advapi32.WmiCloseBlock.restype = ctypes.c_uint32 + return advapi32 + + +def _windows_error(code: int | None = None) -> OSError: + if sys.platform == "win32": + return ctypes.WinError(ctypes.get_last_error() if code is None else code) + return OSError("the Windows API is unavailable") + + +def _enable_disk_performance() -> int: + """Open the disk performance WMI data block and return its handle. + + Windows collects disk counters only on demand, and an open block is + such a demand: until it closes, IOCTL_DISK_PERFORMANCE returns counters + instead of failing. The block grants query access to every account. + """ + guid = _Guid.from_buffer_copy(_DISK_PERFORMANCE_WMI_GUID.bytes_le) + handle = ctypes.c_void_p() + status = _advapi32().WmiOpenBlock( + ctypes.byref(guid), _WMIGUID_QUERY, ctypes.byref(handle) + ) + if status != 0 or not handle.value: + raise _windows_error(status) + return handle.value + + +def _open_device(path: str) -> int: + # Zero access opens a volume or disk for query IOCTLs without privileges. + handle = _kernel32().CreateFileW( + path, 0, _FILE_SHARE_READ_WRITE, None, _OPEN_EXISTING, 0, None + ) + if handle is None or handle == _INVALID_HANDLE_VALUE: + raise _windows_error() + return int(handle) + + +def _close_device(handle: int) -> None: + _kernel32().CloseHandle(handle) + + +def _device_io_control(handle: int, code: int, output: ctypes.Structure) -> None: + returned = ctypes.c_uint32() + if not _kernel32().DeviceIoControl( + handle, + code, + None, + 0, + ctypes.byref(output), + ctypes.sizeof(output), + ctypes.byref(returned), + None, + ): + raise _windows_error() + + +def _windows_mount_point(path: Path) -> str: + buffer = ctypes.create_unicode_buffer(32768) + if not _kernel32().GetVolumePathNameW(str(path), buffer, len(buffer)): + raise _windows_error() + return buffer.value + + +def _windows_volume_device(mount_point: str) -> str: + buffer = ctypes.create_unicode_buffer(64) + if not _kernel32().GetVolumeNameForVolumeMountPointW( + mount_point, buffer, len(buffer) + ): + raise _windows_error() + # Without its trailing separator, the GUID path opens the volume device + # rather than its root directory. + return buffer.value.rstrip("\\") + + +def _volume_key(mount_point: str) -> str: + key = mount_point.rstrip("\\/") or mount_point + # Paths keep the drive-letter case of their spelling. + return key.upper() if re.fullmatch(r"[A-Za-z]:", key) else key + + +def _windows_volume_provenance(path: Path) -> dict[str, object]: + mount_point = _windows_mount_point(path) + key = _volume_key(mount_point) + result: dict[str, object] = {"path": str(path), "volume": key} + filesystem = ctypes.create_unicode_buffer(64) + if _kernel32().GetVolumeInformationW( + mount_point, None, 0, None, None, None, filesystem, len(filesystem) + ): + result["filesystem"] = filesystem.value + result["size_bytes"] = shutil.disk_usage(mount_point).total + system_root = os.environ.get("SystemRoot") + if system_root: + system_key = _volume_key(_windows_mount_point(Path(system_root))) + result["system_volume"] = system_key.casefold() == key.casefold() + try: + result.update(_windows_disk_identity(_windows_volume_device(mount_point))) + except OSError: + pass + return result + + +def _windows_disk_identity(volume_device: str) -> dict[str, object]: + """Return the disk number and SCSI address behind a single-disk volume, + which identify the Azure OS disk, data-disk LUN, or temporary disk.""" + number = _StorageDeviceNumber() + handle = _open_device(volume_device) + try: + _device_io_control(handle, _IOCTL_STORAGE_GET_DEVICE_NUMBER, number) + finally: + _close_device(handle) + identity: dict[str, object] = {"disk_number": int(number.device_number)} + address = _ScsiAddress() + try: + handle = _open_device(rf"\\.\PhysicalDrive{number.device_number}") + try: + _device_io_control(handle, _IOCTL_SCSI_GET_ADDRESS, address) + finally: + _close_device(handle) + except OSError: + return identity + identity["scsi_address"] = ( + f"{address.port_number}:{address.path_id}:{address.target_id}:{address.lun}" + ) + return identity + + +# SYSTEM_PERFORMANCE_INFORMATION offsets. winternl.h documents the structure +# only as a 312-byte reserved block; these counters keep the NT kernel's +# layout, which Windows 10 extended with the system cache's dirty pages. +_SYSTEM_PERFORMANCE_COUNTERS = ( + ("available_pages", 44), + ("modified_pages_written", 96), + ("modified_page_writes", 100), + ("mapped_pages_written", 104), + ("mapped_page_writes", 108), + ("lazy_write_ios", 280), + ("lazy_write_pages", 284), + ("cache_flushes", 288), + ("cache_flush_pages", 292), +) +_SYSTEM_PERFORMANCE_BASE_SIZE = 312 +_SYSTEM_PERFORMANCE_CACHE_COUNTERS = ( + ("cache_dirty_pages", 312), + ("cache_dirty_page_threshold", 320), +) +_SYSTEM_PERFORMANCE_SIZE = 344 + +# x64 SYSTEM_PROCESS_INFORMATION offsets from winternl.h. Its first reserved +# block holds the user and kernel times, and its last one the I/O operation +# and transfer counts. +_PROCESS_ENTRY_SIZE = 256 +_PROCESS_TIMES_OFFSET = 40 +_PROCESS_IMAGE_NAME_OFFSET = 56 +_PROCESS_TRANSFERS_OFFSET = 232 +_DEFENDER_NAME_LENGTHS = frozenset(len(name) * 2 for name in DEFENDER_PROCESS_NAMES) +_MAX_PROCESS_INFORMATION_BYTES = 64 * MIB + + +def parse_system_performance(data: bytes) -> dict[str, int]: + """Return the writeback and memory counters of SystemPerformanceInformation.""" + if len(data) < _SYSTEM_PERFORMANCE_BASE_SIZE: + raise ValueError(f"system performance information has only {len(data)} bytes") + counters: dict[str, int] = { + name: struct.unpack_from("= _SYSTEM_PERFORMANCE_SIZE: + counters.update( + (name, struct.unpack_from(" dict[str, int]: + """Sum the CPU time and I/O transfers of Microsoft Defender processes in + an x64 SystemProcessInformation buffer that was filled at + ``base_address``.""" + totals: dict[str, int] = { + "processes": 0, + "cpu_ticks": 0, + "read_bytes": 0, + "write_bytes": 0, + "other_bytes": 0, + } + offset = 0 + while True: + if offset + _PROCESS_ENTRY_SIZE > len(data): + raise ValueError("process information entry is truncated") + (next_entry,) = struct.unpack_from(" len(data): + raise ValueError("process image name lies outside the buffer") + name = data[start : start + length].decode("utf-16-le").casefold() + if name in DEFENDER_PROCESS_NAMES: + user, kernel = struct.unpack_from( + " None: + super().__init__(f"system information requires {required} bytes") + self.required = required + + +def _query_system_information( + information_class: int, buffer: ctypes.Array[ctypes.c_char] +) -> bytes: + returned = ctypes.c_uint32() + status = _ntdll().NtQuerySystemInformation( + information_class, buffer, len(buffer), ctypes.byref(returned) + ) + if status & 0xFFFFFFFF == _STATUS_INFO_LENGTH_MISMATCH: + raise _BufferTooSmallError(returned.value) + if status != 0: + raise OSError( + f"NtQuerySystemInformation({information_class}) failed with status " + f"0x{status & 0xFFFFFFFF:08X}" + ) + return ctypes.string_at(buffer, returned.value) + + +def _volume_counters(handle: int) -> dict[str, int]: + counters = _DiskPerformance() + _device_io_control(handle, _IOCTL_DISK_PERFORMANCE, counters) + return { + "bytes_read": counters.bytes_read, + "bytes_written": counters.bytes_written, + "read_ticks": counters.read_time, + "write_ticks": counters.write_time, + "idle_ticks": counters.idle_time, + "reads": counters.read_count, + "writes": counters.write_count, + "queue_depth": counters.queue_depth, + "query_ticks": counters.query_time, + } + + +def _cpu_counters() -> dict[str, int]: + idle, kernel, user = ctypes.c_uint64(), ctypes.c_uint64(), ctypes.c_uint64() + if not _kernel32().GetSystemTimes( + ctypes.byref(idle), ctypes.byref(kernel), ctypes.byref(user) + ): + raise _windows_error() + return { + "idle_ticks": idle.value, + "kernel_ticks": kernel.value, + "user_ticks": user.value, + } + + +class CounterReader(Protocol): + def description(self) -> dict[str, object]: + """Return the reader's source, units, and monitored volumes.""" + ... + + def read(self, *, processes: bool) -> tuple[dict[str, object], list[str]]: + """Return one sample of cumulative counters and any read errors. + + ``processes`` adds the costlier per-process counters. + """ + ... + + def close(self) -> None: ... + + +class WindowsCounterReader: + """Read volume, CPU, writeback, and Defender counters without privileges.""" + + def __init__(self, volumes: Mapping[str, Path]) -> None: + self._handles: dict[str, int] = {} + self._roles: dict[str, list[str]] = {} + self._errors: list[str] = [] + self._system_buffer = ctypes.create_string_buffer(_SYSTEM_PERFORMANCE_SIZE) + self._process_buffer = ctypes.create_string_buffer(MIB) + self._disk_performance: int | None = None + try: + self._disk_performance = _enable_disk_performance() + except OSError as error: + self._errors.append(f"disk counters could not be enabled: {error}") + keys: dict[str, str] = {} + for role, path in volumes.items(): + try: + mount_point = _windows_mount_point(path) + device = _windows_volume_device(mount_point) + if device not in keys: + handle = _open_device(device) + keys[device] = _volume_key(mount_point) + self._handles[keys[device]] = handle + except OSError as error: + self._errors.append(f"{role} volume counters are unavailable: {error}") + continue + self._roles.setdefault(keys[device], []).append(role) + + def description(self) -> dict[str, object]: + return { + "source": "windows", + "tick_ns": 100, + "page_size_bytes": mmap.PAGESIZE, + "volumes": {key: list(roles) for key, roles in self._roles.items()}, + } + + def read(self, *, processes: bool) -> tuple[dict[str, object], list[str]]: + errors = list(self._errors) + volumes: dict[str, object] = {} + for key, handle in self._handles.items(): + try: + volumes[key] = _volume_counters(handle) + except OSError as error: + errors.append(f"{key} volume counters failed: {error}") + sample: dict[str, object] = {"volumes": volumes} + readers = [("cpu", _cpu_counters), ("memory", self._memory_counters)] + if processes: + readers.append(("defender", self._defender_counters)) + for name, read in readers: + try: + sample[name] = read() + except (OSError, ValueError, struct.error) as error: + errors.append(f"{name} counters failed: {error}") + return sample, errors + + def close(self) -> None: + for handle in self._handles.values(): + _close_device(handle) + self._handles.clear() + if self._disk_performance is not None: + _advapi32().WmiCloseBlock(self._disk_performance) + self._disk_performance = None + + def _memory_counters(self) -> dict[str, int]: + try: + data = _query_system_information( + _SYSTEM_PERFORMANCE_INFORMATION, self._system_buffer + ) + except _BufferTooSmallError as error: + # Windows before 10 accepts only its shorter structure. + self._system_buffer = ctypes.create_string_buffer(error.required) + data = _query_system_information( + _SYSTEM_PERFORMANCE_INFORMATION, self._system_buffer + ) + return parse_system_performance(data) + + def _defender_counters(self) -> dict[str, int]: + while True: + try: + data = _query_system_information( + _SYSTEM_PROCESS_INFORMATION, self._process_buffer + ) + except _BufferTooSmallError as error: + if error.required > _MAX_PROCESS_INFORMATION_BYTES: + raise + # Leave room for processes that start before the next query. + size = error.required + MIB // 4 + self._process_buffer = ctypes.create_string_buffer(size) + continue + return parse_defender_processes( + data, ctypes.addressof(self._process_buffer) + ) + + +class StorageTelemetry: + """Sample host counters on a fixed cadence between start() and stop(). + + Sample times are nanoseconds after ``origin_ns`` on the + ``time.perf_counter_ns`` clock, the base of the snapshot profile's + ``observer_elapsed_ns``. Sampling never raises; failures are recorded. + """ + + def __init__( + self, + reader: CounterReader, + origin_ns: int, + *, + interval_seconds: float = STORAGE_TELEMETRY_INTERVAL_SECONDS, + process_interval_seconds: float = STORAGE_TELEMETRY_PROCESS_INTERVAL_SECONDS, + max_samples: int = STORAGE_TELEMETRY_MAX_SAMPLES, + clock: Callable[[], int] = time.perf_counter_ns, + ) -> None: + self._reader = reader + self._origin_ns = origin_ns + self._clock = clock + self._interval_seconds = interval_seconds + self._process_interval_ns = int(process_interval_seconds * 1e9) + self._max_samples = max_samples + self._samples: list[dict[str, object]] = [] + self._errors: list[str] = [] + self._truncated = False + self._last_process_sample_ns: int | None = None + self._stop_requested = threading.Event() + self._thread: threading.Thread | None = None + self._stopped = False + + def start(self) -> None: + if self._thread is not None or self._stopped: + return + self._sample(final=False) + self._thread = threading.Thread( + target=self._run, name="nvx-storage-telemetry", daemon=True + ) + self._thread.start() + + def stop(self) -> None: + if self._stopped: + return + self._stopped = True + if self._thread is not None: + self._stop_requested.set() + self._thread.join() + self._sample(final=True) + self._reader.close() + + def result(self, windows: Mapping[str, tuple[int, int]]) -> dict[str, object]: + """Stop sampling and summarize each telemetry phase in ``windows``. + + ``windows`` maps a profile phase to its start and end on the sample + clock. + """ + self.stop() + description = self._reader.description() + page_size = description.get("page_size_bytes") + phases: dict[str, object] = {} + for phase in STORAGE_TELEMETRY_PHASES: + if phase not in windows: + continue + summary = summarize_storage_window( + self._samples, + *windows[phase], + page_size=page_size if isinstance(page_size, int) else mmap.PAGESIZE, + ) + if summary is not None: + phases[phase] = summary + return { + "schema_version": STORAGE_TELEMETRY_SCHEMA_VERSION, + **description, + "interval_ms": self._interval_seconds * 1000, + "process_interval_ms": self._process_interval_ns / 1e6, + "samples": self._samples, + "truncated": self._truncated, + "errors": self._errors, + "phases": phases, + } + + def _run(self) -> None: + while not self._stop_requested.wait(self._interval_seconds): + if not self._sample(final=False): + return + + def _sample(self, *, final: bool) -> bool: + # The final sample always lands, so phases that end before the source + # exits stay bracketed after a long capture truncates sampling. + if not final and len(self._samples) >= self._max_samples: + self._truncated = True + return False + elapsed_ns = self._clock() - self._origin_ns + processes = ( + final + or self._last_process_sample_ns is None + or elapsed_ns - self._last_process_sample_ns >= self._process_interval_ns + ) + if processes: + self._last_process_sample_ns = elapsed_ns + try: + counters, errors = self._reader.read(processes=processes) + except Exception as error: + counters, errors = {}, [f"counter sampling failed: {error}"] + for error in errors: + if error not in self._errors: + self._errors.append(error) + self._samples.append({"elapsed_ns": elapsed_ns, **counters}) + return True + + +def create_storage_telemetry( + volumes: Mapping[str, Path], origin_ns: int +) -> StorageTelemetry | None: + """Return a sampler for ``volumes``, or None where it is unsupported. + + Only Windows captures flush a file-backed guest RAM image, so other hosts + record no storage telemetry. + """ + if sys.platform != "win32" or ctypes.sizeof(ctypes.c_void_p) != 8: + return None + return StorageTelemetry(WindowsCounterReader(volumes), origin_ns) + + +class _Endpoints(NamedTuple): + """The samples nearest to a phase's boundaries that carry one counter + group, and every sample of the group between them.""" + + start_ns: int + end_ns: int + before: dict[str, object] + after: dict[str, object] + covered: list[dict[str, object]] + + @property + def seconds(self) -> float: + return (self.end_ns - self.start_ns) / 1e9 + + +def summarize_storage_window( + samples: Sequence[Mapping[str, object]], + start_ns: int, + end_ns: int, + *, + page_size: int = mmap.PAGESIZE, +) -> dict[str, object] | None: + """Summarize the counters over the phase from ``start_ns`` to ``end_ns``. + + Each counter group uses the samples nearest to the phase's start and end + that carry it, so a failed read does not hide what neighboring samples + recorded. Rates and averages span those samples, and gauges report their + extreme sample between them. The summary reports the interval between the + samples nearest to the phase, and a group whose samples differ, such as + Defender's sparser process counters, reports its own. Returns None unless + two distinct samples bound the phase. + """ + timed = [ + (elapsed, _string_keys(sample)) + for sample in samples + if (elapsed := _counter(sample, "elapsed_ns")) is not None + ] + overall = _endpoints(timed, start_ns, end_ns, lambda _: True) + if overall is None: + return None + + def bounded( + summary: dict[str, object] | None, endpoints: _Endpoints + ) -> dict[str, object] | None: + if summary is not None and (endpoints.start_ns, endpoints.end_ns) != ( + overall.start_ns, + overall.end_ns, + ): + summary["sampled_start_ns"] = endpoints.start_ns + summary["sampled_end_ns"] = endpoints.end_ns + return summary + + def memory(endpoints: _Endpoints) -> dict[str, object] | None: + return _memory_window(endpoints, page_size) + + volumes: dict[str, object] = {} + keys = dict.fromkeys( + key for _, sample in timed for key in _group(sample, "volumes") + ) + for key in keys: + endpoints = _endpoints( + timed, + start_ns, + end_ns, + lambda sample, key=key: key in _group(sample, "volumes"), + ) + if endpoints is not None: + summary = bounded(_volume_window(key, endpoints), endpoints) + if summary is not None: + volumes[key] = summary + summarizers: tuple[ + tuple[str, Callable[[_Endpoints], dict[str, object] | None]], ... + ] = (("cpu", _cpu_window), ("memory", memory), ("defender", _defender_window)) + groups: dict[str, object] = {} + for name, summarize in summarizers: + endpoints = _endpoints( + timed, start_ns, end_ns, lambda sample, name=name: name in sample + ) + groups[name] = ( + None if endpoints is None else bounded(summarize(endpoints), endpoints) + ) + return { + "start_ns": start_ns, + "end_ns": end_ns, + "sampled_start_ns": overall.start_ns, + "sampled_end_ns": overall.end_ns, + "volumes": volumes, + **groups, + } + + +def _endpoints( + timed: Sequence[tuple[int, dict[str, object]]], + start_ns: int, + end_ns: int, + carries: Callable[[dict[str, object]], bool], +) -> _Endpoints | None: + candidates = [(elapsed, sample) for elapsed, sample in timed if carries(sample)] + bounds = _nearest_samples([elapsed for elapsed, _ in candidates], start_ns, end_ns) + if bounds is None: + return None + first, last = bounds + return _Endpoints( + candidates[first][0], + candidates[last][0], + candidates[first][1], + candidates[last][1], + [sample for _, sample in candidates[first : last + 1]], + ) + + +def _nearest_samples( + times: Sequence[int], start_ns: int, end_ns: int +) -> tuple[int, int] | None: + if not times: + return None + first = min(range(len(times)), key=lambda index: abs(times[index] - start_ns)) + last = min(range(len(times)), key=lambda index: abs(times[index] - end_ns)) + if times[last] <= times[first]: + return None + return first, last + + +def _group(sample: Mapping[str, object], name: str) -> dict[str, object]: + return _string_keys(sample.get(name)) + + +def _counter(group: Mapping[str, object], name: str) -> int | None: + value = group.get(name) + if isinstance(value, bool) or not isinstance(value, int): + return None + return value + + +def _increases( + before: Mapping[str, object], + after: Mapping[str, object], + names: Sequence[str], + *, + modulus: int | None = None, +) -> tuple[int, ...] | None: + """Return each counter's increase, or None if any is missing or fell.""" + increases: list[int] = [] + for name in names: + start, end = _counter(before, name), _counter(after, name) + if start is None or end is None: + return None + increase = end - start if modulus is None else (end - start) % modulus + if increase < 0: + return None + increases.append(increase) + return tuple(increases) + + +def _rounded(value: float) -> float: + return round(value, 3) + + +def _percent(value: float) -> float: + return _rounded(min(100.0, max(0.0, value))) + + +def _volume_window(key: str, endpoints: _Endpoints) -> dict[str, object] | None: + first = _string_keys(_group(endpoints.before, "volumes").get(key)) + last = _string_keys(_group(endpoints.after, "volumes").get(key)) + totals = _increases( + first, + last, + ( + "bytes_read", + "bytes_written", + "read_ticks", + "write_ticks", + "idle_ticks", + "query_ticks", + ), + ) + # DISK_PERFORMANCE counts requests in 32-bit fields that wrap. + counts = _increases(first, last, ("reads", "writes"), modulus=1 << 32) + if totals is None or counts is None or totals[5] <= 0: + return None + read_bytes, write_bytes, read_ticks, write_ticks, idle_ticks, query = totals + reads, writes = counts + seconds = endpoints.seconds + depths = [ + depth + for sample in endpoints.covered + if ( + depth := _counter( + _string_keys(_group(sample, "volumes").get(key)), "queue_depth" + ) + ) + is not None + ] + return { + "read_mib_per_second": _rounded(read_bytes / MIB / seconds), + "write_mib_per_second": _rounded(write_bytes / MIB / seconds), + "read_iops": _rounded(reads / seconds), + "write_iops": _rounded(writes / seconds), + "average_read_latency_ms": ( + _rounded(read_ticks / reads / 10_000) if reads else None + ), + "average_write_latency_ms": ( + _rounded(write_ticks / writes / 10_000) if writes else None + ), + "average_queue_length": _rounded((read_ticks + write_ticks) / query), + "busy_percent": _percent(100 * (1 - idle_ticks / query)), + "max_queue_depth": max(depths, default=None), + } + + +def _cpu_window(endpoints: _Endpoints) -> dict[str, object] | None: + times = _increases( + _group(endpoints.before, "cpu"), + _group(endpoints.after, "cpu"), + ("idle_ticks", "kernel_ticks", "user_ticks"), + ) + if times is None: + return None + idle, kernel, user = times + # Windows includes idle time in kernel time. + if kernel + user <= 0: + return None + return {"busy_percent": _percent(100 * (1 - idle / (kernel + user)))} + + +def _memory_window(endpoints: _Endpoints, page_size: int) -> dict[str, object] | None: + start, end = _group(endpoints.before, "memory"), _group(endpoints.after, "memory") + seconds = endpoints.seconds + + def mib(pages: int) -> float: + return _rounded(pages * page_size / MIB) + + def rate(name: str) -> float | None: + pages = _increases(start, end, (name,), modulus=1 << 32) + return None if pages is None else _rounded(mib(pages[0]) / seconds) + + def extreme(name: str, largest: bool) -> float | None: + values = [ + value + for sample in endpoints.covered + if (value := _counter(_group(sample, "memory"), name)) is not None + ] + if not values: + return None + return mib(max(values) if largest else min(values)) + + threshold = _counter(end, "cache_dirty_page_threshold") + return { + "available_mib_min": extreme("available_pages", largest=False), + "cache_dirty_mib_max": extreme("cache_dirty_pages", largest=True), + "cache_dirty_threshold_mib": None if threshold is None else mib(threshold), + "lazy_write_mib_per_second": rate("lazy_write_pages"), + "cache_flush_mib_per_second": rate("cache_flush_pages"), + "modified_write_mib_per_second": rate("modified_pages_written"), + "mapped_write_mib_per_second": rate("mapped_pages_written"), + } + + +def _defender_window(endpoints: _Endpoints) -> dict[str, object] | None: + start = _group(endpoints.before, "defender") + end = _group(endpoints.after, "defender") + processes = _counter(end, "processes") + # A Defender process that started or exited would skew the summed totals. + if processes is None or processes != _counter(start, "processes"): + return None + totals = _increases( + start, end, ("cpu_ticks", "read_bytes", "write_bytes", "other_bytes") + ) + if totals is None: + return None + cpu_ticks, read_bytes, write_bytes, other_bytes = totals + seconds = endpoints.seconds + return { + "processes": processes, + "cpu_percent": _rounded(100 * cpu_ticks / TICKS_PER_SECOND / seconds), + "read_mib_per_second": _rounded(read_bytes / MIB / seconds), + "write_mib_per_second": _rounded(write_bytes / MIB / seconds), + "other_mib_per_second": _rounded(other_bytes / MIB / seconds), + } + + +def describe_storage_window( + window: Mapping[str, object], volume_roles: Mapping[str, object] +) -> str: + """Return a one-line description of a summarized telemetry window.""" + + def number(value: object, unit: str, digits: int = 1) -> str: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return "n/a" + return f"{value:.{digits}f}{unit}" + + def roles(key: str) -> list[str]: + value = volume_roles.get(key) + if not isinstance(value, list): + return [] + return [str(role) for role in cast(list[object], value)] + + parts: list[str] = [] + volumes = _group(window, "volumes") + for key in sorted(volumes, key=lambda key: "scratch" not in roles(key)): + stats = _string_keys(volumes[key]) + label = "/".join(roles(key)) or "volume" + parts.append( + f"{label} {key} write " + f"{number(stats.get('write_mib_per_second'), ' MiB/s')} at " + f"{number(stats.get('average_write_latency_ms'), ' ms')}, queue " + f"{number(stats.get('average_queue_length'), '')}, " + f"{number(stats.get('busy_percent'), '%', 0)} busy" + ) + parts.append( + f"CPU {number(_group(window, 'cpu').get('busy_percent'), '%', 0)} busy" + ) + memory = _group(window, "memory") + if memory: + parts.append( + f"cache dirty {number(memory.get('cache_dirty_mib_max'), ' MiB')}, " + f"lazy writes {number(memory.get('lazy_write_mib_per_second'), ' MiB/s')}" + ) + defender = _group(window, "defender") + if defender: + parts.append(f"Defender {number(defender.get('cpu_percent'), '%')} CPU") + return "; ".join(parts) diff --git a/scripts/nvx_tools/performance.py b/scripts/nvx_tools/performance.py index be392355e..b781e3340 100644 --- a/scripts/nvx_tools/performance.py +++ b/scripts/nvx_tools/performance.py @@ -13,11 +13,12 @@ import tempfile from collections import defaultdict, deque from collections.abc import Callable, Iterable, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from typing import cast -from .common import bytes_to_mib, positive_int +from . import host_telemetry +from .common import bytes_to_mib, positive_int, sha256_file LEGACY_CSV_FIELDS = ["commit", "metric", "unit", "direction", "p50"] CSV_FIELDS = [ @@ -77,6 +78,10 @@ LIFECYCLE_SNAPSHOT_MAX_P50_OVER_P25 = 1.25 LIFECYCLE_SNAPSHOT_MAX_CLUSTER_GAP = 1.25 UNSTABLE_LIFECYCLE_EXIT_CODE = 75 +LIFECYCLE_ATTEMPT_LOG_SCHEMA_VERSION = 1 +LIFECYCLE_ATTEMPT_STATUSES = {0: "valid", UNSTABLE_LIFECYCLE_EXIT_CODE: "unstable"} +"""Validation statuses by exit status; any other status is ``invalid``.""" +SHA256_HEX = re.compile(r"[0-9a-f]{64}") NUMBER = r"[0-9]+(?:,[0-9]{3})*(?:\.[0-9]+)?" ANSI_ESCAPE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") SHELL_SNAPSHOT_MEMORIES_MIB = (128, 256, 512) @@ -138,6 +143,20 @@ class BenchmarkDimensions: processors: int +@dataclass(frozen=True) +class LifecycleAttempt: + """One bounded lifecycle measurement and its validation outcome.""" + + number: int + file: str + sha256: str | None + status: str + exit_status: int + message: str | None + published: bool = False + document: dict[str, object] | None = None + + MetricValue = tuple[str, str, float] Parser = Callable[[str], dict[str, MetricValue]] @@ -1177,12 +1196,170 @@ def read_lifecycle_data(platform: str, input_path: Path) -> LifecycleData: ) +def record_lifecycle_attempt( + log_path: Path, + platform: str, + input_path: Path, + exit_status: int, + message: str | None, +) -> LifecycleAttempt: + """Append a lifecycle validation outcome to the attempt log. + + The log keeps every bounded remeasurement in order, so collection can + name the attempt that supplied the published lifecycle metrics. + """ + attempts = ( + read_lifecycle_attempt_log(log_path, platform) if log_path.exists() else [] + ) + try: + location = Path( + os.path.relpath(input_path.resolve(), log_path.resolve().parent) + ).as_posix() + except ValueError: + location = str(input_path.resolve()) + attempt = LifecycleAttempt( + number=len(attempts) + 1, + file=location, + sha256=sha256_file(input_path) if input_path.is_file() else None, + status=LIFECYCLE_ATTEMPT_STATUSES.get(exit_status, "invalid"), + exit_status=exit_status, + message=message, + ) + attempts.append(attempt) + document = { + "schema_version": LIFECYCLE_ATTEMPT_LOG_SCHEMA_VERSION, + "platform": platform, + "attempts": [ + { + "attempt": item.number, + "file": item.file, + "sha256": item.sha256, + "status": item.status, + "exit_status": item.exit_status, + "message": item.message, + } + for item in attempts + ], + } + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8") + return attempt + + +def read_lifecycle_attempt_log(log_path: Path, platform: str) -> list[LifecycleAttempt]: + try: + document = _json_object( + json.loads(log_path.read_text(encoding="utf-8")), str(log_path) + ) + except FileNotFoundError as error: + raise PerformanceError( + f"lifecycle attempt log not found: {log_path}" + ) from error + except (UnicodeError, json.JSONDecodeError) as error: + raise PerformanceError( + f"invalid lifecycle attempt log {log_path}: {error}" + ) from error + if document.get("schema_version") != LIFECYCLE_ATTEMPT_LOG_SCHEMA_VERSION: + raise PerformanceError( + f"{log_path} has unsupported schema_version " + f"{document.get('schema_version')!r}" + ) + if document.get("platform") != platform: + raise PerformanceError( + f"{log_path} records platform {document.get('platform')!r}, " + f"expected {platform!r}" + ) + entries = document.get("attempts") + if not isinstance(entries, list): + raise PerformanceError(f"expected an attempt list at {log_path}:attempts") + attempts: list[LifecycleAttempt] = [] + for index, value in enumerate(cast(list[object], entries)): + location = f"{log_path}:attempts[{index}]" + entry = _json_object(value, location) + number = entry.get("attempt") + file = entry.get("file") + digest = entry.get("sha256") + status = entry.get("status") + exit_status = entry.get("exit_status") + message = entry.get("message") + if ( + isinstance(number, bool) + or number != index + 1 + or not isinstance(file, str) + or not file + or isinstance(exit_status, bool) + or not isinstance(exit_status, int) + or status != LIFECYCLE_ATTEMPT_STATUSES.get(exit_status, "invalid") + or not (digest is None or isinstance(digest, str)) + or (isinstance(digest, str) and not SHA256_HEX.fullmatch(digest)) + or not (message is None or isinstance(message, str)) + ): + raise PerformanceError(f"invalid lifecycle attempt at {location}") + attempts.append( + LifecycleAttempt( + number=index + 1, + file=file, + sha256=digest, + status=str(status), + exit_status=exit_status, + message=message, + ) + ) + return attempts + + +def read_lifecycle_attempts( + log_path: Path, platform: str, published_path: Path +) -> list[LifecycleAttempt]: + """Read every logged attempt and mark the one published as + ``published_path``, which must be a byte-identical copy of exactly one + valid attempt.""" + attempts = read_lifecycle_attempt_log(log_path, platform) + try: + digest = sha256_file(published_path) + except OSError as error: + raise PerformanceError( + f"cannot read lifecycle result {published_path}: {error}" + ) from error + published = [ + attempt + for attempt in attempts + if attempt.status == "valid" and attempt.sha256 == digest + ] + if len(published) != 1: + raise PerformanceError( + f"{published_path} matches {len(published)} valid attempts in " + f"{log_path}; expected exactly one" + ) + return [ + replace( + attempt, + published=attempt is published[0], + document=_read_attempt_document(log_path.parent / attempt.file), + ) + for attempt in attempts + ] + + +def _read_attempt_document(path: Path) -> dict[str, object] | None: + # A rejected attempt's diagnostics must not block publication. + try: + document = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + return cast(dict[str, object], document) if isinstance(document, dict) else None + + def append_openvmm_diagnostics( path: Path, platform: str, lifecycle: LifecycleData, source: Path, + attempts: Sequence[LifecycleAttempt] = (), ) -> None: + """Append lifecycle diagnostics, the benchmark host, and storage + telemetry. ``attempts`` adds every logged lifecycle attempt, including + rejected ones, to the attempt and telemetry tables.""" document = lifecycle.document backend = lifecycle.backend cold_start = _openvmm_value(document, "backends", backend, "p50_ms", source) @@ -1291,10 +1468,203 @@ def append_openvmm_diagnostics( "", ] ) + lines.extend(_lifecycle_attempt_lines(attempts, backend)) + host = _optional_object(document.get("host")) + if host: + lines.extend(_host_provenance_lines(host)) + sources = ( + [ + (attempt.document, _attempt_label(attempt)) + for attempt in attempts + if attempt.document is not None + ] + if attempts + else [(document, None)] + ) + lines.extend(_storage_telemetry_lines(sources, backend)) with path.open("a", encoding="utf-8", newline="\n") as output: output.write("\n" + "\n".join(lines)) +def _optional_object(value: object) -> dict[str, object]: + return cast(dict[str, object], value) if isinstance(value, dict) else {} + + +def _markdown_cell(value: str) -> str: + return value.replace("|", "\\|").replace("\n", " ") + + +def _attempt_label(attempt: LifecycleAttempt) -> str: + published = ", published" if attempt.published else "" + return f"Attempt {attempt.number}: {attempt.status}{published} (`{attempt.file}`)" + + +def _lifecycle_attempt_lines( + attempts: Sequence[LifecycleAttempt], backend: str +) -> list[str]: + if not attempts: + return [] + lines = [ + "### Lifecycle attempts", + "", + "| Attempt | File | Result | Snapshot generation p50 | Published |", + "| ---: | --- | --- | ---: | --- |", + ] + for attempt in attempts: + capture = _optional_object( + _optional_object((attempt.document or {}).get("snapshot_capture")).get( + backend + ) + ) + result = attempt.status + if attempt.message is not None: + result = f"{result}: {attempt.message}" + lines.append( + f"| {attempt.number} | `{attempt.file}` | {_markdown_cell(result)} | " + f"{_summary_number(capture.get('p50_ms'), ' ms', 2)} | " + f"{'yes' if attempt.published else 'no'} |" + ) + return [*lines, ""] + + +def _host_provenance_lines(host: dict[str, object]) -> list[str]: + lines = ["### Benchmark host", "", "| Property | Value |", "| --- | --- |"] + lines.extend( + f"| {label} | {_markdown_cell(value)} |" + for label, value in host_telemetry.host_provenance_rows(host) + ) + return [*lines, ""] + + +def _summary_number(value: object, unit: str, digits: int) -> str: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return "n/a" + return f"{value:.{digits}f}{unit}" + + +def _storage_telemetry_lines( + sources: Sequence[tuple[dict[str, object], str | None]], backend: str +) -> list[str]: + sections: list[str] = [] + for document, label in sources: + table = _storage_telemetry_table(document, backend) + if not table: + continue + if label is not None: + sections.extend([f"#### {label}", ""]) + sections.extend([*table, ""]) + if not sections: + return [] + return [ + "### Snapshot generation storage telemetry", + "", + ( + "Each row describes one measured capture. Rates and averages span " + "the telemetry samples nearest to the start and end of its " + "`capture.mapped_memory_flush` phase. Other volumes are the " + "monitored workspace and system volumes that do not hold scratch." + ), + "", + *sections, + ] + + +def _storage_telemetry_table(document: dict[str, object], backend: str) -> list[str]: + capture = _optional_object( + _optional_object(document.get("snapshot_capture")).get(backend) + ) + raw_samples = _optional_object(capture.get("profile")).get("raw_samples") + if not isinstance(raw_samples, list): + return [] + rows: list[str] = [] + scratch_volumes: set[str] = set() + for number, sample in enumerate( + map(_optional_object, cast(list[object], raw_samples)), 1 + ): + telemetry = _optional_object(sample.get("storage_telemetry")) + if not telemetry: + continue + window = _optional_object( + _optional_object(telemetry.get("phases")).get("capture.mapped_memory_flush") + ) + roles = _optional_object(telemetry.get("volumes")) + volumes = _optional_object(window.get("volumes")) + scratch_keys = { + key + for key, value in roles.items() + if isinstance(value, list) and "scratch" in cast(list[object], value) + } + scratch_volumes.update(scratch_keys) + scratch = _optional_object( + next((volumes[key] for key in scratch_keys if key in volumes), None) + ) + other_writes = [ + value + for key, stats in volumes.items() + if key not in scratch_keys + and isinstance( + value := _optional_object(stats).get("write_mib_per_second"), + (int, float), + ) + ] + memory = _optional_object(window.get("memory")) + defender = _optional_object(window.get("defender")) + generation = sample.get("generation_duration_ns") + records = sample.get("records") + flush = next( + ( + record.get("duration_ns") + for record in map( + _optional_object, + cast(list[object], records) if isinstance(records, list) else [], + ) + if record.get("operation") == "capture" + and record.get("phase") == "mapped_memory_flush" + ), + None, + ) + cells = [ + str(number), + _summary_number( + generation / 1_000_000 if isinstance(generation, int) else None, + " ms", + 2, + ), + _summary_number( + flush / 1_000_000 if isinstance(flush, int) else None, " ms", 2 + ), + _summary_number(scratch.get("write_mib_per_second"), " MiB/s", 1), + _summary_number(scratch.get("average_write_latency_ms"), " ms", 2), + _summary_number(scratch.get("average_queue_length"), "", 1), + _summary_number(scratch.get("busy_percent"), "%", 0), + _summary_number(sum(other_writes) if other_writes else None, " MiB/s", 1), + _summary_number( + _optional_object(window.get("cpu")).get("busy_percent"), "%", 0 + ), + _summary_number(memory.get("cache_dirty_mib_max"), " MiB", 1), + _summary_number(memory.get("lazy_write_mib_per_second"), " MiB/s", 1), + _summary_number(defender.get("cpu_percent"), "%", 1), + ] + rows.append("| " + " | ".join(cells) + " |") + if not rows: + return [] + scratch_names = ", ".join(f"`{name}`" for name in sorted(scratch_volumes)) + return [ + f"Scratch volume: {scratch_names or 'not monitored'}.", + "", + ( + "| Sample | Generation | Flush | Scratch write | Write latency | " + "Queue | Busy | Other write | CPU busy | Dirty cache | Lazy writes | " + "Defender CPU |" + ), + ( + "| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: " + "| ---: | ---: |" + ), + *rows, + ] + + def collect_openvmm_results( platform: str, commit: str, @@ -1343,11 +1713,14 @@ def collect_results( summary_path: Path | None = None, lifecycle_input: Path | None = None, require_shell_snapshot_restore_512: bool = False, + lifecycle_attempt_log: Path | None = None, ) -> Path: if not platform or "/" in platform or platform in {".", ".."}: raise PerformanceError(f"invalid platform name: {platform!r}") if not commit: raise PerformanceError("commit must not be empty") + if lifecycle_attempt_log is not None and lifecycle_input is None: + raise PerformanceError("a lifecycle attempt log requires a lifecycle input") if require_shell_snapshot_restore_512 and ( require_network or require_shell_snapshot @@ -1425,6 +1798,17 @@ def collect_results( if lifecycle_input is not None else None ) + attempts: list[LifecycleAttempt] = [] + if lifecycle_attempt_log is not None: + assert lifecycle_input is not None + attempts = read_lifecycle_attempts( + lifecycle_attempt_log, platform, lifecycle_input + ) + source = next(attempt for attempt in attempts if attempt.published) + print( + f"Lifecycle metrics come from attempt {source.number} of " + f"{len(attempts)}: {source.file}" + ) if lifecycle is not None: lifecycle_dimensions = ( lifecycle.microvm_abi_version, @@ -1608,6 +1992,7 @@ def collect_results( platform, lifecycle, lifecycle_input, + attempts, ) print(f"Collected {len(results)} p50 metric(s) for {platform}: {output_path}") return output_path @@ -1964,6 +2349,14 @@ def configure_parser(parser: argparse.ArgumentParser) -> None: type=Path, help="merge a 128 MiB e2e lifecycle benchmark JSON result", ) + collect.add_argument( + "--lifecycle-attempt-log", + type=Path, + help=( + "lifecycle attempt log that must name --lifecycle-input as exactly " + "one valid attempt; the summary reports every attempt" + ), + ) collect.add_argument("--summary", type=Path) validate_openvmm = commands.add_parser( @@ -1971,6 +2364,11 @@ def configure_parser(parser: argparse.ArgumentParser) -> None: ) validate_openvmm.add_argument("--platform", required=True) validate_openvmm.add_argument("--input", type=Path, required=True) + validate_openvmm.add_argument( + "--attempt-log", + type=Path, + help="append the validation outcome to this lifecycle attempt log", + ) collect_openvmm = commands.add_parser( "collect-openvmm", help="convert an OpenVMM benchmark JSON result to p50 CSV" @@ -2042,15 +2440,28 @@ def command_performance(args: argparse.Namespace) -> int: require_shell_snapshot_restore_512=( args.require_shell_snapshot_restore_512 ), + lifecycle_attempt_log=args.lifecycle_attempt_log, ) return 0 if args.performance_command == "validate-openvmm": + status, message = 0, None try: read_lifecycle_data(args.platform, args.input) except UnstablePerformanceError as error: - print(f"ERROR: {error}", file=sys.stderr) - return UNSTABLE_LIFECYCLE_EXIT_CODE - return 0 + status, message = UNSTABLE_LIFECYCLE_EXIT_CODE, str(error) + except PerformanceError as error: + status, message = 2, str(error) + if message is not None: + print(f"ERROR: {message}", file=sys.stderr) + if args.attempt_log is not None: + attempt = record_lifecycle_attempt( + args.attempt_log, args.platform, args.input, status, message + ) + print( + f"Recorded lifecycle attempt {attempt.number} " + f"({attempt.status}) in {args.attempt_log}" + ) + return status if args.performance_command == "collect-openvmm": collect_openvmm_results( args.platform, diff --git a/scripts/setup/README.md b/scripts/setup/README.md index d81a141ef..ddd7f15fb 100644 --- a/scripts/setup/README.md +++ b/scripts/setup/README.md @@ -74,7 +74,10 @@ publishes it as the machine-level `NVX_BENCHMARK_SCRATCH` variable. Network Service receives Modify access to that directory tree, as for `_work/_sccache`. CI places benchmark snapshots and guest RAM backing files there so their flushes avoid the burst-limited system disk. Check mode requires the directory -when a data volume exists. +when a data volume exists. Changing the storage behind benchmark scratch or the +runner workspace is a performance-platform change: apply it to every Windows +runner at once and reset the affected history, as +[CI collection](../../doc/benchmarks.md#ci-collection) describes. Persistent runners do not have Docker access. Guest artifacts are built with Docker on a GitHub-hosted runner instead. Linux provisioning runs through the SSH administrator, but the listener and diff --git a/scripts/test_host_telemetry.py b/scripts/test_host_telemetry.py new file mode 100644 index 000000000..79e9061c7 --- /dev/null +++ b/scripts/test_host_telemetry.py @@ -0,0 +1,929 @@ +#!/usr/bin/env python3 +# pyright: reportPrivateUsage=false + +"""Tests for benchmark host provenance and storage telemetry.""" + +import http.client +import json +import os +import struct +import sys +import tempfile +import time +import unittest +import urllib.request +from pathlib import Path +from typing import cast +from unittest.mock import MagicMock, patch + +sys.path.insert(0, str(Path(__file__).parent)) +from nvx_tools import host_telemetry # noqa: E402 + +MIB = 1024 * 1024 +GIB = 1024 * MIB +TICKS_PER_SECOND = 10_000_000 +SUBSCRIPTION = "00000000-1111-2222-3333-444444444444" +RESOURCE_GROUP = f"/subscriptions/{SUBSCRIPTION}/resourceGroups/nvx-runners-rg" +IMDS_COMPUTE: dict[str, object] = { + "azEnvironment": "AzurePublicCloud", + "location": "westus2", + "name": "runner-vm-name", + "osProfile": { + "adminUsername": "runneradmin", + "computerName": "VMSSVAOAN000002", + }, + "publicKeys": [{"keyData": "ssh-rsa AAAA", "path": "/home/runneradmin"}], + "resourceGroupName": "nvx-runners-rg", + "resourceId": f"{RESOURCE_GROUP}/providers/Microsoft.Compute/virtualMachines/x", + "subscriptionId": SUBSCRIPTION, + "tags": "owner:someone", + "tagsList": [{"name": "owner", "value": "someone"}], + "userData": "c2VjcmV0", + "vmId": "55555555-6666-7777-8888-999999999999", + "vmScaleSetName": "nvx-windows", + "vmSize": "Standard_D16ds_v5", + "storageProfile": { + "dataDisks": [ + { + "bytesPerSecondThrottle": "", + "caching": "None", + "createOption": "Empty", + "diskCapacityBytes": "", + "diskSizeGB": "512", + "isSharedDisk": "false", + "isUltraDisk": "false", + "lun": "0", + "managedDisk": { + "id": f"{RESOURCE_GROUP}/providers/Microsoft.Compute/disks/data", + "storageAccountType": "Premium_LRS", + }, + "name": "data-disk-name", + "opsPerSecondThrottle": "", + "vhd": {"uri": ""}, + "writeAcceleratorEnabled": "false", + }, + { + "bytesPerSecondThrottle": "979202048", + "caching": "None", + "diskSizeGB": "1024", + "isUltraDisk": "true", + "lun": "1", + "managedDisk": { + "id": f"{RESOURCE_GROUP}/providers/Microsoft.Compute/disks/ultra", + "storageAccountType": "UltraSSD_LRS", + }, + "name": "ultra-disk-name", + "opsPerSecondThrottle": "65280", + "writeAcceleratorEnabled": "false", + }, + ], + "imageReference": {"offer": "WindowsServer", "sku": "2022-datacenter"}, + "osDisk": { + "caching": "ReadWrite", + "createOption": "FromImage", + "diffDiskSettings": {"option": ""}, + "diskSizeGB": "128", + "encryptionSettings": {"enabled": "false"}, + "managedDisk": { + "id": f"{RESOURCE_GROUP}/providers/Microsoft.Compute/disks/os", + "storageAccountType": "Premium_LRS", + }, + "name": "os-disk-name", + "osType": "Windows", + "writeAcceleratorEnabled": "false", + }, + "resourceDisk": {"size": "614400"}, + }, +} + + +class AzureProvenanceTests(unittest.TestCase): + def setUp(self) -> None: + host_telemetry._azure_vm_provenance.cache_clear() + self.addCleanup(host_telemetry._azure_vm_provenance.cache_clear) + + def test_keeps_only_vm_size_and_disk_performance_settings(self): + provenance = host_telemetry.azure_compute_provenance(IMDS_COMPUTE) + + self.assertEqual( + provenance, + { + "vm_size": "Standard_D16ds_v5", + "os_disk": { + "storage_account_type": "Premium_LRS", + "size_gib": 128, + "caching": "ReadWrite", + "write_accelerator_enabled": False, + "ephemeral_option": None, + }, + "data_disks": [ + { + "lun": 0, + "storage_account_type": "Premium_LRS", + "size_gib": 512, + "caching": "None", + "write_accelerator_enabled": False, + "ephemeral_option": None, + "bytes_per_second_throttle": None, + "ops_per_second_throttle": None, + }, + { + "lun": 1, + "storage_account_type": "UltraSSD_LRS", + "size_gib": 1024, + "caching": "None", + "write_accelerator_enabled": False, + "ephemeral_option": None, + "bytes_per_second_throttle": 979202048, + "ops_per_second_throttle": 65280, + }, + ], + "resource_disk_size_kib": 614400, + }, + ) + serialized = json.dumps(provenance) + for sensitive in ( + SUBSCRIPTION, + "nvx-runners-rg", + "disk-name", + "runneradmin", + "ssh-rsa", + "someone", + "55555555", + "westus2", + "nvx-windows", + "c2VjcmV0", + "VMSSVAOAN000002", + "WindowsServer", + ): + self.assertNotIn(sensitive, serialized) + + def test_rejects_metadata_without_a_vm_size(self): + cases: tuple[object, ...] = ([], {}, {"vmSize": ""}) + for compute in cases: + with ( + self.subTest(compute=compute), + self.assertRaisesRegex(ValueError, "no vmSize"), + ): + host_telemetry.azure_compute_provenance(compute) + + def test_queries_instance_metadata_once_and_only_on_a_possible_azure_vm(self): + with ( + patch.object(host_telemetry, "may_be_azure_vm", return_value=False), + patch.object(host_telemetry, "read_azure_compute_metadata") as read, + ): + self.assertIsNone(host_telemetry.azure_vm_provenance()) + read.assert_not_called() + + host_telemetry._azure_vm_provenance.cache_clear() + with ( + patch.object(host_telemetry, "may_be_azure_vm", return_value=True), + patch.object( + host_telemetry, "read_azure_compute_metadata", return_value=IMDS_COMPUTE + ) as read, + ): + first = host_telemetry.azure_vm_provenance() + assert first is not None + first["vm_size"] = "mutated by a caller" + second = host_telemetry.azure_vm_provenance() + read.assert_called_once_with() + assert second is not None + self.assertEqual(second["vm_size"], "Standard_D16ds_v5") + + host_telemetry._azure_vm_provenance.cache_clear() + with ( + patch.object(host_telemetry, "may_be_azure_vm", return_value=True), + patch.object( + host_telemetry, + "read_azure_compute_metadata", + side_effect=TimeoutError("timed out"), + ), + ): + self.assertEqual( + host_telemetry.azure_vm_provenance(), + {"error": "Azure instance metadata is unavailable: timed out"}, + ) + + # Another service on the link-local address must not fail the run. + host_telemetry._azure_vm_provenance.cache_clear() + with ( + patch.object(host_telemetry, "may_be_azure_vm", return_value=True), + patch.object( + host_telemetry, + "read_azure_compute_metadata", + side_effect=http.client.BadStatusLine("SSH-2.0"), + ), + ): + provenance = host_telemetry.azure_vm_provenance() + assert provenance is not None + self.assertIn("SSH-2.0", str(provenance["error"])) + + def test_metadata_request_bypasses_proxies(self): + class Response: + def __enter__(self) -> "Response": + return self + + def __exit__(self, *_: object) -> None: + return None + + def read(self, _size: int) -> bytes: + return b'{"vmSize": "Standard_D4s_v5"}' + + opener = MagicMock() + opener.open.return_value = Response() + with patch.object( + host_telemetry.urllib.request, "build_opener", return_value=opener + ) as build: + document = host_telemetry.read_azure_compute_metadata(timeout=1.5) + + handler = build.call_args.args[0] + self.assertIsInstance(handler, urllib.request.ProxyHandler) + self.assertEqual(vars(handler)["proxies"], {}) + request = cast(urllib.request.Request, opener.open.call_args.args[0]) + self.assertEqual(request.full_url, host_telemetry.AZURE_IMDS_URL) + self.assertIn("api-version=2021-05-01", request.full_url) + self.assertEqual(request.get_header("Metadata"), "true") + self.assertEqual(opener.open.call_args.kwargs, {"timeout": 1.5}) + self.assertEqual(document, {"vmSize": "Standard_D4s_v5"}) + + +class HostProvenanceTests(unittest.TestCase): + def test_records_runner_cpu_memory_and_volume_roles(self): + with ( + tempfile.TemporaryDirectory() as temporary, + patch.dict(os.environ, {"RUNNER_NAME": "azure-windows-3"}), + patch.object( + host_telemetry, + "azure_vm_provenance", + return_value={"vm_size": "Standard_D16ds_v5"}, + ), + ): + scratch = Path(temporary).resolve() + provenance = host_telemetry.host_provenance( + {"workspace": Path(__file__).resolve().parent, "scratch": scratch} + ) + + json.dumps(provenance) + self.assertEqual(provenance["schema_version"], 1) + self.assertEqual(provenance["runner_name"], "azure-windows-3") + self.assertEqual(provenance["logical_processors"], os.cpu_count()) + self.assertEqual(provenance["azure"], {"vm_size": "Standard_D16ds_v5"}) + memory = provenance["memory_bytes"] + self.assertIsInstance(memory, int) + self.assertGreater(cast(int, memory), 0) + volumes = cast(dict[str, dict[str, object]], provenance["volumes"]) + self.assertEqual(set(volumes), {"workspace", "scratch"}) + self.assertEqual(volumes["scratch"]["path"], str(scratch)) + self.assertNotIn("error", volumes["scratch"]) + self.assertIn("volume", volumes["scratch"]) + self.assertGreater(cast(int, volumes["scratch"]["size_bytes"]), 0) + + def test_volume_errors_are_recorded_instead_of_raised(self): + reader = ( + "_windows_volume_provenance" + if sys.platform == "win32" + else "_linux_volume_provenance" + ) + with patch.object( + host_telemetry, reader, side_effect=PermissionError("access denied") + ): + self.assertEqual( + host_telemetry.volume_provenance(Path("scratch")), + {"path": "scratch", "error": "access denied"}, + ) + + def test_linux_mount_selects_the_deepest_mount_of_the_device(self): + mountinfo = "\n".join( + ( + "22 1 8:1 / / rw,relatime shared:1 - ext4 /dev/sda1 rw", + "30 22 8:17 / /mnt/data rw,relatime shared:2 - xfs /dev/sdb1 rw", + "31 30 8:17 /nested /mnt/data/my\\040scratch rw shared:3 - xfs " + "/dev/sdb1 rw", + "40 22 0:45 / /tmp rw,nosuid - tmpfs tmpfs rw", + ) + ) + + self.assertEqual( + host_telemetry.linux_mount("/mnt/data/my scratch/job", "8:17", mountinfo), + ("/mnt/data/my scratch", "xfs"), + ) + self.assertEqual( + host_telemetry.linux_mount("/mnt/data/other", "8:17", mountinfo), + ("/mnt/data", "xfs"), + ) + self.assertEqual( + host_telemetry.linux_mount("/home/runner", "8:1", mountinfo), + ("/", "ext4"), + ) + self.assertEqual( + host_telemetry.linux_mount("/tmp/x", "0:45", mountinfo), ("/tmp", "tmpfs") + ) + self.assertIsNone( + host_telemetry.linux_mount("/mnt/database", "8:17", mountinfo) + ) + + def test_volume_keys_ignore_drive_letter_case_and_trailing_separators(self): + for mount_point, key in ( + ("c:\\", "C:"), + ("F:\\", "F:"), + ("C:\\mnt\\data\\", "C:\\mnt\\data"), + ("/", "/"), + ("/mnt/data/", "/mnt/data"), + ): + with self.subTest(mount_point=mount_point): + self.assertEqual(host_telemetry._volume_key(mount_point), key) + + def test_provenance_rows_describe_runner_azure_disks_and_volumes(self): + host: dict[str, object] = { + "schema_version": 1, + "runner_name": "azure-windows-3", + "machine_name": "VMSSVAOAN000002", + "os": "Windows-2022Server-10.0.20348-SP0", + "cpu_model": "Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz", + "logical_processors": 16, + "memory_bytes": 64 * GIB, + "azure": host_telemetry.azure_compute_provenance(IMDS_COMPUTE), + "volumes": { + "workspace": { + "path": r"C:\actions-runner\_work\nvx\nvx", + "volume": "C:", + "filesystem": "NTFS", + "size_bytes": 127 * GIB, + "system_volume": True, + "disk_number": 0, + "scsi_address": "0:0:0:0", + }, + "temporary": {"path": r"C:\Windows\TEMP", "error": "access denied"}, + "scratch": { + "path": r"F:\nvx-benchmark-scratch\job", + "volume": "F:", + "filesystem": "NTFS", + "size_bytes": 512 * GIB, + "system_volume": False, + "disk_number": 2, + "scsi_address": "1:0:0:0", + }, + }, + } + + rows = dict(host_telemetry.host_provenance_rows(host)) + + self.assertEqual( + rows, + { + "Runner": "azure-windows-3", + "Machine": "VMSSVAOAN000002", + "Operating system": "Windows-2022Server-10.0.20348-SP0", + "CPU": ( + "Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz, " + "16 logical processors" + ), + "Memory": "64.0 GiB", + "Azure VM size": "Standard_D16ds_v5", + "Azure OS disk": "Premium_LRS, 128 GiB, ReadWrite caching", + "Azure data disk LUN 0": "Premium_LRS, 512 GiB, None caching", + "Azure data disk LUN 1": ( + "UltraSSD_LRS, 1024 GiB, None caching, 934 MiB/s limit, " + "65280 IOPS limit" + ), + "Azure temporary disk": "0.6 GiB", + "Workspace volume": ( + "C: NTFS, 127.0 GiB, system volume, disk 0, SCSI 0:0:0:0, " + r"at C:\actions-runner\_work\nvx\nvx" + ), + "Temporary-directory volume": "unavailable (access denied)", + "Benchmark scratch volume": ( + "F: NTFS, 512.0 GiB, disk 2, SCSI 1:0:0:0, " + r"at F:\nvx-benchmark-scratch\job" + ), + }, + ) + self.assertTrue( + host_telemetry.describe_host_provenance(host).startswith( + "Runner: azure-windows-3; Machine: VMSSVAOAN000002; " + ) + ) + + def test_provenance_rows_report_missing_runner_and_metadata_errors(self): + rows = dict( + host_telemetry.host_provenance_rows( + {"azure": {"error": "Azure instance metadata is unavailable: x"}} + ) + ) + + self.assertEqual(rows["Runner"], "not recorded") + self.assertEqual(rows["CPU"], "unknown CPU") + self.assertEqual( + rows["Azure metadata"], "Azure instance metadata is unavailable: x" + ) + + +class CounterParserTests(unittest.TestCase): + def test_parses_writeback_counters_at_kernel_offsets(self): + data = bytearray(344) + expected = { + "available_pages": (44, 4_000_000), + "modified_pages_written": (96, 11), + "modified_page_writes": (100, 12), + "mapped_pages_written": (104, 13), + "mapped_page_writes": (108, 14), + "lazy_write_ios": (280, 15), + "lazy_write_pages": (284, 16), + "cache_flushes": (288, 17), + "cache_flush_pages": (292, 18), + } + for offset, value in expected.values(): + struct.pack_into(" dict[str, object]: + """Return counters that grow linearly: the scratch volume writes 120 + MiB/s at 500 IOPS with 30 ms latency and 90% busy time, the workspace + volume writes 2 MiB/s, CPUs are 25% busy, the lazy writer writes 10 + MiB/s, and Defender uses a quarter of one CPU and reads 1 MiB/s.""" + writes = elapsed_ms // 2 + sample: dict[str, object] = { + "elapsed_ns": elapsed_ms * 1_000_000, + "volumes": { + "F:": { + "bytes_read": 0, + "bytes_written": 120 * MIB * elapsed_ms // 1000, + "read_ticks": 0, + "write_ticks": writes * 300_000, + "idle_ticks": elapsed_ms * 1_000, + "reads": 0, + "writes": writes, + "queue_depth": 8 if elapsed_ms == 200 else 4, + "query_ticks": elapsed_ms * 10_000, + }, + "C:": { + "bytes_read": 0, + "bytes_written": 2 * MIB * elapsed_ms // 1000, + "read_ticks": 0, + "write_ticks": elapsed_ms * 100, + "idle_ticks": elapsed_ms * 9_900, + "reads": 0, + "writes": elapsed_ms // 10, + "queue_depth": 0, + "query_ticks": elapsed_ms * 10_000, + }, + }, + "cpu": { + "idle_ticks": elapsed_ms * 120_000, + "kernel_ticks": elapsed_ms * 136_000, + "user_ticks": elapsed_ms * 24_000, + }, + "memory": { + "available_pages": 1_000_000 - elapsed_ms, + "modified_pages_written": 7, + "mapped_pages_written": 9, + "lazy_write_pages": elapsed_ms * 256 // 100, + "cache_flush_pages": elapsed_ms * 3072 // 100, + "cache_dirty_pages": 51_200 if elapsed_ms == 200 else 25_600, + "cache_dirty_page_threshold": 262_144, + }, + } + if processes: + sample["defender"] = { + "processes": 2, + "cpu_ticks": elapsed_ms * 2_500, + "read_bytes": MIB * elapsed_ms // 1000, + "write_bytes": 0, + "other_bytes": 0, + } + return sample + + +def counter_series(*elapsed_ms: int) -> list[dict[str, object]]: + process_samples = {elapsed_ms[0], elapsed_ms[-1]} + return [ + counters_at(elapsed, processes=elapsed in process_samples) + for elapsed in elapsed_ms + ] + + +class StorageWindowTests(unittest.TestCase): + def test_summarizes_disk_cpu_writeback_and_defender_activity(self): + samples = counter_series(*range(0, 501, 50)) + + window = host_telemetry.summarize_storage_window( + samples, 100_000_000, 400_000_000 + ) + + self.assertEqual( + window, + { + "start_ns": 100_000_000, + "end_ns": 400_000_000, + "sampled_start_ns": 100_000_000, + "sampled_end_ns": 400_000_000, + "volumes": { + "F:": { + "read_mib_per_second": 0.0, + "write_mib_per_second": 120.0, + "read_iops": 0.0, + "write_iops": 500.0, + "average_read_latency_ms": None, + "average_write_latency_ms": 30.0, + "average_queue_length": 15.0, + "busy_percent": 90.0, + "max_queue_depth": 8, + }, + "C:": { + "read_mib_per_second": 0.0, + "write_mib_per_second": 2.0, + "read_iops": 0.0, + "write_iops": 100.0, + "average_read_latency_ms": None, + "average_write_latency_ms": 0.1, + "average_queue_length": 0.01, + "busy_percent": 1.0, + "max_queue_depth": 0, + }, + }, + "cpu": {"busy_percent": 25.0}, + "memory": { + "available_mib_min": 3904.688, + "cache_dirty_mib_max": 200.0, + "cache_dirty_threshold_mib": 1024.0, + "lazy_write_mib_per_second": 10.0, + "cache_flush_mib_per_second": 120.0, + "modified_write_mib_per_second": 0.0, + "mapped_write_mib_per_second": 0.0, + }, + # Process counters come only from the first and last samples. + "defender": { + "processes": 2, + "cpu_percent": 25.0, + "read_mib_per_second": 1.0, + "write_mib_per_second": 0.0, + "other_mib_per_second": 0.0, + "sampled_start_ns": 0, + "sampled_end_ns": 500_000_000, + }, + }, + ) + + def test_each_counter_group_uses_its_nearest_complete_samples(self): + samples = counter_series(*range(0, 501, 50)) + # Transient read failures at the phase's nearest samples. + failed_start, failed_end = samples[2], samples[8] + failed_start["volumes"] = {} + del failed_start["cpu"] + del failed_end["memory"] + + window = host_telemetry.summarize_storage_window( + samples, 100_000_000, 400_000_000 + ) + + assert window is not None + self.assertEqual(window["sampled_start_ns"], 100_000_000) + self.assertEqual(window["sampled_end_ns"], 400_000_000) + volumes = cast(dict[str, dict[str, object]], window["volumes"]) + self.assertEqual(volumes["F:"]["write_mib_per_second"], 120.0) + self.assertEqual( + (volumes["F:"]["sampled_start_ns"], volumes["F:"]["sampled_end_ns"]), + (50_000_000, 400_000_000), + ) + self.assertEqual( + window["cpu"], + { + "busy_percent": 25.0, + "sampled_start_ns": 50_000_000, + "sampled_end_ns": 400_000_000, + }, + ) + memory = cast(dict[str, object], window["memory"]) + self.assertEqual(memory["lazy_write_mib_per_second"], 10.0) + self.assertEqual( + (memory["sampled_start_ns"], memory["sampled_end_ns"]), + (100_000_000, 350_000_000), + ) + + def test_uses_the_samples_nearest_to_each_phase_boundary(self): + samples = counter_series(0, 60, 120, 180, 240, 300, 360) + + window = host_telemetry.summarize_storage_window( + samples, 100_000_000, 290_000_000 + ) + + assert window is not None + self.assertEqual(window["sampled_start_ns"], 120_000_000) + self.assertEqual(window["sampled_end_ns"], 300_000_000) + self.assertIsNone( + host_telemetry.summarize_storage_window(samples, 10_000_000, 20_000_000) + ) + self.assertIsNone( + host_telemetry.summarize_storage_window(samples[:1], 0, 100_000_000) + ) + + def test_request_counts_wrap_at_32_bits(self): + samples = counter_series(0, 100, 200) + for sample, writes in zip(samples, (2**32 - 30, 2**32 - 10, 40), strict=True): + volumes = cast(dict[str, dict[str, int]], sample["volumes"]) + volumes["F:"]["writes"] = writes + + window = host_telemetry.summarize_storage_window(samples, 0, 200_000_000) + + assert window is not None + volumes = cast(dict[str, dict[str, object]], window["volumes"]) + self.assertEqual(volumes["F:"]["write_iops"], 350.0) + + def test_skips_defender_totals_when_a_defender_process_starts_or_exits(self): + samples = counter_series(0, 100, 200) + defender = cast(dict[str, int], samples[-1]["defender"]) + defender["processes"] = 3 + + window = host_telemetry.summarize_storage_window(samples, 0, 200_000_000) + + assert window is not None + self.assertIsNone(window["defender"]) + + def test_describes_a_window_with_the_scratch_volume_first(self): + window = host_telemetry.summarize_storage_window( + counter_series(*range(0, 501, 50)), 100_000_000, 400_000_000 + ) + assert window is not None + + self.assertEqual( + host_telemetry.describe_storage_window( + window, {"C:": ["workspace", "system"], "F:": ["scratch"]} + ), + "scratch F: write 120.0 MiB/s at 30.0 ms, queue 15.0, 90% busy; " + "workspace/system C: write 2.0 MiB/s at 0.1 ms, queue 0.0, 1% busy; " + "CPU 25% busy; cache dirty 200.0 MiB, lazy writes 10.0 MiB/s; " + "Defender 25.0% CPU", + ) + + +class FakeReader: + def __init__(self, *, failures: frozenset[int] = frozenset()) -> None: + self.reads: list[bool] = [] + self.closed = 0 + self.failures = failures + + def description(self) -> dict[str, object]: + return { + "source": "fake", + "page_size_bytes": 4096, + "volumes": {"F:": ["scratch"]}, + } + + def read(self, *, processes: bool) -> tuple[dict[str, object], list[str]]: + self.reads.append(processes) + if len(self.reads) in self.failures: + raise RuntimeError("counter read failed") + return {"processes": processes}, ["F: volume counters failed: busy"] + + def close(self) -> None: + self.closed += 1 + + +class StorageTelemetrySamplerTests(unittest.TestCase): + def test_reads_process_counters_on_their_own_cadence(self): + reader = FakeReader() + times = iter( + 1_000 + milliseconds * 1_000_000 for milliseconds in (0, 100, 300, 550, 600) + ) + telemetry = host_telemetry.StorageTelemetry( + reader, + 1_000, + interval_seconds=3600, + process_interval_seconds=0.5, + clock=lambda: next(times), + ) + + telemetry.start() + self.assertTrue(telemetry._sample(final=False)) + self.assertTrue(telemetry._sample(final=False)) + self.assertTrue(telemetry._sample(final=False)) + telemetry.stop() + telemetry.stop() + result = telemetry.result({}) + + self.assertEqual(reader.reads, [True, False, False, True, True]) + self.assertEqual(reader.closed, 1) + self.assertEqual( + [ + sample["elapsed_ns"] + for sample in cast(list[dict[str, object]], result["samples"]) + ], + [0, 100_000_000, 300_000_000, 550_000_000, 600_000_000], + ) + self.assertEqual(result["errors"], ["F: volume counters failed: busy"]) + self.assertEqual(result["source"], "fake") + self.assertEqual(result["interval_ms"], 3_600_000) + self.assertEqual(result["process_interval_ms"], 500) + self.assertFalse(result["truncated"]) + + def test_truncation_still_records_the_final_sample(self): + reader = FakeReader() + times = iter(range(0, 10_000_000_000, 1_000_000)) + telemetry = host_telemetry.StorageTelemetry( + reader, 0, interval_seconds=3600, max_samples=2, clock=lambda: next(times) + ) + + telemetry.start() + self.assertTrue(telemetry._sample(final=False)) + self.assertFalse(telemetry._sample(final=False)) + telemetry.stop() + result = telemetry.result({}) + + self.assertTrue(result["truncated"]) + self.assertEqual(len(cast(list[object], result["samples"])), 3) + + def test_counter_failures_are_recorded_without_stopping_sampling(self): + reader = FakeReader(failures=frozenset({2})) + times = iter(range(0, 10_000_000_000, 1_000_000)) + telemetry = host_telemetry.StorageTelemetry( + reader, 0, interval_seconds=3600, clock=lambda: next(times) + ) + + telemetry.start() + telemetry._sample(final=False) + telemetry._sample(final=False) + telemetry.stop() + result = telemetry.result({}) + + samples = cast(list[dict[str, object]], result["samples"]) + self.assertEqual(len(samples), 4) + self.assertEqual(samples[1], {"elapsed_ns": 1_000_000}) + self.assertEqual( + result["errors"], + [ + "F: volume counters failed: busy", + "counter sampling failed: counter read failed", + ], + ) + + def test_result_summarizes_only_telemetry_phases(self): + times = iter((0, 50_000_000, 100_000_000)) + telemetry = host_telemetry.StorageTelemetry( + FakeReader(), 0, interval_seconds=3600, clock=lambda: next(times) + ) + telemetry.start() + telemetry._sample(final=False) + + result = telemetry.result( + { + "capture.mapped_memory_flush": (0, 100_000_000), + "capture.save_state": (0, 10_000_000), + } + ) + + self.assertEqual( + list(cast(dict[str, object], result["phases"])), + ["capture.mapped_memory_flush"], + ) + + def test_background_sampling_runs_until_stopped(self): + reader = FakeReader() + telemetry = host_telemetry.StorageTelemetry( + reader, + time.perf_counter_ns(), + interval_seconds=0.005, + process_interval_seconds=3600, + ) + + telemetry.start() + deadline = time.monotonic() + 10 + while len(reader.reads) < 4 and time.monotonic() < deadline: + time.sleep(0.005) + telemetry.stop() + count = len(reader.reads) + time.sleep(0.05) + + self.assertGreaterEqual(count, 5) + self.assertEqual(len(reader.reads), count) + self.assertTrue(reader.reads[0]) + self.assertTrue(reader.reads[-1]) + self.assertFalse(any(reader.reads[1:-1])) + + +class PlatformTelemetryTests(unittest.TestCase): + def test_windows_reader_keeps_disk_counters_enabled_until_closed(self): + advapi32 = MagicMock() + with ( + patch.object(host_telemetry, "_enable_disk_performance", return_value=1234), + patch.object(host_telemetry, "_advapi32", return_value=advapi32), + ): + reader = host_telemetry.WindowsCounterReader({}) + advapi32.WmiCloseBlock.assert_not_called() + reader.close() + reader.close() + + advapi32.WmiCloseBlock.assert_called_once_with(1234) + + def test_windows_reader_records_unavailable_disk_counters(self): + with patch.object( + host_telemetry, + "_enable_disk_performance", + side_effect=PermissionError("access denied"), + ): + reader = host_telemetry.WindowsCounterReader({}) + reader.close() + + self.assertEqual( + reader._errors, ["disk counters could not be enabled: access denied"] + ) + + @unittest.skipUnless(sys.platform == "win32", "requires Windows") + def test_windows_reads_counters_without_privileges(self): + reader = host_telemetry.WindowsCounterReader( + { + "scratch": Path(tempfile.gettempdir()), + "system": Path(os.environ["SystemRoot"]), + } + ) + try: + first, first_errors = reader.read(processes=True) + second, second_errors = reader.read(processes=False) + finally: + reader.close() + + self.assertEqual(first_errors, []) + self.assertEqual(second_errors, []) + roles = cast(dict[str, list[str]], reader.description()["volumes"]) + self.assertEqual( + sorted(role for values in roles.values() for role in values), + ["scratch", "system"], + ) + volumes = cast(dict[str, dict[str, int]], first["volumes"]) + self.assertEqual(set(volumes), set(roles)) + for counters in volumes.values(): + self.assertGreater(counters["query_ticks"], 0) + cpu = cast(dict[str, int], first["cpu"]) + self.assertGreaterEqual(cpu["kernel_ticks"], cpu["idle_ticks"]) + memory = cast(dict[str, int], first["memory"]) + self.assertGreater(memory["available_pages"], 0) + self.assertIn("cache_dirty_page_threshold", memory) + self.assertEqual( + set(cast(dict[str, int], first["defender"])), + {"processes", "cpu_ticks", "read_bytes", "write_bytes", "other_bytes"}, + ) + self.assertNotIn("defender", second) + + @unittest.skipIf(sys.platform == "win32", "Windows samples storage counters") + def test_other_hosts_record_no_storage_telemetry(self): + self.assertIsNone( + host_telemetry.create_storage_telemetry( + {"scratch": Path(tempfile.gettempdir())}, 0 + ) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/test_nvx_tools.py b/scripts/test_nvx_tools.py index 34f453159..20ec8b699 100644 --- a/scripts/test_nvx_tools.py +++ b/scripts/test_nvx_tools.py @@ -4951,6 +4951,23 @@ def test_ci_remeasures_only_unstable_lifecycle_results(self): script.count("Lifecycle snapshot generation was unstable"), 1, ) + self.assertEqual(script.count("--attempt-log"), 1) + self.assertIn("acceptance-attempts.json", script) + # Publication names the attempt that supplied the lifecycle metrics. + for step, log in ( + ( + "Publish Linux performance results", + '--lifecycle-attempt-log "${run_dir}/acceptance-attempts.json"', + ), + ( + "Publish Windows performance results", + '--lifecycle-attempt-log "$RunDir/acceptance-attempts.json"', + ), + ): + with self.subTest(step=step): + script = _composite_action_script(action, step) + self.assertEqual(script.count(log), 1) + self.assertLess(script.index("--lifecycle-input"), script.index(log)) def test_windows_benchmarks_use_provisioned_data_volume_scratch(self): action = ( @@ -5385,6 +5402,8 @@ def test_windows_acceptance_preserves_attempts_and_publishes_only_valid_data(sel $global:LASTEXITCODE = $script:BenchmarkStatus } elseif ($args[1] -eq "performance") { + $log = $args[[Array]::IndexOf($args, "--attempt-log") + 1] + Add-Content -LiteralPath "attempt-logs.txt" -Value $log $global:LASTEXITCODE = $script:ValidationStatuses[$script:ValidationCount] $script:ValidationCount += 1 } @@ -5413,6 +5432,13 @@ def test_windows_acceptance_preserves_attempts_and_publishes_only_valid_data(sel output.mkdir(parents=True) accepted = output / "acceptance.json" accepted.write_text('{"attempt":"stale"}', encoding="utf-8") + # A rerun in the same directory must not keep a previous + # run's attempt log or later attempts. + attempt_log = output / "acceptance-attempts.json" + attempt_log.write_text('{"attempts":"stale"}', encoding="utf-8") + (output / "acceptance-attempt-2.json").write_text( + '{"attempt":"stale"}', encoding="utf-8" + ) setup = ( "$ValidationStatuses = @(" + ", ".join(str(status) for status in statuses) @@ -5434,6 +5460,20 @@ def test_windows_acceptance_preserves_attempts_and_publishes_only_valid_data(sel ) expected_status = benchmark_status or statuses[-1] self.assertEqual(result.returncode, expected_status, result.stderr) + self.assertFalse(attempt_log.exists()) + validations = root / "attempt-logs.txt" + self.assertEqual( + ( + validations.read_text(encoding="utf-8").splitlines() + if validations.exists() + else [] + ), + [ + "data/runs/windows-whp-virtual-machine/microvm-v2/1vcpu/" + "acceptance-attempts.json" + ] + * (0 if benchmark_status else len(statuses)), + ) self.assertEqual( sorted(path.name for path in output.glob("acceptance-attempt-*")), [ @@ -5477,12 +5517,13 @@ def test_linux_acceptance_preserves_attempts_and_publishes_only_valid_data(self) benchmark_count=0 validation_count=0 python3() { - local argument output="" input="" platform="" previous="" + local argument output="" input="" platform="" attempt_log="" previous="" for argument in "$@"; do case "${previous}" in --output) output="${argument}" ;; --input) input="${argument}" ;; --platform) platform="${argument}" ;; + --attempt-log) attempt_log="${argument}" ;; esac previous="${argument}" done @@ -5495,7 +5536,7 @@ def test_linux_acceptance_preserves_attempts_and_publishes_only_valid_data(self) fi if [[ "${2-} ${3-}" == "performance validate-openvmm" ]]; then validation_count=$((validation_count + 1)) - echo "validate ${platform} ${input}" + echo "validate ${platform} ${input} ${attempt_log}" return "${VALIDATION_STATUSES[validation_count - 1]}" fi echo "unexpected command: $*" >&2 @@ -5525,6 +5566,13 @@ def test_linux_acceptance_preserves_attempts_and_publishes_only_valid_data(self) output.mkdir(parents=True) accepted = output / "acceptance.json" accepted.write_text('{"attempt":"stale"}', encoding="utf-8") + # A rerun in the same directory must not keep a previous + # run's attempt log or later attempts. + attempt_log = output / "acceptance-attempts.json" + attempt_log.write_text('{"attempts":"stale"}', encoding="utf-8") + (output / "acceptance-attempt-2.json").write_text( + '{"attempt":"stale"}', encoding="utf-8" + ) setup = ( "VALIDATION_STATUSES=(" + " ".join(str(status) for status in statuses) @@ -5549,6 +5597,7 @@ def test_linux_acceptance_preserves_attempts_and_publishes_only_valid_data(self) ) expected_status = benchmark_status or statuses[-1] self.assertEqual(result.returncode, expected_status, result.stderr) + self.assertFalse(attempt_log.exists()) expected_stdout: list[str] = [] for index in range(1, len(statuses) + 1): attempt = f"{run_dir}/acceptance-attempt-{index}.json" @@ -5559,7 +5608,10 @@ def test_linux_acceptance_preserves_attempts_and_publishes_only_valid_data(self) ) expected_stdout.append(f"benchmark {platform} {attempt}") if not benchmark_status: - expected_stdout.append(f"validate {platform} {attempt}") + expected_stdout.append( + f"validate {platform} {attempt} " + f"{run_dir}/acceptance-attempts.json" + ) self.assertEqual(result.stdout.splitlines(), expected_stdout) self.assertEqual( sorted(path.name for path in output.glob("acceptance-attempt-*")), @@ -14259,6 +14311,301 @@ def close(self) -> None: for line in [*records.splitlines(), committed.removesuffix(b"\n")]: self.assertTrue(benchmark.contains_output_line(log, line), line) + def test_capture_samples_storage_from_the_request_through_source_exit(self): + records = self.incident_capture_records() + snapshot_requested = threading.Event() + + class FakeProcess: + pid = BenchmarkTests.INCIDENT_PID + returncode = 0 + + def poll(self) -> None: + return None + + def wait(self) -> int: + return 0 + + class FakeInteraction: + def __init__(self, snapshot: Path) -> None: + self.process = FakeProcess() + self.snapshot = snapshot + self.writes: list[bytes] = [] + + def read_output(self, chunks: queue.Queue[bytes | None]) -> None: + chunks.put(benchmark.BOOT_MARKER + b"\r\n") + chunks.put(benchmark.SMP_PROBE_COMPLETION_MARKER + b"\r\n") + if snapshot_requested.wait(5): + chunks.put( + b"nvx-snapshot\r\n" + + benchmark.SNAPSHOT_GUEST_DISPATCH_MARKER + + b"\r\n" + ) + chunks.put(None) + + def read_stderr(self, chunks: queue.Queue[bytes | None]) -> None: + if snapshot_requested.wait(5): + chunks.put(records) + self.snapshot.mkdir() + chunks.put(None) + + def write_input(self, data: bytes) -> None: + self.writes.append(data) + if data == b"nvx-snapshot\n": + snapshot_requested.set() + + def close(self) -> None: + pass + + class FakeTelemetry: + def __init__(self, interaction: FakeInteraction) -> None: + self.interaction = interaction + self.events: list[str] = [] + self.windows: dict[str, tuple[int, int]] = {} + + def start(self) -> None: + self.events.append(f"start after {len(self.interaction.writes)}") + + def stop(self) -> None: + self.events.append("stop") + + def result(self, windows: dict[str, tuple[int, int]]) -> dict[str, object]: + self.events.append("result") + self.windows = windows + return {"phases": {}} + + for retained in (True, False): + with ( + self.subTest(retained=retained), + tempfile.TemporaryDirectory() as temporary, + ): + snapshot_requested.clear() + snapshot = Path(temporary) / "snapshot" + interaction = FakeInteraction(snapshot) + telemetry = FakeTelemetry(interaction) + profiles: list[dict[str, object]] = [] + with ( + patch.object( + benchmark, "InteractiveProcess", return_value=interaction + ), + patch.object(benchmark, "_try_peak_rss", return_value=1024), + patch.object( + benchmark, "process_resource_counters", return_value={} + ), + patch.object(benchmark, "terminate"), + patch.object( + benchmark.host_telemetry, + "create_storage_telemetry", + return_value=telemetry, + ) as create, + ): + benchmark.capture_snapshot( + ["openvmm"], + snapshot, + processors=1, + timeout=10, + snapshot_profile=True, + profile_sink=profiles if retained else None, + ) + + if not retained: + # Warmups keep no profile, so they sample nothing. + create.assert_not_called() + continue + volumes, origin = create.call_args.args + self.assertEqual(volumes["scratch"], Path(temporary)) + self.assertEqual(volumes["workspace"], BuildConstants.REPO_ROOT) + self.assertIsInstance(origin, int) + # Sampling starts after the SMP probe script, before the request. + self.assertEqual( + telemetry.events, ["start after 1", "stop", "result", "stop"] + ) + self.assertEqual(interaction.writes[1:], [b"nvx-snapshot\n"]) + flush_start, flush_end = telemetry.windows[ + "capture.mapped_memory_flush" + ] + self.assertEqual(flush_end - flush_start, 2700) + generation_start, generation_end = telemetry.windows[ + "capture.snapshot_generation" + ] + self.assertEqual(generation_end - generation_start, 1_903_503) + self.assertEqual(profiles[0]["storage_telemetry"], {"phases": {}}) + + def test_profile_phase_windows_use_the_tightest_clock_offset(self): + records: list[dict[str, object]] = [ + { + "operation": "capture", + "phase": "save_state", + "duration_ns": 1_000, + "process_elapsed_ns": 5_000, + "observer_elapsed_ns": 20_500, + "source": "openvmm", + }, + { + "operation": "capture", + "phase": "mapped_memory_flush", + "duration_ns": 900_000, + "process_elapsed_ns": 905_000, + "observer_elapsed_ns": 920_000, + "source": "openvmm", + }, + { + "operation": "capture", + "phase": "snapshot_generation", + "duration_ns": 950_000, + "process_elapsed_ns": 955_000, + "source": "openvmm_clock", + }, + { + "operation": "capture", + "phase": "source_teardown", + "duration_ns": 9_000, + "observer_elapsed_ns": 990_000, + "source": "benchmark_observer", + }, + ] + + # The save_state record arrived 0.5 us late, so the flush record's + # 15 us offset bounds OpenVMM's clock. + self.assertEqual( + benchmark.snapshot_profile_phase_windows(records), + { + "capture.save_state": (19_000, 20_000), + "capture.mapped_memory_flush": (20_000, 920_000), + "capture.snapshot_generation": (20_000, 970_000), + "capture.source_teardown": (981_000, 990_000), + }, + ) + + def test_capture_telemetry_line_describes_the_ram_flush_window(self): + records = [ + { + "operation": "capture", + "phase": "mapped_memory_flush", + "duration_ns": 1_165_314_800, + } + ] + sample: dict[str, object] = { + "records": records, + "storage_telemetry": { + "volumes": {"F:": ["scratch"]}, + "phases": { + "capture.mapped_memory_flush": { + "volumes": { + "F:": { + "write_mib_per_second": 128.44, + "average_write_latency_ms": 31.2, + "average_queue_length": 3.94, + "busy_percent": 99.4, + } + }, + "cpu": {"busy_percent": 6.2}, + } + }, + }, + } + + self.assertEqual( + benchmark.describe_capture_telemetry(sample), + "mapped_memory_flush 1165.315 ms: scratch F: write 128.4 MiB/s at " + "31.2 ms, queue 3.9, 99% busy; CPU 6% busy", + ) + self.assertIsNone(benchmark.describe_capture_telemetry({"records": records})) + + def test_result_documents_record_host_provenance(self): + args = nvx.parse_args(["benchmark"]) + with ( + patch.object(benchmark, "_git_revision", return_value="revision"), + patch.object( + benchmark, + "benchmark_host_provenance", + return_value={"runner_name": "azure-windows-3"}, + ) as host, + ): + document = benchmark.result_document(args, None, None) + + host.assert_called_once_with(args) + self.assertEqual(document["host"], {"runner_name": "azure-windows-3"}) + + def test_host_provenance_names_workspace_temporary_and_scratch_volumes(self): + with tempfile.TemporaryDirectory() as temporary: + scratch = Path(temporary).resolve() / "scratch" + scratch.mkdir() + args = nvx.parse_args(["benchmark", "--scratch-dir", str(scratch)]) + system_temporary = Path(tempfile.gettempdir()).resolve() + with ( + benchmark.benchmark_scratch_directory(args), + patch.object( + benchmark.host_telemetry, + "host_provenance", + return_value={"schema_version": 1}, + ) as provenance, + ): + host = benchmark.benchmark_host_provenance(args) + + self.assertEqual(host, {"schema_version": 1}) + provenance.assert_called_once_with( + { + "workspace": BuildConstants.REPO_ROOT, + "temporary": system_temporary, + "scratch": scratch, + } + ) + + def test_device_io_resume_ignores_host_provenance(self): + args = nvx.parse_args( + [ + "benchmark", + "--suite", + "device-io", + "--backend", + "whp", + "--platform", + "windows-whp-baremetal", + ] + ) + benchmark.apply_benchmark_suite_defaults(args) + hosts = iter(({"machine_name": "first"}, {"machine_name": "second"})) + + def host_provenance(_: argparse.Namespace) -> dict[str, str]: + return next(hosts, {"machine_name": "third"}) + + with ( + tempfile.TemporaryDirectory() as temporary, + patch.object( + benchmark, "benchmark_host_provenance", side_effect=host_provenance + ), + patch.object(benchmark, "sha256_file", return_value="0" * 64), + patch.object( + benchmark, + "_device_io_helper_provenance", + return_value={"source": "helper"}, + ), + patch.object(benchmark, "_git_revision", return_value="revision"), + patch.object(benchmark, "_git_status", return_value=[]), + ): + output = Path(temporary) + + def write() -> Path: + return benchmark.write_benchmark_metadata( + args, + output, + Path("openvmm.exe"), + Path("vmlinux"), + Path("initramfs.cpio.gz"), + "whp", + ) + + path = write() + written = path.read_text(encoding="utf-8") + self.assertEqual(json.loads(written)["host"], {"machine_name": "first"}) + self.assertEqual(write(), path) + self.assertEqual(path.read_text(encoding="utf-8"), written) + args.device_io_port += 1 + with self.assertRaisesRegex( + ValueError, "resume controls or provenance do not match" + ): + write() + def test_measure_once_observes_restore_marker_split_by_profile_record(self): record = ( b"OPENVMM_SNAPSHOT_PROFILE_V1 operation=restore " diff --git a/scripts/test_performance.py b/scripts/test_performance.py index 3c8453069..1818ee084 100644 --- a/scripts/test_performance.py +++ b/scripts/test_performance.py @@ -16,6 +16,7 @@ sys.path.insert(0, str(Path(__file__).parent)) import nvx # noqa: E402 from nvx_tools import benchmark, performance # noqa: E402 +from nvx_tools.common import sha256_file # noqa: E402 COLD_START_LOG = """ base : 101.0 ms (min 100, max 102, n=5) @@ -196,6 +197,132 @@ def samples(minimum: int | float, median: int | float, maximum: int | float): } +# Snapshot generation from run 35393916034, whose minority fast path the +# stability guard rejects. +MINORITY_FAST_PATH_MS = [ + 763.3528, + 1219.9358, + 1242.3189, + 1256.6633, + 1275.0555, + 1274.4436, + 737.5818, + 1257.1264, + 1240.7679, + 1231.9344, +] +BENCHMARK_HOST: dict[str, object] = { + "schema_version": 1, + "runner_name": "azure-windows-3", + "machine_name": "VMSSVAOAN000002", + "os": "Windows-2022Server-10.0.20348-SP0", + "cpu_model": "Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz", + "logical_processors": 16, + "memory_bytes": 64 * 1024**3, + "azure": { + "vm_size": "Standard_D16ds_v5", + "os_disk": { + "storage_account_type": "Premium_LRS", + "size_gib": 128, + "caching": "ReadWrite", + }, + "data_disks": [], + "resource_disk_size_kib": 0, + }, + "volumes": { + "scratch": { + "path": "F:\\nvx-benchmark-scratch\\job", + "volume": "F:", + "filesystem": "NTFS", + "size_bytes": 512 * 1024**3, + } + }, +} + + +def profiled_lifecycle_document( + samples_ms: list[float] | None = None, +) -> dict[str, object]: + """Return a Windows lifecycle result with host provenance and storage + telemetry whose scratch volume writes 160 MiB/s during fast flushes and + 96 MiB/s during slow ones.""" + document = lifecycle_document("whp") + capture = cast( + dict[str, object], cast(dict[str, object], document["snapshot_capture"])["whp"] + ) + if samples_ms is not None: + capture.update( + samples_ms=samples_ms, + p50_ms=statistics.median(samples_ms), + min_ms=min(samples_ms), + max_ms=max(samples_ms), + ) + raw_samples: list[dict[str, object]] = [] + for generation_ms in cast(list[float], capture["samples_ms"]): + raw_samples.append( + { + "generation_duration_ns": round(generation_ms * 1_000_000), + "records": [ + { + "operation": "capture", + "phase": "mapped_memory_flush", + "exclusive": True, + "duration_ns": round((generation_ms - 20) * 1_000_000), + "process_elapsed_ns": 2_000_000_000, + "pid": 4242, + "source": "openvmm", + } + ], + "storage_telemetry": { + "volumes": {"F:": ["scratch"], "C:": ["workspace", "system"]}, + "phases": { + "capture.mapped_memory_flush": { + "volumes": { + "F:": { + "write_mib_per_second": ( + 160.0 if generation_ms < 1000 else 96.0 + ), + "average_write_latency_ms": 31.25, + "average_queue_length": 3.94, + "busy_percent": 99.4, + }, + "C:": {"write_mib_per_second": 0.25}, + }, + "cpu": {"busy_percent": 6.2}, + "memory": { + "cache_dirty_mib_max": 12.34, + "lazy_write_mib_per_second": 0.0, + }, + "defender": {"cpu_percent": 0.04}, + } + }, + }, + } + ) + capture["profile"] = {"schema_version": 1, "raw_samples": raw_samples} + document["host"] = BENCHMARK_HOST + return document + + +def validate_attempt(source: Path, log: Path) -> int: + with ( + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + return nvx.main( + [ + "performance", + "validate-openvmm", + "--platform", + "windows-whp-virtual-machine", + "--input", + str(source), + "--attempt-log", + str(log), + ] + ) + + class PerformanceTests(unittest.TestCase): def test_ci_one_vcpu_metric_count_matches_collectors(self): expected = len( @@ -752,6 +879,284 @@ def test_invalid_openvmm_result_uses_regular_failure_exit_code(self): self.assertEqual(status, 2) self.assertIn("does not match platform", stderr.getvalue()) + def test_validate_openvmm_records_every_attempt_outcome(self): + invalid = lifecycle_document("whp") + cast(dict[str, object], invalid["controls"])["backend"] = "kvm" + documents = ( + profiled_lifecycle_document(MINORITY_FAST_PATH_MS), + invalid, + profiled_lifecycle_document(), + ) + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + log = root / "acceptance-attempts.json" + statuses: list[int] = [] + digests: list[str] = [] + for index, document in enumerate(documents, 1): + source = root / f"acceptance-attempt-{index}.json" + source.write_text(json.dumps(document), encoding="utf-8") + digests.append(sha256_file(source)) + statuses.append(validate_attempt(source, log)) + recorded = json.loads(log.read_text(encoding="utf-8")) + + self.assertEqual(statuses, [75, 2, 0]) + self.assertEqual(recorded["schema_version"], 1) + self.assertEqual(recorded["platform"], "windows-whp-virtual-machine") + attempts = recorded["attempts"] + self.assertEqual( + [ + (attempt["attempt"], attempt["file"], attempt["sha256"]) + for attempt in attempts + ], + [ + (index, f"acceptance-attempt-{index}.json", digest) + for index, digest in enumerate(digests, 1) + ], + ) + self.assertEqual( + [(attempt["status"], attempt["exit_status"]) for attempt in attempts], + [("unstable", 75), ("invalid", 2), ("valid", 0)], + ) + self.assertRegex(attempts[0]["message"], r"split 2/8.*59\.8% gap") + self.assertIn("does not match platform", attempts[1]["message"]) + self.assertIsNone(attempts[2]["message"]) + + def test_collect_names_the_published_attempt_and_reports_every_attempt(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + logs = root / "logs" + logs.mkdir() + (logs / "cold-start.log").write_text(COLD_START_LOG, encoding="utf-8") + (logs / "virtfs.log").write_text(VIRTFS_LOG, encoding="utf-8") + log = root / "acceptance-attempts.json" + for index, document in enumerate( + ( + profiled_lifecycle_document(MINORITY_FAST_PATH_MS), + profiled_lifecycle_document(), + ), + 1, + ): + source = root / f"acceptance-attempt-{index}.json" + source.write_text(json.dumps(document), encoding="utf-8") + validate_attempt(source, log) + published = root / "acceptance.json" + published.write_bytes((root / "acceptance-attempt-2.json").read_bytes()) + summary = root / "summary.md" + stdout = io.StringIO() + + with contextlib.redirect_stdout(stdout): + result_path = performance.collect_results( + "windows-whp-virtual-machine", + "abc123", + logs, + root / "results", + summary_path=summary, + lifecycle_input=published, + lifecycle_attempt_log=log, + ) + markdown = summary.read_text(encoding="utf-8") + + self.assertLessEqual( + performance.LIFECYCLE_METRICS, + {result.metric for result in performance.read_results(result_path)}, + ) + self.assertIn( + "Lifecycle metrics come from attempt 2 of 2: acceptance-attempt-2.json", + stdout.getvalue(), + ) + self.assertIn("### Lifecycle attempts", markdown) + self.assertRegex( + markdown, + r"\| 1 \| `acceptance-attempt-1\.json` \| unstable: unstable snapshot " + r"generation at .*split 2/8.* \| 1241\.54 ms \| no \|", + ) + self.assertIn( + "| 2 | `acceptance-attempt-2.json` | valid | 31.00 ms | yes |", markdown + ) + self.assertIn("### Benchmark host", markdown) + self.assertIn("| Runner | azure-windows-3 |", markdown) + self.assertIn("| Azure VM size | Standard_D16ds_v5 |", markdown) + self.assertIn( + "| Azure OS disk | Premium_LRS, 128 GiB, ReadWrite caching |", markdown + ) + self.assertIn("### Snapshot generation storage telemetry", markdown) + self.assertIn( + "#### Attempt 1: unstable (`acceptance-attempt-1.json`)", markdown + ) + self.assertIn( + "#### Attempt 2: valid, published (`acceptance-attempt-2.json`)", markdown + ) + self.assertIn("Scratch volume: `F:`.", markdown) + self.assertIn( + "| 1 | 763.35 ms | 743.35 ms | 160.0 MiB/s | 31.25 ms | 3.9 | 99% | " + "0.2 MiB/s | 6% | 12.3 MiB | 0.0 MiB/s | 0.0% |", + markdown, + ) + self.assertIn( + "| 2 | 1219.94 ms | 1199.94 ms | 96.0 MiB/s | 31.25 ms | 3.9 | 99% | ", + markdown, + ) + self.assertEqual(markdown.count("| Sample | Generation | Flush |"), 2) + + def test_collect_rejects_a_lifecycle_input_that_no_valid_attempt_supplied(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + logs = root / "logs" + logs.mkdir() + (logs / "cold-start.log").write_text(COLD_START_LOG, encoding="utf-8") + (logs / "virtfs.log").write_text(VIRTFS_LOG, encoding="utf-8") + log = root / "acceptance-attempts.json" + unstable = root / "acceptance-attempt-1.json" + document = profiled_lifecycle_document(MINORITY_FAST_PATH_MS) + unstable.write_text(json.dumps(document), encoding="utf-8") + validate_attempt(unstable, log) + published = root / "acceptance.json" + published.write_text( + json.dumps(profiled_lifecycle_document()), encoding="utf-8" + ) + results = root / "results" + + with self.assertRaisesRegex( + performance.PerformanceError, "matches 0 valid attempts" + ): + performance.collect_results( + "windows-whp-virtual-machine", + "abc123", + logs, + results, + lifecycle_input=published, + lifecycle_attempt_log=log, + ) + self.assertFalse(results.exists()) + with self.assertRaisesRegex( + performance.PerformanceError, "requires a lifecycle input" + ): + performance.collect_results( + "windows-whp-virtual-machine", + "abc123", + logs, + results, + lifecycle_attempt_log=log, + ) + + def test_attempt_log_rejects_malformed_entries(self): + valid = { + "attempt": 1, + "file": "acceptance-attempt-1.json", + "sha256": "a" * 64, + "status": "valid", + "exit_status": 0, + "message": None, + } + cases: dict[str, tuple[dict[str, object], str]] = { + "platform": ( + {"schema_version": 1, "platform": "linux-kvm", "attempts": [valid]}, + "records platform 'linux-kvm'", + ), + "schema": ( + { + "schema_version": 2, + "platform": "windows-whp-virtual-machine", + "attempts": [valid], + }, + "unsupported schema_version 2", + ), + "status": ( + { + "schema_version": 1, + "platform": "windows-whp-virtual-machine", + "attempts": [{**valid, "exit_status": 75}], + }, + r"invalid lifecycle attempt at .*attempts\[0\]", + ), + "order": ( + { + "schema_version": 1, + "platform": "windows-whp-virtual-machine", + "attempts": [{**valid, "attempt": 2}], + }, + r"invalid lifecycle attempt at .*attempts\[0\]", + ), + "digest": ( + { + "schema_version": 1, + "platform": "windows-whp-virtual-machine", + "attempts": [{**valid, "sha256": "not-a-digest"}], + }, + r"invalid lifecycle attempt at .*attempts\[0\]", + ), + } + with tempfile.TemporaryDirectory() as temporary: + log = Path(temporary) / "acceptance-attempts.json" + for name, (document, error) in cases.items(): + with self.subTest(name=name): + log.write_text(json.dumps(document), encoding="utf-8") + with self.assertRaisesRegex(performance.PerformanceError, error): + performance.read_lifecycle_attempt_log( + log, "windows-whp-virtual-machine" + ) + + def test_lifecycle_diagnostics_report_host_and_telemetry_without_a_log(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "acceptance.json" + source.write_text( + json.dumps(profiled_lifecycle_document()), encoding="utf-8" + ) + summary = root / "summary.md" + + performance.collect_openvmm_results( + "windows-whp-virtual-machine", + "abc123", + source, + root / "results", + summary, + ) + markdown = summary.read_text(encoding="utf-8") + + self.assertNotIn("### Lifecycle attempts", markdown) + self.assertIn("| Machine | VMSSVAOAN000002 |", markdown) + self.assertIn("### Snapshot generation storage telemetry", markdown) + self.assertNotIn("#### Attempt", markdown) + self.assertEqual(markdown.count("| 160.0 MiB/s |"), 10) + + def test_collect_cli_accepts_lifecycle_attempt_logs(self): + collect = nvx.parse_args( + [ + "performance", + "collect", + "--platform", + "windows-whp-virtual-machine", + "--commit", + "abc123", + "--input-dir", + "logs", + "--output-dir", + "results", + "--lifecycle-input", + "acceptance.json", + "--lifecycle-attempt-log", + "acceptance-attempts.json", + ] + ) + validate = nvx.parse_args( + [ + "performance", + "validate-openvmm", + "--platform", + "windows-whp-virtual-machine", + "--input", + "acceptance-attempt-1.json", + "--attempt-log", + "acceptance-attempts.json", + ] + ) + + self.assertEqual( + collect.lifecycle_attempt_log, Path("acceptance-attempts.json") + ) + self.assertEqual(validate.attempt_log, Path("acceptance-attempts.json")) + def test_collect_appends_ci_benchmark_table(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary)