aya-log: Remove tokio dep

Require the caller to provide their own executor.

aya-log/CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - The implementation is now backed by a ring buffer rather than a perf event array. This should
   improve performance but increases the minimum supported kernel version to 5.8.
 
+- Drop the built-in `tokio` dependency. Users must now BYOR (bring your own runtime).
+
 ## v0.2.1 (2024-10-09)
 
 ### Chore

aya-log/Cargo.toml

@@ -21,11 +21,11 @@ aya = { path = "../aya", version = "^0.13.1", default-features = false }
 aya-log-common = { path = "../aya-log-common", version = "^0.1.15", default-features = false }
 log = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, features = ["net", "rt"] }
 
 [dev-dependencies]
 env_logger = { workspace = true }
 testing_logger = { workspace = true }
+tokio = { workspace = true, features = ["net", "rt"] }
 
 [lib]
 path = "src/lib.rs"

aya-log/README.md

@@ -38,7 +38,15 @@ use aya_log::EbpfLogger;
 env_logger::init();
 
 // Will log using the default logger, which is TermLogger in this case
-EbpfLogger::init(&mut bpf).unwrap();
+let logger = EbpfLogger::init(&mut bpf).unwrap();
+let mut logger = tokio::io::unix::AsyncFd::with_interest(logger, tokio::io::Interest::READABLE).unwrap();
+tokio::task::spawn(async move { 
+    loop {
+        let mut guard = logger.readable_mut().await.unwrap();
+        guard.get_inner_mut().flush();
+        guard.clear_ready();
+    }
+});
 ```
 
 ### eBPF code

aya-log/src/lib.rs

@@ -3,8 +3,8 @@
 //! This is the user space side of the [Aya] logging framework. For the eBPF
 //! side, see the `aya-log-ebpf` crate.
 //!
-//! `aya-log` provides the [EbpfLogger] type, which reads log records created by
-//! `aya-log-ebpf` and logs them using the [log] crate. Any logger that
+//! `aya-log` provides functions which read log records created by
+//! `aya-log-ebpf` and log them using the [log] crate. Any logger that
 //! implements the [Log] trait can be used with this crate.
 //!
 //! # Example:
@@ -19,7 +19,15 @@
 //! env_logger::init();
 //!
 //! // start reading aya-log records and log them using the default logger
-//! EbpfLogger::init(&mut bpf).unwrap();
+//! let logger = EbpfLogger::init(&mut bpf).unwrap();
+//! let mut logger = tokio::io::unix::AsyncFd::with_interest(logger, tokio::io::Interest::READABLE).unwrap();
+//! tokio::task::spawn(async move {
+//!     loop {
+//!         let mut guard = logger.readable_mut().await.unwrap();
+//!         guard.get_inner_mut().flush();
+//!         guard.clear_ready();
+//!     }
+//! });
 //! ```
 //!
 //! With the following eBPF code:
@@ -51,8 +59,9 @@
 //!
 use std::{
     fmt::{LowerHex, UpperHex},
-    io, mem,
+    mem,
     net::{Ipv4Addr, Ipv6Addr},
+    os::fd::AsRawFd,
     ptr, str,
 };
 
@@ -84,45 +93,54 @@ unsafe impl Pod for DisplayHintWrapper {}
 /// Log messages generated by `aya_log_ebpf` using the [log] crate.
 ///
 /// For more details see the [module level documentation](crate).
-pub struct EbpfLogger;
+pub struct EbpfLogger<T> {
+    ring_buf: RingBuf<MapData>,
+    logger: T,
+}
+
+impl<T> AsRawFd for EbpfLogger<T> {
+    fn as_raw_fd(&self) -> std::os::unix::prelude::RawFd {
+        let Self {
+            ring_buf,
+            logger: _,
+        } = self;
+        ring_buf.as_raw_fd()
+    }
+}
 
 /// Log messages generated by `aya_log_ebpf` using the [log] crate.
 #[deprecated(since = "0.2.1", note = "Use `aya_log::EbpfLogger` instead")]
-pub type BpfLogger = EbpfLogger;
+pub type BpfLogger<T> = EbpfLogger<T>;
 
