runtime-rs: support containerd binary loggers

Support containerd binary:// stdio URIs in runtime-rs by spawning the
logger with stdout, stderr, and readiness on file descriptors 3, 4, and 5.
Preserve URI arguments and container metadata, and stream output through
nonblocking pipes with backpressure.

Processes using a binary logger fall back from passfd to the existing agent
streams so the host shim can feed the logger. After EOF, allow the logger to
drain before using SIGTERM and SIGKILL as bounded fallbacks.

Fixes: #4391
Related: #4420
Signed-off-by: cayoub-oai <276123840+cayoub-oai@users.noreply.github.com>
Assisted-By: OpenAI Codex

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"));
+    }
+}