Merge pull request #270 from Dstack-TEE/passt

Support for using passt as network egress

Author: kvinwang

supervisor/client/src/lib.rs

@@ -110,7 +110,7 @@ impl SupervisorClient {
 
 // Async API
 impl SupervisorClient {
-    pub async fn deploy(&self, config: ProcessConfig) -> Result<()> {
+    pub async fn deploy(&self, config: &ProcessConfig) -> Result<()> {
         self.http_request("POST", "/deploy", config).await
     }
 

vmm/src/app.rs

@@ -1,4 +1,4 @@
-use crate::config::{Config, Protocol};
+use crate::config::{Config, ProcessAnnotation, Protocol};
 
 use anyhow::{bail, Context, Result};
 use bon::Builder;
@@ -174,7 +174,6 @@ impl App {
                 manifest,
                 image,
                 cid,
-                networking: self.config.networking.clone(),
                 workdir: vm_work_dir.path().to_path_buf(),
                 gateway_enabled: app_compose.gateway_enabled(),
             };
@@ -213,6 +212,11 @@ impl App {
             vm_state.config.clone()
         };
         if !is_running {
+            // Try to stop passt if already running
+            if self.config.cvm.networking.is_passt() {
+                self.supervisor.stop(&format!("passt-{}", id)).await.ok();
+            }
+
             let work_dir = self.work_dir(id);
             for path in [work_dir.serial_pty(), work_dir.qmp_socket()] {
                 if path.symlink_metadata().is_ok() {
@@ -221,11 +225,13 @@ impl App {
             }
 
             let devices = self.try_allocate_gpus(&vm_config.manifest)?;
-            let process_config = vm_config.config_qemu(&work_dir, &self.config.cvm, &devices)?;
-            self.supervisor
-                .deploy(process_config)
-                .await
-                .with_context(|| format!("Failed to start VM {id}"))?;
+            let processes = vm_config.config_qemu(&work_dir, &self.config.cvm, &devices)?;
+            for process in processes {
+                self.supervisor
+                    .deploy(&process)
+                    .await
+                    .with_context(|| format!("Failed to start process {}", process.id))?;
+            }
 
             let mut state = self.lock();
             let vm_state = state.get_mut(id).context("VM not found")?;
@@ -259,6 +265,16 @@ impl App {
                 self.supervisor.stop(id).await?;
             }
             self.supervisor.remove(id).await?;
+            if self.config.cvm.networking.is_passt() {
+                let passt_id = format!("passt-{}", id);
+                let info = self.supervisor.info(&passt_id).await.ok().flatten();
+                if let Some(info) = info {
+                    if info.state.status.is_running() {
+                        self.supervisor.stop(&passt_id).await?;
+                    }
+                    self.supervisor.remove(&passt_id).await?;
+                }
+            }
         }
 
         {
@@ -276,9 +292,14 @@ 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: Vec<(ProcessAnnotation, _)> = running_vms
+            .into_iter()
+            .map(|p| (serde_json::from_str(&p.config.note).unwrap_or_default(), p))
+            .collect();
         let occupied_cids = running_vms
             .iter()
-            .flat_map(|p| p.config.cid.map(|cid| (p.config.id.clone(), cid)))
+            .filter(|(note, _)| note.is_cvm())
+            .flat_map(|(_, p)| p.config.cid.map(|cid| (p.config.id.clone(), cid)))
             .collect::<HashMap<_, _>>();
         {
             let mut state = self.lock();

vmm/src/app/qemu.rs

@@ -1,7 +1,7 @@
 //! QEMU related code
 use crate::{
     app::Manifest,
-    config::{CvmConfig, GatewayConfig, Networking},
+    config::{CvmConfig, GatewayConfig, Networking, PasstNetworking, ProcessAnnotation, Protocol},
 };
 use std::{collections::HashMap, os::unix::fs::PermissionsExt};
 use std::{
@@ -54,7 +54,6 @@ pub struct VmConfig {
     pub manifest: Manifest,
     pub image: Image,
     pub cid: u32,
-    pub networking: Networking,
     pub workdir: PathBuf,
     pub gateway_enabled: bool,
 }
@@ -220,12 +219,117 @@ impl VmState {
 }
 
 impl VmConfig {
+    fn config_passt(&self, workdir: &VmWorkDir, netcfg: &PasstNetworking) -> Result<ProcessConfig> {
+        let PasstNetworking {
+            passt_exec,
+            interface,
+            address,
+            netmask,
+            gateway,
+            dns,
+            map_host_loopback,
+            map_guest_addr,
+            no_map_gw,
+            ipv4_only,
+        } = netcfg;
+
+        let passt_socket = workdir.passt_socket();
+        if passt_socket.exists() {
+            fs_err::remove_file(&passt_socket).context("Failed to remove passt socket")?;
+        }
+        let passt_exec = if passt_exec.is_empty() {
+            "passt"
+        } else {
+            passt_exec
+        };
+
+        let passt_log = workdir.passt_log();
+
+        let mut passt_cmd = Command::new(passt_exec);
+        passt_cmd.arg("--socket").arg(&passt_socket);
+        passt_cmd.arg("--log-file").arg(&passt_log);
+
+        if !interface.is_empty() {
+            passt_cmd.arg("--interface").arg(interface);
+        }
+        if !address.is_empty() {
+            passt_cmd.arg("--address").arg(address);
+        }
+        if !netmask.is_empty() {
+            passt_cmd.arg("--netmask").arg(netmask);
+        }
+        if !gateway.is_empty() {
+            passt_cmd.arg("--gateway").arg(gateway);
+        }
+        for dns in dns {
+            passt_cmd.arg("--dns").arg(dns);
+        }
+        if !map_host_loopback.is_empty() {
+            passt_cmd.arg("--map-host-loopback").arg(map_host_loopback);
+        }
+        if !map_guest_addr.is_empty() {
+            passt_cmd.arg("--map-guest-addr").arg(map_guest_addr);
+        }
+        if *no_map_gw {
+            passt_cmd.arg("--no-map-gw");
+        }
+        if *ipv4_only {
+            passt_cmd.arg("--ipv4-only");
+        }
+        // Group port mappings by protocol
+        let mut tcp_ports = Vec::new();
+        let mut udp_ports = Vec::new();
+
+        for pm in &self.manifest.port_map {
+            let port_spec = format!("{}/{}:{}", pm.address, pm.from, pm.to);
+            match pm.protocol {
+                Protocol::Tcp => tcp_ports.push(port_spec),
+                Protocol::Udp => udp_ports.push(port_spec),
+            }
+        }
+        // Add TCP port forwarding if any
+        if !tcp_ports.is_empty() {
+            passt_cmd.arg("--tcp-ports").arg(tcp_ports.join(","));
+        }
+        // Add UDP port forwarding if any
+        if !udp_ports.is_empty() {
+            passt_cmd.arg("--udp-ports").arg(udp_ports.join(","));
+        }
+        passt_cmd.arg("-f").arg("-1");
+
+        let args = passt_cmd
+            .get_args()
+            .map(|arg| arg.to_string_lossy().to_string())
+            .collect::<Vec<_>>();
+        let stdout_path = workdir.passt_stdout();
+        let stderr_path = workdir.passt_stderr();
+        let note = ProcessAnnotation {
+            kind: "passt".to_string(),
+            live_for: Some(self.manifest.id.clone()),
+        };
+        let note = serde_json::to_string(&note)?;
+        let process_config = ProcessConfig {
+            id: format!("passt-{}", self.manifest.id),
+            args,
+            name: format!("passt-{}", self.manifest.name),
+            command: passt_exec.to_string(),
+            env: Default::default(),
+            cwd: workdir.to_string_lossy().to_string(),
+            stdout: stdout_path.to_string_lossy().to_string(),
+            stderr: stderr_path.to_string_lossy().to_string(),
+            pidfile: Default::default(),
+            cid: None,
+            note,
+        };
+        Ok(process_config)
+    }
+
     pub fn config_qemu(
         &self,
         workdir: impl AsRef<Path>,
         cfg: &CvmConfig,
         gpus: &GpuConfig,
-    ) -> Result<ProcessConfig> {
+    ) -> Result<Vec<ProcessConfig>> {
         let workdir = VmWorkDir::new(workdir);
         let serial_file = workdir.serial_file();
         let serial_pty = workdir.serial_pty();
@@ -304,12 +408,13 @@ impl VmConfig {
                 }
             }
         }
+        let mut processes = vec![];
         command
             .arg("-drive")
             .arg(format!("file={},if=none,id=hd1", hda_path.display()))
             .arg("-device")
             .arg("virtio-blk-pci,drive=hd1");
-        let netdev = match &self.networking {
+        let netdev = match &cfg.networking {
             Networking::User(netcfg) => {
                 let mut netdev = format!(
                     "user,id=net0,net={},dhcpstart={},restrict={}",
@@ -328,6 +433,16 @@ impl VmConfig {
                 }
                 netdev
             }
+            Networking::Passt(netcfg) => {
+                processes.push(
+                    self.config_passt(&workdir, netcfg)
+                        .context("Failed to configure passt")?,
+                );
+                format!(
+                    "stream,id=net0,server=off,addr.type=unix,addr.path={}",
+                    workdir.passt_socket().display()
+                )
+            }
             Networking::Custom(netcfg) => netcfg.netdev.clone(),
         };
         command.arg("-netdev").arg(netdev);
@@ -538,7 +653,10 @@ impl VmConfig {
         }
 
         let command = cmd_args.remove(0);
-        let note = "{}".to_string();
+        let note = ProcessAnnotation {
+            kind: "cvm".to_string(),
+            live_for: None,
+        };
         let note = serde_json::to_string(&note)?;
         let process_config = ProcessConfig {
             id: self.manifest.id.clone(),
@@ -553,8 +671,9 @@ impl VmConfig {
             cid: Some(self.cid),
             note,
         };
+        processes.push(process_config);
 
-        Ok(process_config)
+        Ok(processes)
     }
 }
 
@@ -730,6 +849,22 @@ impl VmWorkDir {
         self.workdir.join("qmp.sock")
     }
 
+    pub fn passt_socket(&self) -> PathBuf {
+        self.workdir.join("passt.sock")
+    }
+
+    pub fn passt_stdout(&self) -> PathBuf {
+        self.workdir.join("passt.stdout")
+    }
+
+    pub fn passt_stderr(&self) -> PathBuf {
+        self.workdir.join("passt.stderr")
+    }
+
+    pub fn passt_log(&self) -> PathBuf {
+        self.workdir.join("passt.log")
+    }
+
     pub fn path(&self) -> &Path {
         &self.workdir
     }

vmm/src/config.rs

@@ -130,6 +130,9 @@ pub struct CvmConfig {
     pub qemu_pci_hole64_size: u64,
     /// QEMU hotplug_off
     pub qemu_hotplug_off: bool,
+
+    /// Networking configuration
+    pub networking: Networking,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -210,9 +213,6 @@ pub struct Config {
     /// Gateway configuration
     pub gateway: GatewayConfig,
 
-    /// Networking configuration
-    pub networking: Networking,
-
     /// Authentication configuration
     pub auth: AuthConfig,
 
@@ -226,6 +226,23 @@ pub struct Config {
     pub key_provider: KeyProviderConfig,
 }
 
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+pub struct ProcessAnnotation {
+    #[serde(default)]
+    pub kind: String,
+    #[serde(default)]
+    pub live_for: Option<String>,
+}
+
+impl ProcessAnnotation {
+    pub fn is_cvm(&self) -> bool {
+        if self.live_for.is_some() {
+            return false;
+        }
+        self.kind.is_empty() || self.kind == "cvm"
+    }
+}
+
 impl Config {
     pub fn abs_path(self) -> Result<Self> {
         Ok(Self {
@@ -240,16 +257,37 @@ impl Config {
 #[serde(tag = "mode", rename_all = "lowercase")]
 pub enum Networking {
     User(UserNetworking),
+    Passt(PasstNetworking),
     Custom(CustomNetworking),
 }
 
+impl Networking {
+    pub fn is_passt(&self) -> bool {
+        matches!(self, Networking::Passt(_))
+    }
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct UserNetworking {
     pub net: String,
     pub dhcp_start: String,
     pub restrict: bool,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct PasstNetworking {
+    pub passt_exec: String,
+    pub interface: String,
+    pub address: String,
+    pub netmask: String,
+    pub gateway: String,
+    pub dns: Vec<String>,
+    pub map_host_loopback: String,
+    pub map_guest_addr: String,
+    pub no_map_gw: bool,
+    pub ipv4_only: bool,
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct CustomNetworking {
     pub netdev: String,