diff --git a/Cargo.lock b/Cargo.lock index a14a8476a3..1d0d0d5810 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2171,15 +2171,6 @@ dependencies = [ "miniz_oxide", ] -[[package]] -name = "fluent-uri" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17c704e9dbe1ddd863da1e6ff3567795087b1eb201ce80d8fa81162e1516500d" -dependencies = [ - "bitflags 1.3.2", -] - [[package]] name = "fluent-uri" version = "0.3.2" @@ -2441,7 +2432,7 @@ dependencies = [ "env_logger 0.10.2", "flate2", "fs2", - "json-patch 4.2.0", + "json-patch", "k8s-cri", "kata-agent-policy", "kata-types", @@ -2817,9 +2808,9 @@ dependencies = [ [[package]] name = "hyper" -version = "1.9.0" +version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" dependencies = [ "atomic-waker", "bytes 1.11.1", @@ -2844,7 +2835,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" dependencies = [ "http 1.4.0", - "hyper 1.9.0", + "hyper 1.10.1", "hyper-util", "log", "rustls", @@ -2873,7 +2864,7 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0" dependencies = [ - "hyper 1.9.0", + "hyper 1.10.1", "hyper-util", "pin-project-lite", "tokio", @@ -2888,7 +2879,7 @@ checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0" dependencies = [ "bytes 1.11.1", "http-body-util", - "hyper 1.9.0", + "hyper 1.10.1", "hyper-util", "native-tls", "tokio", @@ -2908,7 +2899,7 @@ dependencies = [ "futures-util", "http 1.4.0", "http-body 1.0.1", - "hyper 1.9.0", + "hyper 1.10.1", "ipnet", "libc", "percent-encoding", @@ -2923,15 +2914,17 @@ dependencies = [ [[package]] name = "hyperlocal" -version = "0.8.0" +version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fafdf7b2b2de7c9784f76e02c0935e65a8117ec3b768644379983ab333ac98c" +checksum = "986c5ce3b994526b3cd75578e62554abd09f0899d6206de48b3e96ab34ccc8c7" dependencies = [ - "futures-util", "hex", - "hyper 0.14.32", - "pin-project", + "http-body-util", + "hyper 1.10.1", + "hyper-util", + "pin-project-lite", "tokio", + "tower-service", ] [[package]] @@ -2941,13 +2934,16 @@ dependencies = [ "actix-rt", "anyhow", "async-trait", + "bytes 1.11.1", "ch-config", "crossbeam-channel", "dbs-utils", "dragonball", "futures", "go-flag", - "hyper 0.14.32", + "http-body-util", + "hyper 1.10.1", + "hyper-util", "hyperlocal", "hypervisor", "kata-sys-util", @@ -3383,25 +3379,13 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "json-patch" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b1fb8864823fad91877e6caea0baca82e49e8db50f8e5c9f9a453e27d3330fc" -dependencies = [ - "jsonptr 0.4.7", - "serde", - "serde_json", - "thiserror 1.0.69", -] - [[package]] name = "json-patch" version = "4.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7421438de105a0827e44fadd05377727847d717c80ce29a229f85fd04c427b72" dependencies = [ - "jsonptr 0.7.1", + "jsonptr", "serde", "serde_json", "thiserror 2.0.18", @@ -3420,17 +3404,6 @@ dependencies = [ "thiserror 2.0.18", ] -[[package]] -name = "jsonptr" -version = "0.4.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1c6e529149475ca0b2820835d3dce8fcc41c6b943ca608d32f35b449255e4627" -dependencies = [ - "fluent-uri 0.1.4", - "serde", - "serde_json", -] - [[package]] name = "jsonptr" version = "0.7.1" @@ -3566,7 +3539,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tokio-vsock 0.3.4", - "toml", + "toml 1.1.2+spec-1.1.0", "tracing", "tracing-opentelemetry 0.17.4", "tracing-subscriber", @@ -3610,7 +3583,7 @@ name = "kata-agent-policy" version = "0.1.0" dependencies = [ "anyhow", - "json-patch 2.0.0", + "json-patch", "libc", "protocols", "regorus", @@ -3628,12 +3601,15 @@ dependencies = [ "agent", "anyhow", "base64 0.13.1", + "bytes 1.11.1", "chrono", "clap", "csv", "epoll", "futures", - "hyper 0.14.32", + "http-body-util", + "hyper 1.10.1", + "hyper-util", "kata-sys-util", "kata-types", "lazy_static", @@ -3664,7 +3640,7 @@ dependencies = [ "test-utils", "thiserror 2.0.18", "tokio", - "toml", + "toml 1.1.2+spec-1.1.0", "url", "virt_container", "vmm-sys-util 0.15.0", @@ -3790,7 +3766,7 @@ dependencies = [ "test-utils", "thiserror 2.0.18", "tokio", - "toml", + "toml 1.1.2+spec-1.1.0", ] [[package]] @@ -3854,7 +3830,7 @@ dependencies = [ "http 1.4.0", "http-body 1.0.1", "http-body-util", - "hyper 1.9.0", + "hyper 1.10.1", "hyper-rustls", "hyper-timeout 0.5.2", "hyper-util", @@ -4658,7 +4634,7 @@ dependencies = [ "serde", "serde_json", "thiserror 1.0.69", - "toml", + "toml 0.5.11", ] [[package]] @@ -6229,7 +6205,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e9c261f7ce75418b3beadfb3f0eb1299fe8eb9640deba45ffa2cb783098697d" dependencies = [ "ahash 0.8.12", - "fluent-uri 0.3.2", + "fluent-uri", "once_cell", "parking_lot", "percent-encoding", @@ -6315,7 +6291,7 @@ dependencies = [ "http 1.4.0", "http-body 1.0.1", "http-body-util", - "hyper 1.9.0", + "hyper 1.10.1", "hyper-rustls", "hyper-tls", "hyper-util", @@ -6359,7 +6335,7 @@ dependencies = [ "http 1.4.0", "http-body 1.0.1", "http-body-util", - "hyper 1.9.0", + "hyper 1.10.1", "hyper-rustls", "hyper-util", "js-sys", @@ -6397,12 +6373,15 @@ dependencies = [ "async-trait", "base64 0.22.1", "byte-unit", + "bytes 1.11.1", "cgroups-rs 0.5.0", "flate2", "futures", "futures-lite 2.6.1", "hex", - "hyper 0.14.32", + "http-body-util", + "hyper 1.10.1", + "hyper-util", "hyperlocal", "hypervisor", "inotify 0.11.1", @@ -6636,9 +6615,12 @@ version = "0.1.0" dependencies = [ "agent", "anyhow", + "bytes 1.11.1", "common", "containerd-shim-protos", - "hyper 0.14.32", + "http-body-util", + "hyper 1.10.1", + "hyper-util", "hypervisor", "kata-sys-util", "kata-types", @@ -7123,6 +7105,15 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -7345,7 +7336,10 @@ name = "shim-interface" version = "0.1.0" dependencies = [ "anyhow", - "hyper 0.14.32", + "bytes 1.11.1", + "http-body-util", + "hyper 1.10.1", + "hyper-util", "hyperlocal", "kata-sys-util", "kata-types", @@ -7633,9 +7627,9 @@ dependencies = [ [[package]] name = "subprocess" -version = "0.2.15" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2c56e8662b206b9892d7a5a3f2ecdbcb455d3d6b259111373b7e08b8055158a8" +checksum = "feb179a3921e9debb9b1479c5a0abf0df75d05a412d8af9c81409c088d449413" dependencies = [ "libc", "winapi", @@ -8098,6 +8092,21 @@ dependencies = [ "serde", ] +[[package]] +name = "toml" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81f3d15e84cbcd896376e6730314d59fb5a87f31e4b038454184435cd57defee" +dependencies = [ + "indexmap 2.14.0", + "serde_core", + "serde_spanned", + "toml_datetime 1.1.1+spec-1.1.0", + "toml_parser", + "toml_writer", + "winnow 1.0.2", +] + [[package]] name = "toml_datetime" version = "0.6.11" @@ -8163,6 +8172,12 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" +[[package]] +name = "toml_writer" +version = "1.1.1+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "756daf9b1013ebe47a8776667b466417e2d4c5679d441c26230efd9ef78692db" + [[package]] name = "tonic" version = "0.9.2" @@ -8205,7 +8220,7 @@ dependencies = [ "http 1.4.0", "http-body 1.0.1", "http-body-util", - "hyper 1.9.0", + "hyper 1.10.1", "hyper-timeout 0.5.2", "hyper-util", "percent-encoding", diff --git a/Cargo.toml b/Cargo.toml index 46a796b950..8fbf549a86 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -164,8 +164,11 @@ containerd-shim-protos = { version = "0.10.0", features = ["async"] } derivative = "2.2.0" futures = "0.3.30" go-flag = "0.1.0" -hyper = "0.14.20" -hyperlocal = "0.8.0" +bytes = "1.0" +http-body-util = "0.1" +hyper = { version = "1.10.1", features = ["http1", "server", "client"] } +hyper-util = { version = "0.1", features = ["tokio", "server-auto", "server-graceful", "client-legacy"] } +hyperlocal = "0.9.1" ipnetwork = "0.17.0" lazy_static = "1.4" libc = "0.2.94" @@ -203,7 +206,7 @@ thiserror = "2.0.18" tokio = "1.46.1" tokio-util = "0.7.17" tokio-vsock = "0.3.4" -toml = "0.5.8" +toml = "1.1.2" tracing = "0.1.44" tracing-opentelemetry = "0.18.0" tracing-subscriber = "0.3.20" diff --git a/src/agent/policy/Cargo.toml b/src/agent/policy/Cargo.toml index c7790166dd..beeeb573e8 100644 --- a/src/agent/policy/Cargo.toml +++ b/src/agent/policy/Cargo.toml @@ -34,7 +34,7 @@ regorus = { git = "https://github.com/microsoft/regorus", rev = "898643129e3652e "regex", "std", ] } -json-patch = "2.0.0" +json-patch = "4.2.0" # POSIX libc.workspace = true diff --git a/src/libs/kata-sys-util/Cargo.toml b/src/libs/kata-sys-util/Cargo.toml index 3538cb3e6a..43c912ad15 100644 --- a/src/libs/kata-sys-util/Cargo.toml +++ b/src/libs/kata-sys-util/Cargo.toml @@ -21,7 +21,7 @@ serde = { version = "1.0.138", features = ["derive"] } serde_json = "1.0.73" slog = "2.5.2" slog-scope = "4.4.0" -subprocess = "0.2.8" +subprocess = "1.2.0" rand = { version = "0.10.1", features = ["std", "std_rng", "thread_rng"] } thiserror = { workspace = true } hex = "0.4.3" diff --git a/src/libs/kata-sys-util/src/hooks.rs b/src/libs/kata-sys-util/src/hooks.rs index cc0e3b6aa7..fd1a65e301 100644 --- a/src/libs/kata-sys-util/src/hooks.rs +++ b/src/libs/kata-sys-util/src/hooks.rs @@ -7,11 +7,11 @@ use std::collections::HashMap; use std::ffi::OsString; use std::hash::{Hash, Hasher}; -use std::io::{self, Read, Result}; +use std::io::{self, Result}; use std::time::Duration; use oci_spec::runtime as oci; -use subprocess::{ExitStatus, Popen, PopenConfig, PopenError, Redirection}; +use subprocess::{Exec, ExitStatus, Job, Redirection}; use crate::sl; use crate::validate::valid_env; @@ -130,34 +130,37 @@ impl HookStates { self.states.insert(hook.into(), HookState::Pending); - let mut executor = HookExecutor::new(hook)?; - let stdin = if state.is_some() { - Redirection::Pipe - } else { - Redirection::None - }; - let mut popen = Popen::create( - &executor.args, - PopenConfig { - stdin, - stdout: Redirection::Pipe, - stderr: Redirection::Pipe, - executable: executor.executable.to_owned(), - detached: true, - env: Some(executor.envs.clone()), - ..Default::default() - }, - ) - .map_err(|e| { - std::io::Error::other(format!( - "failed to create subprocess for hook {hook:?}: {e}" - )) - })?; + let executor = HookExecutor::new(hook)?; + let mut job = executor.spawn(state.as_ref())?; - if let Some(state) = state { - executor.execute_with_input(&mut popen, state)?; + let communicate_result = job + .communicate()? + .limit_time(Duration::from_secs(executor.timeout)) + .read_string(); + + match communicate_result { + Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { + error!(sl!(), "hook poll failed, kill it"); + let _ = job.kill(); + let _ = job.wait(); + return Err(io::Error::from(io::ErrorKind::TimedOut)); + } + Err(e) => { + return Err(std::io::Error::other(format!( + "communicate hook {hook:?}: {e}" + ))); + } + Ok((ref stdout, ref stderr)) => { + if !stderr.is_empty() { + error!(sl!(), "hook {} stderr: {}", hook.path().display(), stderr); + } + if !stdout.is_empty() { + info!(sl!(), "hook {} stdout: {}", hook.path().display(), stdout); + } + } } - executor.execute_and_wait(&mut popen)?; + + executor.wait_and_check(&mut job)?; info!(sl!(), "hook {} finished", hook.path().display()); self.states.insert(hook.into(), HookState::Done); @@ -191,8 +194,8 @@ impl HookStates { struct HookExecutor<'a> { hook: &'a oci::Hook, - executable: Option, - args: Vec, + executable: OsString, + args: Vec, envs: Vec<(OsString, OsString)>, timeout: u64, } @@ -200,24 +203,23 @@ struct HookExecutor<'a> { impl<'a> HookExecutor<'a> { fn new(hook: &'a oci::Hook) -> Result { // Ensure Hook.path is present and is an absolute path. - let executable = if !hook.path().exists() { + let path = hook.path(); + if !path.exists() { return Err(std::io::Error::other(format!( "path of hook {hook:?} is empty" ))); - } else { - let path = hook.path(); - if !path.is_absolute() { - return Err(std::io::Error::other(format!( - "path of hook {hook:?} is not absolute" - ))); - } - Some(path.as_os_str().to_os_string()) - }; + } + if !path.is_absolute() { + return Err(std::io::Error::other(format!( + "path of hook {hook:?} is not absolute" + ))); + } + let executable = path.as_os_str().to_os_string(); // Hook.args is optional, use Hook.path as arg0 if Hook.args is empty. - let args = match hook.args() { - Some(args) => args.clone(), - None => vec![hook.path().display().to_string()], + let args: Vec = match hook.args() { + Some(args) => args.iter().map(OsString::from).collect(), + None => vec![OsString::from(hook.path())], }; let mut envs: Vec<(OsString, OsString)> = Vec::new(); @@ -246,110 +248,85 @@ impl<'a> HookExecutor<'a> { }) } - fn execute_with_input(&mut self, popen: &mut Popen, state: runtime_spec::State) -> Result<()> { - let st = serde_json::to_string(&state)?; - let (stdout, stderr) = popen - .communicate_start(Some(st.as_bytes().to_vec())) - .limit_time(Duration::from_secs(self.timeout)) - .read_string() - .map_err(|e| e.error)?; - if let Some(err) = stderr { - if !err.is_empty() { - error!( - sl!(), - "hook {} exec failed: {}", - self.hook.path().display(), - err - ); - } + /// Spawn the subprocess, optionally feeding `state` to its stdin. + fn spawn(&self, state: Option<&runtime_spec::State>) -> Result { + let mut exec = Exec::cmd(&self.args[0]) + .args(&self.args[1..]) + .arg0(&self.executable) + .env_clear() + .env_extend( + self.envs + .iter() + .map(|(k, v)| (k.as_os_str(), v.as_os_str())), + ) + .detached() + .stdout(Redirection::Pipe) + .stderr(Redirection::Pipe); + + if let Some(st) = state { + let json = serde_json::to_string(st)?; + exec = exec.stdin(json.into_bytes()); + } else { + exec = exec.stdin(Redirection::None); } - if let Some(out) = stdout { - if !out.is_empty() { + + exec.start().map_err(|e| { + std::io::Error::other(format!( + "failed to create subprocess for hook {:?}: {e}", + self.hook + )) + }) + } + + /// Wait for the process to finish and check its exit status. + fn wait_and_check(&self, job: &mut Job) -> Result<()> { + match job.wait_timeout(Duration::from_secs(1)) { + Ok(Some(exit_status)) => { info!( sl!(), - "hook {} exec stdout: {}", - self.hook.path().display(), - out + "exit status of hook {:?} : {:?}", self.hook, exit_status ); + self.check_exit_status(exit_status) + } + Ok(None) => { + // Timeout — kill the process. + error!(sl!(), "hook poll failed, kill it"); + let _ = job.kill(); + let _ = job.wait(); + Err(io::Error::from(io::ErrorKind::TimedOut)) + } + Err(e) => { + error!(sl!(), "wait_timeout for hook {:?} failed: {}", self.hook, e); + Err(std::io::Error::other(format!( + "wait_timeout for hook {:?} failed: {}", + self.hook, e, + ))) } } - // Give a grace period for `execute_and_wait()`. - self.timeout = 1; - Ok(()) } - fn execute_and_wait(&mut self, popen: &mut Popen) -> Result<()> { - match popen.wait_timeout(Duration::from_secs(self.timeout)) { - Ok(v) => self.handle_exit_status(v, popen), - Err(e) => self.handle_popen_wait_error(e, popen), - } - } - - fn handle_exit_status(&mut self, result: Option, popen: &mut Popen) -> Result<()> { - if let Some(exit_status) = result { - // the process has finished - info!( - sl!(), - "exit status of hook {:?} : {:?}", self.hook, exit_status - ); - self.print_result(popen); - match exit_status { - subprocess::ExitStatus::Exited(code) => { - if code == 0 { - info!(sl!(), "hook {:?} succeeds", self.hook); - Ok(()) - } else { - warn!(sl!(), "hook {:?} exit status with {}", self.hook, code,); - Err(std::io::Error::other(format!( - "hook {:?} exit status with {}", - self.hook, code, - ))) - } - } - _ => { - error!( - sl!(), - "no exit code for hook {:?}: {:?}", self.hook, exit_status - ); - Err(std::io::Error::other(format!( - "no exit code for hook {:?}: {:?}", - self.hook, exit_status, - ))) - } + fn check_exit_status(&self, exit_status: ExitStatus) -> Result<()> { + match exit_status.code() { + Some(0) => { + info!(sl!(), "hook {:?} succeeds", self.hook); + Ok(()) } - } else { - // may be timeout - error!(sl!(), "hook poll failed, kill it"); - // it is still running, kill it - popen.kill()?; - let _ = popen.wait(); - self.print_result(popen); - Err(io::Error::from(io::ErrorKind::TimedOut)) - } - } - - fn handle_popen_wait_error(&mut self, e: PopenError, popen: &mut Popen) -> Result<()> { - self.print_result(popen); - error!(sl!(), "wait_timeout for hook {:?} failed: {}", self.hook, e); - Err(std::io::Error::other(format!( - "wait_timeout for hook {:?} failed: {}", - self.hook, e, - ))) - } - - fn print_result(&mut self, popen: &mut Popen) { - if let Some(file) = popen.stdout.as_mut() { - let mut buffer = String::new(); - file.read_to_string(&mut buffer).ok(); - if !buffer.is_empty() { - info!(sl!(), "hook stdout: {}", buffer); + Some(code) => { + warn!(sl!(), "hook {:?} exit status with {}", self.hook, code); + Err(std::io::Error::other(format!( + "hook {:?} exit status with {}", + self.hook, code, + ))) } - } - if let Some(file) = popen.stderr.as_mut() { - let mut buffer = String::new(); - file.read_to_string(&mut buffer).ok(); - if !buffer.is_empty() { - info!(sl!(), "hook stderr: {}", buffer); + None => { + error!( + sl!(), + "no exit code for hook {:?}: {:?}", self.hook, exit_status + ); + Err(std::io::Error::other(format!( + "no exit code for hook {:?}: {:?}", + self.hook, exit_status, + ))) } } } diff --git a/src/libs/kata-types/Cargo.toml b/src/libs/kata-types/Cargo.toml index 94e58e9c01..242a669d47 100644 --- a/src/libs/kata-types/Cargo.toml +++ b/src/libs/kata-types/Cargo.toml @@ -24,7 +24,7 @@ slog = "2.5.2" slog-scope = "4.4.0" serde_json = "1.0.73" thiserror = { workspace = true } -toml = "0.5.8" +toml = { workspace = true } serde-enum-str = "0.5" sysinfo = "0.34.2" sha2 = "0.10.8" diff --git a/src/libs/kata-types/src/config/drop_in.rs b/src/libs/kata-types/src/config/drop_in.rs index bcd99033ec..6208a8c73d 100644 --- a/src/libs/kata-types/src/config/drop_in.rs +++ b/src/libs/kata-types/src/config/drop_in.rs @@ -240,7 +240,8 @@ mod drop_in_directory_handling { )); } let dropin_contents = fs::read_to_string(dropin_file.path())?; - let dropin_config: toml::Value = toml::from_str(&dropin_contents)?; + let dropin_config: toml::Value = + toml::from_str(&dropin_contents).map_err(std::io::Error::other)?; super::toml_tree_ops::merge(base_config, dropin_config); Ok(()) } @@ -267,12 +268,13 @@ mod drop_in_directory_handling { pub fn load(base_cfg_file_path: &Path) -> Result { let base_toml_str = fs::read_to_string(base_cfg_file_path)?; - let mut base_config: toml::Value = toml::from_str(&base_toml_str)?; + let mut base_config: toml::Value = + toml::from_str(&base_toml_str).map_err(std::io::Error::other)?; let dropin_dir = get_dropin_dir_path(base_cfg_file_path)?; update_from_dropins(&mut base_config, &dropin_dir)?; - let config: TomlConfig = base_config.try_into()?; + let config: TomlConfig = base_config.try_into().map_err(std::io::Error::other)?; Ok(config) } diff --git a/src/libs/kata-types/src/config/mod.rs b/src/libs/kata-types/src/config/mod.rs index 1ba04e98f0..18f69dd379 100644 --- a/src/libs/kata-types/src/config/mod.rs +++ b/src/libs/kata-types/src/config/mod.rs @@ -179,7 +179,7 @@ impl TomlConfig { /// This function only works with `configuration.toml` and does not handle /// drop-in config file fragments in config.d/. pub fn load(content: &str) -> Result { - let mut config: TomlConfig = toml::from_str(content)?; + let mut config: TomlConfig = toml::from_str(content).map_err(std::io::Error::other)?; config.adjust_config()?; info!(sl!(), "get kata config: {:?}", config); Ok(config) diff --git a/src/libs/kata-types/src/initdata.rs b/src/libs/kata-types/src/initdata.rs index c6841e1c28..34cf33f34f 100644 --- a/src/libs/kata-types/src/initdata.rs +++ b/src/libs/kata-types/src/initdata.rs @@ -93,7 +93,7 @@ impl InitData { /// serialize it to Vec pub fn to_vec(&self) -> Result> { - Ok(toml::to_vec(&self)?) + Ok(toml::to_string(&self)?.into_bytes()) } /// serialize config to TOML string @@ -411,8 +411,8 @@ key = "value" // Test TOML serialization let toml_str = init_data.to_string().unwrap(); - assert!(toml_str.contains("initdata_key = 'initdata_value'\n")); - assert!(toml_str.starts_with("version = '0.1.0'")); + assert!(toml_str.contains("initdata_key = \"initdata_value\"\n")); + assert!(toml_str.starts_with("version = \"0.1.0\"")); } /// Test calculate_digest with different algorithms @@ -481,8 +481,8 @@ key = "value" // Test processing let result = add_hypervisor_initdata_overrides(&b64_annotation).unwrap(); - assert!(result.contains("hypervisor_key = 'config_value'\n")); - assert!(result.contains("algorithm = 'sha512'\n")); + assert!(result.contains("hypervisor_key = \"config_value\"\n")); + assert!(result.contains("algorithm = \"sha512\"\n")); } /// Test input validation diff --git a/src/libs/shim-interface/Cargo.toml b/src/libs/shim-interface/Cargo.toml index f26b94db23..4af10d979e 100644 --- a/src/libs/shim-interface/Cargo.toml +++ b/src/libs/shim-interface/Cargo.toml @@ -15,8 +15,11 @@ edition = "2018" anyhow = "^1.0" nix = { workspace = true } tokio = { version = "1.44.2", features = ["rt-multi-thread"] } -hyper = { version = "0.14.20", features = ["stream", "server", "http1"] } -hyperlocal = "0.8" +bytes = { workspace = true } +http-body-util = { workspace = true } +hyper = { workspace = true } +hyper-util = { workspace = true } +hyperlocal = { workspace = true } kata-types = { path = "../kata-types" } kata-sys-util = { path = "../kata-sys-util" } diff --git a/src/libs/shim-interface/src/shim_mgmt/client.rs b/src/libs/shim-interface/src/shim_mgmt/client.rs index ae4f02fb9f..11a1dbd843 100644 --- a/src/libs/shim-interface/src/shim_mgmt/client.rs +++ b/src/libs/shim-interface/src/shim_mgmt/client.rs @@ -13,7 +13,10 @@ use std::{path::Path, path::PathBuf, time::Duration}; use crate::mgmt_socket_addr; use anyhow::{anyhow, Context, Result}; -use hyper::{Body, Client, Method, Request, Response}; +use bytes::Bytes; +use http_body_util::Full; +use hyper::{body::Incoming, Method, Request, Response}; +use hyper_util::client::legacy::Client; use hyperlocal::{UnixClientExt, UnixConnector, Uri}; /// Shim management client with timeout @@ -22,7 +25,7 @@ pub struct MgmtClient { sock_path: PathBuf, /// The http client connect to the long standing shim mgmt server - client: Client, + client: Client>, /// Timeout value for each dial, usually 200ms will be enough /// For heavier workload, you may want longer timeout @@ -47,12 +50,12 @@ impl MgmtClient { /// The http GET method for client, return a raw response. Further handling should be done by caller. /// Parameter uri should be like "/agent-url" etc. - pub async fn get(&self, uri: &str) -> Result> { + pub async fn get(&self, uri: &str) -> Result> { let url: hyper::Uri = Uri::new(&self.sock_path, uri).into(); let req = Request::builder() .method(Method::GET) .uri(url) - .body(Body::empty())?; + .body(Full::new(Bytes::new()))?; self.send_request(req).await } @@ -62,11 +65,11 @@ impl MgmtClient { uri: &str, content_type: &str, content: &str, - ) -> Result> { + ) -> Result> { let url: hyper::Uri = Uri::new(&self.sock_path, uri).into(); // build body from content - let body = Body::from(content.to_string()); + let body = Full::new(Bytes::from(content.to_string())); let req = Request::builder() .method(Method::POST) .uri(url) @@ -76,16 +79,16 @@ impl MgmtClient { } /// The http PUT method for client - pub async fn put(&self, uri: &str, data: Vec) -> Result> { + pub async fn put(&self, uri: &str, data: Vec) -> Result> { let url: hyper::Uri = Uri::new(&self.sock_path, uri).into(); let req = Request::builder() .method(Method::PUT) .uri(url) - .body(Body::from(data))?; + .body(Full::new(Bytes::from(data)))?; self.send_request(req).await } - async fn send_request(&self, req: Request) -> Result> { + async fn send_request(&self, req: Request>) -> Result> { let msg = format!("Request ({:?}) to uri {:?}", req.method(), req.uri()); let resp = self.client.request(req); match self.timeout { diff --git a/src/runtime-rs/crates/hypervisor/Cargo.toml b/src/runtime-rs/crates/hypervisor/Cargo.toml index 86b6eeaa84..09c5f76830 100644 --- a/src/runtime-rs/crates/hypervisor/Cargo.toml +++ b/src/runtime-rs/crates/hypervisor/Cargo.toml @@ -33,7 +33,10 @@ crossbeam-channel = "0.5.6" qapi = { version = "0.15", features = ["qmp", "async-tokio-all"] } qapi-spec = "0.3.2" qapi-qmp = "0.15.0" -hyper = { workspace = true, features = ["client"] } +bytes = { workspace = true } +http-body-util = { workspace = true } +hyper = { workspace = true } +hyper-util = { workspace = true } regex = "1" once_cell = "1.21.3" # Local dependencies diff --git a/src/runtime-rs/crates/hypervisor/src/firecracker/fc_api.rs b/src/runtime-rs/crates/hypervisor/src/firecracker/fc_api.rs index 0170bd10ee..cb2d4efa72 100644 --- a/src/runtime-rs/crates/hypervisor/src/firecracker/fc_api.rs +++ b/src/runtime-rs/crates/hypervisor/src/firecracker/fc_api.rs @@ -12,8 +12,10 @@ use crate::{ NetworkConfig, Param, }; use anyhow::{anyhow, Context, Result}; +use bytes::Bytes; use dbs_utils::net::MacAddr; -use hyper::{Body, Method, Request, Response}; +use http_body_util::{BodyExt, Full}; +use hyper::{body::Incoming, Method, Request, Response}; use hyperlocal::Uri; use kata_sys_util::mount; use kata_types::config::hypervisor::RateLimiterConfig; @@ -256,7 +258,7 @@ impl FcInner { .uri(uri.clone()) .header("Accept", "application/json") .header("Content-Type", "application/json") - .body(Body::from(data.clone()))?; + .body(Full::new(Bytes::from(data.clone())))?; match self.send_request(req).await { Ok(resp) => { @@ -276,7 +278,10 @@ impl FcInner { )) } - pub(crate) async fn send_request(&self, req: Request) -> Result> { + pub(crate) async fn send_request( + &self, + req: Request>, + ) -> Result> { let resp = self.client.request(req).await?; let status = resp.status(); @@ -284,7 +289,7 @@ impl FcInner { if status.is_success() { return Ok(resp); } else { - let body = hyper::body::to_bytes(resp.into_body()).await?; + let body = resp.into_body().collect().await?.to_bytes(); if body.is_empty() { debug!(sl(), "Request FAILED WITH STATUS: {:?}", status); None diff --git a/src/runtime-rs/crates/hypervisor/src/firecracker/inner.rs b/src/runtime-rs/crates/hypervisor/src/firecracker/inner.rs index 84c4e56cb0..d841aba617 100644 --- a/src/runtime-rs/crates/hypervisor/src/firecracker/inner.rs +++ b/src/runtime-rs/crates/hypervisor/src/firecracker/inner.rs @@ -10,7 +10,9 @@ use crate::{device::DeviceType, VmmState}; use crate::{selinux, HypervisorState}; use anyhow::{anyhow, Context, Result}; use async_trait::async_trait; -use hyper::Client; +use bytes::Bytes; +use http_body_util::Full; +use hyper_util::client::legacy::Client; use hyperlocal::{UnixClientExt, UnixConnector}; use kata_types::{ capabilities::{Capabilities, CapabilityBits}, @@ -37,7 +39,7 @@ pub struct FcInner { pub(crate) pid: Option, pub(crate) vm_path: String, pub(crate) netns: Option, - pub(crate) client: Client, + pub(crate) client: Client>, pub(crate) jailer_root: String, pub(crate) jailed: bool, pub(crate) run_dir: String, diff --git a/src/runtime-rs/crates/resource/Cargo.toml b/src/runtime-rs/crates/resource/Cargo.toml index 29dfa609fc..6ee758d479 100644 --- a/src/runtime-rs/crates/resource/Cargo.toml +++ b/src/runtime-rs/crates/resource/Cargo.toml @@ -39,7 +39,10 @@ flate2 = "1.1" tempfile = "3.19.1" hex = "0.4" base64 = { workspace = true } -hyper = { workspace = true, features = ["client", "http1"] } +bytes = { workspace = true } +http-body-util = { workspace = true } +hyper = { workspace = true } +hyper-util = { workspace = true } hyperlocal = { workspace = true } ## Dependencies from `rust-netlink` diff --git a/src/runtime-rs/crates/resource/src/share_fs/nydus/nydus_client.rs b/src/runtime-rs/crates/resource/src/share_fs/nydus/nydus_client.rs index 76f300aa3f..65a4507f13 100644 --- a/src/runtime-rs/crates/resource/src/share_fs/nydus/nydus_client.rs +++ b/src/runtime-rs/crates/resource/src/share_fs/nydus/nydus_client.rs @@ -6,8 +6,11 @@ use std::path::{Path, PathBuf}; use anyhow::{anyhow, Context, Result}; -use hyper::{body::to_bytes, Body, Client, Method, Request, StatusCode}; -use hyperlocal::{UnixClientExt, Uri}; +use bytes::Bytes; +use http_body_util::{BodyExt, Full}; +use hyper::{Method, Request, StatusCode}; +use hyper_util::client::legacy::Client; +use hyperlocal::{UnixClientExt, UnixConnector, Uri}; use serde::{Deserialize, Serialize}; use tokio::time::{timeout, Duration}; @@ -54,7 +57,7 @@ pub struct ErrorMessage { pub struct NydusClient { sock_path: PathBuf, - client: Client, + client: Client>, } impl NydusClient { @@ -91,10 +94,10 @@ impl NydusClient { let req = match body { Some(b) => request_builder - .body(Body::from(b.to_string())) + .body(Full::new(Bytes::from(b.to_string()))) .context("failed to build HTTP request with body")?, None => request_builder - .body(Body::empty()) + .body(Full::new(Bytes::new())) .context("failed to build HTTP request")?, }; @@ -104,9 +107,7 @@ impl NydusClient { .context("failed to send HTTP request")?; let status = response.status(); - let body_bytes = to_bytes(response.into_body()) - .await - .context("failed to read response body")?; + let body_bytes = response.into_body().collect().await?.to_bytes(); Ok((status, body_bytes.to_vec())) } diff --git a/src/runtime-rs/crates/runtimes/Cargo.toml b/src/runtime-rs/crates/runtimes/Cargo.toml index f3f0f0f9bd..d3526e8822 100644 --- a/src/runtime-rs/crates/runtimes/Cargo.toml +++ b/src/runtime-rs/crates/runtimes/Cargo.toml @@ -26,7 +26,10 @@ opentelemetry-jaeger = { version = "0.17.0", features = [ "collector_client", ] } tracing-subscriber = { version = "0.3", features = ["registry", "std"] } -hyper = { workspace = true, features = ["stream", "server", "http1"] } +bytes = { workspace = true } +http-body-util = { workspace = true } +hyper = { workspace = true } +hyper-util = { workspace = true } serde_json = { workspace = true } nix = { workspace = true } url = { workspace = true } diff --git a/src/runtime-rs/crates/runtimes/src/shim_mgmt/handlers.rs b/src/runtime-rs/crates/runtimes/src/shim_mgmt/handlers.rs index a2c9949c95..e1d0b3cc70 100644 --- a/src/runtime-rs/crates/runtimes/src/shim_mgmt/handlers.rs +++ b/src/runtime-rs/crates/runtimes/src/shim_mgmt/handlers.rs @@ -10,8 +10,10 @@ use crate::shim_metrics::get_shim_metrics; use agent::ResizeVolumeRequest; use anyhow::{anyhow, Context, Result}; +use bytes::Bytes; use common::Sandbox; -use hyper::{Body, Method, Request, Response, StatusCode}; +use http_body_util::{BodyExt, Full}; +use hyper::{body::Incoming, Method, Request, Response, StatusCode}; use std::{str, sync::Arc}; use url::Url; @@ -24,8 +26,8 @@ use shim_interface::shim_mgmt::{ // http arrival which invokes the corresponding handler function pub(crate) async fn handler_mux( sandbox: Arc, - req: Request, -) -> Result> { + req: Request, +) -> Result>> { info!( sl!(), "mgmt-svr(mux): recv req, method: {}, uri: {}", @@ -51,35 +53,38 @@ pub(crate) async fn handler_mux( } // url not found -async fn not_found(_req: Request) -> Response { +async fn not_found(_req: Request) -> Response> { Response::builder() .status(StatusCode::NOT_FOUND) - .body(Body::from("URL NOT FOUND")) + .body(Full::new(Bytes::from("URL NOT FOUND"))) .unwrap() } // returns the url for agent async fn agent_url_handler( sandbox: Arc, - _req: Request, -) -> Result> { + _req: Request, +) -> Result>> { let agent_sock = sandbox .agent_sock() .await .unwrap_or_else(|_| String::from("")); - Ok(Response::new(Body::from(agent_sock))) + Ok(Response::new(Full::new(Bytes::from(agent_sock)))) } /// the ipv4 handler of iptable operation -async fn ip_table_handler(sandbox: Arc, req: Request) -> Result> { +async fn ip_table_handler( + sandbox: Arc, + req: Request, +) -> Result>> { generic_ip_table_handler(sandbox, req, false).await } /// the ipv6 handler of iptable operation async fn ipv6_table_handler( sandbox: Arc, - req: Request, -) -> Result> { + req: Request, +) -> Result>> { generic_ip_table_handler(sandbox, req, true).await } @@ -87,24 +92,24 @@ async fn ipv6_table_handler( /// this requires iptables-series binaries to be inside guest rootfs async fn generic_ip_table_handler( sandbox: Arc, - req: Request, + req: Request, is_ipv6: bool, -) -> Result> { +) -> Result>> { info!(sl!(), "handler: iptable ipv6?: {}", is_ipv6); match *req.method() { Method::GET => match sandbox.get_iptables(is_ipv6).await { Ok(data) => { - let body = Body::from(data); + let body = Full::new(Bytes::from(data)); Response::builder().body(body).map_err(|e| anyhow!(e)) } _ => Err(anyhow!("Failed to get iptable")), }, Method::PUT => { - let data = hyper::body::to_bytes(req.into_body()).await?; + let data = req.into_body().collect().await?.to_bytes(); match sandbox.set_iptables(is_ipv6, data.to_vec()).await { Ok(resp_data) => Response::builder() - .body(Body::from(resp_data)) + .body(Full::new(Bytes::from(resp_data))) .map_err(|e| anyhow!(e)), _ => Err(anyhow!("Failed to set iptable")), } @@ -116,8 +121,8 @@ async fn generic_ip_table_handler( async fn direct_volume_stats_handler( sandbox: Arc, - req: Request, -) -> Result> { + req: Request, +) -> Result>> { let params = Url::parse(&req.uri().to_string()) .map_err(|e| anyhow!(e))? .query_pairs() @@ -128,16 +133,16 @@ async fn direct_volume_stats_handler( .context("shim-mgmt: volume path key not found in request params")?; let result = sandbox.direct_volume_stats(volume_path).await; match result { - Ok(stats) => Ok(Response::new(Body::from(stats))), + Ok(stats) => Ok(Response::new(Full::new(Bytes::from(stats)))), _ => Err(anyhow!("handler: Failed to get volume stats")), } } async fn direct_volume_resize_handler( sandbox: Arc, - req: Request, -) -> Result> { - let body = hyper::body::to_bytes(req.into_body()).await?; + req: Request, +) -> Result>> { + let body = req.into_body().collect().await?.to_bytes(); // unserialize json body into resizeRequest struct let resize_req: ResizeVolumeRequest = @@ -145,7 +150,7 @@ async fn direct_volume_resize_handler( let result = sandbox.direct_volume_resize(resize_req).await; match result { - Ok(_) => Ok(Response::new(Body::from(""))), + Ok(_) => Ok(Response::new(Full::new(Bytes::from("")))), _ => Err(anyhow!("handler: Failed to resize volume")), } } @@ -153,32 +158,32 @@ async fn direct_volume_resize_handler( // returns the url for metrics async fn metrics_url_handler( sandbox: Arc, - _req: Request, -) -> Result> { + _req: Request, +) -> Result>> { // get metrics from agent, hypervisor, and shim let agent_metrics = sandbox.agent_metrics().await.unwrap_or_default(); let hypervisor_metrics = sandbox.hypervisor_metrics().await.unwrap_or_default(); let shim_metrics = get_shim_metrics().unwrap_or_default(); - Ok(Response::new(Body::from(format!( + Ok(Response::new(Full::new(Bytes::from(format!( "{agent_metrics}{hypervisor_metrics}{shim_metrics}" - )))) + ))))) } /// The set agent policy handler, for setting agent policy async fn set_agent_policy_handler( sandbox: Arc, - req: Request, -) -> Result> { + req: Request, +) -> Result>> { match *req.method() { Method::PUT => { - let data = hyper::body::to_bytes(req.into_body()).await?; + let data = req.into_body().collect().await?.to_bytes(); let policy: &str = str::from_utf8(&data)?; sandbox .set_policy(policy) .await .context("set agent policy handler failed")?; - Ok(Response::new(Body::from(""))) + Ok(Response::new(Full::new(Bytes::from("")))) } _ => Err(anyhow!("Set agent policy only takes PUT method")), } diff --git a/src/runtime-rs/crates/runtimes/src/shim_mgmt/server.rs b/src/runtime-rs/crates/runtimes/src/shim_mgmt/server.rs index 3484e9e826..f50ef7e5e7 100644 --- a/src/runtime-rs/crates/runtimes/src/shim_mgmt/server.rs +++ b/src/runtime-rs/crates/runtimes/src/shim_mgmt/server.rs @@ -15,7 +15,9 @@ use std::{fs, path::Path, sync::Arc}; use anyhow::{Context, Result}; use common::Sandbox; -use hyper::{server::conn::Http, service::service_fn}; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper_util::rt::TokioIo; use shim_interface::{sb_storage_path, SHIM_MGMT_SOCK_NAME}; use tokio::net::UnixListener; @@ -52,12 +54,13 @@ impl MgmtServer { // start an infinite loop, which serves the incomming uds stream loop { let (stream, _) = listener.accept().await.unwrap(); + let io = TokioIo::new(stream); let me = self.clone(); // spawn a light weight thread to multiplex to the handler tokio::task::spawn(async move { - if let Err(err) = Http::new() + if let Err(err) = http1::Builder::new() .serve_connection( - stream, + io, service_fn(|request| handler_mux(me.sandbox.clone(), request)), ) .await diff --git a/src/runtime-rs/crates/runtimes/virt_container/src/sandbox.rs b/src/runtime-rs/crates/runtimes/virt_container/src/sandbox.rs index 083b0f341e..b486063c19 100644 --- a/src/runtime-rs/crates/runtimes/virt_container/src/sandbox.rs +++ b/src/runtime-rs/crates/runtimes/virt_container/src/sandbox.rs @@ -33,7 +33,6 @@ use hypervisor::ch::CloudHypervisor; use hypervisor::device::topology::PCIePort; use hypervisor::device::util::{get_host_path, DEVICE_TYPE_CHAR}; use hypervisor::remote::Remote; -use hypervisor::{is_vfio_ap_device, VfioDeviceBase}; use hypervisor::VsockConfig; use hypervisor::HYPERVISOR_REMOTE; #[cfg(all( @@ -43,6 +42,7 @@ use hypervisor::HYPERVISOR_REMOTE; use hypervisor::{dragonball::Dragonball, HYPERVISOR_DRAGONBALL}; #[cfg(any(target_arch = "x86_64", target_arch = "aarch64"))] use hypervisor::{firecracker::Firecracker, HYPERVISOR_FIRECRACKER}; +use hypervisor::{is_vfio_ap_device, VfioDeviceBase}; use hypervisor::{qemu::Qemu, HYPERVISOR_QEMU}; use hypervisor::{ utils::{get_hvsock_path, uses_native_ccw_bus}, @@ -56,12 +56,12 @@ use kata_sys_util::protection::{available_guest_protection, GuestProtection}; use kata_sys_util::spec::load_oci_spec; use kata_types::capabilities::CapabilityBits; use kata_types::config::hypervisor::Hypervisor as HypervisorConfig; -use kata_types::config::hypervisor::{VIRTIO_BLK_CCW, VIRTIO_BLK_PCI}; #[cfg(all( feature = "cloud-hypervisor", any(target_arch = "x86_64", target_arch = "aarch64") ))] use kata_types::config::hypervisor::HYPERVISOR_NAME_CH; +use kata_types::config::hypervisor::{VIRTIO_BLK_CCW, VIRTIO_BLK_PCI}; use kata_types::config::{hypervisor::Factory, TomlConfig}; use kata_types::initdata::{calculate_initdata_digest, ProtectedPlatform}; use oci_spec::runtime as oci; @@ -287,7 +287,9 @@ impl VirtSandbox { // avoiding unnecessary file I/O and OCI spec parsing in the common K8s case. let mut vfio_devices = self.prepare_coldplug_cdi_devices(sandbox_config).await?; if vfio_devices.is_empty() { - let raw_vfio = self.prepare_coldplug_raw_vfio_devices(sandbox_config).await?; + let raw_vfio = self + .prepare_coldplug_raw_vfio_devices(sandbox_config) + .await?; vfio_devices.extend(raw_vfio); } if !vfio_devices.is_empty() { @@ -473,7 +475,9 @@ impl VirtSandbox { Err(e) => { warn!( sl!(), - "failed to resolve host path for {:?}: {:?}", d.path(), e + "failed to resolve host path for {:?}: {:?}", + d.path(), + e ); continue; } diff --git a/src/tools/kata-ctl/Cargo.toml b/src/tools/kata-ctl/Cargo.toml index 461d3a667d..b0cb50e4dd 100644 --- a/src/tools/kata-ctl/Cargo.toml +++ b/src/tools/kata-ctl/Cargo.toml @@ -41,7 +41,10 @@ libc.workspace = true logging.workspace = true slog.workspace = true slog-scope.workspace = true -hyper.workspace = true +bytes = { workspace = true } +http-body-util = { workspace = true } +hyper = { workspace = true } +hyper-util = { workspace = true } tokio = { workspace = true, features = ["signal"] } prometheus.workspace = true diff --git a/src/tools/kata-ctl/src/monitor/http_server.rs b/src/tools/kata-ctl/src/monitor/http_server.rs index ee19ac0f99..e9f5d9eddd 100644 --- a/src/tools/kata-ctl/src/monitor/http_server.rs +++ b/src/tools/kata-ctl/src/monitor/http_server.rs @@ -8,18 +8,28 @@ use crate::sl; use crate::utils::TIMEOUT; use anyhow::{anyhow, Context, Result}; -use hyper::body; -use hyper::service::{make_service_fn, service_fn}; -use hyper::{Body, Method, Request, Response, Server, StatusCode}; +use bytes::Bytes; +use http_body_util::{BodyExt, Full}; +use hyper::body::Body; +use hyper::service::service_fn; +use hyper::{Method, Request, Response, StatusCode}; +use hyper_util::rt::{TokioExecutor, TokioIo}; +use hyper_util::server::conn::auto::Builder as AutoBuilder; use shim_interface::shim_mgmt::client::MgmtClient; use slog::{self, info}; use std::collections::HashMap; use std::net::SocketAddr; +use tokio::net::TcpListener; const ROOT_URI: &str = "/"; const METRICS_URI: &str = "/metrics"; -async fn handler_mux(req: Request) -> Result> { +async fn handler_mux(req: Request) -> Result>> +where + B: Body + Send + 'static, + B::Data: Send, + B::Error: std::error::Error + Send + Sync, +{ info!( sl!(), "mgmt-svr(mux): recv req, method: {}, uri: {}", @@ -36,7 +46,7 @@ async fn handler_mux(req: Request) -> Result> { |e| { Response::builder() .status(StatusCode::INTERNAL_SERVER_ERROR) - .body(Body::from(format!("{e:?}\n"))) + .body(Full::new(Bytes::from(format!("{e:?}\n")))) .map_err(|e| anyhow!("Failed to Build Response {:?}", e)) }, Ok, @@ -48,26 +58,39 @@ pub async fn http_server_setup(socket_addr: &str) -> Result<()> { .parse() .context("failed to parse http socket address")?; - let make_svc = - make_service_fn(|_conn| async { Ok::<_, anyhow::Error>(service_fn(handler_mux)) }); + let listener = TcpListener::bind(addr) + .await + .context("failed to bind TCP listener")?; - Server::bind(&addr).serve(make_svc).await?; + let builder = AutoBuilder::new(TokioExecutor::new()); - Ok(()) + loop { + let (stream, _) = listener + .accept() + .await + .context("failed to accept connection")?; + let io = TokioIo::new(stream); + let builder = builder.clone(); + tokio::spawn(async move { + if let Err(e) = builder.serve_connection(io, service_fn(handler_mux)).await { + info!(sl!(), "http server connection error: {:?}", e); + } + }); + } } -async fn root_uri_handler(_req: Request) -> Result> { +async fn root_uri_handler(_req: Request) -> Result>> { Response::builder() .status(StatusCode::OK) - .body(Body::from( + .body(Full::new(Bytes::from( r#"Available HTTP endpoints: /metrics : Get metrics from sandboxes. "#, - )) + ))) .map_err(|e| anyhow!("Failed to Build Response {:?}", e)) } -async fn metrics_uri_handler(req: Request) -> Result> { +async fn metrics_uri_handler(req: Request) -> Result>> { let mut response_body = String::new(); response_body += &get_monitor_metrics().context("Failed to Get Monitor Metrics")?; @@ -82,7 +105,7 @@ async fn metrics_uri_handler(req: Request) -> Result> { Response::builder() .status(StatusCode::OK) - .body(Body::from(response_body)) + .body(Full::new(Bytes::from(response_body))) .map_err(|e| anyhow!("Failed to Build Response {:?}", e)) } @@ -98,16 +121,23 @@ async fn get_runtime_metrics(sandbox_id: &str) -> Result { .context("failed to get METRICS_URI")?; // get runtime_metrics - let runtime_metrics = String::from_utf8(body::to_bytes(shim_response).await?.to_vec()) - .context("failed to get runtime_metrics")?; + let runtime_metrics = String::from_utf8( + shim_response + .into_body() + .collect() + .await? + .to_bytes() + .to_vec(), + ) + .context("failed to get runtime_metrics")?; Ok(runtime_metrics) } -async fn not_found_uri_handler(_req: Request) -> Result> { +async fn not_found_uri_handler(_req: Request) -> Result>> { Response::builder() .status(StatusCode::NOT_FOUND) - .body(Body::from("NOT FOUND")) + .body(Full::new(Bytes::from("NOT FOUND"))) .map_err(|e| anyhow!("Failed to Build Response {:?}", e)) } @@ -140,7 +170,7 @@ mod tests { Request::builder() .method("GET") .uri("/") - .body(hyper::Body::from("")) + .body(Full::new(Bytes::new())) .unwrap(), ) .await @@ -155,7 +185,7 @@ mod tests { Request::builder() .method("GET") .uri("/metrics?sandbox=demo_sandbox") - .body(hyper::Body::from("")) + .body(Full::new(Bytes::new())) .unwrap(), ) .await @@ -170,7 +200,7 @@ mod tests { Request::builder() .method("POST") .uri("/metrics?sandbox=demo_sandbox") - .body(hyper::Body::from("")) + .body(Full::new(Bytes::new())) .unwrap(), ) .await diff --git a/src/tools/kata-ctl/src/ops/exec_ops.rs b/src/tools/kata-ctl/src/ops/exec_ops.rs index 57e3a4fb95..dc92fe6d94 100644 --- a/src/tools/kata-ctl/src/ops/exec_ops.rs +++ b/src/tools/kata-ctl/src/ops/exec_ops.rs @@ -17,6 +17,7 @@ use std::{ }; use anyhow::{anyhow, Context}; +use http_body_util::BodyExt; use hyper::StatusCode; use nix::sys::socket::{connect, socket, AddressFamily, SockFlag, SockType, VsockAddr}; use slog::{debug, error, o}; @@ -342,7 +343,7 @@ async fn get_agent_socket(sandbox_id: &str) -> anyhow::Result { return Err(anyhow!("shim client get connection failed: {:?} ", status)); } - let body = hyper::body::to_bytes(response.into_body()).await?; + let body = response.into_body().collect().await?.to_bytes(); let agent_sock = String::from_utf8(body.to_vec())?; Ok(agent_sock)