feat: add --replay mode for JSONL event replay without eBPF

CHANGELOG.md

@@ -6,6 +6,7 @@ possible include a PR number for easier tracking.
 
 ## Next
 
+* feat: add --replay mode for JSONL event replay without eBPF (#1010)
 * feat(output): add basic opentelemetry output (#971)
 * feat(grpc): add exponential backoff for reconnection attempts (#789)
 * feat: run integration tests on more platforms (#760)

fact-ebpf/src/lib.rs

@@ -4,7 +4,11 @@ use std::{error::Error, ffi::c_char, fmt::Display, hash::Hash, path::PathBuf};
 
 use aya::{maps::lpm_trie, Pod};
 use libc::memcpy;
-use serde::{ser::SerializeStruct, Serialize};
+use serde::{
+    de::{self, MapAccess, Visitor},
+    ser::SerializeStruct,
+    Deserialize, Serialize,
+};
 
 include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
 
@@ -101,6 +105,51 @@ impl Serialize for inode_key_t {
     }
 }
 
+impl<'de> Deserialize<'de> for inode_key_t {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: de::Deserializer<'de>,
+    {
+        struct InodeKeyVisitor;
+
+        impl<'de> Visitor<'de> for InodeKeyVisitor {
+            type Value = inode_key_t;
+
+            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+                f.write_str("a struct with inode and dev fields")
+            }
+
+            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
+            where
+                A: MapAccess<'de>,
+            {
+                let mut inode = None;
+                let mut dev = None;
+                while let Some(key) = map.next_key::<String>()? {
+                    match key.as_str() {
+                        "inode" => inode = Some(map.next_value()?),
+                        "dev" => dev = Some(map.next_value()?),
+                        _ => {
+                            map.next_value::<de::IgnoredAny>()?;
+                        }
+                    }
+                }
+
+                let Some(inode) = inode else {
+                    return Err(de::Error::missing_field("inode"));
+                };
+                let Some(dev) = dev else {
+                    return Err(de::Error::missing_field("dev"));
+                };
+
+                Ok(inode_key_t { inode, dev })
+            }
+        }
+
+        deserializer.deserialize_struct("inode_key_t", &["inode", "dev"], InodeKeyVisitor)
+    }
+}
+
 unsafe impl Pod for inode_key_t {}
 
 impl Default for monitored_t {
@@ -124,6 +173,25 @@ impl Serialize for monitored_t {
     }
 }
 
+impl<'de> Deserialize<'de> for monitored_t {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: de::Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        match s.as_str() {
+            "not monitored" => Ok(monitored_t::NOT_MONITORED),
+            "by inode" => Ok(monitored_t::MONITORED_BY_INODE),
+            "by path" => Ok(monitored_t::MONITORED_BY_PATH),
+            "by parent" => Ok(monitored_t::MONITORED_BY_PARENT),
+            _ => Err(de::Error::unknown_variant(
+                &s,
+                &["not monitored", "by inode", "by path", "by parent"],
+            )),
+        }
+    }
+}
+
 impl metrics_by_hook_t {
     fn accumulate(mut self, other: &metrics_by_hook_t) -> metrics_by_hook_t {
         self.total += other.total;

fact/src/bpf/mod.rs

@@ -13,7 +13,7 @@ use log::{error, info, warn};
 use tokio::{
     io::unix::AsyncFd,
     sync::{mpsc, watch},
-    task::JoinHandle,
+    task::JoinSet,
 };
 
 use crate::{config::BpfConfig, event::Event, host_info, metrics::EventCounter};
@@ -225,12 +225,13 @@ impl Bpf {
     // Gather events from the ring buffer and print them out.
     pub fn start(
         mut self,
+        task_set: &mut JoinSet<anyhow::Result<()>>,
         mut running: watch::Receiver<bool>,
         event_counter: EventCounter,
-    ) -> JoinHandle<anyhow::Result<()>> {
+    ) {
         info!("Starting BPF worker...");
 
-        tokio::spawn(async move {
+        task_set.spawn(async move {
             let rb = self.take_ringbuffer()?;
             let mut fd = AsyncFd::new(rb)?;
 
@@ -283,7 +284,7 @@ impl Bpf {
             }
 
             Ok(())
-        })
+        });
     }
 }
 
@@ -322,9 +323,10 @@ mod bpf_tests {
             Bpf::new(reloader.paths(), &reloader.config().bpf).expect("Failed to load BPF code");
         let (run_tx, run_rx) = watch::channel(true);
         // Create a metrics exporter, but don't start it
-        let exporter = Exporter::new(bpf.take_metrics().unwrap());
+        let exporter = Exporter::new(Some(bpf.take_metrics().unwrap()));
+        let mut task_set = JoinSet::new();
 
-        let handle = bpf.start(run_rx, exporter.metrics.bpf_worker.clone());
+        bpf.start(&mut task_set, run_rx, exporter.metrics.bpf_worker.clone());
 
         tokio::time::sleep(Duration::from_millis(500)).await;
 
@@ -412,7 +414,7 @@ mod bpf_tests {
 
         tokio::select! {
             res = wait => res.unwrap(),
-            res = handle => res.unwrap().unwrap(),
+            res = task_set.join_next() => res.unwrap().unwrap().unwrap(),
         }
 
         run_tx.send(false).unwrap();

fact/src/config/mod.rs

@@ -42,6 +42,7 @@ pub struct FactConfig {
     hotreload: Option<bool>,
     scan_interval: Option<Duration>,
     rate_limit: Option<u64>,
+    replay: Option<PathBuf>,
 }
 
 impl FactConfig {
@@ -108,6 +109,10 @@ impl FactConfig {
         if let Some(rate_limit) = from.rate_limit {
             self.rate_limit = Some(rate_limit);
         }
+
+        if let Some(replay) = from.replay.as_deref() {
+            self.replay = Some(replay.to_path_buf());
+        }
     }
 
     pub fn paths(&self) -> &[PathBuf] {
@@ -134,6 +139,10 @@ impl FactConfig {
         self.rate_limit.unwrap_or(0)
     }
 
+    pub fn replay(&self) -> Option<&Path> {
+        self.replay.as_deref()
+    }
+
     #[cfg(test)]
     pub fn set_paths(&mut self, paths: Vec<PathBuf>) {
         self.paths = Some(paths);
@@ -247,6 +256,12 @@ impl TryFrom<Vec<Yaml>> for FactConfig {
                     }
                     config.rate_limit = Some(rate_limit as u64);
                 }
+                "replay" => {
+                    let Some(replay) = v.as_str() else {
+                        bail!("replay field has incorrect type: {v:?}")
+                    };
+                    config.replay = Some(PathBuf::from(replay));
+                }
                 name => bail!("Invalid field '{name}' with value: {v:?}"),
             }
         }
@@ -760,6 +775,13 @@ pub struct FactCli {
     /// Default value is 0 (unlimited)
     #[arg(long, short = 'l', env = "FACT_RATE_LIMIT")]
     rate_limit: Option<u64>,
+
+    /// Replay events from a JSONL file instead of loading eBPF programs.
+    ///
+    /// This mode skips BPF loading and HostScanner, reading pre-recorded
+    /// events for profiling purposes (e.g. valgrind, DHAT).
+    #[arg(long, env = "FACT_REPLAY")]
+    replay: Option<PathBuf>,
 }
 
 impl FactCli {
@@ -794,6 +816,7 @@ impl FactCli {
             hotreload: resolve_bool_arg(self.hotreload, self.no_hotreload),
             scan_interval: self.scan_interval,
             rate_limit: self.rate_limit,
+            replay: self.replay.clone(),
         }
     }
 }

fact/src/config/reloader.rs

@@ -5,7 +5,6 @@ use std::{
 use log::{debug, info, warn};
 use tokio::{
     sync::{Notify, watch},
-    task::JoinHandle,
     time::interval,
 };
 
@@ -34,13 +33,13 @@ impl Reloader {
     ///
     /// If hotreload is disabled on startup the task will not be
     /// spawned.
-    pub fn start(mut self, mut running: watch::Receiver<bool>) -> Option<JoinHandle<()>> {
+    pub fn start(mut self, mut running: watch::Receiver<bool>) {
         if !self.config.hotreload() {
             info!("Configuration hotreload is disabled, changes will require a restart.");
-            return None;
+            return;
         }
 
-        let handle = tokio::spawn(async move {
+        tokio::spawn(async move {
             let mut ticker = interval(Duration::from_secs(10));
             loop {
                 tokio::select! {
@@ -55,7 +54,6 @@ impl Reloader {
                 }
             }
         });
-        Some(handle)
     }
 
     pub fn config(&self) -> &FactConfig {

fact/src/config/tests.rs

@@ -393,6 +393,13 @@ fn parsing() {
                 ..Default::default()
             },
         ),
+        (
+            "replay: /some/path",
+            FactConfig {
+                replay: Some(PathBuf::from("/some/path")),
+                ..Default::default()
+            },
+        ),
         (
             r#"
             paths:
@@ -420,6 +427,7 @@ fn parsing() {
             hotreload: false
             scan_interval: 60
             rate_limit: 50000
+            replay: /some/path.jsonl
             "#,
             FactConfig {
                 paths: Some(vec![PathBuf::from("/etc")]),
@@ -451,6 +459,7 @@ fn parsing() {
                 hotreload: Some(false),
                 scan_interval: Some(Duration::from_secs(60)),
                 rate_limit: Some(50000),
+                replay: Some(PathBuf::from("/some/path.jsonl")),
             },
         ),
     ];
@@ -790,6 +799,10 @@ paths:
             "rate_limit field has incorrect type: Real(\"1000.0\")",
         ),
         ("rate_limit: -1000", "invalid rate_limit: -1000"),
+        (
+            "replay: true",
+            "replay field has incorrect type: Boolean(true)",
+        ),
         ("unknown:", "Invalid field 'unknown' with value: Null"),
     ];
     for (input, expected) in tests {
@@ -1600,6 +1613,25 @@ fn update() {
                 ..Default::default()
             },
         ),
+        (
+            "replay: /some/path.jsonl",
+            FactConfig::default(),
+            FactConfig {
+                replay: Some("/some/path.jsonl".into()),
+                ..Default::default()
+            },
+        ),
+        (
+            "replay: /some/path.jsonl",
+            FactConfig {
+                replay: Some("/initial/path.jsonl".into()),
+                ..Default::default()
+            },
+            FactConfig {
+                replay: Some("/some/path.jsonl".into()),
+                ..Default::default()
+            },
+        ),
         (
             r#"
             paths:
@@ -1658,6 +1690,7 @@ fn update() {
                 hotreload: Some(true),
                 scan_interval: Some(Duration::from_secs(30)),
                 rate_limit: Some(5000),
+                replay: None,
             },
             FactConfig {
                 paths: Some(vec![PathBuf::from("/etc")]),
@@ -1689,6 +1722,7 @@ fn update() {
                 hotreload: Some(false),
                 scan_interval: Some(Duration::from_secs(60)),
                 rate_limit: Some(1000),
+                replay: None,
             },
         ),
     ];
@@ -1728,6 +1762,7 @@ fn defaults() {
     assert_eq!(config.otel.endpoint(), None);
     assert_eq!(config.scan_interval(), Duration::from_secs(30));
     assert_eq!(config.rate_limit(), 0);
+    assert!(config.replay().is_none());
 }
 
 static ENV_MUTEX: Mutex<()> = Mutex::new(());