diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md new file mode 100644 index 000000000..e2a7549c1 --- /dev/null +++ b/docs/experimental-systemd-vm-processes.md @@ -0,0 +1,105 @@ +# Experimental systemd VM process manager + +The VMM can experimentally launch each VM as a transient systemd service instead +of sending the process to the standalone dstack supervisor. This gives every VM +its own cgroup and lets systemd retain ownership while QEMU performs a long +kernel-side shutdown, such as encrypted-memory teardown. + +Enable it in the VMM configuration: + +```toml +[cvm] +pm = "auto" + +[systemd] +unit_prefix = "dstack-vm" +state_dir = "/var/lib/dstack-vmm/systemd-processes" +stop_timeout = "infinity" +``` + +The three process-manager modes are: + +- `supervisor`: launch and manage every VM through the standalone Supervisor. +- `systemd`: launch and manage every VM as a transient systemd service. +- `auto`: use systemd for every new launch. When the VMM starts, VM processes + already running in Supervisor are pinned to Supervisor for their remaining + lifecycle. Their next VM launch removes the stopped Supervisor record and + migrates them to systemd. + +The default is `supervisor`, preserving existing deployments. Use `auto` for +transitions from Supervisor. Direct `systemd` mode refuses to start when it can +verify that Supervisor still owns running VMs. + +## Runtime model + +The VMM invokes `systemd-run` directly. A service is named from the configured +prefix and the SHA-256 digest of the VM ID: + +```text +dstack-vm-.service +``` + +For a software-TPM VM, the service cgroup contains: + +```text +vm-launcher +├── qemu +└── swtpm +``` + +The transient service uses these properties: + +```ini +Type=exec +ExitType=cgroup +KillMode=mixed +KillSignal=SIGTERM +SendSIGKILL=yes +TimeoutStopSec= +Restart=no +``` + +The existing launcher remains responsible for swtpm readiness and graceful +child shutdown. systemd owns the final cgroup lifetime. A stop request is +submitted asynchronously so the VMM can report a VM as stopping while QEMU is +still completing kernel teardown. + +The default stop timeout is `infinity` because large encrypted-memory guests +can spend hours in kernel teardown. Operators that prefer bounded escalation +can set a systemd time span such as `stop_timeout = "30min"`. + +Process metadata is persisted in `systemd.state_dir`. It is required because a +successful transient unit may be garbage-collected after exit, while the VMM +still needs the original process annotation and CID during reconciliation. +When left empty, it defaults to `~/.dstack-vmm/systemd-processes`. + +## Inspecting a VM + +```bash +systemctl list-units 'dstack-vm-*.service' --all +systemctl show dstack-vm-.service \ + -p ActiveState -p SubState -p MainPID -p ControlGroup +systemd-cgls /system.slice/dstack-vm-.service +``` + +The implementation currently uses the `systemd-run` and `systemctl` CLIs. A +future production implementation should use the systemd D-Bus API directly for +atomic property handling and event-driven state updates. + +## Limitations + +- The host must run systemd with support for `ExitType=cgroup` and + `StandardOutput=append:`. +- The VMM must be authorized to create and stop system services. +- Transient services inherit the systemd manager environment rather than the + VMM environment. Variables in `ProcessConfig.env` are forwarded; unrelated + inherited variables are not. +- Unit status is currently polled through `systemctl show`. +- If Supervisor becomes unavailable during an `auto` migration, pinned VMs + retain their cached state to prevent double launch and CID reuse. Their + stop/removal may remain pending until Supervisor is restored. +- `systemd.stop_timeout` syntax is validated by systemd when the first VM is + launched; an invalid time span causes that launch to fail. +- Start and stop are not yet transactional with the metadata file. +- A host reboot removes transient units; normal VMM workdir recovery recreates + services for VMs marked for automatic start. diff --git a/dstack/supervisor/client/src/lib.rs b/dstack/supervisor/client/src/lib.rs index 53257e5c9..84a6a8535 100644 --- a/dstack/supervisor/client/src/lib.rs +++ b/dstack/supervisor/client/src/lib.rs @@ -33,12 +33,15 @@ impl SupervisorClient { ) -> Result { let uri = format!("unix:{}", uds.as_ref().display()); let client = Self::new(&uri); - if client.probe(Duration::from_millis(100)).await.is_ok() { - info!("Connected to supervisor at {uri}"); - return Ok(client); - } + let probe_error = match client.probe(Duration::from_millis(100)).await { + Ok(()) => { + info!("Connected to supervisor at {uri}"); + return Ok(client); + } + Err(error) => error, + }; if !auto_start { - anyhow::bail!("Failed to connect to supervisor at {uri}"); + return Err(probe_error).with_context(|| format!("failed to connect to {uri}")); } info!("Failed to connect to supervisor at {uri}, trying to start supervisor"); // if the uds exists, remove it @@ -146,11 +149,9 @@ impl SupervisorClient { } pub async fn probe(&self, timeout: Duration) -> Result<()> { - let response = tokio::time::timeout(timeout, self.ping()).await; - if matches!(response, Ok(Ok(_))) { - Ok(()) - } else { - anyhow::bail!("failed to probe supervisor") + match tokio::time::timeout(timeout, self.ping()).await { + Ok(result) => result.map(|_| ()).context("failed to probe supervisor"), + Err(error) => Err(error).context("supervisor probe timed out"), } } diff --git a/dstack/vmm/src/app.rs b/dstack/vmm/src/app.rs index 90b9bcc1d..bc88f0d1e 100644 --- a/dstack/vmm/src/app.rs +++ b/dstack/vmm/src/app.rs @@ -8,6 +8,7 @@ use crate::{ netd::{self, InterfaceIdentity, PrepareRequest, Request as NetdRequest}, }; +use crate::process_manager::ProcessManager; use anyhow::{bail, Context, Result}; use bon::Builder; use dstack_kms_rpc::kms_client::KmsClient; @@ -32,7 +33,6 @@ use std::net::IpAddr; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::SystemTime; -use supervisor_client::SupervisorClient; use tracing::{debug, error, info, warn}; pub use image::{Image, ImageInfo}; @@ -291,7 +291,7 @@ pub(crate) enum PullStatus { #[derive(Clone)] pub struct App { pub config: Arc, - pub supervisor: SupervisorClient, + pub process_manager: ProcessManager, state: Arc>, /// Pull status for registry images: tag → status. pub(crate) pull_status: Arc>>, @@ -311,12 +311,12 @@ impl App { Ok(VmWorkDir::new(self.config.run_path.join(id))) } - pub fn new(config: Config, supervisor: SupervisorClient) -> Self { + pub fn new(config: Config, process_manager: ProcessManager) -> Self { let cid_start = config.cvm.cid_start; let cid_end = cid_start.saturating_add(config.cvm.cid_pool_size); let cid_pool = IdPool::new(cid_start, cid_end); Self { - supervisor: supervisor.clone(), + process_manager, state: Arc::new(Mutex::new(AppState { cid_pool, vms: HashMap::new(), @@ -413,7 +413,7 @@ impl App { } self.sync_dynamic_config(id)?; let is_running = self - .supervisor + .process_manager .info(id) .await? .is_some_and(|info| info.state.status.is_running()); @@ -464,7 +464,7 @@ impl App { vm_state.state.runtime_networks = runtime_networks.clone(); } for process in processes { - if let Err(err) = self.supervisor.deploy(&process).await { + if let Err(err) = self.process_manager.deploy(&process).await { if let Err(cleanup_error) = self .remove_filtered_networks(&vm_config.manifest.id, &runtime_networks) .await @@ -611,37 +611,37 @@ impl App { } pub(crate) async fn stop_vm_process(&self, id: &str) -> Result<()> { - let Some(info) = self.supervisor.info(id).await? else { + let Some(info) = self.process_manager.info(id).await? else { return Ok(()); }; // Non-TPM VMs run QEMU directly and keep the existing Supervisor stop // path. Only the TPM launcher's hidden subcommand implements graceful // child-process shutdown. if info.config.args.first().map(String::as_str) != Some("vm-launcher") { - return self.supervisor.stop(id).await; + return self.process_manager.stop(id).await; } if info.state.status.is_running() { let pid = info.state.pid.context("running VM launcher has no PID")?; if let Err(error) = signal_pidfd(pid, libc::SIGTERM) { warn!(id, %pid, %error, "failed to signal VM launcher gracefully; forcing shutdown"); - return self.supervisor.stop(id).await; + return self.process_manager.stop(id).await; } for _ in 0..150 { tokio::time::sleep(std::time::Duration::from_millis(100)).await; let running = self - .supervisor + .process_manager .info(id) .await? .is_some_and(|info| info.state.status.is_running()); if !running { // Synchronize Supervisor's `started` flag after the launcher // completed its graceful child cleanup. - return self.supervisor.stop(id).await; + return self.process_manager.stop(id).await; } } warn!(id, "VM launcher did not stop gracefully; forcing shutdown"); } - self.supervisor.stop(id).await + self.process_manager.stop(id).await } pub async fn remove_vm(&self, id: &str) -> Result<()> { @@ -687,7 +687,7 @@ impl App { // Some VMs take a long time to stop (e.g. 2+ hours), so we wait indefinitely. let mut poll_count: u64 = 0; loop { - match self.supervisor.info(id).await { + match self.process_manager.info(id).await { Ok(Some(info)) if info.state.status.is_running() => { tokio::time::sleep(std::time::Duration::from_secs(2)).await; poll_count += 1; @@ -700,7 +700,7 @@ impl App { } Ok(Some(_)) => { // Not running — remove from supervisor - if let Err(err) = self.supervisor.remove(id).await { + if let Err(err) = self.process_manager.remove(id).await { warn!("supervisor.remove({id}) failed: {err:?}"); } break; @@ -778,7 +778,11 @@ impl App { pub async fn reload_vms(&self) -> Result<()> { let vm_path = self.vm_dir(); - let running_vms = self.supervisor.list().await.context("Failed to list VMs")?; + let running_vms = self + .process_manager + .list() + .await + .context("Failed to list VMs")?; let running_vms: Vec<(ProcessAnnotation, _)> = running_vms .into_iter() .map(|p| (serde_json::from_str(&p.config.note).unwrap_or_default(), p)) @@ -852,7 +856,11 @@ impl App { let mut removed = 0u32; // Get running VMs to preserve CIDs and process info - let running_vms = self.supervisor.list().await.context("Failed to list VMs")?; + let running_vms = self + .process_manager + .list() + .await + .context("Failed to list VMs")?; let running_vms_map: HashMap = running_vms .into_iter() .map(|p| (p.config.id.clone(), p)) @@ -1065,7 +1073,7 @@ impl App { pub async fn list_vms(&self, request: StatusRequest) -> Result { let vms = self - .supervisor + .process_manager .list() .await .context("Failed to list VMs")? @@ -1126,7 +1134,7 @@ impl App { } pub async fn vm_info(&self, id: &str) -> Result> { - let proc_state = self.supervisor.info(id).await?; + let proc_state = self.process_manager.info(id).await?; let state = self.lock(); let Some(vm_state) = state.get(id) else { return Ok(None); @@ -1309,7 +1317,7 @@ impl App { } let max_backups = self.config.cvm.log.max_backups; let running = self - .supervisor + .process_manager .list() .await .context("failed to list VMs")? @@ -1338,7 +1346,7 @@ impl App { pub(crate) async fn try_restart_exited_vms(&self) -> Result<()> { let running_vms = self - .supervisor + .process_manager .list() .await .context("Failed to list VMs")? @@ -1452,9 +1460,8 @@ fn append_boot_separator(path: &std::path::Path) { /// Logs a CVM writes into its work directory, subject to retention. /// -/// stdout and stderr are written by the supervisor, which always opens them -/// with `append(true)` and reopens them when they change, so they satisfy -/// [`crate::logrotate`]'s contract no matter which VMM launched the VM. +/// stdout and stderr are opened in append mode by every process-manager +/// backend, so in-place truncation satisfies [`crate::logrotate`]'s contract. /// serial.log is written by QEMU, whose fd only appends when *we* passed /// `logappend=on`, so it is included only when `serial` says so. fn rotatable_logs(work_dir: &VmWorkDir, serial: bool) -> Vec { diff --git a/dstack/vmm/src/config.rs b/dstack/vmm/src/config.rs index f33b9bae4..f3ca78d3b 100644 --- a/dstack/vmm/src/config.rs +++ b/dstack/vmm/src/config.rs @@ -294,6 +294,9 @@ impl TdxAttestationVariantConfig { #[derive(Debug, Clone, Deserialize)] pub struct CvmConfig { + /// Process manager used to launch and monitor VM processes. + #[serde(default)] + pub pm: ProcessManagerBackend, /// TEE platform to use when launching CVMs. Omit (or set `auto`) to detect /// the host TEE from /proc/cpuinfo (AMD SEV-SNP vs Intel TDX); set `tdx` or /// `amd-sev-snp` to force a platform. @@ -477,6 +480,31 @@ pub struct AuthConfig { pub htpasswd_file: PathBuf, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProcessManagerBackend { + /// Launch and manage every VM through the standalone Supervisor service. + #[default] + Supervisor, + /// Launch and manage every VM as a transient systemd service. + Systemd, + /// Launch new VMs through systemd, but keep VMs found running in the + /// standalone Supervisor there until each VM is restarted. + Auto, +} + +fn default_systemd_unit_prefix() -> String { + "dstack-vm".into() +} + +fn default_systemd_state_dir() -> PathBuf { + PathBuf::new() +} + +fn default_systemd_stop_timeout() -> String { + "infinity".into() +} + #[derive(Debug, Clone, Default, Deserialize)] pub struct SupervisorConfig { pub exe: String, @@ -487,6 +515,26 @@ pub struct SupervisorConfig { pub auto_start: bool, } +#[derive(Debug, Clone, Deserialize)] +pub struct SystemdConfig { + #[serde(default = "default_systemd_unit_prefix")] + pub unit_prefix: String, + #[serde(default = "default_systemd_state_dir")] + pub state_dir: PathBuf, + #[serde(default = "default_systemd_stop_timeout")] + pub stop_timeout: String, +} + +impl Default for SystemdConfig { + fn default() -> Self { + Self { + unit_prefix: default_systemd_unit_prefix(), + state_dir: default_systemd_state_dir(), + stop_timeout: default_systemd_stop_timeout(), + } + } +} + #[derive(Debug, Clone, Deserialize)] pub struct GatewayConfig { pub base_domain: String, @@ -527,10 +575,13 @@ pub struct Config { /// CVM configuration pub cvm: CvmConfig, - /// Privileged host networking service configuration. #[serde(default)] pub netd: NetdConfig, + + /// Experimental systemd process manager configuration + #[serde(default)] + pub systemd: SystemdConfig, /// Gateway configuration pub gateway: GatewayConfig, @@ -635,6 +686,7 @@ impl Config { pub fn abs_path(mut self) -> Result { self.image.path = self.image.path.absolutize()?.to_path_buf(); self.run_path = self.run_path.absolutize()?.to_path_buf(); + self.systemd.state_dir = self.systemd.state_dir.absolutize()?.to_path_buf(); Ok(self) } @@ -675,11 +727,17 @@ impl Config { } validate_networking(&self.cvm.networking)?; + if self.cvm.pm != ProcessManagerBackend::Systemd { + anyhow::ensure!( + !self.supervisor.sock.trim().is_empty(), + "supervisor.sock must not be empty unless cvm.pm = \"systemd\"" + ); + } anyhow::ensure!( - !self.supervisor.sock.trim().is_empty(), - "supervisor.sock must not be empty" + !self.systemd.stop_timeout.trim().is_empty(), + "systemd.stop_timeout must not be empty" ); - if self.supervisor.auto_start { + if self.cvm.pm == ProcessManagerBackend::Supervisor && self.supervisor.auto_start { for (name, value) in [ ("supervisor.exe", self.supervisor.exe.as_str()), ("supervisor.pid_file", self.supervisor.pid_file.as_str()), @@ -897,6 +955,9 @@ impl Config { if me.run_path == PathBuf::default() { me.run_path = app_home.join("vm"); } + if me.systemd.state_dir == PathBuf::default() { + me.systemd.state_dir = app_home.join("systemd-processes"); + } if me.cvm.qemu_path == PathBuf::default() { // Prefer the path from dstack client config if present if let Some(qemu_path) = read_qemu_path_from_client_conf() { @@ -1086,6 +1147,16 @@ mod tests { default_config().validate().unwrap(); } + #[test] + fn process_manager_modes_parse() { + let parse = |mode: &str| { + serde_json::from_str::(&format!("\"{mode}\"")).unwrap() + }; + assert_eq!(parse("supervisor"), ProcessManagerBackend::Supervisor); + assert_eq!(parse("systemd"), ProcessManagerBackend::Systemd); + assert_eq!(parse("auto"), ProcessManagerBackend::Auto); + } + #[test] fn config_validation_rejects_invalid_static_invariants() { let mut config = default_config(); diff --git a/dstack/vmm/src/main.rs b/dstack/vmm/src/main.rs index 9e97b0d6e..da640e196 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -32,6 +32,7 @@ mod main_service; mod netd; mod one_shot; mod openapi; +mod process_manager; mod vm_launcher; const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -41,6 +42,14 @@ fn app_version() -> String { dstack_build_info::app_version!() } +fn is_connection_refused(error: &anyhow::Error) -> bool { + error.chain().any(|cause| { + cause + .downcast_ref::() + .is_some_and(|error| error.kind() == std::io::ErrorKind::ConnectionRefused) + }) +} + #[derive(Parser)] #[command(author, version, about, long_version = app_version())] struct Args { @@ -328,8 +337,16 @@ async fn main() -> Result<()> { token, or bind `address` to localhost / a Unix socket." ); } - let supervisor = { - let cfg = &config.supervisor; + let systemd_manager = || { + process_manager::ProcessManager::systemd_backend( + config.systemd.state_dir.clone(), + config.systemd.unit_prefix.clone(), + config.systemd.stop_timeout.clone(), + ) + }; + let supervisor_config = &config.supervisor; + let connect_supervisor = |auto_start| async move { + let cfg = supervisor_config; let abs_exe = Path::new(&cfg.exe).absolutize()?; SupervisorClient::start_and_connect_uds( &abs_exe, @@ -337,12 +354,59 @@ async fn main() -> Result<()> { &cfg.pid_file, &cfg.log_file, cfg.detached, - cfg.auto_start, + auto_start, ) .await - .context("Failed to connect to supervisor")? + .context("failed to connect to supervisor") + }; + let legacy_socket_exists = Path::new(&supervisor_config.sock).exists(); + let process_manager = match config.cvm.pm { + config::ProcessManagerBackend::Supervisor => process_manager::ProcessManager::supervisor( + connect_supervisor(supervisor_config.auto_start).await?, + ), + config::ProcessManagerBackend::Systemd => { + if legacy_socket_exists { + match connect_supervisor(false).await { + Ok(client) => anyhow::ensure!( + !client + .list() + .await? + .iter() + .any(|process| process.state.status.is_running()), + "running Supervisor VMs detected; use cvm.pm = \"auto\" for migration" + ), + Err(error) if is_connection_refused(&error) => { + warn!(%error, "ignoring stale legacy Supervisor socket") + } + Err(error) => return Err(error).context( + "supervisor socket exists but its state cannot be verified in systemd mode; \ + if Supervisor is definitely not running, remove the stale socket and restart", + ), + } + } + process_manager::ProcessManager::systemd(systemd_manager()?) + } + config::ProcessManagerBackend::Auto => { + let legacy_supervisor = if legacy_socket_exists { + match connect_supervisor(false).await { + Ok(client) => Some(client), + Err(error) if is_connection_refused(&error) => { + warn!(%error, "ignoring stale legacy Supervisor socket"); + None + } + Err(error) => return Err(error).context( + "supervisor socket exists but its state cannot be verified in auto mode; \ + if Supervisor is definitely not running, remove the stale socket and restart", + ), + } + } else { + info!("legacy supervisor socket is absent; using systemd for all VMs"); + None + }; + process_manager::ProcessManager::auto(systemd_manager()?, legacy_supervisor).await? + } }; - let state = app::App::new(config, supervisor); + let state = app::App::new(config, process_manager); state.reload_vms().await.context("Failed to reload VMs")?; tokio::spawn(auto_restart_task(state.clone())); tokio::spawn(log_rotation_task(state.clone())); diff --git a/dstack/vmm/src/main_service.rs b/dstack/vmm/src/main_service.rs index 839fb990a..ef52eed4c 100644 --- a/dstack/vmm/src/main_service.rs +++ b/dstack/vmm/src/main_service.rs @@ -710,7 +710,7 @@ impl VmmRpc for RpcHandler { }; let is_running = self .app - .supervisor + .process_manager .info(&request.id) .await? .is_some_and(|info| info.state.status.is_running()); @@ -884,7 +884,7 @@ impl VmmRpc for RpcHandler { async fn sv_list(self) -> Result { use supervisor_client::supervisor::ProcessStatus; - let list = self.app.supervisor.list().await?; + let list = self.app.process_manager.list().await?; let processes = list .into_iter() .map(|p| { @@ -913,7 +913,7 @@ impl VmmRpc for RpcHandler { // same helper preserves generic Supervisor stop semantics for every // other process type. self.app - .supervisor + .process_manager .info(&request.id) .await? .context("Supervisor process not found")?; @@ -921,7 +921,7 @@ impl VmmRpc for RpcHandler { } async fn sv_remove(self, request: Id) -> Result<()> { - self.app.supervisor.remove(&request.id).await?; + self.app.process_manager.remove(&request.id).await?; Ok(()) } diff --git a/dstack/vmm/src/process_manager.rs b/dstack/vmm/src/process_manager.rs new file mode 100644 index 000000000..e102dd2ee --- /dev/null +++ b/dstack/vmm/src/process_manager.rs @@ -0,0 +1,596 @@ +// SPDX-FileCopyrightText: © 2026 Phala Network +// +// SPDX-License-Identifier: Apache-2.0 + +use std::path::PathBuf; +use std::time::{Duration, SystemTime}; +use std::{collections::HashMap, sync::Arc}; + +use anyhow::{bail, Context, Result}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use supervisor_client::supervisor::{ProcessConfig, ProcessInfo, ProcessState, ProcessStatus}; +use supervisor_client::SupervisorClient; +use tokio::process::Command; +use tokio::sync::RwLock; +use tracing::warn; + +#[derive(Clone)] +pub enum ProcessManager { + Supervisor(SupervisorClient), + Systemd(Arc), + Auto(Arc), +} + +impl ProcessManager { + pub fn supervisor(client: SupervisorClient) -> Self { + Self::Supervisor(client) + } + + pub fn systemd_backend( + state_dir: PathBuf, + unit_prefix: String, + stop_timeout: String, + ) -> Result> { + Ok(Arc::new(SystemdProcessManager::new( + state_dir, + unit_prefix, + stop_timeout, + )?)) + } + + pub fn systemd(backend: Arc) -> Self { + Self::Systemd(backend) + } + + pub async fn auto( + systemd: Arc, + supervisor: Option, + ) -> Result { + let mut supervisor_processes = HashMap::new(); + if let Some(client) = &supervisor { + for process in client.list().await? { + supervisor_processes.insert(process.config.id.clone(), process); + } + } + Ok(Self::Auto(Arc::new(AutoProcessManager { + systemd, + supervisor, + supervisor_processes: RwLock::new(supervisor_processes), + }))) + } + + pub async fn deploy(&self, config: &ProcessConfig) -> Result<()> { + match self { + Self::Supervisor(client) => client.deploy(config).await, + Self::Systemd(manager) => manager.deploy(config).await, + Self::Auto(manager) => manager.deploy(config).await, + } + } + + pub async fn stop(&self, id: &str) -> Result<()> { + match self { + Self::Supervisor(client) => client.stop(id).await, + Self::Systemd(manager) => manager.stop(id).await, + Self::Auto(manager) => manager.stop(id).await, + } + } + + pub async fn remove(&self, id: &str) -> Result<()> { + match self { + Self::Supervisor(client) => client.remove(id).await, + Self::Systemd(manager) => manager.remove(id).await, + Self::Auto(manager) => manager.remove(id).await, + } + } + + pub async fn list(&self) -> Result> { + match self { + Self::Supervisor(client) => client.list().await, + Self::Systemd(manager) => manager.list().await, + Self::Auto(manager) => manager.list().await, + } + } + + pub async fn info(&self, id: &str) -> Result> { + match self { + Self::Supervisor(client) => client.info(id).await, + Self::Systemd(manager) => manager.info(id).await, + Self::Auto(manager) => manager.info(id).await, + } + } +} + +pub struct AutoProcessManager { + systemd: Arc, + supervisor: Option, + /// VM processes found running in Supervisor when the VMM started. They + /// stay pinned to Supervisor until their next deploy. + supervisor_processes: RwLock>, +} + +impl AutoProcessManager { + async fn is_supervisor_process(&self, id: &str) -> bool { + self.supervisor_processes.read().await.contains_key(id) + } + + fn supervisor(&self) -> Result<&SupervisorClient> { + self.supervisor + .as_ref() + .context("legacy supervisor is unavailable") + } + + async fn deploy(&self, config: &ProcessConfig) -> Result<()> { + // Concurrent deploys of the same ID are not serialized here; one wins + // the handoff and the other fails on record/unit removal or collision. + if self.is_supervisor_process(&config.id).await { + let supervisor = self.supervisor()?; + match supervisor.info(&config.id).await? { + Some(info) if info.state.status.is_running() => { + bail!("process is already running") + } + Some(_) => { + // Natural exits leave Supervisor's `started` flag set. + // Normalize it before removing the legacy record. + supervisor.stop(&config.id).await?; + supervisor.remove(&config.id).await?; + } + None => { + // Supervisor may have restarted or the record may have + // been removed out of band. It is already safe to migrate. + } + } + self.supervisor_processes.write().await.remove(&config.id); + } + self.systemd.deploy(config).await + } + + async fn stop(&self, id: &str) -> Result<()> { + if self.is_supervisor_process(id).await { + self.supervisor()?.stop(id).await + } else { + self.systemd.stop(id).await + } + } + + async fn remove(&self, id: &str) -> Result<()> { + if self.is_supervisor_process(id).await { + self.supervisor()?.remove(id).await?; + self.supervisor_processes.write().await.remove(id); + Ok(()) + } else { + self.systemd.remove(id).await + } + } + + async fn list(&self) -> Result> { + let mut processes = self.systemd.list().await?; + let pinned = self.supervisor_processes.read().await.clone(); + if !pinned.is_empty() { + let supervisor = self.supervisor()?; + match supervisor.list().await { + Ok(legacy) => { + let legacy = legacy + .into_iter() + .filter(|process| pinned.contains_key(&process.config.id)) + .collect::>(); + let mut cache = self.supervisor_processes.write().await; + let present = legacy + .iter() + .map(|process| process.config.id.as_str()) + .collect::>(); + cache.retain(|id, _| present.contains(id.as_str())); + for process in &legacy { + if let Some(cached) = cache.get_mut(&process.config.id) { + *cached = process.clone(); + } + } + drop(cache); + processes.extend(legacy); + } + Err(error) => { + warn!(%error, "legacy supervisor is unavailable; using cached pinned VM state"); + processes.extend(pinned.into_values()); + } + } + } + Ok(processes) + } + + async fn info(&self, id: &str) -> Result> { + if self.is_supervisor_process(id).await { + match self.supervisor()?.info(id).await { + Ok(info) => { + if let Some(process) = &info { + if let Some(cached) = self.supervisor_processes.write().await.get_mut(id) { + *cached = process.clone(); + } + } else { + self.supervisor_processes.write().await.remove(id); + } + Ok(info) + } + Err(error) => { + warn!(%id, %error, "legacy supervisor is unavailable; using cached pinned VM state"); + Ok(self.supervisor_processes.read().await.get(id).cloned()) + } + } + } else { + self.systemd.info(id).await + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct ProcessRecord { + config: ProcessConfig, + started: bool, +} + +fn system_time_from_monotonic_micros(value: &str) -> Option { + let target = value.parse::().ok().filter(|value| *value != 0)?; + let mut now = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // SAFETY: `now` points to a valid timespec. systemd's monotonic timestamp + // properties use CLOCK_MONOTONIC through its dual_timestamp helpers. + if unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut now) } != 0 { + return None; + } + let now_micros = (now.tv_sec as u64) + .saturating_mul(1_000_000) + .saturating_add((now.tv_nsec as u64) / 1_000); + SystemTime::now().checked_sub(Duration::from_micros(now_micros.saturating_sub(target))) +} + +fn state_from_systemd_properties( + properties: &str, + started: bool, +) -> ( + ProcessStatus, + Option, + Option, + Option, +) { + let value = |name: &str| { + properties + .lines() + .find_map(|line| line.strip_prefix(&format!("{name}="))) + .unwrap_or_default() + }; + let load_state = value("LoadState"); + let active_state = value("ActiveState"); + let sub_state = value("SubState"); + let running = matches!(active_state, "active" | "activating" | "deactivating"); + let plain_status = value("ExecMainStatus").parse::().unwrap_or_default(); + let raw_status = match value("ExecMainCode") { + "exited" => plain_status << 8, + "dumped" => plain_status | 0x80, + _ => plain_status, + }; + let status = if running { + ProcessStatus::Running + } else if !started && raw_status == 0 { + ProcessStatus::Stopped + } else if load_state == "not-found" || matches!(active_state, "inactive" | "failed") { + // A collected unit has no status properties, so a started record can + // only be represented as a clean exit after daemon reload/reboot. + ProcessStatus::Exited(raw_status) + } else { + ProcessStatus::Error(format!( + "systemd unit is {active_state}/{sub_state} (code={}, status={})", + value("ExecMainCode"), + value("ExecMainStatus") + )) + }; + let pid = value("MainPID").parse().ok().filter(|pid| *pid != 0); + let started_at = system_time_from_monotonic_micros(value("ExecMainStartTimestampMonotonic")); + let stopped_at = system_time_from_monotonic_micros(value("InactiveEnterTimestampMonotonic")); + (status, pid, started_at, stopped_at) +} + +pub struct SystemdProcessManager { + state_dir: PathBuf, + unit_prefix: String, + stop_timeout: String, +} + +impl SystemdProcessManager { + fn new(state_dir: PathBuf, unit_prefix: String, stop_timeout: String) -> Result { + anyhow::ensure!( + !unit_prefix.is_empty(), + "systemd unit prefix must not be empty" + ); + anyhow::ensure!( + unit_prefix + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_'), + "systemd unit prefix contains unsupported characters" + ); + fs_err::create_dir_all(&state_dir).context("failed to create systemd process state dir")?; + Ok(Self { + state_dir, + unit_prefix, + stop_timeout, + }) + } + + fn key(id: &str) -> String { + hex::encode(Sha256::digest(id.as_bytes())) + } + + fn unit(&self, id: &str) -> String { + format!("{}-{}.service", self.unit_prefix, Self::key(id)) + } + + fn record_path(&self, id: &str) -> PathBuf { + self.state_dir.join(format!("{}.json", Self::key(id))) + } + + fn read_record(&self, id: &str) -> Result { + let path = self.record_path(id); + let raw = + fs_err::read(&path).with_context(|| format!("process record not found for {id}"))?; + serde_json::from_slice(&raw).context("failed to parse systemd process record") + } + + fn write_record(&self, record: &ProcessRecord) -> Result<()> { + let path = self.record_path(&record.config.id); + safe_write::safe_write(path, serde_json::to_vec_pretty(record)?) + .context("failed to persist systemd process record") + } + + async fn command(mut command: Command, operation: &str) -> Result { + let output = command + .output() + .await + .with_context(|| format!("failed to execute {operation}"))?; + if !output.status.success() { + bail!( + "{operation} failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + } + Ok(output) + } + + async fn launch(&self, config: &ProcessConfig) -> Result<()> { + let unit = self.unit(&config.id); + // Failed transient units remain loaded until reset and otherwise + // prevent automatic restart from reusing the unit name. + let mut reset = Command::new("systemctl"); + reset.arg("reset-failed").arg(&unit); + let _ = reset.output().await; + let mut command = Command::new("systemd-run"); + command + .arg("--quiet") + .arg("--unit") + .arg(&unit) + .arg("--service-type=exec") + .arg("--property=KillMode=mixed") + .arg("--property=KillSignal=SIGTERM") + .arg("--property=SendSIGKILL=yes") + .arg(format!("--property=TimeoutStopSec={}", self.stop_timeout)) + .arg("--property=ExitType=cgroup") + .arg("--property=Restart=no") + .arg(format!("--description=dstack VM process {}", config.id)); + + if !config.cwd.is_empty() { + command.arg(format!("--working-directory={}", config.cwd)); + } + if config.stdout.is_empty() { + command.arg("--property=StandardOutput=null"); + } else { + command.arg(format!( + "--property=StandardOutput=append:{}", + config.stdout + )); + } + if config.stderr.is_empty() { + command.arg("--property=StandardError=null"); + } else { + command.arg(format!("--property=StandardError=append:{}", config.stderr)); + } + for (key, value) in &config.env { + command.arg(format!("--setenv={key}={value}")); + } + command.arg("--").arg(&config.command).args(&config.args); + Self::command(command, "systemd-run").await?; + + if !config.pidfile.is_empty() { + if let Some(info) = self.info(&config.id).await? { + if let Some(pid) = info.state.pid { + fs_err::write(&config.pidfile, pid.to_string()) + .context("failed to write systemd process pidfile")?; + } + } + } + Ok(()) + } + + async fn deploy(&self, config: &ProcessConfig) -> Result<()> { + if self + .info(&config.id) + .await? + .is_some_and(|info| info.state.status.is_running()) + { + bail!("process is already running"); + } + let record = ProcessRecord { + config: config.clone(), + started: true, + }; + self.write_record(&record)?; + self.launch(config).await + } + + async fn stop(&self, id: &str) -> Result<()> { + let mut record = self.read_record(id)?; + record.started = false; + self.write_record(&record)?; + + if self + .info(id) + .await? + .is_some_and(|info| info.state.status.is_running()) + { + let mut command = Command::new("systemctl"); + command.arg("stop").arg("--no-block").arg(self.unit(id)); + if let Err(error) = Self::command(command, "systemctl stop").await { + // The unit may have exited and been collected between the + // preceding status query and this stop request. + if self + .info(id) + .await? + .is_some_and(|info| info.state.status.is_running()) + { + return Err(error); + } + } + } + Ok(()) + } + + async fn remove(&self, id: &str) -> Result<()> { + if self + .info(id) + .await? + .is_some_and(|info| info.state.status.is_running()) + { + bail!("process is running"); + } + let record = self.read_record(id)?; + if record.started { + bail!("process is started"); + } + let mut command = Command::new("systemctl"); + command.arg("reset-failed").arg(self.unit(id)); + let _ = command.output().await; + fs_err::remove_file(self.record_path(id)).context("failed to remove process record") + } + + async fn list(&self) -> Result> { + let mut processes = Vec::new(); + for entry in fs_err::read_dir(&self.state_dir)? { + let entry = entry?; + if entry.path().extension().and_then(|value| value.to_str()) != Some("json") { + continue; + } + let path = entry.path(); + let record = fs_err::read(&path) + .context("failed to read process record") + .and_then(|raw| serde_json::from_slice::(&raw).map_err(Into::into)); + let record = match record { + Ok(record) => record, + Err(error) => { + warn!(path = %path.display(), %error, "skipping invalid process record"); + continue; + } + }; + // A per-record parse error is isolated above. A systemd-wide + // query error is propagated so callers cannot lose CID ownership + // and mistake a running VM for a stopped one. + processes.push(self.info_from_record(record).await?); + } + Ok(processes) + } + + async fn info(&self, id: &str) -> Result> { + match self.read_record(id) { + Ok(record) => self.info_from_record(record).await.map(Some), + Err(error) + if error + .downcast_ref::() + .is_some_and(|error| error.kind() == std::io::ErrorKind::NotFound) => + { + Ok(None) + } + Err(error) => Err(error), + } + } + + async fn info_from_record(&self, record: ProcessRecord) -> Result { + let unit = self.unit(&record.config.id); + let mut command = Command::new("systemctl"); + command + .arg("show") + .arg(&unit) + .arg( + "--property=LoadState,ActiveState,SubState,MainPID,ExecMainCode,ExecMainStatus,ExecMainStartTimestampMonotonic,InactiveEnterTimestampMonotonic", + ); + // A bus failure must not be mapped to a stopped VM: callers rely on an + // error here to avoid rotating logs and attempting duplicate restarts. + let output = Self::command(command, "systemctl show").await?; + let properties = String::from_utf8_lossy(&output.stdout); + let (status, pid, started_at, stopped_at) = + state_from_systemd_properties(&properties, record.started); + Ok(ProcessInfo { + config: record.config, + state: ProcessState { + status, + started: record.started, + pid, + started_at, + stopped_at, + }, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unit_names_are_stable_and_do_not_embed_process_ids() { + let dir = tempfile::tempdir().unwrap(); + let manager = SystemdProcessManager::new( + dir.path().to_path_buf(), + "dstack-vm".into(), + "infinity".into(), + ) + .unwrap(); + assert_eq!(manager.unit("vm/one"), manager.unit("vm/one")); + assert_ne!(manager.unit("vm/one"), manager.unit("vm-two")); + assert!(!manager.unit("vm/one").contains("vm/one")); + } + + #[test] + fn maps_systemd_states_to_process_status() { + let state = |properties, started| state_from_systemd_properties(properties, started).0; + assert!(matches!( + state("ActiveState=active\nSubState=running\nMainPID=42", true), + ProcessStatus::Running + )); + assert!(matches!( + state("ActiveState=deactivating\nSubState=stop-sigterm", false), + ProcessStatus::Running + )); + assert!(matches!( + state("LoadState=not-found\nActiveState=inactive", false), + ProcessStatus::Stopped + )); + assert!(matches!( + state( + "LoadState=loaded\nActiveState=failed\nExecMainCode=exited\nExecMainStatus=3", + true + ), + ProcessStatus::Exited(768) + )); + assert!(matches!( + state( + "LoadState=loaded\nActiveState=failed\nExecMainCode=exited\nExecMainStatus=3", + false + ), + ProcessStatus::Exited(768) + )); + assert!(matches!( + state( + "LoadState=loaded\nActiveState=failed\nExecMainCode=killed\nExecMainStatus=9", + true + ), + ProcessStatus::Exited(9) + )); + } +} diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index 06782be2d..74826f634 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -21,6 +21,12 @@ node_name = "" registry = "" [cvm] +# Process manager modes: +# - "supervisor": launch and manage every VM through standalone Supervisor. +# - "systemd": launch and manage every VM as a transient systemd service. +# - "auto": use systemd for new launches, but keep VMs already running in +# Supervisor there until each VM is restarted. +pm = "supervisor" # TEE platform: "auto", "tdx", or "amd-sev-snp". Auto selects AMD SEV-SNP when host CPU flags include sev_snp, otherwise TDX. platform = "auto" qemu_path = "" @@ -185,6 +191,18 @@ log_file = "./run/supervisor.log" detached = false auto_start = true +[systemd] +# Used when cvm.pm = "systemd" or "auto". Transient services are named +# -.service. +unit_prefix = "dstack-vm" +# Process metadata used to reconcile transient services after a VMM restart. +# Empty defaults to ~/.dstack-vmm/systemd-processes. +state_dir = "" +# How long systemd waits after SIGTERM before escalating to SIGKILL. Large TDX +# guests can spend hours tearing down encrypted memory, so the safe default is +# unbounded. Set a systemd time span such as "30min" to enable escalation. +stop_timeout = "infinity" + [host_api] ident = "dstack VMM" address = "vsock:2"