aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--crates/hule-vmm/Cargo.toml5+5 −0
-rw-r--r--crates/hule-vmm/src/backend/qemu/cli.rs88+88 −0
-rw-r--r--crates/hule-vmm/src/backend/qemu/mod.rs237+208 −29
-rw-r--r--crates/hule-vmm/src/backend/qemu/qga.rs233+233 −0
-rw-r--r--crates/hule-vmm/src/backend/qemu/qmp.rs87+87 −0
-rw-r--r--crates/hule-vmm/src/backend/qemu/supervise.rs50+50 −0
-rw-r--r--crates/hule-vmm/src/machine.rs6+2 −4
-rw-r--r--crates/hule-vmm/tests/qemu_lifecycle.rs136+136 −0
8 files changed, 809 insertions, 33 deletions
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<PathBuf>,
+ /// `-qmp tcp:127.0.0.1:<port>,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<u16>,
+ /// Host side of the qemu-guest-agent virtio-serial channel.
+ pub qga_port: Option<u16>,
+ /// 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<ChildState>,
+ qga_lock: Mutex<()>,
+ qmp_client: Mutex<Option<qmp::Qmp>>,
}
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:<port>` 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<u16> {
+ 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<u16> {
+ self.read_port(&self.qmp_port_path(), "qmp").await
+ }
+
+ async fn qga_port(&self) -> R<u16> {
+ self.read_port(&self.qga_port_path(), "qga").await
+ }
+
+ async fn console_port(&self) -> R<u16> {
+ self.read_port(&self.console_port_path(), "console").await
+ }
+
+ async fn qmp_execute(&self, command: &str) -> R<serde_json::Value> {
+ 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<serde_json::Value>,
+ ) -> R<serde_json::Value> {
+ 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<supervise::Pid> {
@@ -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<Stats> {
- 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<Vec<u8>> {
- 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<Output> {
- todo!("needs the access channel (ssh today) wired up here")
+ async fn exec(&self, cmd: &[String]) -> R<Output> {
+ 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<Box<dyn Console>> {
- 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
--- /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<serde_json::Value>,
+) -> R<serde_json::Value> {
+ 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<serde_json::Value>,
+) -> R<serde_json::Value> {
+ 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<serde_json::Value> {
+ 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<std::process::Output> {
+ 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<std::process::Output> {
+ 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<Vec<u8>> {
+ 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<std::process::ExitStatus> {
+ #[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
--- /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:<port>,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<OwnedReadHalf>,
+ 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<Self> {
+ 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<Value> {
+ 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<Value>) -> R<Value> {
+ 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<u16> {
+ 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<Child> {
.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<Child> {
.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<u64> {
+ 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::<u64>().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<T: AsyncRead + AsyncWrite + Send + Unpin> 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<ExitStatus>;
async fn exec(&self, cmd: &[String]) -> R<Output>;
-
+ async fn wait(&mut self) -> R<ExitStatus>;
async fn attach(&self) -> R<Box<dyn Console>>;
- 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
--- /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");
+}
generated by Gilti