Merge pull request #13351 from cayoub-oai/runtime-rs-binary-logger

runtime-rs: support containerd binary loggers

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<bool> {
+    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.

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<dyn AsyncWrite + Send + Unpin>,
+    pub(crate) stderr: Box<dyn AsyncWrite + Send + Unpin>,
+    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<OwnedFd> {
+    // 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<Box<dyn AsyncWrite + Send + Unpin>> {
+    Ok(Box::new(
+        Sender::from_owned_fd(fd).context("register logger pipe")?,
+    ))
+}
+
+fn command_args(uri: &Url) -> Vec<String> {
+    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<PathBuf> {
+    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<BinaryIo> {
+    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();
+    }
+}

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;

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<Box<dyn AsyncRead + Send + Unpin>>,
     pub stdout: Option<Box<dyn AsyncWrite + Send + Unpin>>,
     pub stderr: Option<Box<dyn AsyncWrite + Send + Unpin>>,
+    pub(crate) binary_logger: Option<BinaryLogger>,
 }
 
 impl ShimIo {
@@ -67,6 +69,8 @@ impl ShimIo {
         stdin: &Option<String>,
         stdout: &Option<String>,
         stderr: &Option<String>,
+        container_id: &str,
+        namespace: &str,
     ) -> Result<Self> {
         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<Url>| -> Option<Box<dyn AsyncWrite + Send + Unpin>> {
             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,
         })
     }
 }

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<File> {
     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<RwLock<HashMap<String, Container>>>,
         agent: Arc<dyn Agent>,
         mut wg: WaitGroup,
+        binary_logger: Option<BinaryLogger>,
     ) -> 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"));
+    }
+}