From 24bd8d5b288110c9c77106a1db9a44cd3e32f3ec 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 + src/backends/fake/mod.rs | 289 +++++++++++++++++++++++++++++++++++++++ src/backends/mod.rs | 2 + 3 files changed, 293 insertions(+) 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/src/backends/fake/mod.rs b/src/backends/fake/mod.rs new file mode 100644 index 00000000..65568a6e --- /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}, + util::Path, +}; + +const BACKEND_NAME: &str = "fake"; + +fn build_fake_backend(dir: &Path) -> 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: Path, +} + +impl FakeConfig { + /// Read `fake.json` when it exists. + fn from_dir(dir: &Path) -> 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: &Path) -> 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: &Path) -> 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"), + managed: vm.managed, + }; + instances.push(Arc::new(Instance::new( + id.to_string(), + Path::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 c63f6fce..07c40b86 100644 --- a/src/backends/mod.rs +++ b/src/backends/mod.rs @@ -142,6 +142,8 @@ fn is_peer_disconnect_kind(kind: ErrorKind) -> bool { use std::{io, sync::Arc}; +#[cfg(feature = "fake-backend")] +pub mod fake; #[cfg(feature = "qemu-backend")] pub mod qemu; From a451ed107b8dc1335c7d5d4ec86e8df5b145189e 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 AI-Generated: Most of it Signed-off-by: Leonardo Forchini --- tests/component/harness/fake_backend.py | 121 ++++++++++++++++++++++++ 1 file changed, 121 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..f6cbafad --- /dev/null +++ b/tests/component/harness/fake_backend.py @@ -0,0 +1,121 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Nutanix, Inc. All rights reserved. +# +# Author: Leonardo Forchini + +import fcntl +import json +import os + +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): + 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, **fields): + """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) + return vm_id + + def set_util(self, util, vm=None): + self._update(vm or self.default_vm, per_thread_util=util) + + def set_threads(self, count, vm=None): + self._update(vm or self.default_vm, thread_count=count) + + def set_vcpu_count(self, count, vm=None): + self._update(vm or self.default_vm, vcpu_count=count) + + def fail_snapshot(self, failed=True, vm=None): + self._update(vm or self.default_vm, fail_snapshot=failed) + + def set_managed(self, managed, vm=None): + self._update(vm or self.default_vm, managed=managed) + + def set_io_counts(self, read=0, write=0, other=0, vm=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, util, io=0, vcpu=16): + """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=None): + return self._read(vm or self.default_vm)["thread_count"] + + def clear_calls(self): + """Forget recorded ``set_thread_count`` calls for every VM.""" + path = self.state_dir / "calls.json" + if path.exists(): + path.unlink() + + def calls(self, vm=None): + """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): + return self.state_dir / ("%s.json" % vm_id) + + def _read(self, vm_id): + return json.loads(self._vm_path(vm_id).read_text()) + + def _update(self, vm_id, **fields): + """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)