diff --git a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/container.rs b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/container.rs index f5e52eb453..45fd37921a 100644 --- a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/container.rs +++ b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/container.rs @@ -54,6 +54,19 @@ pub struct Container { pub(crate) passfd_listener_addr: Option<(String, u32)>, } +fn process_uses_passfd_io(inner: &ContainerInner, process: &ContainerProcess) -> Result { + match process.process_type { + ProcessType::Container => Ok(inner.init_process.passfd_io.is_some()), + ProcessType::Exec => Ok(inner + .exec_processes + .get(&process.exec_id) + .ok_or_else(|| Error::ProcessNotFound(process.clone()))? + .process + .passfd_io + .is_some()), + } +} + impl Container { pub async fn new( pid: u32, @@ -317,7 +330,7 @@ impl Container { return Err(err); } - if self.passfd_listener_addr.is_some() { + if process_uses_passfd_io(&inner, process)? { inner .init_process .passfd_io_wait(containers, self.agent.clone()) @@ -364,7 +377,7 @@ impl Container { } } - if self.passfd_listener_addr.is_some() { + if process_uses_passfd_io(&inner, process)? { // In passfd io mode, we don't bother with the IO. // We send `WaitProcessRequest` immediately to the agent // and wait for the response in a separate thread. diff --git a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/binary_io.rs b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/binary_io.rs new file mode 100644 index 0000000000..bd8fda2afa --- /dev/null +++ b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/binary_io.rs @@ -0,0 +1,268 @@ +// Copyright (c) 2026 Kata Contributors +// +// SPDX-License-Identifier: Apache-2.0 + +use std::{ + collections::HashSet, + io, + os::fd::{AsRawFd, FromRawFd, OwnedFd}, + path::PathBuf, + process::Stdio, + time::Duration, +}; + +use anyhow::{anyhow, Context, Result}; +use nix::{ + fcntl::{fcntl, FcntlArg, OFlag}, + sys::signal::{kill, Signal}, + unistd::{pipe2, Pid}, +}; +use tokio::{ + fs::File, + io::{AsyncReadExt, AsyncWrite}, + net::unix::pipe::Sender, + process::{Child, Command}, +}; +use url::Url; + +const BINARY_IO_PROC_DRAIN_TIMEOUT: Duration = Duration::from_secs(12); +const BINARY_IO_PROC_TERM_TIMEOUT: Duration = Duration::from_secs(12); + +pub(crate) struct BinaryIo { + pub(crate) stdout: Box, + pub(crate) stderr: Box, + pub(crate) logger: BinaryLogger, +} + +pub(crate) struct BinaryLogger { + child: Child, +} + +impl BinaryLogger { + pub(crate) async fn shutdown(mut self) { + match tokio::time::timeout(BINARY_IO_PROC_DRAIN_TIMEOUT, self.child.wait()).await { + Ok(Ok(status)) => { + info!(sl!(), "binary logger exited with {}", status); + return; + } + Ok(Err(err)) => { + warn!(sl!(), "failed to wait for binary logger: {}", err); + return; + } + Err(_) => warn!( + sl!(), + "binary logger did not exit after EOF; terminating it" + ), + } + + if let Some(pid) = self.child.id() { + if let Err(err) = kill(Pid::from_raw(pid as i32), Signal::SIGTERM) { + warn!(sl!(), "failed to terminate binary logger: {}", err); + let _ = self.child.start_kill(); + } + } + + match tokio::time::timeout(BINARY_IO_PROC_TERM_TIMEOUT, self.child.wait()).await { + Ok(Ok(status)) => info!(sl!(), "binary logger exited with {}", status), + Ok(Err(err)) => warn!( + sl!(), + "failed to wait for binary logger after SIGTERM: {}", err + ), + Err(_) => { + warn!(sl!(), "binary logger ignored SIGTERM; killing it"); + let _ = self.child.kill().await; + let _ = self.child.wait().await; + } + } + } +} + +fn pipe() -> Result<(OwnedFd, OwnedFd)> { + pipe2(OFlag::O_CLOEXEC).context("create pipe") +} + +fn duplicate_for_child(fd: &OwnedFd) -> Result { + // Keep the source descriptors away from their fd 3/4/5 destinations so + // the dup2 calls cannot overwrite one another. + let duplicated = fcntl(fd, FcntlArg::F_DUPFD_CLOEXEC(6)).context("duplicate logger fd")?; + + // SAFETY: F_DUPFD_CLOEXEC returns a new descriptor owned by the caller. + Ok(unsafe { OwnedFd::from_raw_fd(duplicated) }) +} + +fn pipe_writer(fd: OwnedFd) -> Result> { + Ok(Box::new( + Sender::from_owned_fd(fd).context("register logger pipe")?, + )) +} + +fn command_args(uri: &Url) -> Vec { + let mut seen = HashSet::new(); + let mut args = Vec::new(); + + // Go's url.Values iteration passes each key and only its first value. + for (key, value) in uri.query_pairs() { + if seen.insert(key.to_string()) { + args.push(key.into_owned()); + args.push(value.into_owned()); + } + } + + args +} + +fn command_path(uri: &Url) -> Result { + if uri.path().is_empty() { + return Err(anyhow!("binary logger URI has an empty path")); + } + + uri.to_file_path() + .map_err(|_| anyhow!("binary logger URI has an invalid path")) +} + +pub(crate) async fn open(uri: &Url, container_id: &str, namespace: &str) -> Result { + let binary = command_path(uri)?; + + let (stdout_read, stdout_write) = pipe().context("create stdout pipe")?; + let (stderr_read, stderr_write) = pipe().context("create stderr pipe")?; + let (ready_read, ready_write) = pipe().context("create readiness pipe")?; + + let child_stdout = duplicate_for_child(&stdout_read)?; + let child_stderr = duplicate_for_child(&stderr_read)?; + let child_ready = duplicate_for_child(&ready_write)?; + let child_fds = [ + child_stdout.as_raw_fd(), + child_stderr.as_raw_fd(), + child_ready.as_raw_fd(), + ]; + + let mut command = Command::new(&binary); + command + .args(command_args(uri)) + .env_clear() + .env("CONTAINER_ID", container_id) + .env("CONTAINER_NAMESPACE", namespace) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + command.kill_on_drop(true); + + // Match containerd's binary logger ABI: stdout, stderr, and readiness are + // inherited as fd 3, 4, and 5 respectively. + // SAFETY: the closure only invokes async-signal-safe descriptor operations. + unsafe { + command.pre_exec(move || { + for (source, destination) in child_fds.iter().copied().zip(3..=5) { + if libc::dup2(source, destination) < 0 { + return Err(io::Error::last_os_error()); + } + } + Ok(()) + }); + } + + let child = command + .spawn() + .with_context(|| format!("start binary logger {}", binary.display()))?; + + drop(stdout_read); + drop(stderr_read); + drop(ready_write); + drop(child_stdout); + drop(child_stderr); + drop(child_ready); + + // The legacy protocol accepts either one byte or EOF as readiness. Like + // the Go runtime, logger startup waits for that handshake. + let mut ready = File::from_std(std::fs::File::from(ready_read)); + let mut byte = [0_u8; 1]; + ready + .read(&mut byte) + .await + .context("wait for binary logger readiness")?; + + Ok(BinaryIo { + stdout: pipe_writer(stdout_write)?, + stderr: pipe_writer(stderr_write)?, + logger: BinaryLogger { child }, + }) +} + +#[cfg(test)] +mod tests { + use std::{fs, os::unix::fs::PermissionsExt, time::SystemTime}; + + use tokio::io::AsyncWriteExt; + + use super::*; + + #[test] + fn query_parameters_match_go_binary_logger_arguments() { + let uri = + Url::parse("binary:///logger?config=%2Frun%2Flog.json&empty=&config=ignored").unwrap(); + assert_eq!(command_args(&uri), ["config", "/run/log.json", "empty", ""]); + } + + #[test] + fn decodes_binary_logger_path_like_go() { + use std::os::unix::ffi::OsStrExt; + + let uri = Url::parse("binary:///opt/log%20helper%25/a%2Fb/%FF").unwrap(); + assert_eq!( + command_path(&uri).unwrap().as_os_str().as_bytes(), + b"/opt/log helper%/a/b/\xff" + ); + } + + #[tokio::test] + async fn streams_to_binary_logger_and_passes_metadata() { + let dir = std::env::temp_dir().join(format!( + "kata-binary-logger-test-{}", + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + fs::create_dir(&dir).unwrap(); + let logger = dir.join("logger helper%"); + let output = dir.join("output"); + fs::write( + &logger, + r#"#!/bin/sh +printf x >&5 +printf '%s\n%s\n%s\n%s\n' "$CONTAINER_ID" "$CONTAINER_NAMESPACE" "$1" "$2" > "$2.meta" +/bin/cat <&3 > "$2.stdout" +/bin/cat <&4 > "$2.stderr" +"#, + ) + .unwrap(); + fs::set_permissions(&logger, fs::Permissions::from_mode(0o755)).unwrap(); + + let uri = Url::parse(&format!( + "binary://{}/logger%20helper%25?output={}", + dir.display(), + output.display() + )) + .unwrap(); + let mut io = open(&uri, "container-id", "k8s.io").await.unwrap(); + io.stdout.write_all(b"stdout data\n").await.unwrap(); + io.stderr.write_all(b"stderr data\n").await.unwrap(); + drop(io.stdout); + drop(io.stderr); + io.logger.shutdown().await; + + assert_eq!( + fs::read(output.with_extension("stdout")).unwrap(), + b"stdout data\n" + ); + assert_eq!( + fs::read(output.with_extension("stderr")).unwrap(), + b"stderr data\n" + ); + assert_eq!( + fs::read_to_string(output.with_extension("meta")).unwrap(), + format!("container-id\nk8s.io\noutput\n{}\n", output.display()) + ); + fs::remove_dir_all(dir).unwrap(); + } +} diff --git a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/mod.rs b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/mod.rs index a72eecac2f..cb792f5b9f 100644 --- a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/mod.rs +++ b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/mod.rs @@ -4,9 +4,11 @@ // SPDX-License-Identifier: Apache-2.0 // +mod binary_io; mod container_io; pub use container_io::ContainerIo; mod passfd_io; mod shim_io; +pub(crate) use binary_io::BinaryLogger; pub use passfd_io::PassfdIo; pub use shim_io::ShimIo; diff --git a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/shim_io.rs b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/shim_io.rs index 1cecbf039b..d344625f9d 100644 --- a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/shim_io.rs +++ b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/io/shim_io.rs @@ -22,6 +22,8 @@ use tokio::{ }; use url::Url; +use super::binary_io::{self, BinaryLogger}; + /// Clear O_NONBLOCK for an fd (turn it into blocking mode). fn set_flag_with_blocking(fd: RawFd) { let flag = unsafe { libc::fcntl(fd, libc::F_GETFL) }; @@ -29,7 +31,6 @@ fn set_flag_with_blocking(fd: RawFd) { error!(sl!(), "failed to fcntl(F_GETFL) fd {} ret {}", fd, flag); return; } - let ret = unsafe { libc::fcntl(fd, libc::F_SETFL, flag & !libc::O_NONBLOCK) }; if ret < 0 { error!(sl!(), "failed to fcntl(F_SETFL) fd {} ret {}", fd, ret); @@ -60,6 +61,7 @@ pub struct ShimIo { pub stdin: Option>, pub stdout: Option>, pub stderr: Option>, + pub(crate) binary_logger: Option, } impl ShimIo { @@ -67,6 +69,8 @@ impl ShimIo { stdin: &Option, stdout: &Option, stderr: &Option, + container_id: &str, + namespace: &str, ) -> Result { info!( sl!(), @@ -116,6 +120,18 @@ impl ShimIo { } }; + let stdout_url = get_url(stdout); + if let Some(uri) = stdout_url.as_ref().filter(|uri| uri.scheme() == "binary") { + let binary = binary_io::open(uri, container_id, namespace) + .await + .context("open binary logger")?; + return Ok(Self { + stdin: stdin_fd, + stdout: Some(binary.stdout), + stderr: Some(binary.stderr), + binary_logger: Some(binary.logger), + }); + } let get_fd = |url: &Option| -> Option> { info!(sl!(), "get fd for {:?}", &url); if let Some(url) = url { @@ -139,6 +155,7 @@ impl ShimIo { stdin: stdin_fd, stdout: get_fd(&stdout_url), stderr: get_fd(&stderr_url), + binary_logger: None, }) } } diff --git a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/process.rs b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/process.rs index cc5e67e496..6798bc8b82 100644 --- a/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/process.rs +++ b/src/runtime-rs/crates/runtimes/virt_container/src/container_manager/process.rs @@ -15,7 +15,7 @@ use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; use tokio::sync::{watch, RwLock}; use super::container::Container; -use super::io::{ContainerIo, PassfdIo, ShimIo}; +use super::io::{BinaryLogger, ContainerIo, PassfdIo, ShimIo}; use super::logger_with_process; /// Exit status returned when containerd/shim is unable to determine @@ -97,6 +97,12 @@ fn open_fifo_write(path: &str) -> Result { open_fifo(path, false, true) } +fn is_binary_stdio(value: &str) -> bool { + url::Url::parse(value) + .map(|uri| uri.scheme() == "binary") + .unwrap_or(false) +} + impl Process { pub fn new( process: &ContainerProcess, @@ -133,12 +139,16 @@ impl Process { pub fn pre_fifos_open(&mut self) -> Result<()> { if let Some(ref stdout) = self.stdout { - self.stdout_r = Some(open_fifo_read(stdout).context("open stdout")?); + if !is_binary_stdio(stdout) { + self.stdout_r = Some(open_fifo_read(stdout).context("open stdout")?); + } } if !self.terminal { if let Some(ref stderr) = self.stderr { - self.stderr_r = Some(open_fifo_read(stderr).context("open stderr")?); + if !is_binary_stdio(stderr) { + self.stderr_r = Some(open_fifo_read(stderr).context("open stderr")?); + } } } @@ -156,6 +166,15 @@ impl Process { pub async fn passfd_io_init(&mut self, hvsock_uds_path: &str, passfd_port: u32) -> Result<()> { info!(self.logger, "passfd io init"); + // Binary loggers need the host shim to feed their pipes, so this + // process uses the legacy agent streams instead of passfd. + if self.stdout.as_deref().is_some_and(is_binary_stdio) + || self.stderr.as_deref().is_some_and(is_binary_stdio) + { + info!(self.logger, "binary logger disables passfd io"); + return Ok(()); + } + let mut passfd_io = PassfdIo::new(self.stdin.clone(), self.stdout.clone(), self.stderr.clone()).await; @@ -249,9 +268,15 @@ impl Process { self.pre_fifos_open()?; // new shim io - let shim_io = ShimIo::new(&self.stdin, &self.stdout, &self.stderr) - .await - .context("new shim io")?; + let shim_io = ShimIo::new( + &self.stdin, + &self.stdout, + &self.stderr, + self.process.container_id(), + &std::env::var("NAMESPACE").unwrap_or_default(), + ) + .await + .context("new shim io")?; self.post_fifos_open()?; // start io copy for stdin @@ -283,7 +308,7 @@ impl Process { } } - self.run_io_wait(containers, agent, wg) + self.run_io_wait(containers, agent, wg, shim_io.binary_logger) .await .context("run io thread")?; Ok(()) @@ -331,6 +356,9 @@ impl Process { } }; + // Close the destination before notifying the waiter. Binary + // loggers must see pipe EOF before their shutdown signal. + drop(writer); if let Some(w) = wgw { w.done() } @@ -347,6 +375,7 @@ impl Process { containers: Arc>>, agent: Arc, mut wg: WaitGroup, + binary_logger: Option, ) -> Result<()> { let logger = self.logger.clone(); info!(logger, "start run io wait"); @@ -361,6 +390,10 @@ impl Process { wg.wait().await; info!(logger, "end wait group for io"); + if let Some(binary_logger) = binary_logger { + binary_logger.shutdown().await; + } + let req = agent::WaitProcessRequest { process_id: process.clone().into(), }; @@ -457,3 +490,17 @@ impl Process { *status = new_status; } } + +#[cfg(test)] +mod tests { + use super::is_binary_stdio; + + #[test] + fn identifies_binary_logger_uri() { + assert!(is_binary_stdio( + "binary:///usr/bin/logger?config=%2Frun%2Flog.json" + )); + assert!(!is_binary_stdio("/run/containerd/io/stdout")); + assert!(!is_binary_stdio("file:///run/container.log")); + } +}