Introduce BPF Streams abstraction

Add an abstraction for reading BPF Streams. This commit
introduces the wrappers around the libbpf-sys method.

Signed-off-by: Emil Tsalapatis <[email protected]>

Author: etsal

libbpf-rs/src/lib.rs

@@ -80,6 +80,7 @@ mod program;
 pub mod query;
 mod ringbuf;
 mod skeleton;
+mod streams;
 mod tc;
 mod tracepoint;
 mod user_ringbuf;

libbpf-rs/src/program.rs

@@ -6,6 +6,7 @@ use std::ffi::c_void;
 use std::ffi::CStr;
 use std::ffi::CString;
 use std::ffi::OsStr;
+use std::io::Read;
 use std::marker::PhantomData;
 use std::mem;
 use std::mem::size_of;
@@ -27,6 +28,7 @@ use std::time::Duration;
 use libbpf_sys::bpf_func_id;
 
 use crate::netfilter;
+use crate::streams::Stream;
 use crate::util;
 use crate::util::validate_bpf_ret;
 use crate::util::BpfObjectType;
@@ -178,7 +180,6 @@ impl From<PerfEventOpts> for libbpf_sys::bpf_perf_event_opts {
     }
 }
 
-
 /// Options used when iterating over a map.
 #[derive(Clone, Debug)]
 pub struct MapIterOpts<'fd> {
@@ -198,7 +199,6 @@ impl<'fd> MapIterOpts<'fd> {
     }
 }
 
-
 /// Iteration order for cgroups.
 #[non_exhaustive]
 #[repr(u32)]
@@ -239,7 +239,6 @@ impl<'fd> CgroupIterOpts<'fd> {
     }
 }
 
-
 /// Options to optionally be provided when attaching to an iterator.
 #[non_exhaustive]
 #[derive(Clone, Debug)]
@@ -277,7 +276,6 @@ impl From<IterOpts<'_>> for libbpf_sys::bpf_iter_link_info {
     }
 }
 
-
 /// An immutable parsed but not yet loaded BPF program.
 pub type OpenProgram<'obj> = OpenProgramImpl<'obj>;
 /// A mutable parsed but not yet loaded BPF program.
@@ -760,7 +758,6 @@ pub type Program<'obj> = ProgramImpl<'obj>;
 /// A mutable loaded BPF program.
 pub type ProgramMut<'obj> = ProgramImpl<'obj, Mut>;
 
-
 /// Represents a loaded [`Program`].
 ///
 /// This struct is not safe to clone because the underlying libbpf resource cannot currently
@@ -1656,6 +1653,16 @@ impl<'obj> ProgramMut<'obj> {
         };
         Ok(output)
     }
+
+    /// Get the stdout BPF stream of the program.
+    pub fn stdout(&self) -> impl Read + '_ {
+        Stream::new(self.as_fd(), Stream::BPF_STDOUT)
+    }
+
+    /// Get the stderr BPF stream of the program.
+    pub fn stderr(&self) -> impl Read + '_ {
+        Stream::new(self.as_fd(), Stream::BPF_STDERR)
+    }
 }
 
 impl<'obj> Deref for ProgramMut<'obj> {

libbpf-rs/src/streams.rs

@@ -0,0 +1,52 @@
+use core::ffi::{c_ulong, c_void};
+use std::io;
+use std::io::Read;
+use std::mem;
+use std::os::fd::{self, AsRawFd};
+
+use crate::libbpf_sys::bpf_prog_stream_read_opts;
+
+/// [`Stream`] instances.
+///
+/// `streams`s are BPF descriptors that provide a character output
+/// interface from the BPF program to userspace.
+#[derive(Debug)]
+pub(crate) struct Stream<'a> {
+    prog_fd: fd::BorrowedFd<'a>,
+    stream_id: u32,
+}
+
+impl<'a> Stream<'a> {
+    /// Default BPF stdout stream id.
+    pub(crate) const BPF_STDOUT: u32 = 1;
+    /// Default BPF stderr stream id.
+    pub(crate) const BPF_STDERR: u32 = 2;
+
+    /// Create a new Stream instance.
+    pub(crate) fn new(prog_fd: fd::BorrowedFd<'a>, stream_id: u32) -> Stream<'a> {
+        Stream { prog_fd, stream_id }
+    }
+}
+
+impl Read for Stream<'_> {
+    /// Fill in a caller-provided buffer with contents from the stream.
+    fn read(&mut self, buf: &mut [u8]) -> std::result::Result<usize, std::io::Error> {
+        let ret = unsafe {
+            let mut c_opts = bpf_prog_stream_read_opts {
+                sz: mem::size_of::<bpf_prog_stream_read_opts>() as c_ulong,
+            };
+            libbpf_sys::bpf_prog_stream_read(
+                self.prog_fd.as_raw_fd(),
+                self.stream_id,
+                buf.as_mut_ptr() as *mut c_void,
+                buf.len().try_into().unwrap(),
+                &mut c_opts as *mut bpf_prog_stream_read_opts,
+            )
+        };
+        if ret < 0 {
+            return Err(io::Error::last_os_error());
+        }
+
+        Ok(ret as usize)
+    }
+}