feat(vmm): add experimental systemd process manager

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-<sha256(vm-id)>.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=<systemd.stop_timeout>
+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-<digest>.service \
+  -p ActiveState -p SubState -p MainPID -p ControlGroup
+systemd-cgls /system.slice/dstack-vm-<digest>.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.

dstack/supervisor/client/src/lib.rs

@@ -33,12 +33,15 @@ impl SupervisorClient {
     ) -> Result<Self> {
         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"),
         }
     }
 

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<Config>,
-    pub supervisor: SupervisorClient,
+    pub process_manager: ProcessManager,
     state: Arc<Mutex<AppState>>,
     /// Pull status for registry images: tag → status.
     pub(crate) pull_status: Arc<Mutex<std::collections::HashMap<String, PullStatus>>>,
@@ -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<String, _> = 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<StatusResponse> {
         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<Option<pb::VmInfo>> {
-        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<PathBuf> {