From e2761449dcaff801cadeb52b08d4233796826840 Mon Sep 17 00:00:00 2001 From: Pedro Henrique Penna Date: Sat, 10 Oct 2026 12:32:52 -0300 Subject: [PATCH] ci: record benchmark host provenance and storage telemetry Windows/WHP lifecycle acceptance split snapshot generation into fast and slow clusters in runs 35382585623 and 35393916034, isolated to capture.mapped_memory_flush. #118 made such a series remeasure once, and b51a0b8c6 moved benchmark scratch off the burst-limited OS disk, but a future split would still leave three gaps (#119): - The published CSV could not be traced to the bounded attempt that supplied it; only an unlabeled copy in acceptance.json remained. - Results did not identify the runner, its Azure VM size and disks, or the volumes behind the workspace, temporary, and scratch directories. - Nothing recorded the host's disk, CPU, writeback, or Defender activity during the flush, so disk throttling, scanning, writeback, and unrelated host load could not be told apart. Attempt identification: - performance validate-openvmm --attempt-log appends each attempt's file, SHA-256 digest, outcome, exit status, and error to acceptance-attempts.json. - performance collect --lifecycle-attempt-log fails before it writes a CSV unless acceptance.json is a byte-identical copy of exactly one valid attempt, prints that attempt, and lists every attempt in the step summary. - Both acceptance steps remove a stale log and stale attempt files before measuring, and both publish steps pass the log. Host provenance: - The new nvx_tools.host_telemetry module records the runner and machine names, OS, CPU model, logical processor count, and memory, and the volume, file system, size, disk number, and SCSI address behind each benchmark directory. - On a possible Azure VM it queries the instance metadata service once, bypassing proxies, and keeps only the VM size and each managed disk's SKU, size, caching, write-accelerator, ephemeral, LUN, and Ultra Disk limit fields, plus the temporary disk size. It never records names, resource or VM identifiers, the subscription, tags, network addresses, or credentials, and records a failed query as azure.error. - Lifecycle JSON results and benchmark-metadata.json record it under host, acceptance and diagnostic suites print it, and summaries add a host table. The device-io resume identity ignores host. Storage telemetry: - With --snapshot-profile on Windows, a coordinator thread samples volume performance IOCTLs, GetSystemTimes, and SystemPerformanceInformation about every 50 ms, and Defender process CPU and I/O from SystemProcessInformation about every 500 ms, from just before the snapshot request until the source process exits. These interfaces need no privileges; the Network Service account of the CI jobs cannot read performance counters. Failed reads are recorded and never fail the benchmark. - Each retained capture maps its capture.mapped_memory_flush and capture.snapshot_generation phases onto the observer clock through the tightest record offset and summarizes throughput, IOPS, latency, queue length, busy time, CPU, writeback, dirty cache, and Defender activity. Each counter group spans its own samples nearest to each phase boundary, so a failed read does not blank the group, and reports its sampled interval when that differs from the phase's. - The job log prints each sample's flush with its host activity, and the step summary tabulates it for every attempt, rejected ones included. Linux captures flush no file-backed guest RAM image and record no storage telemetry. The benchmark documentation explains how to read the telemetry, keeps runner identity diagnostic only, and requires any storage change to apply to every pooled runner together with a history reset of the affected metrics. Validation: - On Windows 11 without elevation, the SystemProcessInformation offsets match GetProcessTimes and GetProcessIoCounters exactly, available pages match GlobalMemoryStatusEx, and a child process's mapped-view flush moved the flush counters by the bytes the volume wrote. - ruff check and format, compileall, and Pyright for Linux and Windows report no findings. The validate-nvx unit tests (932 on Windows), the adversarial (70) and host inventory (4) tests, the help smoke checks, and git diff --check pass. - On bare-metal Linux/MSHV (Azure Linux 3.0, Python 3.12) and Linux/KVM (Ubuntu, Python 3.14) hosts, test_host_telemetry, test_performance, and test_nvx_tools pass (554), including the bash acceptance-step test. The CI acceptance and performance-suite steps, run verbatim with CI inputs, recorded one valid attempt, and collect named it for all 29 metrics and listed the attempt and host in the summary. - On bare-metal Windows Server 2025/WHP, the verbatim acceptance step rejected attempt 1 as unstable (a 2/8 split between 488 and 725 ms), recorded it, and remeasured; collect named attempt 2 of 2 and reported both attempts' telemetry. That run showed that Windows Server keeps disk counters on demand: the volume IOCTL failed until the sampler held the disk performance WMI block open, after which the Windows counter test passes there too. - A rerun overlapped another session's cargo build of OpenVMM. During its flushes, CPU busy time rose to 7-70% (1-2% idle), the dirty cache to 205-450 MiB (0.1 MiB idle), lazy writes to 38 MiB/s, and write latency to 30-240 ms as flushes grew from 477 to 1509 ms. - Fourteen interleaved e2e runs on the otherwise idle host gave a snapshot-generation p50 median of 527 ms with telemetry and 522 ms without; cold start (476 ms) and restore (94 ms) matched. - In PR CI run 38067120152, the Azure Windows runner, whose jobs run as Network Service, recorded every counter group without errors, and collect named the published attempt. During each flush, the 512 GiB Premium_LRS (P20) scratch disk wrote 148-165 MiB/s at 94-100% busy, near that tier's 170 MB/s burst limit. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .github/actions/run-benchmark/action.yml | 23 +- .github/actions/validate-nvx/action.yml | 4 +- .github/skills/nvx-benchmark/SKILL.md | 2 + doc/benchmarks.md | 86 ++ doc/usage.md | 20 +- scripts/nvx_tools/benchmark.py | 161 ++- scripts/nvx_tools/host_telemetry.py | 1397 ++++++++++++++++++++++ scripts/nvx_tools/performance.py | 421 ++++++- scripts/setup/README.md | 5 +- scripts/test_host_telemetry.py | 929 ++++++++++++++ scripts/test_nvx_tools.py | 353 +++++- scripts/test_performance.py | 405 +++++++ 12 files changed, 3785 insertions(+), 21 deletions(-) create mode 100644 scripts/nvx_tools/host_telemetry.py create mode 100644 scripts/test_host_telemetry.py 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)