From 785d71566acb410721e638ee3fafe1b5b26cc5b7 Mon Sep 17 00:00:00 2001 From: Leonardo Forchini Date: Fri, 25 Sep 2026 08:46:30 +0000 Subject: [PATCH 1/2] Add a fake backend for component tests The harness needs a VM the daemon can discover and scale without libvirt. JSON files carry the snapshot, and set_thread_count records each target so tests can assert behavior. Signed-off-by: Leonardo Forchini --- Cargo.toml | 2 + Makefile | 2 +- src/backends/fake/mod.rs | 289 +++++++++++++++++++++++++++++++++++++++ src/backends/mod.rs | 2 + 4 files changed, 294 insertions(+), 1 deletion(-) create mode 100644 src/backends/fake/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 2c82a4d2..750b3a5c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,8 @@ path = "src/lib.rs" default = ["threshold-engine", "qemu-backend", "tui"] threshold-engine = [] qemu-backend = ["dep:rustix", "dep:uuid", "dep:virt"] +# In-process VM stand-in used by component tests. Not part of a production build. +fake-backend = [] # TUI (`iothread-tui` binary), enabled by default. tui = ["dep:anyhow", "dep:ratatui", "dep:crossterm"] diff --git a/Makefile b/Makefile index 116f15a0..b89e6d8e 100644 --- a/Makefile +++ b/Makefile @@ -51,7 +51,7 @@ unit-test: .PHONY: component-test component-test: cargo build --locked --no-default-features \ - --features threshold-engine + --features threshold-engine,fake-backend python3 -m pytest tests/component .PHONY: test diff --git a/src/backends/fake/mod.rs b/src/backends/fake/mod.rs new file mode 100644 index 00000000..e1db21d9 --- /dev/null +++ b/src/backends/fake/mod.rs @@ -0,0 +1,289 @@ +// SPDX-License-Identifier: BSD-3-Clause +// Copyright (c) 2026 Nutanix, Inc. +// +// Author: Leonardo Forchini + +//! File-backed backend for component tests. +//! +//! Instead of requiring a running QEMU instance, this backend manages +//! state through JSON files in `state_dir`. This allows for the external +//! Python component testing harness to manipulate "VM"s as necessary. +//! Similarly, controller actuations are preserved in `calls.json`, which +//! the harness can use in assertions. +//! +//! VM JSON files consist of the following which are used to construct a +//! [`ThreadPoolSnapshot`]: +//! * thread_count - Number of worker threads assigned to this VM +//! * vcpu_count - Number of vCPUs +//! * read_io_count - Cumulative read IO operations +//! * write_io_count - Cumulative write IO operations +//! * other_io_count - Cumulative other IO operations +//! +//! As well as some control fields: +//! * fail_snapshot - An error should be injected when attempting to get VMs +//! ThreadPoolSnapshot. +//! +//! `call.json` is used to give the external testing harness visibility over +//! [`InstanceClient::set_thread_count`] calls by the controller. It keeps +//! track of the following: +//! * vm - Actuated VM +//! * thread_count - New VM worker thread count + +use std::{ + fs::{self, File, OpenOptions}, + io::{Error as IoError, Write}, + sync::Arc, +}; + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; + +use crate::{ + backends::{Backend, BackendClientError, BackendRegistration}, + config::ConfigError, + instance::{Instance, InstanceClient, InstancePerfSample, ThreadPoolSnapshot}, + path::PrefixedPath, +}; + +const BACKEND_NAME: &str = "fake"; + +fn build_fake_backend(dir: &PrefixedPath) -> Result, BackendClientError> { + Ok(Box::new(FakeBackend::from_config_dir(dir)?)) +} + +#[linkme::distributed_slice(crate::backends::BACKENDS)] +static FAKE_BACKEND: BackendRegistration = BackendRegistration { + name: BACKEND_NAME, + build: build_fake_backend, +}; + +/// `/fake.json`. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct FakeConfig { + /// Directory of `.json` files and `calls.json`. + state_dir: PrefixedPath, +} + +impl FakeConfig { + /// Read `fake.json` when it exists. + fn from_dir(dir: &PrefixedPath) -> Result, ConfigError> { + let path = dir.join("fake.json"); + if !path.exists() { + return Ok(None); + } + + let data = fs::read_to_string(&path)?; + Ok(Some(serde_json::from_str(&data)?)) + } +} + +/// One component-test VM, as stored on disk. +/// +/// Missing fields take [`VmFile::default`]: one thread, one vCPU, and +/// `managed` so the controller may scale the VM. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields, default)] +struct VmFile { + thread_count: u32, + vcpu_count: u32, + per_thread_util: f64, + /// First-snapshot ownership. `true` lets the controller scale the VM. + managed: bool, + fail_snapshot: bool, + read_io_count: u64, + write_io_count: u64, + other_io_count: u64, +} + +impl Default for VmFile { + fn default() -> Self { + Self { + thread_count: 1, + vcpu_count: 1, + per_thread_util: 0.0, + managed: true, + fail_snapshot: false, + read_io_count: 0, + write_io_count: 0, + other_io_count: 0, + } + } +} + +/// Fleet adapter that discovers JSON VM files. +pub struct FakeBackend { + cfg: Option, +} + +impl FakeBackend { + /// Build from `/fake.json`, or an empty inventory when it is absent. + pub fn from_config_dir(dir: &PrefixedPath) -> Result { + Ok(Self { + cfg: FakeConfig::from_dir(dir)?, + }) + } +} + +#[async_trait] +impl Backend for FakeBackend { + fn name(&self) -> &'static str { + BACKEND_NAME + } + + async fn discover(&self) -> Vec> { + let Some(cfg) = &self.cfg else { + return Vec::new(); + }; + discover_vms(&cfg.state_dir) + } + + fn watch_paths(&self) -> Vec { + self.cfg + .as_ref() + .map(|cfg| vec![cfg.state_dir.clone()]) + .unwrap_or_default() + } +} + +/// Load every `.json` file in `state_dir`. +#[tracing::instrument(skip(state_dir), fields(directory = %state_dir))] +fn discover_vms(state_dir: &PrefixedPath) -> Vec> { + let entries = fs::read_dir(state_dir).expect("fake backend state directory is not readable"); + let mut instances = Vec::new(); + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().and_then(|ext| ext.to_str()) != Some("json") { + continue; + } + let Some(id) = path.file_stem().and_then(|stem| stem.to_str()) else { + continue; + }; + let Some(vm) = read_vm(&path, id) else { + continue; + }; + let client = FakeClient { + vm_path: path.clone(), + calls_path: state_dir.join("calls.json").to_path_buf(), + managed: vm.managed, + }; + instances.push(Arc::new(Instance::new( + id.to_string(), + PrefixedPath::new(id), + 0, + client, + ))); + } + instances +} + +/// Read one VM file, or log and return `None` when it cannot be parsed. +#[tracing::instrument(skip(path, id), fields(vm = id))] +fn read_vm(path: &std::path::Path, id: &str) -> Option { + let _guard = LockGuard::acquire(&lock_path_for(path)).ok()?; + VmFile::read(path).ok() +} + +/// Per-VM client. Paths are real filesystem paths, not logical `util::Path`s. +struct FakeClient { + vm_path: std::path::PathBuf, + calls_path: std::path::PathBuf, + managed: bool, +} + +/// `.json` shares `.lock` with the pytest harness. +fn lock_path_for(vm_path: &std::path::Path) -> std::path::PathBuf { + vm_path.with_extension("lock") +} + +/// Exclusive `flock` held until dropped. Matches `fcntl.flock(LOCK_EX)`. +struct LockGuard { + file: File, +} + +impl LockGuard { + /// Get the lock on the given file. Blocks until it gets the lock. + fn acquire(path: &std::path::Path) -> Result { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let file = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(path)?; + file.lock()?; + Ok(Self { file }) + } +} + +impl Drop for LockGuard { + fn drop(&mut self) { + let _ = self.file.unlock(); + } +} + +impl VmFile { + fn read(path: &std::path::Path) -> Result { + let data = fs::read_to_string(path)?; + Ok(serde_json::from_str(&data)?) + } + + fn write_atomic(path: &std::path::Path, vm: &Self) -> Result<(), BackendClientError> { + let tmp = path.with_extension("json.tmp"); + fs::write(&tmp, serde_json::to_vec_pretty(vm)?)?; + fs::rename(&tmp, path)?; + Ok(()) + } +} + +#[async_trait] +impl InstanceClient for FakeClient { + async fn set_thread_count(&self, count: u32) -> Result<(), BackendClientError> { + let _guard = LockGuard::acquire(&lock_path_for(&self.vm_path))?; + let mut vm = VmFile::read(&self.vm_path)?; + vm.thread_count = count; + VmFile::write_atomic(&self.vm_path, &vm)?; + let mut calls = OpenOptions::new() + .create(true) + .append(true) + .open(&self.calls_path)?; + let line = serde_json::json!({ + "vm": self.vm_path.file_stem().and_then(|stem| stem.to_str()).unwrap_or(""), + "threads": count, + }); + writeln!(calls, "{line}")?; + Ok(()) + } + + async fn get_thread_pool_snapshot(&self) -> Result { + let _guard = LockGuard::acquire(&lock_path_for(&self.vm_path))?; + let vm = VmFile::read(&self.vm_path)?; + if vm.fail_snapshot { + return Err(BackendClientError::InvalidState(format!( + "fake snapshot failed for {}", + self.vm_path.display() + ))); + } + Ok(ThreadPoolSnapshot { + thread_count: vm.thread_count, + vcpu_count: vm.vcpu_count, + perf: Some(InstancePerfSample { + read_io_count: vm.read_io_count, + write_io_count: vm.write_io_count, + other_io_count: vm.other_io_count, + read_bytes_total: 0, + write_bytes_total: 0, + ..Default::default() + }), + per_thread_util: Some(vm.per_thread_util), + }) + } + + fn initial_pool_is_managed(&self, _thread_count: u32, _vcpu_count: u32) -> bool { + self.managed + } + + async fn close(&self) {} +} diff --git a/src/backends/mod.rs b/src/backends/mod.rs index 013e72dc..9251ea1d 100644 --- a/src/backends/mod.rs +++ b/src/backends/mod.rs @@ -148,6 +148,8 @@ use std::{ sync::{Arc, LazyLock, Mutex}, }; +#[cfg(feature = "fake-backend")] +pub mod fake; #[cfg(feature = "qemu-backend")] pub mod qemu; From bd32a4337c418896fa0e37c901da7767af6fa212 Mon Sep 17 00:00:00 2001 From: Leonardo Forchini Date: Fri, 23 Oct 2026 09:06:26 +0100 Subject: [PATCH 2/2] Add fake backend to test harness Add a Python component test interface for the FakeBackend. It is able to read the JSON files used by the backend and provides utilities to modify the 'VM's as required by tests. AI-Generated: Most of it Signed-off-by: Leonardo Forchini --- tests/component/harness/fake_backend.py | 135 ++++++++++++++++++++++++ 1 file changed, 135 insertions(+) create mode 100644 tests/component/harness/fake_backend.py diff --git a/tests/component/harness/fake_backend.py b/tests/component/harness/fake_backend.py new file mode 100644 index 00000000..8cc417c0 --- /dev/null +++ b/tests/component/harness/fake_backend.py @@ -0,0 +1,135 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Nutanix, Inc. All rights reserved. +# +# Author: Leonardo Forchini + +import fcntl +import json +import os +from pathlib import Path +from typing import Any + +from harness.common import physical + +DEFAULT_VM = "vm-a" +STATE_DIR = "/var/fake" +OWNERSHIP_FILE = "/run/io-thread-controller/vm-ownership.json" + + +class FakeBackend: + """JSON VM directory the ``fake`` backend discovers.""" + + def __init__(self, root: Path): + """Create a new FakeBackend interface at the given path.""" + self.root = root + physical(root, STATE_DIR).mkdir(parents=True, exist_ok=True) + self.state_dir = physical(root, STATE_DIR) + self.default_vm = DEFAULT_VM + self.add_vm(DEFAULT_VM) + + def add_vm(self, vm_id: str, **fields: Any) -> None: + """Create ``vm_id`` with idle defaults overridden by ``fields``.""" + defaults = { + "thread_count": 1, + "vcpu_count": 1, + "per_thread_util": 0.0, + "managed": True, + "fail_snapshot": False, + "read_io_count": 0, + "write_io_count": 0, + "other_io_count": 0, + } + defaults.update(fields) + self._update(vm_id, **defaults) + + def set_util(self, util: float, vm: str | None = None) -> None: + """Set utilisation for a given VM.""" + self._update(vm or self.default_vm, per_thread_util=util) + + def set_threads(self, count: int, vm: str | None = None) -> None: + """Set thread count for a given VM.""" + self._update(vm or self.default_vm, thread_count=count) + + def set_vcpu_count(self, count: int, vm: str | None = None) -> None: + """Set vCPU cound for a given VM.""" + self._update(vm or self.default_vm, vcpu_count=count) + + def fail_snapshot(self, failed: bool = True, vm: str | None = None) -> None: + """Set whether the next snapshot should fail for the given VM.""" + self._update(vm or self.default_vm, fail_snapshot=failed) + + def set_managed(self, managed: bool, vm: str | None = None) -> None: + """Set whether the given VM is managed.""" + self._update(vm or self.default_vm, managed=managed) + + def set_io_counts( + self, read: int = 0, write: int = 0, other: int = 0, vm: str | None = None + ) -> None: + """Set cumulative read, write, and other I/O counts.""" + self._update( + vm or self.default_vm, + read_io_count=read, + write_io_count=write, + other_io_count=other, + ) + + def prepare_healthy_vm( + self, threads: int, util: float, io: int = 0, vcpu: int = 16 + ) -> None: + """Set the default VM's threads, utilisation, vCPU cap, and read I/O.""" + self.set_threads(threads) + self.set_vcpu_count(vcpu) + self.set_util(util) + self.set_io_counts(io, 0, 0) + + def thread_count(self, vm: str | None = None) -> int: + """Get the number of threads assigned to the given VM.""" + return self._read(vm or self.default_vm)["thread_count"] + + def clear_calls(self) -> None: + """Forget recorded ``set_thread_count`` calls for every VM.""" + path = self.state_dir / "calls.json" + if path.exists(): + path.unlink() + + def calls(self, vm: str | None = None) -> list[int]: + """Thread-count targets ``set_thread_count`` recorded, in order.""" + vm_id = vm or self.default_vm + path = self.state_dir / "calls.json" + if not path.exists(): + return [] + recorded = [] + for line in path.read_text().splitlines(): + if not line.strip(): + continue + try: + item = json.loads(line) + except json.JSONDecodeError: + continue + if item.get("vm") == vm_id: + recorded.append(item["threads"]) + return recorded + + def _vm_path(self, vm_id: str) -> Path: + """Get the path to the JSON file of the given VM.""" + return self.state_dir / f"{vm_id}.json" + + def _read(self, vm_id: str) -> Any: + """Get the contents of the given VM's JSON file.""" + return json.loads(self._vm_path(vm_id).read_text()) + + def _update(self, vm_id: str, **fields: Any) -> None: + """Update the contents of ``vm_id`` with ``fields`` atomically.""" + path = self._vm_path(vm_id) + lock_path = path.with_suffix(".lock") + path.parent.mkdir(parents=True, exist_ok=True) + with open(lock_path, "a+") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + data = {} + if path.exists(): + data = json.loads(path.read_text()) + data.update(fields) + temporary = path.with_suffix(".json.tmp-pytest") + temporary.write_text(json.dumps(data)) + os.replace(temporary, path) + fcntl.flock(lock, fcntl.LOCK_UN)