runtime-rs: support vhost-vsock

Rename old VsockConfig to HybridVsockConfig. And add VsockConfig to
support vhost-vsock. We follow kata's old way to try random vhost fd
for 50 times to generate uniqe fd.

Fixes: #5654

Signed-off-by: Yipeng Yin <yinyipeng@bytedance.com>

src/libs/kata-types/src/config/agent.rs

@@ -10,6 +10,7 @@ use crate::config::{ConfigOps, TomlConfig};
 pub use vendor::AgentVendor;
 
 use super::default::{DEFAULT_AGENT_LOG_PORT, DEFAULT_AGENT_VSOCK_PORT};
+use crate::eother;
 
 /// agent name of Kata agent.
 pub const AGENT_NAME_KATA: &str = "kata";
@@ -108,6 +109,16 @@ fn default_health_check_timeout() -> u32 {
     90_000
 }
 
+impl Agent {
+    fn validate(&self) -> Result<()> {
+        if self.dial_timeout_ms == 0 {
+            return Err(eother!("dial_timeout_ms couldn't be 0."));
+        }
+
+        Ok(())
+    }
+}
+
 impl ConfigOps for Agent {
     fn adjust_config(conf: &mut TomlConfig) -> Result<()> {
         AgentVendor::adjust_config(conf)?;
@@ -116,6 +127,9 @@ impl ConfigOps for Agent {
 
     fn validate(conf: &TomlConfig) -> Result<()> {
         AgentVendor::validate(conf)?;
+        for (_, agent_config) in conf.agent.iter() {
+            agent_config.validate()?;
+        }
         Ok(())
     }
 }

src/runtime-rs/Cargo.lock

@@ -48,6 +48,7 @@ dependencies = [
  "kata-types",
  "log",
  "logging",
+ "nix 0.24.2",
  "oci",
  "protobuf",
  "protocols",
@@ -1212,6 +1213,7 @@ dependencies = [
  "logging",
  "nix 0.24.2",
  "persist",
+ "rand 0.8.5",
  "seccompiler",
  "serde",
  "serde_json",

src/runtime-rs/crates/agent/Cargo.toml

@@ -20,6 +20,7 @@ slog-scope = "4.4.0"
 ttrpc = { version = "0.6.1" }
 tokio = { version = "1.8.0", features = ["fs", "rt"] }
 url = "2.2.2"
+nix = "0.24.2"
 
 kata-types = { path = "../../../libs/kata-types"}
 logging = { path = "../../../libs/logging"}

src/runtime-rs/crates/agent/src/sock/mod.rs

@@ -35,8 +35,8 @@ pub enum Stream {
     // model, and mediates communication between AF_UNIX sockets (on the host end)
     // and AF_VSOCK sockets (on the guest end).
     Unix(UnixStream),
-    // TODO: support vsock
     // vsock://<cid>:<port>
+    Vsock(UnixStream),
 }
 
 impl Stream {
@@ -47,7 +47,7 @@ impl Stream {
     ) -> Poll<std::io::Result<()>> {
         // Safety: `UnixStream::read` correctly handles reads into uninitialized memory
         match self {
-            Stream::Unix(stream) => Pin::new(stream).poll_read(cx, buf),
+            Stream::Unix(stream) | Stream::Vsock(stream) => Pin::new(stream).poll_read(cx, buf),
         }
     }
 }
@@ -55,7 +55,7 @@ impl Stream {
 impl IntoRawFd for Stream {
     fn into_raw_fd(self) -> RawFd {
         match self {
-            Stream::Unix(stream) => match stream.into_std() {
+            Stream::Unix(stream) | Stream::Vsock(stream) => match stream.into_std() {
                 Ok(stream) => stream.into_raw_fd(),
                 Err(err) => {
                     error!(sl!(), "failed to into std unix stream {:?}", err);

src/runtime-rs/crates/agent/src/sock/vsock.rs

@@ -4,8 +4,15 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
-use anyhow::Result;
+use std::{
+    os::unix::prelude::{AsRawFd, FromRawFd},
+    time::Duration,
+};
+
+use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
+use nix::sys::socket::{connect, socket, AddressFamily, SockFlag, SockType, VsockAddr};
+use tokio::net::UnixStream;
 
 use super::{ConnectConfig, Sock, Stream};
 
@@ -26,7 +33,50 @@ impl Vsock {
 
 #[async_trait]
 impl Sock for Vsock {
-    async fn connect(&self, _config: &ConnectConfig) -> Result<Stream> {
-        todo!()
+    async fn connect(&self, config: &ConnectConfig) -> Result<Stream> {
+        let retry_times = config.reconnect_timeout_ms / config.dial_timeout_ms;
+        let sock_addr = VsockAddr::new(self.vsock_cid, self.port);
+        let connect_once = || {
+            // Create socket fd
+            let socket = socket(
+                AddressFamily::Vsock,
+                SockType::Stream,
+                SockFlag::empty(),
+                None,
+            )
+            .context("failed to create vsock socket")?;
+
+            // Wrap the socket fd in a UnixStream, so that it is closed when
+            // anything fails.
+            // We MUST NOT reuse a vsock socket which has failed a connection
+            // attempt before, since a ECONNRESET error marks the whole socket as
+            // broken and non-reusable.
+            let socket = unsafe { std::os::unix::net::UnixStream::from_raw_fd(socket) };
+
+            // Connect the socket to vsock server.
+            connect(socket.as_raw_fd(), &sock_addr)
+                .with_context(|| format!("failed to connect to {}", sock_addr))?;
+
+            // Finally, convert the std UnixSocket to tokio's UnixSocket.
+            UnixStream::from_std(socket).context("from_std")
+        };
+
+        for i in 0..retry_times {
+            match connect_once() {
+                Ok(stream) => {
+                    info!(
+                        sl!(),
+                        "connect vsock success on {} current client fd {}",
+                        i,
+                        stream.as_raw_fd()
+                    );
+                    return Ok(Stream::Vsock(stream));
+                }
+                Err(_) => {
+                    tokio::time::sleep(Duration::from_millis(config.dial_timeout_ms)).await;
+                }
+            }
+        }
+        Err(anyhow!("cannot connect to agent ttrpc server {:?}", config))
     }
 }

src/runtime-rs/crates/hypervisor/Cargo.toml

@@ -23,6 +23,7 @@ slog-scope = "4.4.0"
 thiserror = "1.0"
 tokio = { version = "1.8.0", features = ["sync"] }
 vmm-sys-util = "0.11.0"
+rand = "0.8.4"
 
 kata-sys-util = { path = "../../../libs/kata-sys-util" }
 kata-types = { path = "../../../libs/kata-types" }

src/runtime-rs/crates/hypervisor/src/device/mod.rs

@@ -15,7 +15,7 @@ pub use vfio::{bind_device_to_host, bind_device_to_vfio, VfioBusMode, VfioConfig
 mod share_fs_mount;
 pub use share_fs_mount::{ShareFsMountConfig, ShareFsMountType, ShareFsOperation};
 mod vsock;
-pub use vsock::VsockConfig;
+pub use vsock::{HybridVsockConfig, VsockConfig};
 
 use std::fmt;
 
@@ -27,6 +27,7 @@ pub enum Device {
     Vfio(VfioConfig),
     ShareFsMount(ShareFsMountConfig),
     Vsock(VsockConfig),
+    HybridVsock(HybridVsockConfig),
 }
 
 impl fmt::Display for Device {

src/runtime-rs/crates/hypervisor/src/device/vsock.rs

@@ -4,8 +4,13 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
+use anyhow::{Context, Result};
+use rand::Rng;
+use std::os::unix::prelude::AsRawFd;
+use tokio::fs::{File, OpenOptions};
+
 #[derive(Debug)]
-pub struct VsockConfig {
+pub struct HybridVsockConfig {
     /// Unique identifier of the device
     pub id: String,
 
@@ -15,3 +20,76 @@ pub struct VsockConfig {
     /// unix domain socket path
     pub uds_path: String,
 }
+
+#[derive(Debug)]
+pub struct VsockConfig {
+    /// Unique identifier of the device
+    pub id: String,
+
+    /// A 32-bit Context Identifier (CID) used to identify the guest.
+    pub guest_cid: u32,
+
+    /// Vhost vsock fd. Hold to ensure CID is not used by other VM.
+    pub vhost_fd: File,
+}
+
+const VHOST_VSOCK_DEVICE: &str = "/dev/vhost-vsock";
+
+// From <linux/vhost.h>
+// Generate a wrapper function for VHOST_VSOCK_SET_GUEST_CID ioctl.
+// It set guest CID for vsock fd, and return error if CID is already
+// in use.
+const VHOST_VIRTIO_IOCTL: u8 = 0xAF;
+const VHOST_VSOCK_SET_GUEST_CID: u8 = 0x60;
+nix::ioctl_write_ptr!(
+    vhost_vsock_set_guest_cid,
+    VHOST_VIRTIO_IOCTL,
+    VHOST_VSOCK_SET_GUEST_CID,
+    u64
+);
+
+const CID_RETRY_COUNT: u32 = 50;
+
+impl VsockConfig {
+    pub async fn new(id: String) -> Result<Self> {
+        let vhost_fd = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(VHOST_VSOCK_DEVICE)
+            .await
+            .context(format!(
+                "failed to open {}, try to run modprobe vhost_vsock.",
+                VHOST_VSOCK_DEVICE
+            ))?;
+        let mut rng = rand::thread_rng();
+
+        // Try 50 times to find a context ID that is not in use.
+        for _ in 0..CID_RETRY_COUNT {
+            // First usable CID above VMADDR_CID_HOST (see vsock(7))
+            let first_usable_cid = 3;
+            let rand_cid = rng.gen_range(first_usable_cid..=(u32::MAX));
+            let guest_cid =
+                unsafe { vhost_vsock_set_guest_cid(vhost_fd.as_raw_fd(), &(rand_cid as u64)) };
+            match guest_cid {
+                Ok(_) => {
+                    return Ok(VsockConfig {
+                        id,
+                        guest_cid: rand_cid,
+                        vhost_fd,
+                    });
+                }
+                Err(nix::Error::EADDRINUSE) => {
+                    // The CID is already in use. Try another one.
+                }
+                Err(err) => {
+                    return Err(err).context("failed to set guest CID");
+                }
+            }
+        }
+
+        anyhow::bail!(
+            "failed to find a free vsock context ID after {} attempts",
+            CID_RETRY_COUNT
+        );
+    }
+}

src/runtime-rs/crates/hypervisor/src/dragonball/inner_device.rs

@@ -15,8 +15,8 @@ use dragonball::api::v1::{
 
 use super::DragonballInner;
 use crate::{
-    device::Device, NetworkConfig, ShareFsDeviceConfig, ShareFsMountConfig, ShareFsMountType,
-    ShareFsOperation, VmmState, VsockConfig,
+    device::Device, HybridVsockConfig, NetworkConfig, ShareFsDeviceConfig, ShareFsMountConfig,
+    ShareFsMountType, ShareFsOperation, VmmState,
 };
 
 const MB_TO_B: u32 = 1024 * 1024;
@@ -56,13 +56,16 @@ impl DragonballInner {
                     config.no_drop,
                 )
                 .context("add block device"),
-            Device::Vsock(config) => self.add_vsock(&config).context("add vsock"),
+            Device::HybridVsock(config) => self.add_hvsock(&config).context("add vsock"),
             Device::ShareFsDevice(config) => self
                 .add_share_fs_device(&config)
                 .context("add share fs device"),
             Device::ShareFsMount(config) => self
                 .add_share_fs_mount(&config)
                 .context("add share fs mount"),
+            Device::Vsock(_) => {
+                todo!()
+            }
         }
     }
 
@@ -139,7 +142,7 @@ impl DragonballInner {
             .context("insert network device")
     }
 
-    fn add_vsock(&mut self, config: &VsockConfig) -> Result<()> {
+    fn add_hvsock(&mut self, config: &HybridVsockConfig) -> Result<()> {
         let vsock_cfg = VsockDeviceConfigInfo {
             id: String::from("root"),
             guest_cid: config.guest_cid,

src/runtime-rs/crates/hypervisor/src/dragonball/inner_hypervisor.rs

@@ -32,7 +32,7 @@ impl DragonballInner {
 
         // prepare vsock
         let uds_path = [&self.jailer_root, DEFAULT_HYBRID_VSOCK_NAME].join("/");
-        let d = crate::device::Device::Vsock(crate::device::VsockConfig {
+        let d = crate::device::Device::HybridVsock(crate::device::HybridVsockConfig {
             id: format!("vsock-{}", &self.id),
             guest_cid: 3,
             uds_path,