runtime-rs: support for remote hypervisors type

This patch adds the support of the remote hypervisor type for runtime-rs.
The cloud-api-adaptor needs the annotations and network namespace path
to create the VMs.
The remote hypervisor opens a UNIX domain socket specified in the config
file, and sends ttrpc requests to a external process to control sandbox
VMs.

Fixes: #10350

Signed-off-by: Chasing1020 <643601464@qq.com>

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

@@ -8,6 +8,8 @@ mod hybrid_vsock;
 pub use hybrid_vsock::HybridVsock;
 mod vsock;
 pub use vsock::Vsock;
+mod remote;
+pub use remote::Remote;
 
 use std::{
     pin::Pin,
@@ -28,6 +30,7 @@ use url::Url;
 
 const VSOCK_SCHEME: &str = "vsock";
 const HYBRID_VSOCK_SCHEME: &str = "hvsock";
+const REMOTE_SCHEME: &str = "remote";
 
 /// Socket stream
 pub enum Stream {
@@ -98,6 +101,7 @@ impl ConnectConfig {
 enum SockType {
     Vsock(Vsock),
     HybridVsock(HybridVsock),
+    Remote(Remote),
 }
 
 #[async_trait]
@@ -114,6 +118,7 @@ pub fn new(address: &str, port: u32) -> Result<Arc<dyn Sock>> {
     match parse(address, port).context("parse url")? {
         SockType::Vsock(sock) => Ok(Arc::new(sock)),
         SockType::HybridVsock(sock) => Ok(Arc::new(sock)),
+        SockType::Remote(sock) => Ok(Arc::new(sock)),
     }
 }
 
@@ -136,6 +141,13 @@ fn parse(address: &str, port: u32) -> Result<SockType> {
             let uds = path[0];
             Ok(SockType::HybridVsock(HybridVsock::new(uds, port)))
         }
+        REMOTE_SCHEME => {
+            let path: Vec<&str> = url.path().split(':').collect();
+            if path.len() != 1 {
+                return Err(anyhow!("invalid path {:?}", path));
+            }
+            Ok(SockType::Remote(Remote::new(path[0].to_string())))
+        }
         _ => Err(anyhow!("Unsupported scheme")),
     }
 }

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

@@ -0,0 +1,61 @@
+// Copyright (c) 2019-2022 Alibaba Cloud
+// Copyright (c) 2019-2022 Ant Group
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+use std::{os::unix::prelude::AsRawFd, path::Path};
+
+use anyhow::{anyhow, Context, Result};
+use async_trait::async_trait;
+use tokio::{io::Interest, net::UnixStream};
+
+use super::{ConnectConfig, Sock, Stream};
+
+#[derive(Debug, PartialEq)]
+pub struct Remote {
+    path: String,
+}
+
+impl Remote {
+    pub fn new(path: String) -> Self {
+        Self { path }
+    }
+}
+
+#[async_trait]
+impl Sock for Remote {
+    async fn connect(&self, config: &ConnectConfig) -> Result<Stream> {
+        let retry_times = config.reconnect_timeout_ms / config.dial_timeout_ms;
+        for i in 0..retry_times {
+            match connect_helper(&self.path).await {
+                Ok(stream) => {
+                    info!(
+                        sl!(),
+                        "remote connect success on {} current client fd {}",
+                        i,
+                        stream.as_raw_fd()
+                    );
+                    return Ok(Stream::Unix(stream));
+                }
+                Err(err) => {
+                    debug!(sl!(), "remote connect on {} err : {:?}", i, err);
+                    tokio::time::sleep(std::time::Duration::from_millis(config.dial_timeout_ms))
+                        .await;
+                    continue;
+                }
+            }
+        }
+        Err(anyhow!("cannot connect to agent ttrpc server {:?}", config))
+    }
+}
+
+async fn connect_helper(address: &str) -> Result<UnixStream> {
+    let stream = UnixStream::connect(Path::new(&address))
+        .await
+        .context("failed to create UnixAddr")?;
+    stream
+        .ready(Interest::READABLE | Interest::WRITABLE)
+        .await?;
+    Ok(stream)
+}