diff --git a/examples/fake_qemu_backend.rs b/examples/fake_qemu_backend.rs new file mode 100644 index 0000000..e20dd9a --- /dev/null +++ b/examples/fake_qemu_backend.rs @@ -0,0 +1,323 @@ +// SPDX-License-Identifier: BSD-3-Clause +// Copyright (c) 2026 Nutanix, Inc. All rights reserved. +// +// Author: Thanos Makatos + +//! End-to-end example: an io-thread-controller binary built by +//! *depending on the crate as a library* that plugs in a fake +//! QEMU-shaped backend for component testing. +//! +//! The point of the fake-QEMU backend is to exercise the +//! *backend-supplied ``per_thread_util``* path (the QEMU +//! backend reads per-iothread jiffies itself and hands the +//! controller a precomputed number, so the /proc walker skips) +//! without depending on a real libvirt+QEMU pair. +//! +//! Layout the example demonstrates: +//! +//! * `FakeQemuConfig` and `FakeQemuBackend` live entirely in this file, i.e. +//! in the consumer crate. Nothing about them touches +//! `io-thread-controller`'s source tree. +//! * The consumer implements `InstanceClient` for the wire client and +//! `Backend` for the backend handle -- exactly the trait objects the +//! shipped controller already speaks. +//! * `main` uses the shipped [`ControllerBuilder`] / [`run`] surface: parse +//! config, materialise the custom backend, hand everything to the daemon. +//! +//! The companion pytest fixture spawns `fake_qemu.py` (a Python +//! process backing a UNIX socket + HTTP admin surface), then +//! runs this example against the socket. The example connects +//! to that socket and drives the fake service's JSON-lines +//! protocol -- which is what proves both the library plug-in +//! path *and* the backend-supplied util path in the controller. + +use std::sync::Arc; + +use async_trait::async_trait; +use clap::Parser; +use serde::{Deserialize, Serialize}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + net::UnixStream, + sync::Mutex, +}; + +use io_thread_controller::{ + backends::Backend, + backends::BackendClientError, + config::{Config, load_config, validate_config}, + daemon::run, + instance::{Instance, InstanceClient, InstancePerfSample, ThreadPoolSnapshot}, + util::Path, +}; + +// --------------------------------------------------------------- +// Backend-specific config +// --------------------------------------------------------------- +// +// Loaded from ``/fake_qemu.json``. Same +// shape (``from_dir`` helper + serde defaults) as the shipped +// QEMU backend so this example mirrors the in-tree +// convention verbatim. + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct FakeQemuConfig { + /// UNIX socket the fake QEMU process listens on. The + /// example spins up one tracked instance per config file; + /// discovery reports it exactly once per tick. + #[serde(default = "default_socket")] + socket_path: Path, + /// Stable id used verbatim as ``Instance::id`` and every + /// log line's ``vm=`` field. Matches the shape of a + /// QEMU vm UUID in real life. + #[serde(default = "default_vm_id")] + vm_id: String, +} + +fn default_socket() -> Path { + Path::new("/tmp/fake_qemu.sock") +} + +fn default_vm_id() -> String { + "vm-fake-qemu".to_string() +} + +impl Default for FakeQemuConfig { + fn default() -> Self { + Self { + socket_path: default_socket(), + vm_id: default_vm_id(), + } + } +} + +impl FakeQemuConfig { + fn from_dir(dir: &Path) -> Self { + let path = dir.join("fake_qemu.json"); + if !path.exists() { + return Self::default(); + } + let body = std::fs::read_to_string(&path).expect("read fake QEMU config"); + serde_json::from_str(&body).expect("deserialize fake QEMU config") + } +} + +// --------------------------------------------------------------- +// InstanceClient: wire client +// --------------------------------------------------------------- + +/// Wire client for one fake-QEMU instance. Owns the connection +/// to the UNIX socket and speaks the JSON-lines protocol +/// (``get_snapshot`` / ``set_thread_count``) with the Python +/// fake. Lazily dials the socket on the first call so a +/// pre-fake-startup instantiation still works. +struct FakeQemuClient { + socket_path: Path, + stream: Mutex>>, +} + +impl FakeQemuClient { + fn new(socket_path: Path) -> Self { + Self { + socket_path, + stream: Mutex::new(None), + } + } + + /// Return an initialised buffered stream, dialing on first + /// use and reconnecting on any transport error the caller + /// observed on the previous round-trip. + async fn connect( + &self, + ) -> Result>>, BackendClientError> + { + let mut guard = self.stream.lock().await; + if guard.is_none() { + let s = UnixStream::connect(&self.socket_path) + .await + .map_err(|e| BackendClientError::Transport(e.to_string()))?; + *guard = Some(BufReader::new(s)); + } + Ok(guard) + } + + /// Send one JSON request line and read one JSON reply line. + /// Drops the cached stream on any transport error so the + /// next call reconnects fresh. + async fn round_trip( + &self, + req: &serde_json::Value, + ) -> Result { + let mut guard = self.connect().await?; + let stream = guard.as_mut().expect("stream just initialised"); + let mut req_line = serde_json::to_vec(req).map_err(BackendClientError::from)?; + req_line.push(b'\n'); + if let Err(e) = stream.get_mut().write_all(&req_line).await { + *guard = None; + return Err(BackendClientError::Transport(e.to_string())); + } + let mut reply = String::new(); + match stream.read_line(&mut reply).await { + Ok(0) => { + *guard = None; + Err(BackendClientError::Transport("fake_qemu closed".into())) + } + Ok(_) => serde_json::from_str(reply.trim()).map_err(BackendClientError::from), + Err(e) => { + *guard = None; + Err(BackendClientError::Transport(e.to_string())) + } + } + } +} + +#[derive(Debug, Deserialize)] +struct SnapshotReply { + num_threads: u32, + per_thread_util: f64, + read_ios: u64, + write_ios: u64, +} + +#[async_trait] +impl InstanceClient for FakeQemuClient { + async fn set_thread_count(&self, n: u32) -> Result<(), BackendClientError> { + let req = serde_json::json!({ + "cmd": "set_thread_count", + "value": n, + }); + self.round_trip(&req).await?; + Ok(()) + } + + async fn get_thread_pool_snapshot(&self) -> Result { + let req = serde_json::json!({"cmd": "get_snapshot"}); + let v = self.round_trip(&req).await?; + let snap: SnapshotReply = serde_json::from_value(v).map_err(BackendClientError::from)?; + Ok(ThreadPoolSnapshot { + thread_count: snap.num_threads, // FIXME snap.num_threads should be snap.thread_count + perf: Some(InstancePerfSample { + read_io_count: snap.read_ios, + write_io_count: snap.write_ios, + ..Default::default() + }), + // The whole point of fake QEMU: hand the controller + // a backend-computed per-thread util so the /proc + // walker is skipped. Mirrors the real QEMU + // backend's ``client::QemuInstanceClient`` semantics. + per_thread_util: Some(snap.per_thread_util), + vcpu_count: 16, + }) + } + + async fn close(&self) { + let mut guard = self.stream.lock().await; + *guard = None; + } +} + +// --------------------------------------------------------------- +// Backend: discovery glue +// --------------------------------------------------------------- + +/// Backend. Holds the config and materialises exactly +/// one [`Instance`] per discovery pass (the fake QEMU is a +/// singleton in every test). We deliberately keep the +/// discovery contract identical to the shipped QEMU backend so +/// the pytest fixture can swap this example in front of the +/// real backend if we ever want an integration test against a +/// live libvirtd. +struct FakeQemuBackend { + cfg: FakeQemuConfig, +} + +impl FakeQemuBackend { + fn from_config_dir(dir: &Path) -> Self { + Self { + cfg: FakeQemuConfig::from_dir(dir), + } + } +} + +#[async_trait] +impl Backend for FakeQemuBackend { + fn name(&self) -> &'static str { + "fake_qemu" + } + + async fn discover(&self) -> Vec> { + // The fake QEMU is always "there" from the controller's + // point of view: a real libvirt/QEMU discovery would + // enumerate active vms here, but for the test we + // hand out one fixed instance. + let client: Box = + Box::new(FakeQemuClient::new(self.cfg.socket_path.clone())); + vec![Arc::new(Instance::new( + self.cfg.vm_id.clone(), + self.cfg.socket_path.clone(), + // Fake QEMU deliberately sets pid=0 so the /proc + // walker is skipped -- same contract as the real + // backend, which does not know the QEMU pid until + // after a QMP round-trip. + 0, + client, + ))] + } + + fn watch_paths(&self) -> Vec { + // Fake QEMU is a persistent singleton; no filesystem + // churn to watch for. + Vec::new() + } +} + +// --------------------------------------------------------------- +// Binary entrypoint +// --------------------------------------------------------------- + +#[derive(Debug, Parser)] +#[command( + name = "fake-qemu-backend", + about = "io-thread-controller variant with a Python-driven fake QEMU \ + backend plugged in via the library API. \ + Used exclusively by tests/component/test_fake_qemu_backend.py." +)] +struct Cli { + #[arg(long, default_value = "/etc/io-thread-controller/config.json")] + config: Path, + + #[arg(long, env = "RUST_LOG", default_value = "info")] + log: String, +} + +fn init_logging(filter: &str) { + let subscriber = tracing_subscriber::fmt() + .with_target(true) + .with_env_filter( + tracing_subscriber::EnvFilter::try_new(filter) + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .finish(); + let _ = tracing::subscriber::set_global_default(subscriber); +} + +#[tokio::main] +async fn main() { + let cli = Cli::parse(); + init_logging(&cli.log); + + let cfg: Config = load_config(&cli.config).expect("load config"); + validate_config(&cfg).expect("validate config"); + + let backend = FakeQemuBackend::from_config_dir(&cfg.backend_config_dir); + tracing::info!( + target: "fake-qemu-backend", + socket = %backend.cfg.socket_path.display(), + vm = %backend.cfg.vm_id, + "fake QEMU backend registered" + ); + + run(cfg, vec![Box::new(backend)]) + .await + .expect("run io-thread-controller daemon"); +}