From 235c5c2e966109841d4371604233b1221284c266 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Thu, 6 Aug 2026 06:27:48 -0700 Subject: [PATCH 1/7] feat(vmm): add experimental systemd process manager --- docs/experimental-systemd-vm-processes.md | 79 +++++ dstack/vmm/src/app.rs | 6 +- dstack/vmm/src/config.rs | 22 ++ dstack/vmm/src/main.rs | 13 +- dstack/vmm/src/process_manager.rs | 333 ++++++++++++++++++++++ dstack/vmm/vmm.toml | 7 + 6 files changed, 454 insertions(+), 6 deletions(-) create mode 100644 docs/experimental-systemd-vm-processes.md create mode 100644 dstack/vmm/src/process_manager.rs diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md new file mode 100644 index 000000000..55b403f63 --- /dev/null +++ b/docs/experimental-systemd-vm-processes.md @@ -0,0 +1,79 @@ +# 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 +[supervisor] +backend = "systemd" +systemd_unit_prefix = "dstack-vm" +systemd_state_dir = "/run/dstack-vmm/systemd-processes" +``` + +The section retains its historical name so existing configurations remain +compatible. The default backend is `supervisor`. + +## 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=infinity +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. + +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. + +## 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. +- Unit status is currently polled through `systemctl show`. +- 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/vmm/src/app.rs b/dstack/vmm/src/app.rs index 90b9bcc1d..f66994aa4 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 supervisor: ProcessManager, state: Arc>, /// Pull status for registry images: tag → status. pub(crate) pull_status: Arc>>, @@ -311,7 +311,7 @@ impl App { Ok(VmWorkDir::new(self.config.run_path.join(id))) } - pub fn new(config: Config, supervisor: SupervisorClient) -> Self { + pub fn new(config: Config, supervisor: 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); diff --git a/dstack/vmm/src/config.rs b/dstack/vmm/src/config.rs index f33b9bae4..d16b0ec81 100644 --- a/dstack/vmm/src/config.rs +++ b/dstack/vmm/src/config.rs @@ -477,14 +477,36 @@ pub struct AuthConfig { pub htpasswd_file: PathBuf, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProcessManagerBackend { + #[default] + Supervisor, + Systemd, +} + +fn default_systemd_unit_prefix() -> String { + "dstack-vm".into() +} + +fn default_systemd_state_dir() -> PathBuf { + "./run/systemd-processes".into() +} + #[derive(Debug, Clone, Default, Deserialize)] pub struct SupervisorConfig { + #[serde(default)] + pub backend: ProcessManagerBackend, pub exe: String, pub sock: String, pub pid_file: String, pub log_file: String, pub detached: bool, pub auto_start: bool, + #[serde(default = "default_systemd_unit_prefix")] + pub systemd_unit_prefix: String, + #[serde(default = "default_systemd_state_dir")] + pub systemd_state_dir: PathBuf, } #[derive(Debug, Clone, Deserialize)] diff --git a/dstack/vmm/src/main.rs b/dstack/vmm/src/main.rs index 9e97b0d6e..744b398c6 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"); @@ -328,10 +329,15 @@ async fn main() -> Result<()> { token, or bind `address` to localhost / a Unix socket." ); } - let supervisor = { + let supervisor = if config.supervisor.backend == config::ProcessManagerBackend::Systemd { + process_manager::ProcessManager::systemd( + config.supervisor.systemd_state_dir.clone(), + config.supervisor.systemd_unit_prefix.clone(), + )? + } else { let cfg = &config.supervisor; let abs_exe = Path::new(&cfg.exe).absolutize()?; - SupervisorClient::start_and_connect_uds( + let client = SupervisorClient::start_and_connect_uds( &abs_exe, &cfg.sock, &cfg.pid_file, @@ -340,7 +346,8 @@ async fn main() -> Result<()> { cfg.auto_start, ) .await - .context("Failed to connect to supervisor")? + .context("Failed to connect to supervisor")?; + process_manager::ProcessManager::supervisor(client) }; let state = app::App::new(config, supervisor); state.reload_vms().await.context("Failed to reload VMs")?; diff --git a/dstack/vmm/src/process_manager.rs b/dstack/vmm/src/process_manager.rs new file mode 100644 index 000000000..e7ea9c11e --- /dev/null +++ b/dstack/vmm/src/process_manager.rs @@ -0,0 +1,333 @@ +// SPDX-FileCopyrightText: © 2026 Phala Network +// +// SPDX-License-Identifier: Apache-2.0 + +use std::path::PathBuf; +use std::process::Stdio; +use std::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; + +#[derive(Clone)] +pub enum ProcessManager { + Supervisor(SupervisorClient), + Systemd(Arc), +} + +impl ProcessManager { + pub fn supervisor(client: SupervisorClient) -> Self { + Self::Supervisor(client) + } + + pub fn systemd(state_dir: PathBuf, unit_prefix: String) -> Result { + Ok(Self::Systemd(Arc::new(SystemdProcessManager::new( + state_dir, + unit_prefix, + )?))) + } + + pub async fn deploy(&self, config: &ProcessConfig) -> Result<()> { + match self { + Self::Supervisor(client) => client.deploy(config).await, + Self::Systemd(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, + } + } + + pub async fn remove(&self, id: &str) -> Result<()> { + match self { + Self::Supervisor(client) => client.remove(id).await, + Self::Systemd(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, + } + } + + pub async fn info(&self, id: &str) -> Result> { + match self { + Self::Supervisor(client) => client.info(id).await, + Self::Systemd(manager) => manager.info(id).await, + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct ProcessRecord { + config: ProcessConfig, + started: bool, +} + +pub struct SystemdProcessManager { + state_dir: PathBuf, + unit_prefix: String, +} + +impl SystemdProcessManager { + fn new(state_dir: PathBuf, unit_prefix: 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, + }) + } + + 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); + 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("--property=TimeoutStopSec=infinity") + .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)?; + if let Err(error) = self.launch(config).await { + let _ = fs_err::remove_file(self.record_path(&config.id)); + return Err(error); + } + Ok(()) + } + + 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)); + Self::command(command, "systemctl stop").await?; + } + 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 raw = fs_err::read(entry.path())?; + let record: ProcessRecord = serde_json::from_slice(&raw)?; + if let Some(info) = self.info_from_record(record).await? { + processes.push(info); + } + } + Ok(processes) + } + + async fn info(&self, id: &str) -> Result> { + let path = self.record_path(id); + if !path.exists() { + return Ok(None); + } + self.info_from_record(self.read_record(id)?).await + } + + 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") + .stdout(Stdio::piped()); + let output = command + .output() + .await + .context("failed to execute systemctl show")?; + let properties = String::from_utf8_lossy(&output.stdout); + 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") + || matches!( + sub_state, + "running" | "start" | "stop-sigterm" | "stop-sigkill" + ); + let status = if running { + ProcessStatus::Running + } else if !record.started { + ProcessStatus::Stopped + } else if load_state == "not-found" || active_state == "inactive" { + ProcessStatus::Exited(value("ExecMainStatus").parse().unwrap_or_default()) + } 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); + Ok(Some(ProcessInfo { + config: record.config, + state: ProcessState { + status, + started: record.started, + pid, + started_at: None, + stopped_at: None, + }, + })) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unit_names_are_stable_and_do_not_embed_process_ids() { + let manager = + SystemdProcessManager::new(PathBuf::from("/tmp/test"), "dstack-vm".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")); + } +} diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index 06782be2d..2703057dd 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -178,6 +178,13 @@ tokens = [] htpasswd_file = "" [supervisor] +# Experimental: set to "systemd" to let dstack-vmm launch each VM as a +# transient systemd service instead of using the standalone supervisor. +backend = "supervisor" +# Transient services are named -.service. +systemd_unit_prefix = "dstack-vm" +# Process metadata used to reconcile transient services after a VMM restart. +systemd_state_dir = "./run/systemd-processes" exe = "./supervisor" sock = "./run/supervisor.sock" pid_file = "./run/supervisor.pid" From bae3ec45eda658daa23f70c9f221a73e264e68fd Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Fri, 7 Aug 2026 09:37:23 +0800 Subject: [PATCH 2/7] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- docs/experimental-systemd-vm-processes.md | 4 ++-- dstack/vmm/src/main.rs | 2 +- dstack/vmm/src/process_manager.rs | 12 +++++------- 3 files changed, 8 insertions(+), 10 deletions(-) diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md index 55b403f63..656dfeab5 100644 --- a/docs/experimental-systemd-vm-processes.md +++ b/docs/experimental-systemd-vm-processes.md @@ -14,8 +14,8 @@ systemd_unit_prefix = "dstack-vm" systemd_state_dir = "/run/dstack-vmm/systemd-processes" ``` -The section retains its historical name so existing configurations remain -compatible. The default backend is `supervisor`. +The section retains its historical name so existing configurations remain compatible. +The default backend is `supervisor`. Keep the existing `supervisor.*` settings (exe/sock/pid_file/log_file/...) in place; they are currently still required even when using `backend = "systemd"`. ## Runtime model diff --git a/dstack/vmm/src/main.rs b/dstack/vmm/src/main.rs index 744b398c6..4c982a585 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -346,7 +346,7 @@ async fn main() -> Result<()> { cfg.auto_start, ) .await - .context("Failed to connect to supervisor")?; + .context("failed to connect to supervisor")?; process_manager::ProcessManager::supervisor(client) }; let state = app::App::new(config, supervisor); diff --git a/dstack/vmm/src/process_manager.rs b/dstack/vmm/src/process_manager.rs index e7ea9c11e..e66a9f0ba 100644 --- a/dstack/vmm/src/process_manager.rs +++ b/dstack/vmm/src/process_manager.rs @@ -191,7 +191,7 @@ impl SystemdProcessManager { .await? .is_some_and(|info| info.state.status.is_running()) { - bail!("Process is already running"); + bail!("process is already running"); } let record = ProcessRecord { config: config.clone(), @@ -272,10 +272,7 @@ impl SystemdProcessManager { .arg(&unit) .arg("--property=LoadState,ActiveState,SubState,MainPID,ExecMainCode,ExecMainStatus") .stdout(Stdio::piped()); - let output = command - .output() - .await - .context("failed to execute systemctl show")?; + let output = Self::command(command, "systemctl show").await?; let properties = String::from_utf8_lossy(&output.stdout); let value = |name: &str| { properties @@ -324,8 +321,9 @@ mod tests { #[test] fn unit_names_are_stable_and_do_not_embed_process_ids() { - let manager = - SystemdProcessManager::new(PathBuf::from("/tmp/test"), "dstack-vm".into()).unwrap(); + let dir = tempfile::tempdir().unwrap(); + let manager = SystemdProcessManager::new(dir.path().to_path_buf(), "dstack-vm".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")); From f86dff8320d6c75fbcddd2a7b81a30d71e9703e0 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Thu, 6 Aug 2026 18:45:53 -0700 Subject: [PATCH 3/7] refactor(vmm): separate process manager configuration --- docs/experimental-systemd-vm-processes.md | 15 +++++++------ dstack/vmm/src/config.rs | 27 ++++++++++++++++++----- dstack/vmm/src/main.rs | 6 ++--- dstack/vmm/vmm.toml | 16 ++++++++------ 4 files changed, 42 insertions(+), 22 deletions(-) diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md index 656dfeab5..6f13dafb5 100644 --- a/docs/experimental-systemd-vm-processes.md +++ b/docs/experimental-systemd-vm-processes.md @@ -8,14 +8,15 @@ kernel-side shutdown, such as encrypted-memory teardown. Enable it in the VMM configuration: ```toml -[supervisor] -backend = "systemd" -systemd_unit_prefix = "dstack-vm" -systemd_state_dir = "/run/dstack-vmm/systemd-processes" +[cvm] +pm = "systemd" + +[systemd] +unit_prefix = "dstack-vm" +state_dir = "/run/dstack-vmm/systemd-processes" ``` -The section retains its historical name so existing configurations remain compatible. -The default backend is `supervisor`. Keep the existing `supervisor.*` settings (exe/sock/pid_file/log_file/...) in place; they are currently still required even when using `backend = "systemd"`. +The default process manager is `supervisor`. ## Runtime model @@ -51,7 +52,7 @@ 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. -Process metadata is persisted in `systemd_state_dir`. It is required because a +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. diff --git a/dstack/vmm/src/config.rs b/dstack/vmm/src/config.rs index d16b0ec81..e261c794c 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. @@ -495,18 +498,29 @@ fn default_systemd_state_dir() -> PathBuf { #[derive(Debug, Clone, Default, Deserialize)] pub struct SupervisorConfig { - #[serde(default)] - pub backend: ProcessManagerBackend, pub exe: String, pub sock: String, pub pid_file: String, pub log_file: String, pub detached: bool, pub auto_start: bool, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct SystemdConfig { #[serde(default = "default_systemd_unit_prefix")] - pub systemd_unit_prefix: String, + pub unit_prefix: String, #[serde(default = "default_systemd_state_dir")] - pub systemd_state_dir: PathBuf, + pub state_dir: PathBuf, +} + +impl Default for SystemdConfig { + fn default() -> Self { + Self { + unit_prefix: default_systemd_unit_prefix(), + state_dir: default_systemd_state_dir(), + } + } } #[derive(Debug, Clone, Deserialize)] @@ -549,10 +563,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, diff --git a/dstack/vmm/src/main.rs b/dstack/vmm/src/main.rs index 4c982a585..1e4673ba3 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -329,10 +329,10 @@ async fn main() -> Result<()> { token, or bind `address` to localhost / a Unix socket." ); } - let supervisor = if config.supervisor.backend == config::ProcessManagerBackend::Systemd { + let supervisor = if config.cvm.pm == config::ProcessManagerBackend::Systemd { process_manager::ProcessManager::systemd( - config.supervisor.systemd_state_dir.clone(), - config.supervisor.systemd_unit_prefix.clone(), + config.systemd.state_dir.clone(), + config.systemd.unit_prefix.clone(), )? } else { let cfg = &config.supervisor; diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index 2703057dd..ed3a7fbf4 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -21,6 +21,8 @@ node_name = "" registry = "" [cvm] +# Process manager: "supervisor" (default) or experimental "systemd". +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 = "" @@ -178,13 +180,6 @@ tokens = [] htpasswd_file = "" [supervisor] -# Experimental: set to "systemd" to let dstack-vmm launch each VM as a -# transient systemd service instead of using the standalone supervisor. -backend = "supervisor" -# Transient services are named -.service. -systemd_unit_prefix = "dstack-vm" -# Process metadata used to reconcile transient services after a VMM restart. -systemd_state_dir = "./run/systemd-processes" exe = "./supervisor" sock = "./run/supervisor.sock" pid_file = "./run/supervisor.pid" @@ -192,6 +187,13 @@ log_file = "./run/supervisor.log" detached = false auto_start = true +[systemd] +# Used when cvm.pm = "systemd". Transient services are named +# -.service. +unit_prefix = "dstack-vm" +# Process metadata used to reconcile transient services after a VMM restart. +state_dir = "./run/systemd-processes" + [host_api] ident = "dstack VMM" address = "vsock:2" From 80ce2662594104b3fe9a9562e3d9987b0b88f4af Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Thu, 6 Aug 2026 18:59:19 -0700 Subject: [PATCH 4/7] feat(vmm): add automatic process manager migration --- docs/experimental-systemd-vm-processes.md | 13 ++- dstack/vmm/src/config.rs | 5 + dstack/vmm/src/main.rs | 33 ++++-- dstack/vmm/src/process_manager.rs | 117 +++++++++++++++++++++- dstack/vmm/vmm.toml | 6 +- 5 files changed, 160 insertions(+), 14 deletions(-) diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md index 6f13dafb5..cb2e2a4a1 100644 --- a/docs/experimental-systemd-vm-processes.md +++ b/docs/experimental-systemd-vm-processes.md @@ -9,14 +9,23 @@ Enable it in the VMM configuration: ```toml [cvm] -pm = "systemd" +pm = "auto" [systemd] unit_prefix = "dstack-vm" state_dir = "/run/dstack-vmm/systemd-processes" ``` -The default process manager is `supervisor`. +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. ## Runtime model diff --git a/dstack/vmm/src/config.rs b/dstack/vmm/src/config.rs index e261c794c..1bea3e71e 100644 --- a/dstack/vmm/src/config.rs +++ b/dstack/vmm/src/config.rs @@ -483,9 +483,14 @@ pub struct AuthConfig { #[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 { diff --git a/dstack/vmm/src/main.rs b/dstack/vmm/src/main.rs index 1e4673ba3..e8b915c01 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -329,25 +329,42 @@ async fn main() -> Result<()> { token, or bind `address` to localhost / a Unix socket." ); } - let supervisor = if config.cvm.pm == config::ProcessManagerBackend::Systemd { + let systemd_manager = || { process_manager::ProcessManager::systemd( config.systemd.state_dir.clone(), config.systemd.unit_prefix.clone(), - )? - } else { - let cfg = &config.supervisor; + ) + }; + let supervisor_config = &config.supervisor; + let connect_supervisor = |auto_start| async move { + let cfg = supervisor_config; let abs_exe = Path::new(&cfg.exe).absolutize()?; - let client = SupervisorClient::start_and_connect_uds( + SupervisorClient::start_and_connect_uds( &abs_exe, &cfg.sock, &cfg.pid_file, &cfg.log_file, cfg.detached, - cfg.auto_start, + auto_start, ) .await - .context("failed to connect to supervisor")?; - process_manager::ProcessManager::supervisor(client) + .context("failed to connect to supervisor") + }; + let supervisor = match config.cvm.pm { + config::ProcessManagerBackend::Supervisor => process_manager::ProcessManager::supervisor( + connect_supervisor(supervisor_config.auto_start).await?, + ), + config::ProcessManagerBackend::Systemd => systemd_manager()?, + config::ProcessManagerBackend::Auto => { + let legacy_supervisor = match connect_supervisor(false).await { + Ok(client) => Some(client), + Err(error) => { + info!(%error, "legacy supervisor is not running; using systemd for all VMs"); + None + } + }; + process_manager::ProcessManager::auto(systemd_manager()?, legacy_supervisor).await? + } }; let state = app::App::new(config, supervisor); state.reload_vms().await.context("Failed to reload VMs")?; diff --git a/dstack/vmm/src/process_manager.rs b/dstack/vmm/src/process_manager.rs index e66a9f0ba..27df665d2 100644 --- a/dstack/vmm/src/process_manager.rs +++ b/dstack/vmm/src/process_manager.rs @@ -4,7 +4,7 @@ use std::path::PathBuf; use std::process::Stdio; -use std::sync::Arc; +use std::{collections::HashSet, sync::Arc}; use anyhow::{bail, Context, Result}; use serde::{Deserialize, Serialize}; @@ -12,11 +12,13 @@ use sha2::{Digest, Sha256}; use supervisor_client::supervisor::{ProcessConfig, ProcessInfo, ProcessState, ProcessStatus}; use supervisor_client::SupervisorClient; use tokio::process::Command; +use tokio::sync::RwLock; #[derive(Clone)] pub enum ProcessManager { Supervisor(SupervisorClient), Systemd(Arc), + Auto(Arc), } impl ProcessManager { @@ -31,10 +33,30 @@ impl ProcessManager { )?))) } + pub async fn auto(systemd: Self, supervisor: Option) -> Result { + let Self::Systemd(systemd) = systemd else { + bail!("auto process manager requires a systemd backend"); + }; + let mut supervisor_processes = HashSet::new(); + if let Some(client) = &supervisor { + for process in client.list().await? { + if process.state.status.is_running() { + supervisor_processes.insert(process.config.id); + } + } + } + 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, } } @@ -42,6 +64,7 @@ impl ProcessManager { match self { Self::Supervisor(client) => client.stop(id).await, Self::Systemd(manager) => manager.stop(id).await, + Self::Auto(manager) => manager.stop(id).await, } } @@ -49,6 +72,7 @@ impl ProcessManager { match self { Self::Supervisor(client) => client.remove(id).await, Self::Systemd(manager) => manager.remove(id).await, + Self::Auto(manager) => manager.remove(id).await, } } @@ -56,6 +80,7 @@ impl ProcessManager { match self { Self::Supervisor(client) => client.list().await, Self::Systemd(manager) => manager.list().await, + Self::Auto(manager) => manager.list().await, } } @@ -63,6 +88,92 @@ impl ProcessManager { 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(id) + } + + fn supervisor(&self) -> Result<&SupervisorClient> { + self.supervisor + .as_ref() + .context("legacy supervisor is unavailable") + } + + async fn deploy(&self, config: &ProcessConfig) -> Result<()> { + if self.is_supervisor_process(&config.id).await { + let supervisor = self.supervisor()?; + if supervisor + .info(&config.id) + .await? + .is_some_and(|info| info.state.status.is_running()) + { + bail!("process is already running"); + } + // Natural exits leave Supervisor's `started` flag set. Normalize + // it before removing the legacy record and migrating this launch. + supervisor.stop(&config.id).await?; + supervisor.remove(&config.id).await?; + 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 ids = self + .supervisor_processes + .read() + .await + .iter() + .cloned() + .collect::>(); + if !ids.is_empty() { + let supervisor = self.supervisor()?; + for id in ids { + if let Some(process) = supervisor.info(&id).await? { + processes.push(process); + } + } + } + Ok(processes) + } + + async fn info(&self, id: &str) -> Result> { + if self.is_supervisor_process(id).await { + self.supervisor()?.info(id).await + } else { + self.systemd.info(id).await } } } @@ -322,8 +433,8 @@ mod tests { #[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()) - .unwrap(); + let manager = + SystemdProcessManager::new(dir.path().to_path_buf(), "dstack-vm".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")); diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index ed3a7fbf4..6697b7241 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -21,7 +21,11 @@ node_name = "" registry = "" [cvm] -# Process manager: "supervisor" (default) or experimental "systemd". +# 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" From 11f6d87f5320a2f6a52224345d78d2f327c5c42d Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Thu, 6 Aug 2026 20:29:31 -0700 Subject: [PATCH 5/7] fix(vmm): harden systemd process lifecycle --- docs/experimental-systemd-vm-processes.md | 12 +- dstack/vmm/src/app.rs | 51 ++--- dstack/vmm/src/config.rs | 28 ++- dstack/vmm/src/main.rs | 35 +++- dstack/vmm/src/main_service.rs | 8 +- dstack/vmm/src/process_manager.rs | 221 +++++++++++++++------- dstack/vmm/vmm.toml | 2 +- 7 files changed, 242 insertions(+), 115 deletions(-) diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md index cb2e2a4a1..0990b5dcd 100644 --- a/docs/experimental-systemd-vm-processes.md +++ b/docs/experimental-systemd-vm-processes.md @@ -13,7 +13,7 @@ pm = "auto" [systemd] unit_prefix = "dstack-vm" -state_dir = "/run/dstack-vmm/systemd-processes" +state_dir = "/var/lib/dstack-vmm/systemd-processes" ``` The three process-manager modes are: @@ -25,7 +25,9 @@ The three process-manager modes are: lifecycle. Their next VM launch removes the stopped Supervisor record and migrates them to systemd. -The default is `supervisor`, preserving existing deployments. +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 @@ -52,7 +54,7 @@ ExitType=cgroup KillMode=mixed KillSignal=SIGTERM SendSIGKILL=yes -TimeoutStopSec=infinity +TimeoutStopSec=30min Restart=no ``` @@ -64,6 +66,7 @@ still completing kernel teardown. 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 @@ -83,6 +86,9 @@ atomic property handling and event-driven state updates. - 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`. - Start and stop are not yet transactional with the metadata file. - A host reboot removes transient units; normal VMM workdir recovery recreates diff --git a/dstack/vmm/src/app.rs b/dstack/vmm/src/app.rs index f66994aa4..bc88f0d1e 100644 --- a/dstack/vmm/src/app.rs +++ b/dstack/vmm/src/app.rs @@ -291,7 +291,7 @@ pub(crate) enum PullStatus { #[derive(Clone)] pub struct App { pub config: Arc, - pub supervisor: ProcessManager, + 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: ProcessManager) -> 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 1bea3e71e..3329ff15f 100644 --- a/dstack/vmm/src/config.rs +++ b/dstack/vmm/src/config.rs @@ -498,7 +498,7 @@ fn default_systemd_unit_prefix() -> String { } fn default_systemd_state_dir() -> PathBuf { - "./run/systemd-processes".into() + PathBuf::new() } #[derive(Debug, Clone, Default, Deserialize)] @@ -679,6 +679,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) } @@ -719,11 +720,13 @@ impl Config { } validate_networking(&self.cvm.networking)?; - anyhow::ensure!( - !self.supervisor.sock.trim().is_empty(), - "supervisor.sock must not be empty" - ); - if self.supervisor.auto_start { + if self.cvm.pm != ProcessManagerBackend::Systemd { + anyhow::ensure!( + !self.supervisor.sock.trim().is_empty(), + "supervisor.sock must not be empty unless cvm.pm = \"systemd\"" + ); + } + 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()), @@ -941,6 +944,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() { @@ -1130,6 +1136,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 e8b915c01..2dba5aaeb 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -350,23 +350,40 @@ async fn main() -> Result<()> { .await .context("failed to connect to supervisor") }; - let supervisor = match config.cvm.pm { + 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 => systemd_manager()?, + config::ProcessManagerBackend::Systemd => { + if legacy_socket_exists { + let client = connect_supervisor(false).await.context( + "supervisor socket exists but its state cannot be verified in systemd mode", + )?; + anyhow::ensure!( + !client + .list() + .await? + .iter() + .any(|process| process.state.status.is_running()), + "running Supervisor VMs detected; use cvm.pm = \"auto\" for migration" + ); + } + systemd_manager()? + } config::ProcessManagerBackend::Auto => { - let legacy_supervisor = match connect_supervisor(false).await { - Ok(client) => Some(client), - Err(error) => { - info!(%error, "legacy supervisor is not running; using systemd for all VMs"); - None - } + let legacy_supervisor = if legacy_socket_exists { + Some(connect_supervisor(false).await.context( + "supervisor socket exists but its state cannot be verified in auto mode", + )?) + } 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 index 27df665d2..f1debb482 100644 --- a/dstack/vmm/src/process_manager.rs +++ b/dstack/vmm/src/process_manager.rs @@ -3,8 +3,7 @@ // SPDX-License-Identifier: Apache-2.0 use std::path::PathBuf; -use std::process::Stdio; -use std::{collections::HashSet, sync::Arc}; +use std::{collections::HashMap, sync::Arc}; use anyhow::{bail, Context, Result}; use serde::{Deserialize, Serialize}; @@ -13,6 +12,7 @@ use supervisor_client::supervisor::{ProcessConfig, ProcessInfo, ProcessState, Pr use supervisor_client::SupervisorClient; use tokio::process::Command; use tokio::sync::RwLock; +use tracing::warn; #[derive(Clone)] pub enum ProcessManager { @@ -37,12 +37,10 @@ impl ProcessManager { let Self::Systemd(systemd) = systemd else { bail!("auto process manager requires a systemd backend"); }; - let mut supervisor_processes = HashSet::new(); + let mut supervisor_processes = HashMap::new(); if let Some(client) = &supervisor { for process in client.list().await? { - if process.state.status.is_running() { - supervisor_processes.insert(process.config.id); - } + supervisor_processes.insert(process.config.id.clone(), process); } } Ok(Self::Auto(Arc::new(AutoProcessManager { @@ -98,12 +96,12 @@ pub struct AutoProcessManager { supervisor: Option, /// VM processes found running in Supervisor when the VMM started. They /// stay pinned to Supervisor until their next deploy. - supervisor_processes: RwLock>, + supervisor_processes: RwLock>, } impl AutoProcessManager { async fn is_supervisor_process(&self, id: &str) -> bool { - self.supervisor_processes.read().await.contains(id) + self.supervisor_processes.read().await.contains_key(id) } fn supervisor(&self) -> Result<&SupervisorClient> { @@ -113,6 +111,8 @@ impl AutoProcessManager { } async fn deploy(&self, config: &ProcessConfig) -> Result<()> { + // VMM start operations are serialized by the caller. If that changes, + // migration should gain a per-ID lock spanning this handoff. if self.is_supervisor_process(&config.id).await { let supervisor = self.supervisor()?; if supervisor @@ -151,18 +151,25 @@ impl AutoProcessManager { async fn list(&self) -> Result> { let mut processes = self.systemd.list().await?; - let ids = self - .supervisor_processes - .read() - .await - .iter() - .cloned() - .collect::>(); - if !ids.is_empty() { + let pinned = self.supervisor_processes.read().await.clone(); + if !pinned.is_empty() { let supervisor = self.supervisor()?; - for id in ids { - if let Some(process) = supervisor.info(&id).await? { - processes.push(process); + 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; + for process in &legacy { + cache.insert(process.config.id.clone(), 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()); } } } @@ -171,7 +178,21 @@ impl AutoProcessManager { async fn info(&self, id: &str) -> Result> { if self.is_supervisor_process(id).await { - self.supervisor()?.info(id).await + match self.supervisor()?.info(id).await { + Ok(info) => { + if let Some(process) = &info { + self.supervisor_processes + .write() + .await + .insert(id.to_string(), process.clone()); + } + 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 } @@ -184,6 +205,39 @@ struct ProcessRecord { started: bool, } +fn state_from_systemd_properties(properties: &str, started: bool) -> (ProcessStatus, 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 status = if running { + ProcessStatus::Running + } else if !started { + ProcessStatus::Stopped + } else if load_state == "not-found" || matches!(active_state, "inactive" | "failed") { + let status = value("ExecMainStatus").parse::().unwrap_or_default(); + ProcessStatus::Exited(if value("ExecMainCode") == "exited" { + status << 8 + } else { + 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); + (status, pid) +} + pub struct SystemdProcessManager { state_dir: PathBuf, unit_prefix: String, @@ -249,6 +303,11 @@ impl SystemdProcessManager { 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") @@ -258,7 +317,7 @@ impl SystemdProcessManager { .arg("--property=KillMode=mixed") .arg("--property=KillSignal=SIGTERM") .arg("--property=SendSIGKILL=yes") - .arg("--property=TimeoutStopSec=infinity") + .arg("--property=TimeoutStopSec=30min") .arg("--property=ExitType=cgroup") .arg("--property=Restart=no") .arg(format!("--description=dstack VM process {}", config.id)); @@ -309,11 +368,7 @@ impl SystemdProcessManager { started: true, }; self.write_record(&record)?; - if let Err(error) = self.launch(config).await { - let _ = fs_err::remove_file(self.record_path(&config.id)); - return Err(error); - } - Ok(()) + self.launch(config).await } async fn stop(&self, id: &str) -> Result<()> { @@ -339,11 +394,11 @@ impl SystemdProcessManager { .await? .is_some_and(|info| info.state.status.is_running()) { - bail!("Process is running"); + bail!("process is running"); } let record = self.read_record(id)?; if record.started { - bail!("Process is started"); + bail!("process is started"); } let mut command = Command::new("systemctl"); command.arg("reset-failed").arg(self.unit(id)); @@ -358,62 +413,57 @@ impl SystemdProcessManager { if entry.path().extension().and_then(|value| value.to_str()) != Some("json") { continue; } - let raw = fs_err::read(entry.path())?; - let record: ProcessRecord = serde_json::from_slice(&raw)?; - if let Some(info) = self.info_from_record(record).await? { - processes.push(info); + 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; + } + }; + match self.info_from_record(record).await { + Ok(info) => processes.push(info), + Err(error) => { + warn!(path = %path.display(), %error, "skipping unavailable systemd process") + } } } Ok(processes) } async fn info(&self, id: &str) -> Result> { - let path = self.record_path(id); - if !path.exists() { - return Ok(None); + 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), } - self.info_from_record(self.read_record(id)?).await } - async fn info_from_record(&self, record: ProcessRecord) -> Result> { + 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") - .stdout(Stdio::piped()); - let output = Self::command(command, "systemctl show").await?; + .arg("--property=LoadState,ActiveState,SubState,MainPID,ExecMainCode,ExecMainStatus"); + // `systemctl show` returns non-zero for a collected transient unit but + // still emits LoadState=not-found, which is a valid stopped state. + let output = command + .output() + .await + .context("failed to execute systemctl show")?; let properties = String::from_utf8_lossy(&output.stdout); - 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") - || matches!( - sub_state, - "running" | "start" | "stop-sigterm" | "stop-sigkill" - ); - let status = if running { - ProcessStatus::Running - } else if !record.started { - ProcessStatus::Stopped - } else if load_state == "not-found" || active_state == "inactive" { - ProcessStatus::Exited(value("ExecMainStatus").parse().unwrap_or_default()) - } 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); - Ok(Some(ProcessInfo { + let (status, pid) = state_from_systemd_properties(&properties, record.started); + Ok(ProcessInfo { config: record.config, state: ProcessState { status, @@ -422,7 +472,7 @@ impl SystemdProcessManager { started_at: None, stopped_at: None, }, - })) + }) } } @@ -439,4 +489,35 @@ mod tests { 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=killed\nExecMainStatus=9", + true + ), + ProcessStatus::Exited(9) + )); + } } diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index 6697b7241..ba9359a89 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -196,7 +196,7 @@ auto_start = true # -.service. unit_prefix = "dstack-vm" # Process metadata used to reconcile transient services after a VMM restart. -state_dir = "./run/systemd-processes" +state_dir = "" [host_api] ident = "dstack VMM" From 29ef71c923d90b19edbb40dfa63fd412fa78389a Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Thu, 6 Aug 2026 21:28:25 -0700 Subject: [PATCH 6/7] fix(vmm): preserve systemd manager availability semantics --- docs/experimental-systemd-vm-processes.md | 7 +- dstack/vmm/src/config.rs | 11 +++ dstack/vmm/src/main.rs | 11 ++- dstack/vmm/src/process_manager.rs | 111 +++++++++++++++------- dstack/vmm/vmm.toml | 7 +- 5 files changed, 106 insertions(+), 41 deletions(-) diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md index 0990b5dcd..40216648e 100644 --- a/docs/experimental-systemd-vm-processes.md +++ b/docs/experimental-systemd-vm-processes.md @@ -14,6 +14,7 @@ pm = "auto" [systemd] unit_prefix = "dstack-vm" state_dir = "/var/lib/dstack-vmm/systemd-processes" +stop_timeout = "infinity" ``` The three process-manager modes are: @@ -54,7 +55,7 @@ ExitType=cgroup KillMode=mixed KillSignal=SIGTERM SendSIGKILL=yes -TimeoutStopSec=30min +TimeoutStopSec= Restart=no ``` @@ -63,6 +64,10 @@ 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. diff --git a/dstack/vmm/src/config.rs b/dstack/vmm/src/config.rs index 3329ff15f..f3ca78d3b 100644 --- a/dstack/vmm/src/config.rs +++ b/dstack/vmm/src/config.rs @@ -501,6 +501,10 @@ 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, @@ -517,6 +521,8 @@ pub struct SystemdConfig { 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 { @@ -524,6 +530,7 @@ impl Default for SystemdConfig { Self { unit_prefix: default_systemd_unit_prefix(), state_dir: default_systemd_state_dir(), + stop_timeout: default_systemd_stop_timeout(), } } } @@ -726,6 +733,10 @@ impl Config { "supervisor.sock must not be empty unless cvm.pm = \"systemd\"" ); } + anyhow::ensure!( + !self.systemd.stop_timeout.trim().is_empty(), + "systemd.stop_timeout must not be empty" + ); if self.cvm.pm == ProcessManagerBackend::Supervisor && self.supervisor.auto_start { for (name, value) in [ ("supervisor.exe", self.supervisor.exe.as_str()), diff --git a/dstack/vmm/src/main.rs b/dstack/vmm/src/main.rs index 2dba5aaeb..2e02877de 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -330,9 +330,10 @@ async fn main() -> Result<()> { ); } let systemd_manager = || { - process_manager::ProcessManager::systemd( + 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; @@ -358,7 +359,8 @@ async fn main() -> Result<()> { config::ProcessManagerBackend::Systemd => { if legacy_socket_exists { let client = connect_supervisor(false).await.context( - "supervisor socket exists but its state cannot be verified in systemd mode", + "supervisor socket exists but its state cannot be verified in systemd mode; \ + if Supervisor is definitely not running, remove the stale socket and restart", )?; anyhow::ensure!( !client @@ -369,12 +371,13 @@ async fn main() -> Result<()> { "running Supervisor VMs detected; use cvm.pm = \"auto\" for migration" ); } - systemd_manager()? + process_manager::ProcessManager::systemd(systemd_manager()?) } config::ProcessManagerBackend::Auto => { let legacy_supervisor = if legacy_socket_exists { Some(connect_supervisor(false).await.context( - "supervisor socket exists but its state cannot be verified in auto mode", + "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"); diff --git a/dstack/vmm/src/process_manager.rs b/dstack/vmm/src/process_manager.rs index f1debb482..af97006ee 100644 --- a/dstack/vmm/src/process_manager.rs +++ b/dstack/vmm/src/process_manager.rs @@ -3,6 +3,7 @@ // 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}; @@ -26,17 +27,26 @@ impl ProcessManager { Self::Supervisor(client) } - pub fn systemd(state_dir: PathBuf, unit_prefix: String) -> Result { - Ok(Self::Systemd(Arc::new(SystemdProcessManager::new( + 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 async fn auto(systemd: Self, supervisor: Option) -> Result { - let Self::Systemd(systemd) = systemd else { - bail!("auto process manager requires a systemd backend"); - }; + 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? { @@ -111,8 +121,8 @@ impl AutoProcessManager { } async fn deploy(&self, config: &ProcessConfig) -> Result<()> { - // VMM start operations are serialized by the caller. If that changes, - // migration should gain a per-ID lock spanning this handoff. + // 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()?; if supervisor @@ -205,7 +215,32 @@ struct ProcessRecord { started: bool, } -fn state_from_systemd_properties(properties: &str, started: bool) -> (ProcessStatus, Option) { +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 and CLOCK_BOOTTIME is a + // process-independent monotonic clock on Linux. + if unsafe { libc::clock_gettime(libc::CLOCK_BOOTTIME, &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() @@ -222,10 +257,10 @@ fn state_from_systemd_properties(properties: &str, started: bool) -> (ProcessSta ProcessStatus::Stopped } else if load_state == "not-found" || matches!(active_state, "inactive" | "failed") { let status = value("ExecMainStatus").parse::().unwrap_or_default(); - ProcessStatus::Exited(if value("ExecMainCode") == "exited" { - status << 8 - } else { - status + ProcessStatus::Exited(match value("ExecMainCode") { + "exited" => status << 8, + "dumped" => status | 0x80, + _ => status, }) } else { ProcessStatus::Error(format!( @@ -235,16 +270,19 @@ fn state_from_systemd_properties(properties: &str, started: bool) -> (ProcessSta )) }; let pid = value("MainPID").parse().ok().filter(|pid| *pid != 0); - (status, pid) + 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) -> Result { + 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" @@ -259,6 +297,7 @@ impl SystemdProcessManager { Ok(Self { state_dir, unit_prefix, + stop_timeout, }) } @@ -317,7 +356,7 @@ impl SystemdProcessManager { .arg("--property=KillMode=mixed") .arg("--property=KillSignal=SIGTERM") .arg("--property=SendSIGKILL=yes") - .arg("--property=TimeoutStopSec=30min") + .arg(format!("--property=TimeoutStopSec={}", self.stop_timeout)) .arg("--property=ExitType=cgroup") .arg("--property=Restart=no") .arg(format!("--description=dstack VM process {}", config.id)); @@ -424,12 +463,10 @@ impl SystemdProcessManager { continue; } }; - match self.info_from_record(record).await { - Ok(info) => processes.push(info), - Err(error) => { - warn!(path = %path.display(), %error, "skipping unavailable systemd process") - } - } + // 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) } @@ -454,23 +491,23 @@ impl SystemdProcessManager { command .arg("show") .arg(&unit) - .arg("--property=LoadState,ActiveState,SubState,MainPID,ExecMainCode,ExecMainStatus"); - // `systemctl show` returns non-zero for a collected transient unit but - // still emits LoadState=not-found, which is a valid stopped state. - let output = command - .output() - .await - .context("failed to execute systemctl show")?; + .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) = state_from_systemd_properties(&properties, record.started); + 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: None, - stopped_at: None, + started_at, + stopped_at, }, }) } @@ -483,8 +520,12 @@ mod tests { #[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()).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")); diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index ba9359a89..74826f634 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -192,11 +192,16 @@ detached = false auto_start = true [systemd] -# Used when cvm.pm = "systemd". Transient services are named +# 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" From 9742a2ac566def2911501bf2122dfa14f6de7d65 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Thu, 6 Aug 2026 22:01:38 -0700 Subject: [PATCH 7/7] fix(vmm): close auto migration lifecycle races --- docs/experimental-systemd-vm-processes.md | 5 ++ dstack/supervisor/client/src/lib.rs | 21 +++--- dstack/vmm/src/main.rs | 52 +++++++++----- dstack/vmm/src/process_manager.rs | 84 ++++++++++++++++------- 4 files changed, 110 insertions(+), 52 deletions(-) diff --git a/docs/experimental-systemd-vm-processes.md b/docs/experimental-systemd-vm-processes.md index 40216648e..e2a7549c1 100644 --- a/docs/experimental-systemd-vm-processes.md +++ b/docs/experimental-systemd-vm-processes.md @@ -95,6 +95,11 @@ atomic property handling and event-driven state updates. 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/main.rs b/dstack/vmm/src/main.rs index 2e02877de..da640e196 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -42,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 { @@ -358,27 +366,39 @@ async fn main() -> Result<()> { ), config::ProcessManagerBackend::Systemd => { if legacy_socket_exists { - let client = connect_supervisor(false).await.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", - )?; - anyhow::ensure!( - !client - .list() - .await? - .iter() - .any(|process| process.state.status.is_running()), - "running Supervisor VMs detected; use cvm.pm = \"auto\" for migration" - ); + 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 { - Some(connect_supervisor(false).await.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", - )?) + 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 diff --git a/dstack/vmm/src/process_manager.rs b/dstack/vmm/src/process_manager.rs index af97006ee..e102dd2ee 100644 --- a/dstack/vmm/src/process_manager.rs +++ b/dstack/vmm/src/process_manager.rs @@ -125,17 +125,21 @@ impl AutoProcessManager { // the handoff and the other fails on record/unit removal or collision. if self.is_supervisor_process(&config.id).await { let supervisor = self.supervisor()?; - if supervisor - .info(&config.id) - .await? - .is_some_and(|info| info.state.status.is_running()) - { - bail!("process is already running"); + 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. + } } - // Natural exits leave Supervisor's `started` flag set. Normalize - // it before removing the legacy record and migrating this launch. - supervisor.stop(&config.id).await?; - supervisor.remove(&config.id).await?; self.supervisor_processes.write().await.remove(&config.id); } self.systemd.deploy(config).await @@ -171,8 +175,15 @@ impl AutoProcessManager { .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 { - cache.insert(process.config.id.clone(), process.clone()); + if let Some(cached) = cache.get_mut(&process.config.id) { + *cached = process.clone(); + } } drop(cache); processes.extend(legacy); @@ -191,10 +202,11 @@ impl AutoProcessManager { match self.supervisor()?.info(id).await { Ok(info) => { if let Some(process) = &info { - self.supervisor_processes - .write() - .await - .insert(id.to_string(), process.clone()); + 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) } @@ -221,9 +233,9 @@ fn system_time_from_monotonic_micros(value: &str) -> Option { tv_sec: 0, tv_nsec: 0, }; - // SAFETY: `now` points to a valid timespec and CLOCK_BOOTTIME is a - // process-independent monotonic clock on Linux. - if unsafe { libc::clock_gettime(libc::CLOCK_BOOTTIME, &mut now) } != 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) @@ -251,17 +263,20 @@ fn state_from_systemd_properties( 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 { + } else if !started && raw_status == 0 { ProcessStatus::Stopped } else if load_state == "not-found" || matches!(active_state, "inactive" | "failed") { - let status = value("ExecMainStatus").parse::().unwrap_or_default(); - ProcessStatus::Exited(match value("ExecMainCode") { - "exited" => status << 8, - "dumped" => status | 0x80, - _ => status, - }) + // 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={})", @@ -422,7 +437,17 @@ impl SystemdProcessManager { { let mut command = Command::new("systemctl"); command.arg("stop").arg("--no-block").arg(self.unit(id)); - Self::command(command, "systemctl stop").await?; + 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(()) } @@ -553,6 +578,13 @@ mod tests { ), 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",