-impl EbpfLogger {
+impl EbpfLogger<&'static dyn Log> {
     /// Starts reading log records created with `aya-log-ebpf` and logs them
     /// with the default logger. See [log::logger].
-    pub fn init(bpf: &mut Ebpf) -> Result<EbpfLogger, Error> {
-        EbpfLogger::init_with_logger(bpf, log::logger())
-    }
-
-    /// Starts reading log records created with `aya-log-ebpf` and logs them
-    /// with the given logger.
-    pub fn init_with_logger<T: Log + 'static>(
-        bpf: &mut Ebpf,
-        logger: T,
-    ) -> Result<EbpfLogger, Error> {
-        let map = bpf.take_map(MAP_NAME).ok_or(Error::MapNotFound)?;
-        Self::read_logs_async(map, logger)
+    pub fn init(bpf: &mut Ebpf) -> Result<Self, Error> {
+        Self::init_with_logger(bpf, log::logger())
     }
 
     /// Attaches to an existing `aya-log-ebpf` instance.
     ///
     /// Attaches to the logs produced by `program_id`. Can be used to read logs generated by a
     /// pinned program. The log records will be written to the default logger. See [log::logger].
-    pub fn init_from_id(program_id: u32) -> Result<EbpfLogger, Error> {
+    pub fn init_from_id(program_id: u32) -> Result<Self, Error> {
         Self::init_from_id_with_logger(program_id, log::logger())
     }
+}
+
+impl<T: Log> EbpfLogger<T> {
+    /// Starts reading log records created with `aya-log-ebpf` and logs them
+    /// with the given logger.
+    pub fn init_with_logger(bpf: &mut Ebpf, logger: T) -> Result<Self, Error> {
+        let map = bpf.take_map(MAP_NAME).ok_or(Error::MapNotFound)?;
+        Self::new(map, logger)
+    }
 
     /// Attaches to an existing `aya-log-ebpf` instance and logs with the given logger.
     ///
     /// Attaches to the logs produced by `program_id`. Can be used to read logs generated by a
     /// pinned program. The log records will be written to the given logger.
-    pub fn init_from_id_with_logger<T: Log + 'static>(
-        program_id: u32,
-        logger: T,
-    ) -> Result<EbpfLogger, Error> {
+    pub fn init_from_id_with_logger(program_id: u32, logger: T) -> Result<Self, Error> {
         let program_info = loaded_programs()
             .filter_map(|info| info.ok())
             .find(|info| info.id() == program_id)
@@ -140,26 +158,21 @@ impl EbpfLogger {
             .ok_or(Error::MapNotFound)?;
         let map = MapData::from_id(map.id())?;
 
-        Self::read_logs_async(Map::RingBuf(map), logger)
+        Self::new(Map::RingBuf(map), logger)
     }
 
-    fn read_logs_async<T: Log + 'static>(map: Map, logger: T) -> Result<Self, Error> {
+    fn new(map: Map, logger: T) -> Result<Self, Error> {
         let ring_buf: RingBuf<_> = map.try_into()?;
-        let mut async_fd =
-            tokio::io::unix::AsyncFd::with_interest(ring_buf, tokio::io::Interest::READABLE)
-                .map_err(Error::AsyncFdNew)?;
-
-        tokio::spawn(async move {
-            loop {
-                let mut guard = async_fd.readable_mut().await.unwrap();
-                while let Some(buf) = guard.get_inner_mut().next() {
-                    log_buf(buf.as_ref(), &logger).unwrap();
-                }
-                guard.clear_ready();
-            }
-        });
 
-        Ok(EbpfLogger {})
+        Ok(EbpfLogger { ring_buf, logger })
+    }
+
+    /// Reads log records from eBPF and writes them to the logger.
+    pub fn flush(&mut self) {
+        let Self { ring_buf, logger } = self;
+        while let Some(buf) = ring_buf.next() {
+            log_buf(buf.as_ref(), logger).unwrap();
+        }
     }
 }
 
@@ -429,9 +442,6 @@ pub enum Error {
     #[error(transparent)]
     MapError(#[from] MapError),
 
-    #[error("tokio::io::unix::AsyncFd::new")]
-    AsyncFdNew(#[source] io::Error),
-
     #[error("program not found")]
     ProgramNotFound,
 

test/integration-test/Cargo.toml

@@ -33,7 +33,12 @@ rbpf = { workspace = true }
 scopeguard = { workspace = true }
 test-case = { workspace = true }
 test-log = { workspace = true, features = ["log"] }
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
+tokio = { workspace = true, features = [
+    "macros",
+    "net",
+    "rt-multi-thread",
+    "time",
+] }
 xdpilone = { workspace = true }
 
 [build-dependencies]

test/integration-test/src/tests/log.rs

@@ -1,7 +1,4 @@
-use std::{
-    borrow::Cow,
-    sync::{Arc, Mutex},
-};
+use std::{borrow::Cow, sync::Mutex};
 
 use aya::{Ebpf, programs::UProbe};
 use aya_log::EbpfLogger;
@@ -15,10 +12,10 @@ pub extern "C" fn trigger_ebpf_program() {
 }
 
 struct TestingLogger<F> {
-    log: F,
+    log: Mutex<F>,
 }
 
-impl<F: Send + Sync + Fn(&Record)> Log for TestingLogger<F> {
+impl<F: Send + FnMut(&Record)> Log for TestingLogger<F> {
     fn enabled(&self, _metadata: &log::Metadata) -> bool {
         true
     }
@@ -27,6 +24,7 @@ impl<F: Send + Sync + Fn(&Record)> Log for TestingLogger<F> {
 
     fn log(&self, record: &Record) {
         let Self { log } = self;
+        let mut log = log.lock().unwrap();
         log(record);
     }
 }
@@ -38,28 +36,21 @@ struct CapturedLog<'a> {
     pub target: Cow<'a, str>,
 }
 
-#[test(tokio::test)]
-async fn log() {
+#[test]
+fn log() {
     let mut bpf = Ebpf::load(crate::LOG).unwrap();
 
-    let captured_logs = Arc::new(Mutex::new(Vec::new()));
-    {
-        let captured_logs = captured_logs.clone();
-        EbpfLogger::init_with_logger(
-            &mut bpf,
-            TestingLogger {
-                log: move |record: &Record| {
-                    let mut logs = captured_logs.lock().unwrap();
-                    logs.push(CapturedLog {
-                        body: format!("{}", record.args()).into(),
-                        level: record.level(),
-                        target: record.target().to_string().into(),
-                    });
-                },
-            },
-        )
-        .unwrap();
-    }
+    let mut captured_logs = Vec::new();
+    let logger = TestingLogger {
+        log: Mutex::new(|record: &Record| {
+            captured_logs.push(CapturedLog {
+                body: format!("{}", record.args()).into(),
+                level: record.level(),
+                target: record.target().to_string().into(),
+            });
+        }),
+    };
+    let mut logger = EbpfLogger::init_with_logger(&mut bpf, &logger).unwrap();
 
     let prog: &mut UProbe = bpf.program_mut("test_log").unwrap().try_into().unwrap();
     prog.load().unwrap();
@@ -69,21 +60,9 @@ async fn log() {
     // Call the function that the uprobe is attached to, so it starts logging.
     trigger_ebpf_program();
 
-    let mut logs = 0;
-    let records = loop {
-        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-        let records = captured_logs.lock().unwrap();
-        let len = records.len();
-        if len == 0 {
-            continue;
-        }
-        if len == logs {
-            break records;
-        }
-        logs = len;
-    };
+    logger.flush();
 
-    let mut records = records.iter();
+    let mut records = captured_logs.iter();
 
     assert_eq!(
         records.next(),

xtask/public-api/aya-log.txt

@@ -1,6 +1,5 @@
 pub mod aya_log
 pub enum aya_log::Error
-pub aya_log::Error::AsyncFdNew(std::io::error::Error)
 pub aya_log::Error::MapError(aya::maps::MapError)
 pub aya_log::Error::MapNotFound
 pub aya_log::Error::ProgramError(aya::programs::ProgramError)
@@ -64,34 +63,38 @@ impl<T> core::borrow::BorrowMut<T> for aya_log::DefaultFormatter where T: ?core:
 pub fn aya_log::DefaultFormatter::borrow_mut(&mut self) -> &mut T
 impl<T> core::convert::From<T> for aya_log::DefaultFormatter
 pub fn aya_log::DefaultFormatter::from(t: T) -> T
-pub struct aya_log::EbpfLogger
-impl aya_log::EbpfLogger
-pub fn aya_log::EbpfLogger::init(bpf: &mut aya::bpf::Ebpf) -> core::result::Result<aya_log::EbpfLogger, aya_log::Error>
-pub fn aya_log::EbpfLogger::init_from_id(program_id: u32) -> core::result::Result<aya_log::EbpfLogger, aya_log::Error>
-pub fn aya_log::EbpfLogger::init_from_id_with_logger<T: log::Log + 'static>(program_id: u32, logger: T) -> core::result::Result<aya_log::EbpfLogger, aya_log::Error>
-pub fn aya_log::EbpfLogger::init_with_logger<T: log::Log + 'static>(bpf: &mut aya::bpf::Ebpf, logger: T) -> core::result::Result<aya_log::EbpfLogger, aya_log::Error>
-impl core::marker::Freeze for aya_log::EbpfLogger
-impl core::marker::Send for aya_log::EbpfLogger
-impl core::marker::Sync for aya_log::EbpfLogger
-impl core::marker::Unpin for aya_log::EbpfLogger
-impl core::panic::unwind_safe::RefUnwindSafe for aya_log::EbpfLogger
-impl core::panic::unwind_safe::UnwindSafe for aya_log::EbpfLogger
-impl<T, U> core::convert::Into<U> for aya_log::EbpfLogger where U: core::convert::From<T>
-pub fn aya_log::EbpfLogger::into(self) -> U
-impl<T, U> core::convert::TryFrom<U> for aya_log::EbpfLogger where U: core::convert::Into<T>
-pub type aya_log::EbpfLogger::Error = core::convert::Infallible
-pub fn aya_log::EbpfLogger::try_from(value: U) -> core::result::Result<T, <T as core::convert::TryFrom<U>>::Error>
-impl<T, U> core::convert::TryInto<U> for aya_log::EbpfLogger where U: core::convert::TryFrom<T>
-pub type aya_log::EbpfLogger::Error = <U as core::convert::TryFrom<T>>::Error
-pub fn aya_log::EbpfLogger::try_into(self) -> core::result::Result<U, <U as core::convert::TryFrom<T>>::Error>
-impl<T> core::any::Any for aya_log::EbpfLogger where T: 'static + ?core::marker::Sized
-pub fn aya_log::EbpfLogger::type_id(&self) -> core::any::TypeId
-impl<T> core::borrow::Borrow<T> for aya_log::EbpfLogger where T: ?core::marker::Sized
-pub fn aya_log::EbpfLogger::borrow(&self) -> &T
-impl<T> core::borrow::BorrowMut<T> for aya_log::EbpfLogger where T: ?core::marker::Sized
-pub fn aya_log::EbpfLogger::borrow_mut(&mut self) -> &mut T
-impl<T> core::convert::From<T> for aya_log::EbpfLogger
-pub fn aya_log::EbpfLogger::from(t: T) -> T
+pub struct aya_log::EbpfLogger<T>
+impl aya_log::EbpfLogger<&'static dyn log::Log>
+pub fn aya_log::EbpfLogger<&'static dyn log::Log>::init(bpf: &mut aya::bpf::Ebpf) -> core::result::Result<Self, aya_log::Error>
+pub fn aya_log::EbpfLogger<&'static dyn log::Log>::init_from_id(program_id: u32) -> core::result::Result<Self, aya_log::Error>
+impl<T: log::Log> aya_log::EbpfLogger<T>
+pub fn aya_log::EbpfLogger<T>::flush(&mut self)
+pub fn aya_log::EbpfLogger<T>::init_from_id_with_logger(program_id: u32, logger: T) -> core::result::Result<Self, aya_log::Error>
+pub fn aya_log::EbpfLogger<T>::init_with_logger(bpf: &mut aya::bpf::Ebpf, logger: T) -> core::result::Result<Self, aya_log::Error>
+impl<T> std::os::fd::raw::AsRawFd for aya_log::EbpfLogger<T>
+pub fn aya_log::EbpfLogger<T>::as_raw_fd(&self) -> std::os::fd::raw::RawFd
+impl<T> core::marker::Freeze for aya_log::EbpfLogger<T> where T: core::marker::Freeze
+impl<T> core::marker::Send for aya_log::EbpfLogger<T> where T: core::marker::Send
+impl<T> core::marker::Sync for aya_log::EbpfLogger<T> where T: core::marker::Sync
+impl<T> core::marker::Unpin for aya_log::EbpfLogger<T> where T: core::marker::Unpin
+impl<T> core::panic::unwind_safe::RefUnwindSafe for aya_log::EbpfLogger<T> where T: core::panic::unwind_safe::RefUnwindSafe
+impl<T> core::panic::unwind_safe::UnwindSafe for aya_log::EbpfLogger<T> where T: core::panic::unwind_safe::UnwindSafe
+impl<T, U> core::convert::Into<U> for aya_log::EbpfLogger<T> where U: core::convert::From<T>
+pub fn aya_log::EbpfLogger<T>::into(self) -> U
+impl<T, U> core::convert::TryFrom<U> for aya_log::EbpfLogger<T> where U: core::convert::Into<T>
+pub type aya_log::EbpfLogger<T>::Error = core::convert::Infallible
+pub fn aya_log::EbpfLogger<T>::try_from(value: U) -> core::result::Result<T, <T as core::convert::TryFrom<U>>::Error>
+impl<T, U> core::convert::TryInto<U> for aya_log::EbpfLogger<T> where U: core::convert::TryFrom<T>
+pub type aya_log::EbpfLogger<T>::Error = <U as core::convert::TryFrom<T>>::Error
+pub fn aya_log::EbpfLogger<T>::try_into(self) -> core::result::Result<U, <U as core::convert::TryFrom<T>>::Error>
+impl<T> core::any::Any for aya_log::EbpfLogger<T> where T: 'static + ?core::marker::Sized
+pub fn aya_log::EbpfLogger<T>::type_id(&self) -> core::any::TypeId
+impl<T> core::borrow::Borrow<T> for aya_log::EbpfLogger<T> where T: ?core::marker::Sized
+pub fn aya_log::EbpfLogger<T>::borrow(&self) -> &T
+impl<T> core::borrow::BorrowMut<T> for aya_log::EbpfLogger<T> where T: ?core::marker::Sized
+pub fn aya_log::EbpfLogger<T>::borrow_mut(&mut self) -> &mut T
+impl<T> core::convert::From<T> for aya_log::EbpfLogger<T>
+pub fn aya_log::EbpfLogger<T>::from(t: T) -> T
 pub struct aya_log::Ipv4Formatter
 impl<T> aya_log::Formatter<T> for aya_log::Ipv4Formatter where T: core::convert::Into<core::net::ip_addr::Ipv4Addr>
 pub fn aya_log::Ipv4Formatter::format(v: T) -> alloc::string::String
@@ -312,4 +315,4 @@ impl<T> aya_log::Formatter<T> for aya_log::LowerHexFormatter where T: core::fmt:
 pub fn aya_log::LowerHexFormatter::format(v: T) -> alloc::string::String
 impl<T> aya_log::Formatter<T> for aya_log::UpperHexFormatter where T: core::fmt::UpperHex
 pub fn aya_log::UpperHexFormatter::format(v: T) -> alloc::string::String
-pub type aya_log::BpfLogger = aya_log::EbpfLogger
+pub type aya_log::BpfLogger<T> = aya_log::EbpfLogger<T>

xtask/public-api/aya.txt

@@ -608,6 +608,8 @@ pub fn aya::maps::ring_buf::RingBuf<&'a aya::maps::MapData>::try_from(map: &'a a
 impl<'a> core::convert::TryFrom<&'a mut aya::maps::Map> for aya::maps::ring_buf::RingBuf<&'a mut aya::maps::MapData>
 pub type aya::maps::ring_buf::RingBuf<&'a mut aya::maps::MapData>::Error = aya::maps::MapError
 pub fn aya::maps::ring_buf::RingBuf<&'a mut aya::maps::MapData>::try_from(map: &'a mut aya::maps::Map) -> core::result::Result<Self, Self::Error>
+impl<T: core::borrow::Borrow<aya::maps::MapData>> std::os::fd::owned::AsFd for aya::maps::ring_buf::RingBuf<T>
+pub fn aya::maps::ring_buf::RingBuf<T>::as_fd(&self) -> std::os::fd::owned::BorrowedFd<'_>
 impl<T: core::borrow::Borrow<aya::maps::MapData>> std::os::fd::raw::AsRawFd for aya::maps::ring_buf::RingBuf<T>
 pub fn aya::maps::ring_buf::RingBuf<T>::as_raw_fd(&self) -> std::os::fd::raw::RawFd
 impl<T> core::marker::Freeze for aya::maps::ring_buf::RingBuf<T> where T: core::marker::Freeze
@@ -2220,6 +2222,8 @@ pub fn aya::maps::ring_buf::RingBuf<&'a aya::maps::MapData>::try_from(map: &'a a
 impl<'a> core::convert::TryFrom<&'a mut aya::maps::Map> for aya::maps::ring_buf::RingBuf<&'a mut aya::maps::MapData>
 pub type aya::maps::ring_buf::RingBuf<&'a mut aya::maps::MapData>::Error = aya::maps::MapError
 pub fn aya::maps::ring_buf::RingBuf<&'a mut aya::maps::MapData>::try_from(map: &'a mut aya::maps::Map) -> core::result::Result<Self, Self::Error>
+impl<T: core::borrow::Borrow<aya::maps::MapData>> std::os::fd::owned::AsFd for aya::maps::ring_buf::RingBuf<T>
+pub fn aya::maps::ring_buf::RingBuf<T>::as_fd(&self) -> std::os::fd::owned::BorrowedFd<'_>
 impl<T: core::borrow::Borrow<aya::maps::MapData>> std::os::fd::raw::AsRawFd for aya::maps::ring_buf::RingBuf<T>
 pub fn aya::maps::ring_buf::RingBuf<T>::as_raw_fd(&self) -> std::os::fd::raw::RawFd
 impl<T> core::marker::Freeze for aya::maps::ring_buf::RingBuf<T> where T: core::marker::Freeze