runtime-rs: shim management

Add shim management http server and boot it as a light-weight thread
when the sandbox is created.

Fixes: #5114
Signed-off-by: Ji-Xinyou <jerryji0414@outlook.com>

src/runtime-rs/Cargo.lock

@@ -1169,12 +1169,62 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "http"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "75f43d41e26995c17e71ee126451dd3941010b0514a81a9d11f3b341debc2399"
+dependencies = [
+ "bytes 1.1.0",
+ "fnv",
+ "itoa",
+]
+
+[[package]]
+name = "http-body"
+version = "0.4.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1"
+dependencies = [
+ "bytes 1.1.0",
+ "http",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "httparse"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d897f394bad6a705d5f4104762e116a75639e470d80901eed05a860a95cb1904"
+
 [[package]]
 name = "httpdate"
 version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c4a1e36c821dbe04574f602848a19f742f4fb3c98d40449f11bcad18d6b17421"
 
+[[package]]
+name = "hyper"
+version = "0.14.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "02c929dc5c39e335a03c405292728118860721b10190d98c2a0f0efd5baafbac"
+dependencies = [
+ "bytes 1.1.0",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "http",
+ "http-body",
+ "httparse",
+ "httpdate",
+ "itoa",
+ "pin-project-lite",
+ "tokio",
+ "tower-service",
+ "tracing",
+ "want",
+]
+
 [[package]]
 name = "hypervisor"
 version = "0.1.0"
@@ -2217,6 +2267,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "common",
+ "hyper",
  "kata-types",
  "lazy_static",
  "linux_container",
@@ -2770,6 +2821,38 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "tower-service"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6bc1c9ce2b5135ac7f93c72918fc37feb872bdc6a5533a8b85eb4b86bfdae52"
+
+[[package]]
+name = "tracing"
+version = "0.1.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a400e31aa60b9d44a52a8ee0343b5b18566b03a8321e0d321f695cf56e940160"
+dependencies = [
+ "cfg-if 1.0.0",
+ "pin-project-lite",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-core"
+version = "0.1.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b7358be39f2f274f322d2aaed611acc57f382e8eb1e5b48cb9ae30933495ce7"
+dependencies = [
+ "once_cell",
+]
+
+[[package]]
+name = "try-lock"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642"
+
 [[package]]
 name = "ttrpc"
 version = "0.6.1"
@@ -2994,6 +3077,16 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d5b2c62b4012a3e1eca5a7e077d13b3bf498c4073e33ccd58626607748ceeca"
 
+[[package]]
+name = "want"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ce8a968cb1cd110d136ff8b819a556d6fb6d919363c61534f6860c7eb172ba0"
+dependencies = [
+ "log",
+ "try-lock",
+]
+
 [[package]]
 name = "wasi"
 version = "0.9.0+wasi-snapshot-preview1"

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

@@ -11,6 +11,7 @@ lazy_static = "1.4.0"
 slog = "2.5.2"
 slog-scope = "4.4.0"
 tokio = { version = "1.8.0", features = ["rt-multi-thread"] }
+hyper = { version = "0.14.20", features = ["stream", "server", "http1"] }
 
 common = { path = "./common" }
 kata-types = { path = "../../../libs/kata-types" }

src/runtime-rs/crates/runtimes/src/lib.rs

@@ -11,4 +11,5 @@ logging::logger_with_subsystem!(sl, "runtimes");
 
 pub mod manager;
 pub use manager::RuntimeHandlerManager;
+mod shim_mgmt;
 mod static_resource;

src/runtime-rs/crates/runtimes/src/manager.rs

@@ -8,7 +8,7 @@ use std::sync::Arc;
 
 use anyhow::{anyhow, Context, Result};
 
-use crate::static_resource::StaticResourceManager;
+use crate::{shim_mgmt::server::MgmtServer, static_resource::StaticResourceManager};
 use common::{
     message::Message,
     types::{Request, Response},
@@ -109,6 +109,11 @@ impl RuntimeHandlerManagerInner {
             .await
             .context("init runtime handler")?;
 
+        // the sandbox creation can reach here only once and the sandbox is created
+        // so we can safely create the shim management socket right now
+        let shim_mgmt_svr = MgmtServer::new(&self.id);
+        shim_mgmt_svr.run().await;
+
         Ok(())
     }
 
@@ -196,6 +201,7 @@ impl RuntimeHandlerManager {
                 .create_container(req, spec)
                 .await
                 .context("create container")?;
+
             Ok(Response::CreateContainer(shim_pid))
         } else {
             self.handler_request(req).await.context("handler request")

src/runtime-rs/crates/runtimes/src/shim_mgmt/handlers.rs

@@ -0,0 +1,40 @@
+// Copyright (c) 2019-2022 Alibaba Cloud
+// Copyright (c) 2019-2022 Ant Group
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+// This defines the handler corresponding to the url
+// when a request is sent to destined url, the handler
+// function should be invoked, and the corresponding
+// data will be in the response's body
+//
+// NOTE: ALL HANDLER SHOULD BE ASYNC UNDER ROUTERIFY
+
+use hyper::{Body, Method, Request, Response, Result, StatusCode};
+
+use super::server::AGENT_URL;
+
+// main router for response, this works as a multiplexer on
+// http arrival which invokes the corresponding handler function
+pub(crate) async fn handler_mux(req: Request<Body>) -> Result<Response<Body>> {
+    info!(sl!(), "mgmt-svr(mux): recv req {:?}", req);
+    match (req.method(), req.uri().path()) {
+        (&Method::GET, AGENT_URL) => agent_url_handler(req).await,
+        _ => Ok(not_found(req).await),
+    }
+}
+
+// url not found
+async fn not_found(_req: Request<Body>) -> Response<Body> {
+    Response::builder()
+        .status(StatusCode::NOT_FOUND)
+        .body(Body::from("URL NOT FOUND"))
+        .unwrap()
+}
+
+// returns the url for agent
+async fn agent_url_handler(_req: Request<Body>) -> Result<Response<Body>> {
+    // todo
+    Ok(Response::new(Body::from("")))
+}

src/runtime-rs/crates/runtimes/src/shim_mgmt/mod.rs

@@ -0,0 +1,8 @@
+// Copyright (c) 2019-2022 Alibaba Cloud
+// Copyright (c) 2019-2022 Ant Group
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+mod handlers;
+pub mod server;

src/runtime-rs/crates/runtimes/src/shim_mgmt/server.rs

@@ -0,0 +1,97 @@
+// Copyright (c) 2019-2022 Alibaba Cloud
+// Copyright (c) 2019-2022 Ant Group
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+// Shim management service, this service starts a management http server on a socket
+// and wire certain URL with a corresponding handler. When a command-line interface
+// or further shim functions want the information corresponding to this, it can just
+// send a GET request to the url, and the info will be in the response
+
+#![allow(dead_code)] // some url's handler are *to be* developed
+
+use std::{fs, path::Path};
+
+use anyhow::{Context, Result};
+use hyper::{server::conn::Http, service::service_fn};
+use tokio::net::UnixListener;
+
+use super::handlers::handler_mux;
+
+pub(crate) const DIRECT_VOLUMN_PATH_KEY: &str = "path";
+pub(crate) const DIRECT_VOLUMN_STAT_URL: &str = "/direct-volumn/stats";
+pub(crate) const DIRECT_VOLUMN_RESIZE_URL: &str = "/direct-volumn/resize";
+pub(crate) const AGENT_URL: &str = "/agent-url";
+pub(crate) const IP_TABLE_URL: &str = "/iptables";
+pub(crate) const IP6_TABLE_URL: &str = "/ip6tables";
+pub(crate) const METRICS_URL: &str = "/metrics";
+
+const SHIM_MGMT_SOCK_NAME: &str = "shim-monitor.sock";
+
+// The management server
+#[derive(Debug)]
+pub struct MgmtServer {
+    // socket address
+    pub s_addr: String,
+}
+
+impl MgmtServer {
+    pub fn new(sid: &str) -> Self {
+        Self {
+            s_addr: mgmt_socket_addr(sid.to_owned()),
+        }
+    }
+
+    // TODO(when metrics is supported): write metric addresses to fs
+    // TODO(when metrics is supported): register shim metrics
+    // TODO(when metrics is supported): register sandbox metrics
+    // start a new thread, running management http server in a dead loop
+    pub async fn run(&self) {
+        let lsnr = lsnr_from_path(self.s_addr.clone()).await.unwrap();
+
+        // start an infinate loop, which serves the incomming uds stream
+        tokio::task::spawn(async move {
+            loop {
+                let (stream, _) = lsnr.accept().await.unwrap();
+                // spawn a light weight thread to multiplex to the handler
+                tokio::task::spawn(async move {
+                    if let Err(err) = Http::new()
+                        .serve_connection(stream, service_fn(handler_mux))
+                        .await
+                    {
+                        warn!(sl!(), "Failed to serve connection: {:?}", err);
+                    }
+                });
+            }
+        });
+    }
+}
+
+// return sandbox's storage path
+pub fn sb_storage_path() -> String {
+    String::from("/run/kata")
+}
+
+// returns the address of the unix domain socket(UDS) for communication with shim
+// management service using http
+// normally returns "unix:///run/kata/{sid}/shim_monitor.sock"
+pub fn mgmt_socket_addr(sid: String) -> String {
+    let p = Path::new(&sb_storage_path())
+        .join(sid)
+        .join(SHIM_MGMT_SOCK_NAME);
+    format!("unix://{}", p.to_string_lossy())
+}
+
+// from path, return a unix listener corresponding to that path,
+// if the path(socket file) is not created, we create that here
+async fn lsnr_from_path(path: String) -> Result<UnixListener> {
+    let trim_path = path.strip_prefix("unix:").context("trim path")?;
+    let file_path = Path::new("/").join(trim_path);
+    let file_path = file_path.as_path();
+    if let Some(parent_dir) = file_path.parent() {
+        fs::create_dir_all(parent_dir).context("create parent dir")?;
+    }
+    info!(sl!(), "mgmt-svr: binding to path {}", path);
+    UnixListener::bind(file_path).context("bind address")
+}