From a6cb176ca5680dcfc2a224be6267e72105eaef44 Mon Sep 17 00:00:00 2001 From: Nikolay Govorov Date: Sun, 12 Jul 2026 22:05:12 +0000 Subject: WIP: QMP, QGA, exec, and serial console --- crates/hule-vmm/Cargo.toml | 5 + crates/hule-vmm/src/backend/qemu/cli.rs | 88 +++++++ crates/hule-vmm/src/backend/qemu/mod.rs | 237 +++++++++++++++--- crates/hule-vmm/src/backend/qemu/qga.rs | 233 +++++++++++++++++ crates/hule-vmm/src/backend/qemu/qmp.rs | 87 +++++++ crates/hule-vmm/src/backend/qemu/supervise.rs | 50 ++++ crates/hule-vmm/src/machine.rs | 6 +- crates/hule-vmm/tests/qemu_lifecycle.rs | 136 ++++++++++ 8 files changed, 809 insertions(+), 33 deletions(-) create mode 100644 crates/hule-vmm/src/backend/qemu/qga.rs create mode 100644 crates/hule-vmm/src/backend/qemu/qmp.rs create mode 100644 crates/hule-vmm/tests/qemu_lifecycle.rs diff --git a/crates/hule-vmm/Cargo.toml b/crates/hule-vmm/Cargo.toml index ed32410..8cebbae 100644 --- a/crates/hule-vmm/Cargo.toml +++ b/crates/hule-vmm/Cargo.toml @@ -17,3 +17,8 @@ hule-image.workspace = true uuid = { version = "1", features = ["v7"] } tokio = { workspace = true, features = ["process", "fs", "io-util", "sync", "net", "time"] } async-trait = "0.1" +serde_json.workspace = true +base64 = "0.22" + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/hule-vmm/src/backend/qemu/cli.rs b/crates/hule-vmm/src/backend/qemu/cli.rs index b17f8e9..1612d13 100644 --- a/crates/hule-vmm/src/backend/qemu/cli.rs +++ b/crates/hule-vmm/src/backend/qemu/cli.rs @@ -114,6 +114,8 @@ pub enum Device { VirtioBlkPci { drive: String }, VirtioRngPci, VirtioBalloon, + VirtioSerialPci, + VirtSerialPort { chardev: String, name: String }, } impl Device { @@ -122,6 +124,10 @@ impl Device { Self::VirtioBlkPci { drive } => format!("virtio-blk-pci,drive={drive}"), Self::VirtioRngPci => "virtio-rng-pci".into(), Self::VirtioBalloon => "virtio-balloon".into(), + Self::VirtioSerialPci => "virtio-serial-pci".into(), + Self::VirtSerialPort { chardev, name } => { + format!("virtserialport,chardev={chardev},name={name}") + } } } } @@ -147,6 +153,15 @@ pub struct Command { /// `-pidfile`: qemu writes its own pid here. The only reliable way to /// signal/check it later from a process that isn't its parent. pub pidfile: Option, + /// `-qmp tcp:127.0.0.1:,server=on,wait=off`: the control channel. + /// TCP loopback, not a unix socket -- the latter doesn't exist on + /// Windows at all (`tokio::net::UnixStream` is unix-only), and this way + /// there's nothing to `#[cfg]` on the transport. + pub qmp_port: Option, + /// Host side of the qemu-guest-agent virtio-serial channel. + pub qga_port: Option, + /// Interactive guest serial console and its persistent output log. + pub console: Option<(u16, PathBuf)>, } impl Command { @@ -187,6 +202,37 @@ impl Command { if let Some(pidfile) = &self.pidfile { argv.extend(["-pidfile".into(), pidfile.display().to_string()]); } + if let Some(port) = self.qmp_port { + argv.extend([ + "-qmp".into(), + format!("tcp:127.0.0.1:{port},server=on,wait=off"), + ]); + } + if let Some(port) = self.qga_port { + argv.extend([ + "-chardev".into(), + format!("socket,id=qga0,host=127.0.0.1,port={port},server=on,wait=off"), + "-device".into(), + Device::VirtioSerialPci.to_arg(), + "-device".into(), + Device::VirtSerialPort { + chardev: "qga0".into(), + name: "org.qemu.guest_agent.0".into(), + } + .to_arg(), + ]); + } + if let Some((port, logfile)) = &self.console { + argv.extend([ + "-chardev".into(), + format!( + "socket,id=console0,host=127.0.0.1,port={port},server=on,wait=off,logfile={},logappend=off", + logfile.display() + ), + "-serial".into(), + "chardev:console0".into(), + ]); + } argv } } @@ -297,4 +343,46 @@ mod tests { .any(|w| w == ["-device", "virtio-blk-pci,drive=root"]) ); } + + #[test] + fn qga_uses_a_dedicated_virtio_serial_channel() { + let argv = Command { + binary: "qemu-system-x86_64".into(), + qga_port: Some(1234), + ..Default::default() + } + .to_argv(); + assert!(argv.windows(2).any(|w| { + w == [ + "-chardev", + "socket,id=qga0,host=127.0.0.1,port=1234,server=on,wait=off", + ] + })); + assert!(argv.windows(2).any(|w| { + w == [ + "-device", + "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0", + ] + })); + } + + #[test] + fn serial_console_has_an_attach_socket_and_log() { + let argv = Command { + binary: "qemu-system-x86_64".into(), + console: Some((4321, "console.log".into())), + ..Default::default() + } + .to_argv(); + assert!(argv.windows(2).any(|w| { + w == [ + "-chardev", + "socket,id=console0,host=127.0.0.1,port=4321,server=on,wait=off,logfile=console.log,logappend=off", + ] + })); + assert!( + argv.windows(2) + .any(|w| w == ["-serial", "chardev:console0"]) + ); + } } diff --git a/crates/hule-vmm/src/backend/qemu/mod.rs b/crates/hule-vmm/src/backend/qemu/mod.rs index 0ae4cd8..9feede2 100644 --- a/crates/hule-vmm/src/backend/qemu/mod.rs +++ b/crates/hule-vmm/src/backend/qemu/mod.rs @@ -1,17 +1,20 @@ // SPDX-FileCopyrightText: 2026 Nikolay Govorov // SPDX-License-Identifier: Apache-2.0 -//! qemu backend. Basic extraction from the old `hule` binary's boot code: -//! `start`/`kill`/`wait`/`state`/`delete` work, everything needing QMP or a -//! console channel (`stop`, `restart`, `pause`, `exec`, `attach`, ...) is -//! `todo!()` -- QMP is opportunistic, for operations signals can't do, not -//! load-bearing for the base lifecycle. +//! QMP is an unconditional implementation detail of this backend. Guest +//! access is independent: an image may be a black box, or may promise QGA; +//! a promised QGA channel must become ready for `start()` to succeed. +//! +//! `logs`/`attach` use a dedicated serial console; `exec` uses QGA. +//! Console resizing and `reattach` are not implemented yet. //! //! `cli` is the typed qemu CLI wrapper; `supervise` gets an exit code out of //! a process even if we stop being its parent; this module only translates //! Hule's image/machine types into both. mod cli; +mod qga; +mod qmp; mod supervise; use std::path::{Path, PathBuf}; @@ -19,7 +22,7 @@ use std::process::{ExitStatus, Output}; use std::time::Duration; use async_trait::async_trait; -use hule_image::{Architecture, Boot, BootLinux, Disk, MachineImage}; +use hule_image::{Access, Architecture, Boot, BootLinux, Disk, MachineImage}; use tokio::process::Child; use tokio::sync::Mutex; @@ -53,7 +56,10 @@ pub struct QemuMachine { settings: Settings, disk: Disk, boot: QemuBoot, + qga: bool, child: Mutex, + qga_lock: Mutex<()>, + qmp_client: Mutex>, } impl QemuMachine { @@ -68,6 +74,81 @@ impl QemuMachine { self.dir.join("qemu.exit-code") } + /// Where the port `start()` picked for `-qmp tcp:127.0.0.1:` is + /// recorded -- chosen by us, not qemu, so (unlike `pid_path`) nothing + /// needs to wait for this file; it exists before qemu is spawned. + fn qmp_port_path(&self) -> PathBuf { + self.dir.join("qemu.qmp-port") + } + + fn qga_port_path(&self) -> PathBuf { + self.dir.join("qemu.qga-port") + } + + fn console_port_path(&self) -> PathBuf { + self.dir.join("qemu.console-port") + } + + fn console_log_path(&self) -> PathBuf { + self.dir.join("console.log") + } + + async fn read_port(&self, path: &Path, name: &str) -> R { + let data = tokio::fs::read_to_string(path).await?; + data.trim() + .parse() + .map_err(|_| Error::InvalidState(format!("invalid {name} port contents"))) + } + + async fn qmp_port(&self) -> R { + self.read_port(&self.qmp_port_path(), "qmp").await + } + + async fn qga_port(&self) -> R { + self.read_port(&self.qga_port_path(), "qga").await + } + + async fn console_port(&self) -> R { + self.read_port(&self.console_port_path(), "console").await + } + + async fn qmp_execute(&self, command: &str) -> R { + self.qmp_execute_with_args(command, None).await + } + + /// Connects on first use (there's a short window after spawn before + /// qemu's TCP listener is actually up, so a couple of retries) and + /// reuses the connection after that. + async fn qmp_execute_with_args( + &self, + command: &str, + arguments: Option, + ) -> R { + let mut client = self.qmp_client.lock().await; + if client.is_none() { + let port = self.qmp_port().await?; + let mut connected = None; + for _ in 0..50 { + match qmp::Qmp::connect(port).await { + Ok(qmp) => { + connected = Some(qmp); + break; + } + Err(_) => tokio::time::sleep(Duration::from_millis(20)).await, + } + } + *client = Some( + connected + .ok_or_else(|| Error::InvalidState("qemu's qmp port never came up".into()))?, + ); + } + client + .as_mut() + .expect("just set above") + .execute(command, arguments) + .await + } + /// Reads qemu's pid, waiting briefly for `-pidfile` to actually appear /// (there's a short window between spawn and qemu writing it). async fn qemu_pid(&self) -> R { @@ -127,23 +208,43 @@ impl Machine for QemuMachine { if let Some(status) = self.read_exit_code().await { return State::Exited(status.code().unwrap_or(-1)); } - match self.qemu_pid().await { - Ok(pid) if supervise::pid_alive(pid).await => State::Running, - Ok(_) => State::Dead, - Err(_) => match &*self.child.lock().await { - ChildState::NotStarted => State::Created, - ChildState::Exited(status) => State::Exited(status.code().unwrap_or(-1)), - ChildState::Running(_) => State::Running, - }, + let pid = match self.qemu_pid().await { + Ok(pid) => pid, + Err(_) => { + return match &*self.child.lock().await { + ChildState::NotStarted => State::Created, + ChildState::Exited(status) => State::Exited(status.code().unwrap_or(-1)), + ChildState::Running(_) => State::Running, + }; + } + }; + if !supervise::pid_alive(pid).await { + return State::Dead; + } + // A live pid alone can't distinguish running from paused; ask QMP. + if let Ok(status) = self.qmp_execute("query-status").await + && status.get("status").and_then(|s| s.as_str()) == Some("paused") + { + return State::Paused; } + State::Running } async fn stats(&self) -> R { - todo!("needs /proc or QMP query-blockstats wiring") + let pid = self.qemu_pid().await?; + let (cpu_time_ns, memory_bytes) = supervise::process_stats(pid).await?; + Ok(Stats { + cpu_time_ns, + memory_bytes, + }) } async fn logs(&self) -> R> { - todo!("needs a captured serial console, not wired up yet") + match tokio::fs::read(self.console_log_path()).await { + Ok(logs) => Ok(logs), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()), + Err(error) => Err(error.into()), + } } async fn start(&mut self) -> R<()> { @@ -153,6 +254,29 @@ impl Machine for QemuMachine { } let _ = tokio::fs::remove_file(self.pid_path()).await; let _ = tokio::fs::remove_file(self.exit_code_path()).await; + let _ = tokio::fs::remove_file(self.qmp_port_path()).await; + let _ = tokio::fs::remove_file(self.qga_port_path()).await; + let _ = tokio::fs::remove_file(self.console_port_path()).await; + let _ = tokio::fs::remove_file(self.console_log_path()).await; + *self.qmp_client.lock().await = None; + + let qmp_port = qmp::free_port().await?; + tokio::fs::write(self.qmp_port_path(), qmp_port.to_string()).await?; + let qga_port = if self.qga { + let mut port = qmp::free_port().await?; + while port == qmp_port { + port = qmp::free_port().await?; + } + tokio::fs::write(self.qga_port_path(), port.to_string()).await?; + Some(port) + } else { + None + }; + let mut console_port = qmp::free_port().await?; + while console_port == qmp_port || Some(console_port) == qga_port { + console_port = qmp::free_port().await?; + } + tokio::fs::write(self.console_port_path(), console_port.to_string()).await?; let kvm = Architecture::host() == Some(self.arch) && cli::kvm_available().await; let sys_arch = match self.arch { @@ -213,15 +337,27 @@ impl Machine for QemuMachine { } } cmd.pidfile = Some(self.pid_path()); + cmd.qmp_port = Some(qmp_port); + cmd.qga_port = qga_port; + cmd.console = Some((console_port, self.console_log_path())); let argv = cmd.to_argv(); let wrapper = supervise::spawn(&argv, &self.exit_code_path())?; *child = ChildState::Running(wrapper); + drop(child); + + if let Some(port) = qga_port + && let Err(error) = qga::wait_until_ready(port, Duration::from_secs(60)).await + { + let _ = self.kill().await; + return Err(error); + } Ok(()) } async fn stop(&mut self) -> R<()> { - todo!("needs QMP system_powerdown, not wired up yet") + self.qmp_execute("system_powerdown").await?; + Ok(()) } async fn kill(&mut self) -> R<()> { @@ -240,7 +376,8 @@ impl Machine for QemuMachine { } async fn restart(&mut self) -> R<()> { - todo!("needs QMP system_reset, not wired up yet") + self.qmp_execute("system_reset").await?; + Ok(()) } async fn delete(&mut self) -> R<()> { @@ -249,19 +386,43 @@ impl Machine for QemuMachine { } let _ = tokio::fs::remove_file(self.pid_path()).await; let _ = tokio::fs::remove_file(self.exit_code_path()).await; + let _ = tokio::fs::remove_file(self.qmp_port_path()).await; + let _ = tokio::fs::remove_file(self.qga_port_path()).await; + let _ = tokio::fs::remove_file(self.console_port_path()).await; + let _ = tokio::fs::remove_file(self.console_log_path()).await; Ok(()) } async fn pause(&mut self) -> R<()> { - todo!("needs QMP stop, not wired up yet") + self.qmp_execute("stop").await?; + Ok(()) } async fn unpause(&mut self) -> R<()> { - todo!("needs QMP cont, not wired up yet") + self.qmp_execute("cont").await?; + Ok(()) } - async fn update(&mut self, _settings: Settings) -> R<()> { - todo!("needs a diff against current settings and QMP/hotplug to apply it") + async fn update(&mut self, settings: Settings) -> R<()> { + if settings.cpu != self.settings.cpu { + return Err(Error::Unsupported( + "qemu backend can't hot-change cpu count (needs maxcpus reserved at boot)".into(), + )); + } + if settings.port_forwards != self.settings.port_forwards { + return Err(Error::Unsupported( + "qemu backend can't hot-change port forwards (fixed at boot via -net user)".into(), + )); + } + if settings.ram != self.settings.ram { + self.qmp_execute_with_args( + "balloon", + Some(serde_json::json!({ "value": settings.ram * 1024 * 1024 })), + ) + .await?; + } + self.settings = settings; + Ok(()) } async fn rename(&mut self, name: Option<&str>) -> R<()> { @@ -290,16 +451,27 @@ impl Machine for QemuMachine { } } - async fn exec(&self, _cmd: &[String]) -> R { - todo!("needs the access channel (ssh today) wired up here") + async fn exec(&self, cmd: &[String]) -> R { + if !self.qga { + return Err(Error::Unsupported( + "this image does not declare qga access".into(), + )); + } + let _guard = self.qga_lock.lock().await; + qga::exec(self.qga_port().await?, cmd).await } async fn attach(&self) -> R> { - todo!("needs a serial console channel, not wired up yet") - } - - async fn resize(&self, _cols: u16, _rows: u16) -> R<()> { - todo!("depends on attach") + let port = self.console_port().await?; + for _ in 0..50 { + match tokio::net::TcpStream::connect(("127.0.0.1", port)).await { + Ok(stream) => return Ok(Box::new(stream)), + Err(_) => tokio::time::sleep(Duration::from_millis(20)).await, + } + } + Err(Error::InvalidState( + "qemu serial console did not become available".into(), + )) } } @@ -359,7 +531,14 @@ impl Hypervisor for QemuHypervisor { settings: settings.clone(), disk, boot: qemu_boot, + qga: image + .machine + .access + .iter() + .any(|access| matches!(access, Access::Qga { .. })), child: Mutex::new(ChildState::NotStarted), + qga_lock: Mutex::new(()), + qmp_client: Mutex::new(None), })) } diff --git a/crates/hule-vmm/src/backend/qemu/qga.rs b/crates/hule-vmm/src/backend/qemu/qga.rs new file mode 100644 index 0000000..a9a57f7 --- /dev/null +++ b/crates/hule-vmm/src/backend/qemu/qga.rs @@ -0,0 +1,233 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: Apache-2.0 + +//! Minimal qemu-guest-agent client over qemu's host-side chardev socket. + +use std::time::{Duration, Instant}; + +use base64::Engine; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::TcpStream; + +use crate::{Error, R}; + +async fn execute( + port: u16, + command: &str, + arguments: Option, +) -> R { + tokio::time::timeout( + Duration::from_secs(5), + execute_without_timeout(port, command, arguments), + ) + .await + .map_err(|_| Error::InvalidState(format!("qga {command} timed out")))? +} + +async fn execute_without_timeout( + port: u16, + command: &str, + arguments: Option, +) -> R { + let stream = TcpStream::connect(("127.0.0.1", port)).await?; + let mut stream = BufReader::new(stream); + let mut request = serde_json::json!({ "execute": command }); + if let Some(arguments) = arguments { + request["arguments"] = arguments; + } + stream + .get_mut() + .write_all( + format!( + "{}\n", + serde_json::to_string(&request) + .map_err(|error| Error::InvalidState(error.to_string()))? + ) + .as_bytes(), + ) + .await?; + + let mut response = String::new(); + stream.read_line(&mut response).await?; + if response.is_empty() { + return Err(Error::InvalidState( + "qemu-guest-agent closed its channel without replying".into(), + )); + } + validate_response(command, &response) +} + +fn validate_response(command: &str, response: &str) -> R { + let response: serde_json::Value = serde_json::from_str(response) + .map_err(|error| Error::InvalidState(format!("invalid qga response: {error}")))?; + if let Some(error) = response.get("error") { + return Err(Error::InvalidState(format!( + "qga {command} failed: {error}" + ))); + } + response + .get("return") + .cloned() + .ok_or_else(|| Error::InvalidState(format!("qga {command} response has no return value"))) +} + +async fn ping(port: u16) -> R<()> { + execute(port, "guest-ping", None).await.map(|_| ()) +} + +pub async fn wait_until_ready(port: u16, timeout: Duration) -> R<()> { + let deadline = Instant::now() + timeout; + loop { + let attempt = tokio::time::timeout(Duration::from_secs(1), ping(port)).await; + if matches!(attempt, Ok(Ok(()))) { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(Error::InvalidState(format!( + "declared qga access did not become ready within {} seconds", + timeout.as_secs() + ))); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +pub async fn exec(port: u16, command: &[String]) -> R { + let (path, args) = command + .split_first() + .ok_or_else(|| Error::InvalidState("exec command must not be empty".into()))?; + let started = execute( + port, + "guest-exec", + Some(serde_json::json!({ + "path": path, + "arg": args, + "capture-output": true + })), + ) + .await?; + let pid = started + .get("pid") + .and_then(serde_json::Value::as_u64) + .ok_or_else(|| Error::InvalidState("qga guest-exec returned no pid".into()))?; + + loop { + let status = execute( + port, + "guest-exec-status", + Some(serde_json::json!({ "pid": pid })), + ) + .await?; + if !status + .get("exited") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + { + tokio::time::sleep(Duration::from_millis(20)).await; + continue; + } + + return output(&status); + } +} + +fn output(status: &serde_json::Value) -> R { + for field in ["out-truncated", "err-truncated"] { + if status + .get(field) + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + { + return Err(Error::InvalidState(format!( + "qga exec output was truncated ({field})" + ))); + } + } + let decode = |field: &str| -> R> { + let Some(encoded) = status.get(field).and_then(serde_json::Value::as_str) else { + return Ok(Vec::new()); + }; + base64::engine::general_purpose::STANDARD + .decode(encoded) + .map_err(|error| Error::InvalidState(format!("invalid qga {field}: {error}"))) + }; + Ok(std::process::Output { + status: exit_status(status)?, + stdout: decode("out-data")?, + stderr: decode("err-data")?, + }) +} + +fn exit_status(status: &serde_json::Value) -> R { + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + if let Some(code) = status.get("exitcode").and_then(serde_json::Value::as_i64) { + let code: i32 = code + .try_into() + .map_err(|_| Error::InvalidState("qga returned an invalid exit code".into()))?; + return Ok(std::process::ExitStatus::from_raw(code << 8)); + } + if let Some(signal) = status.get("signal").and_then(serde_json::Value::as_i64) { + let signal: i32 = signal + .try_into() + .map_err(|_| Error::InvalidState("qga returned an invalid signal".into()))?; + return Ok(std::process::ExitStatus::from_raw(signal & 0x7f)); + } + Err(Error::InvalidState( + "qga exec status has neither exitcode nor signal".into(), + )) + } + #[cfg(windows)] + { + use std::os::windows::process::ExitStatusExt; + let code = status + .get("exitcode") + .and_then(serde_json::Value::as_i64) + .ok_or_else(|| Error::InvalidState("qga exec status has no exitcode".into()))?; + let code: u32 = code + .try_into() + .map_err(|_| Error::InvalidState("qga returned an invalid exit code".into()))?; + Ok(std::process::ExitStatus::from_raw(code)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn readiness_requires_a_successful_guest_ping_response() { + validate_response("guest-ping", r#"{"return":{}}"#).unwrap(); + assert!( + validate_response("guest-ping", r#"{"error":{"class":"CommandNotFound"}}"#).is_err() + ); + assert!(validate_response("guest-ping", r#"{"event":"something"}"#).is_err()); + } + + #[test] + fn exec_output_is_decoded_without_losing_exit_status() { + let output = output(&serde_json::json!({ + "exited": true, + "exitcode": 7, + "out-data": "aGVsbG8K", + "err-data": "b29wcwo=" + })) + .unwrap(); + assert_eq!(output.status.code(), Some(7)); + assert_eq!(output.stdout, b"hello\n"); + assert_eq!(output.stderr, b"oops\n"); + } + + #[test] + fn truncated_exec_output_is_an_error() { + assert!( + output(&serde_json::json!({ + "exited": true, + "exitcode": 0, + "out-truncated": true + })) + .is_err() + ); + } +} diff --git a/crates/hule-vmm/src/backend/qemu/qmp.rs b/crates/hule-vmm/src/backend/qemu/qmp.rs new file mode 100644 index 0000000..017f4fc --- /dev/null +++ b/crates/hule-vmm/src/backend/qemu/qmp.rs @@ -0,0 +1,87 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: Apache-2.0 + +//! Minimal QMP client: line-delimited JSON commands/responses over the TCP +//! loopback channel qemu exposes via +//! `-qmp tcp:127.0.0.1:,server=on,wait=off`. Events are read and +//! discarded, not surfaced. + +use serde_json::{Value, json}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::TcpStream; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; + +use crate::{Error, R}; + +pub struct Qmp { + reader: BufReader, + writer: OwnedWriteHalf, +} + +impl Qmp { + /// Connects and completes the capabilities handshake (greeting -> + /// `qmp_capabilities`), after which arbitrary commands are allowed. + pub async fn connect(port: u16) -> R { + let (read_half, write_half) = TcpStream::connect(("127.0.0.1", port)).await?.into_split(); + let mut qmp = Self { + reader: BufReader::new(read_half), + writer: write_half, + }; + qmp.read_message().await?; // greeting + qmp.execute("qmp_capabilities", None).await?; + Ok(qmp) + } + + async fn read_message(&mut self) -> R { + loop { + let mut line = String::new(); + if self.reader.read_line(&mut line).await? == 0 { + return Err(Error::InvalidState("qmp connection closed".into())); + } + + let line = line.trim(); + if line.is_empty() { + continue; + } + + let value: Value = serde_json::from_str(line) + .map_err(|e| Error::InvalidState(format!("invalid qmp message: {e}")))?; + + // Events arrive unprompted between a command and its response; + // we don't surface them (yet), so skip past them here. + if value.get("event").is_some() { + continue; + } + + return Ok(value); + } + } + + pub async fn execute(&mut self, command: &str, arguments: Option) -> R { + let mut request = json!({ "execute": command }); + if let Some(arguments) = arguments { + request["arguments"] = arguments; + } + let mut bytes = serde_json::to_vec(&request).expect("serializable request"); + bytes.push(b'\n'); + self.writer.write_all(&bytes).await?; + let response = self.read_message().await?; + if let Some(error) = response.get("error") { + return Err(Error::InvalidState(format!( + "qmp '{command}' failed: {error}" + ))); + } + Ok(response.get("return").cloned().unwrap_or(Value::Null)) + } +} + +/// Picks a free TCP port on 127.0.0.1: binds to port 0 (OS assigns one), +/// reads it back, then drops the listener before qemu binds it for real. +/// Small TOCTOU race in principle; standard practice in the absence of a +/// way to ask qemu what port it actually bound. +pub async fn free_port() -> std::io::Result { + Ok(tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await? + .local_addr()? + .port()) +} diff --git a/crates/hule-vmm/src/backend/qemu/supervise.rs b/crates/hule-vmm/src/backend/qemu/supervise.rs index 53f890e..c799bb2 100644 --- a/crates/hule-vmm/src/backend/qemu/supervise.rs +++ b/crates/hule-vmm/src/backend/qemu/supervise.rs @@ -54,6 +54,9 @@ pub fn spawn(argv: &[String], exit_code_file: &Path) -> io::Result { .arg(&argv[0]) .arg(exit_code_file) .args(&argv[1..]) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) .spawn() } #[cfg(windows)] @@ -73,6 +76,9 @@ pub fn spawn(argv: &[String], exit_code_file: &Path) -> io::Result { .arg(&argv[0]) .arg(exit_code_file) .args(&argv[1..]) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) .spawn() } } @@ -138,3 +144,47 @@ pub async fn kill(pid: Pid) -> io::Result<()> { } } } + +/// (cpu_time_ns, memory_bytes) for `pid`, read straight from the host +/// process -- no QMP/guest cooperation needed. +pub async fn process_stats(pid: Pid) -> io::Result<(u64, u64)> { + #[cfg(target_os = "linux")] + { + let stat = tokio::fs::read_to_string(format!("/proc/{pid}/stat")).await?; + // `comm` (field 2) can itself contain spaces/parens; splitting after + // the last ')' reliably skips past it regardless. + let after_comm = stat + .rsplit_once(')') + .map(|(_, rest)| rest) + .ok_or_else(|| io::Error::other("unexpected /proc/pid/stat format"))?; + let fields: Vec<&str> = after_comm.split_whitespace().collect(); + let field = |i: usize| -> io::Result { + fields + .get(i) + .and_then(|s| s.parse().ok()) + .ok_or_else(|| io::Error::other("unexpected /proc/pid/stat format")) + }; + // Fields are 0-indexed from `state` (original field 3); utime/stime + // are original fields 14/15. 100 (USER_HZ) is the practically + // universal Linux clock tick rate. + let cpu_time_ns = (field(11)? + field(12)?) * 10_000_000; + + let status = tokio::fs::read_to_string(format!("/proc/{pid}/status")).await?; + let memory_bytes = status + .lines() + .find_map(|line| line.strip_prefix("VmRSS:")) + .and_then(|rest| rest.split_whitespace().next()) + .and_then(|kb| kb.parse::().ok()) + .map(|kb| kb * 1024) + .ok_or_else(|| io::Error::other("VmRSS not found in /proc/pid/status"))?; + + Ok((cpu_time_ns, memory_bytes)) + } + #[cfg(not(target_os = "linux"))] + { + let _ = pid; + Err(io::Error::other( + "process stats aren't implemented for this host yet", + )) + } +} diff --git a/crates/hule-vmm/src/machine.rs b/crates/hule-vmm/src/machine.rs index 914bd4d..26af2f1 100644 --- a/crates/hule-vmm/src/machine.rs +++ b/crates/hule-vmm/src/machine.rs @@ -41,7 +41,7 @@ impl Console for T {} /// [`hule_image::MachineImage`], which only declares a cpu/ram *range* and /// knows nothing about e.g. host ports. Always concrete: callers resolve /// image defaults before constructing this. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq)] pub struct Settings { pub cpu: u32, pub ram: u64, // megabytes @@ -90,9 +90,7 @@ pub trait Machine: Send + Sync { async fn update(&mut self, settings: Settings) -> R<()>; async fn rename(&mut self, name: Option<&str>) -> R<()>; - async fn wait(&mut self) -> R; async fn exec(&self, cmd: &[String]) -> R; - + async fn wait(&mut self) -> R; async fn attach(&self) -> R>; - async fn resize(&self, cols: u16, rows: u16) -> R<()>; } diff --git a/crates/hule-vmm/tests/qemu_lifecycle.rs b/crates/hule-vmm/tests/qemu_lifecycle.rs new file mode 100644 index 0000000..d30ac61 --- /dev/null +++ b/crates/hule-vmm/tests/qemu_lifecycle.rs @@ -0,0 +1,136 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: Apache-2.0 + +//! Boots a real qemu VM against the repo's alpine fixture and drives it +//! through lifecycle, QGA exec, and serial-console methods. Needs qemu-system-x86_64 and +//! `images/alpine/3.24/amd64` on disk; `#[ignore]`d so a normal `cargo +//! test` run doesn't depend on either. + +use std::path::Path; +use std::time::Duration; + +use hule_image::MachineImage; +use hule_vmm::backend::qemu::QemuHypervisor; +use hule_vmm::{Hypervisor, MachineId, Settings, State}; +use uuid::Uuid; + +#[tokio::test] +#[ignore = "needs qemu-system-x86_64 and the alpine fixture on disk"] +async fn pause_unpause_restart_and_kill_track_a_real_qemu() { + let fixture = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../images/alpine/3.24/amd64"); + let config = std::fs::read_to_string(fixture.join("config.json")) + .expect("read fixture config.json") + .replace( + r#"{ "type": "ssh", "port": 22, "user": "build", "auth": "empty-password" }"#, + r#"{ "type": "qga" }"#, + ); + let image = MachineImage::from_json(config.as_bytes()).expect("valid fixture config.json"); + + let scratch = std::env::temp_dir().join(format!("hule-vmm-test-{}", std::process::id())); + std::fs::create_dir_all(&scratch).unwrap(); + for disk in &image.disks { + let src = fixture.join(&disk.path); + let dest = scratch.join(&disk.path); + std::fs::hard_link(&src, &dest) + .or_else(|_| std::fs::copy(&src, &dest).map(|_| ())) + .expect("stage disk into scratch dir"); + } + + let hv = QemuHypervisor; + let boot = image + .machine + .boot + .iter() + .find(|b| hv.supports(b)) + .expect("fixture declares a boot protocol qemu supports"); + let settings = Settings { + cpu: 1, + ram: 512, + port_forwards: vec![], + }; + + let mut machine = hv + .create( + MachineId(Uuid::now_v7()), + None, + &image, + &scratch, + boot, + &settings, + ) + .await + .expect("create"); + + machine.start().await.expect("start"); + + let output = machine + .exec(&["/bin/echo".into(), "hello from qga".into()]) + .await + .expect("qga exec"); + assert!(output.status.success()); + assert_eq!(output.stdout, b"hello from qga\n"); + assert!(!machine.logs().await.expect("serial logs").is_empty()); + drop(machine.attach().await.expect("attach serial console")); + + assert_eq!(machine.state().await, State::Running); + + machine.pause().await.expect("pause"); + assert_eq!(machine.state().await, State::Paused); + + machine.unpause().await.expect("unpause"); + assert_eq!(machine.state().await, State::Running); + + machine.restart().await.expect("restart"); + tokio::time::sleep(Duration::from_millis(500)).await; + assert_eq!(machine.state().await, State::Running); + + let stats = machine.stats().await.expect("stats"); + assert!(stats.cpu_time_ns > 0); + assert!(stats.memory_bytes > 0); + + let mut new_settings = machine.settings(); + new_settings.ram = 384; + machine + .update(new_settings) + .await + .expect("update (balloon)"); + assert_eq!(machine.settings().ram, 384); + + machine.kill().await.expect("kill"); + assert!(matches!(machine.state().await, State::Exited(_))); + + let _ = std::fs::remove_dir_all(&scratch); +} + +/// QMP is internal to the qemu backend, so a black-box image is valid. +#[tokio::test] +async fn create_accepts_an_image_without_access() { + let fixture = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../images/alpine/3.24/amd64"); + let config = std::fs::read(fixture.join("config.json")).expect("read fixture config.json"); + let mut image = MachineImage::from_json(&config).expect("valid fixture config.json"); + image.machine.access.clear(); + + let hv = QemuHypervisor; + let boot = image + .machine + .boot + .iter() + .find(|b| hv.supports(b)) + .expect("fixture declares a boot protocol qemu supports"); + let settings = Settings { + cpu: 1, + ram: 512, + port_forwards: vec![], + }; + + hv.create( + MachineId(Uuid::now_v7()), + None, + &image, + &fixture, + boot, + &settings, + ) + .await + .expect("black-box images don't need to declare qemu's QMP channel"); +} -- Gilti