From 68ec0c48770553c40b69656bf8e2786c25b3cffa Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Tue, 7 Jul 2026 16:02:19 +0200 Subject: [PATCH 1/2] feat: add --replay mode for JSONL event replay without eBPF Add a --replay CLI flag that reads pre-recorded events from a JSONL file instead of loading eBPF programs. This allows profiling the entire userspace pipeline with tools like valgrind DHAT that don't support BPF syscalls. In replay mode, the BPF loader and HostScanner are bypassed entirely. Events flow directly from the JSONL reader through the RateLimiter and Output stages. Changes: - Add Deserialize impls for inode_key_t and monitored_t (fact-ebpf) - Add Deserialize derives to Event, FileData, Process and related types - Make kernel metrics optional in Exporter (None in replay mode) - Migrate task handles to a shared JoinSet for cleaner lifecycle mgmt - Add replay module with async JSONL file reader - Add --replay flag to CLI and config Assisted-by: `agent-mode` --- fact-ebpf/src/lib.rs | 70 +++++++++++++++++++++- fact/src/bpf/mod.rs | 16 ++--- fact/src/config/mod.rs | 17 ++++++ fact/src/config/reloader.rs | 15 +++-- fact/src/config/tests.rs | 3 + fact/src/event/mod.rs | 25 ++++---- fact/src/event/process.rs | 7 ++- fact/src/host_scanner.rs | 8 +-- fact/src/lib.rs | 111 +++++++++++++++++++---------------- fact/src/metrics/exporter.rs | 11 ++-- fact/src/output/mod.rs | 9 +-- fact/src/rate_limiter.rs | 12 ++-- fact/src/replay.rs | 61 +++++++++++++++++++ 13 files changed, 267 insertions(+), 98 deletions(-) create mode 100644 fact/src/replay.rs diff --git a/fact-ebpf/src/lib.rs b/fact-ebpf/src/lib.rs index 4117e983..c1390cac 100644 --- a/fact-ebpf/src/lib.rs +++ b/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(deserializer: D) -> Result + 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(self, mut map: A) -> Result + where + A: MapAccess<'de>, + { + let mut inode = None; + let mut dev = None; + while let Some(key) = map.next_key::()? { + match key.as_str() { + "inode" => inode = Some(map.next_value()?), + "dev" => dev = Some(map.next_value()?), + _ => { + map.next_value::()?; + } + } + } + + 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(deserializer: D) -> Result + 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; diff --git a/fact/src/bpf/mod.rs b/fact/src/bpf/mod.rs index b07c86c8..6c81a577 100644 --- a/fact/src/bpf/mod.rs +++ b/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>, mut running: watch::Receiver, event_counter: EventCounter, - ) -> JoinHandle> { + ) { 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(); diff --git a/fact/src/config/mod.rs b/fact/src/config/mod.rs index ff829eb5..91651be2 100644 --- a/fact/src/config/mod.rs +++ b/fact/src/config/mod.rs @@ -41,6 +41,7 @@ pub struct FactConfig { hotreload: Option, scan_interval: Option, rate_limit: Option, + replay: Option, } impl FactConfig { @@ -106,6 +107,10 @@ impl FactConfig { if let Some(rate_limit) = from.rate_limit { self.rate_limit = Some(rate_limit); } + + if let Some(ref replay) = from.replay { + self.replay = Some(replay.clone()); + } } pub fn paths(&self) -> &[PathBuf] { @@ -132,6 +137,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) { self.paths = Some(paths); @@ -708,6 +717,13 @@ pub struct FactCli { /// Default value is 0 (unlimited) #[arg(long, short = 'l', env = "FACT_RATE_LIMIT")] rate_limit: Option, + + /// 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, } impl FactCli { @@ -739,6 +755,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(), } } } diff --git a/fact/src/config/reloader.rs b/fact/src/config/reloader.rs index 2ac14742..126fd95c 100644 --- a/fact/src/config/reloader.rs +++ b/fact/src/config/reloader.rs @@ -5,7 +5,7 @@ use std::{ use log::{debug, info, warn}; use tokio::{ sync::{Notify, watch}, - task::JoinHandle, + task::JoinSet, time::interval, }; @@ -31,13 +31,17 @@ impl Reloader { /// /// If hotreload is disabled on startup the task will not be /// spawned. - pub fn start(mut self, mut running: watch::Receiver) -> Option> { + pub fn start( + mut self, + task_set: &mut JoinSet>, + mut running: watch::Receiver, + ) { if !self.config.hotreload() { info!("Configuration hotreload is disabled, changes will require a restart."); - return None; + return; } - let handle = tokio::spawn(async move { + task_set.spawn(async move { let mut ticker = interval(Duration::from_secs(10)); loop { tokio::select! { @@ -46,13 +50,12 @@ impl Reloader { _ = running.changed() => { if !*running.borrow() { info!("Stopping config reloader..."); - return; + return Ok(()); } } } } }); - Some(handle) } pub fn config(&self) -> &FactConfig { diff --git a/fact/src/config/tests.rs b/fact/src/config/tests.rs index 01edbbbf..02dd8744 100644 --- a/fact/src/config/tests.rs +++ b/fact/src/config/tests.rs @@ -433,6 +433,7 @@ fn parsing() { hotreload: Some(false), scan_interval: Some(Duration::from_secs(60)), rate_limit: None, + replay: None, }, ), ]; @@ -1555,6 +1556,7 @@ fn update() { hotreload: Some(true), scan_interval: Some(Duration::from_secs(30)), rate_limit: None, + replay: None, }, FactConfig { paths: Some(vec![PathBuf::from("/etc")]), @@ -1583,6 +1585,7 @@ fn update() { hotreload: Some(false), scan_interval: Some(Duration::from_secs(60)), rate_limit: None, + replay: None, }, ), ]; diff --git a/fact/src/event/mod.rs b/fact/src/event/mod.rs index 5792e174..c084deb2 100644 --- a/fact/src/event/mod.rs +++ b/fact/src/event/mod.rs @@ -7,7 +7,7 @@ use std::{ }; use globset::GlobSet; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use fact_ebpf::{ PATH_MAX, XATTR_NAME_MAX_LEN, event_t, file_activity_type_t, inode_key_t, monitored_t, @@ -70,9 +70,10 @@ pub(crate) enum EventTestData { Rename(PathBuf), } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct Event { timestamp: u64, + #[serde(skip_deserializing)] hostname: &'static str, process: Process, file: FileData, @@ -370,7 +371,7 @@ impl PartialEq for Event { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum FileData { Open(BaseFileData), Creation(BaseFileData), @@ -544,7 +545,7 @@ impl PartialEq for FileData { } } -#[derive(Debug, Clone, Serialize, Default)] +#[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct BaseFileData { pub filename: PathBuf, host_file: PathBuf, @@ -586,7 +587,7 @@ impl From for fact_api::FileActivityBase { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChmodFileData { inner: BaseFileData, new_mode: u16, @@ -617,7 +618,7 @@ impl PartialEq for ChmodFileData { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChownFileData { inner: BaseFileData, new_uid: u32, @@ -656,13 +657,13 @@ impl From for fact_api::FileOwnershipChange { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct RenameFileData { new: BaseFileData, old: BaseFileData, } -#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] pub enum AclTag { UserObj, User, @@ -694,7 +695,7 @@ impl AclTag { } } -#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct AclEntry { tag: AclTag, perm: u16, @@ -713,13 +714,13 @@ impl AclEntry { } } -#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] pub enum AclType { Access, Default, } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct AclSetFileData { inner: BaseFileData, acl_type: AclType, @@ -788,7 +789,7 @@ impl PartialEq for RenameFileData { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct XattrFileData { inner: BaseFileData, xattr_name: String, diff --git a/fact/src/event/process.rs b/fact/src/event/process.rs index f295ac66..fd6432da 100644 --- a/fact/src/event/process.rs +++ b/fact/src/event/process.rs @@ -1,14 +1,14 @@ use std::{ffi::CStr, path::PathBuf}; use fact_ebpf::{lineage_t, process_t}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use uuid::Uuid; use crate::host_info; use super::{sanitize_d_path, slice_to_string}; -#[derive(Debug, Clone, Default, PartialEq, Serialize)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] pub struct Lineage { uid: u32, exe_path: PathBuf, @@ -38,13 +38,14 @@ impl From for fact_api::process_signal::LineageInfo { } } -#[derive(Debug, Clone, Default, Serialize)] +#[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct Process { comm: String, args: Vec, exe_path: PathBuf, container_id: Option, uid: u32, + #[serde(skip_deserializing)] username: &'static str, gid: u32, login_uid: u32, diff --git a/fact/src/host_scanner.rs b/fact/src/host_scanner.rs index 273bd5df..1a096912 100644 --- a/fact/src/host_scanner.rs +++ b/fact/src/host_scanner.rs @@ -37,7 +37,7 @@ use globset::{Glob, GlobSet, GlobSetBuilder}; use log::{debug, info, warn}; use tokio::{ sync::{Notify, mpsc, watch}, - task::JoinHandle, + task::JoinSet, }; use crate::{ @@ -409,7 +409,7 @@ You can increase this limit with: }); } - pub fn start(mut self) -> JoinHandle> { + pub fn start(mut self, task_set: &mut JoinSet>) { let scan_interval_value = *self.scan_interval.borrow(); let scan_trigger = Arc::new(Notify::new()); @@ -419,7 +419,7 @@ You can increase this limit with: self.start_scan_notifier(scan_trigger.clone()); } - tokio::spawn(async move { + task_set.spawn(async move { info!("Starting host scanner..."); loop { @@ -491,6 +491,6 @@ You can increase this limit with: info!("Stopping host scanner"); Ok(()) - }) + }); } } diff --git a/fact/src/lib.rs b/fact/src/lib.rs index 8ec187f6..65c8005c 100644 --- a/fact/src/lib.rs +++ b/fact/src/lib.rs @@ -1,6 +1,6 @@ -use std::{borrow::BorrowMut, io::Write, str::FromStr}; +use std::{io::Write, str::FromStr}; -use anyhow::{Context, bail}; +use anyhow::Context; use bpf::Bpf; use host_info::{SystemInfo, get_distro, get_hostname}; use host_scanner::HostScanner; @@ -9,8 +9,8 @@ use metrics::exporter::Exporter; use rate_limiter::RateLimiter; use tokio::{ signal::unix::{SignalKind, signal}, - sync::watch, - task::JoinError, + sync::{mpsc, watch}, + task::JoinSet, }; mod bpf; @@ -23,10 +23,13 @@ mod metrics; mod output; mod pre_flight; mod rate_limiter; +mod replay; use config::FactConfig; use pre_flight::pre_flight; +use crate::event::Event; + pub fn init_log() -> anyhow::Result<()> { let log_level = std::env::var("FACT_LOGLEVEL").unwrap_or("info".to_owned()); let log_level = LevelFilter::from_str(&log_level)?; @@ -65,46 +68,31 @@ pub fn log_system_information() { info!("Hostname: {}", get_hostname()); } -fn flatten_task_result( - component: &str, - res: Result, JoinError>, -) -> anyhow::Result<()> { - match res { - Ok(Ok(_)) => Ok(()), - Ok(Err(e)) => { - bail!("{component} worker errored out: {e:?}"); - } - Err(e) => bail!("{component} task errored out: {e:?}"), - } -} - pub async fn run(config: FactConfig) -> anyhow::Result<()> { // Log system information as early as possible so we have it // available in case of a crash log_system_information(); let (running, _) = watch::channel(true); - - if !config.skip_pre_flight() { - debug!("Performing pre-flight checks"); - pre_flight().context("Pre-flight checks failed")?; - } else { - debug!("Skipping pre-flight checks"); - } - let reloader = config::reloader::Reloader::from(config); let config_trigger = reloader.get_trigger(); + let mut task_set = JoinSet::new(); - let (mut bpf, rx) = Bpf::new(reloader.paths(), &reloader.config().bpf)?; - let exporter = Exporter::new(bpf.take_metrics()?); + let (exporter, rx) = match reloader.config().replay() { + Some(replay_file) => { + let rx = replay::start(&mut task_set, replay_file, running.subscribe())?; + (Exporter::new(None), rx) + } + None => { + if !reloader.config().skip_pre_flight() { + debug!("Performing pre-flight checks"); + pre_flight().context("Pre-flight checks failed")?; + } else { + debug!("Skipping pre-flight checks"); + } - let (host_scanner, rx) = HostScanner::new( - &mut bpf, - rx, - reloader.paths(), - reloader.scan_interval(), - running.subscribe(), - exporter.metrics.host_scanner.clone(), - )?; + bpf_input(&mut task_set, &reloader, running.subscribe())? + } + }; let (rate_limiter, rx) = RateLimiter::new( rx, @@ -113,18 +101,18 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { exporter.metrics.rate_limiter.clone(), )?; - let mut output_handle = output::start( + output::start( + &mut task_set, rx, running.subscribe(), exporter.metrics.output.clone(), reloader.grpc(), reloader.config().json(), ); - let mut host_scanner_handle = host_scanner.start(); - let mut rate_limiter_handle = rate_limiter.start(); + + rate_limiter.start(&mut task_set); endpoints::Server::new(exporter.clone(), reloader.endpoint(), running.subscribe()).start(); - let mut bpf_handle = bpf.start(running.subscribe(), exporter.metrics.bpf_worker.clone()); - reloader.start(running.subscribe()); + reloader.start(&mut task_set, running.subscribe()); let mut sigterm = signal(SignalKind::terminate())?; let mut sighup = signal(SignalKind::hangup())?; @@ -133,23 +121,42 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { _ = tokio::signal::ctrl_c() => break Ok(()), _ = sigterm.recv() => break Ok(()), _ = sighup.recv() => config_trigger.notify_one(), - task_res = bpf_handle.borrow_mut() => { - break flatten_task_result("BPF", task_res); - } - task_res = host_scanner_handle.borrow_mut() => { - break flatten_task_result("HostScanner", task_res); - } - task_res = rate_limiter_handle.borrow_mut() => { - break flatten_task_result("Rate limiter", task_res); - } - task_res = output_handle.borrow_mut() => { - break flatten_task_result("Output", task_res); + task_res = task_set.join_next() => { + let Some(task_res) = task_res else { + unreachable!("No task in task_set"); + }; + match task_res { + Ok(Ok(_)) => break Ok(()), + Ok(Err(e)) => break Err(e), + Err(e) => break Err(e.into()), + } } } }; running.send(false)?; info!("Exiting..."); - res } + +fn bpf_input( + task_set: &mut JoinSet>, + reloader: &config::reloader::Reloader, + running: watch::Receiver, +) -> anyhow::Result<(Exporter, mpsc::Receiver)> { + let (mut bpf, rx) = Bpf::new(reloader.paths(), &reloader.config().bpf)?; + let exporter = Exporter::new(Some(bpf.take_metrics()?)); + + let (host_scanner, rx) = HostScanner::new( + &mut bpf, + rx, + reloader.paths(), + reloader.scan_interval(), + running.clone(), + exporter.metrics.host_scanner.clone(), + )?; + + bpf.start(task_set, running, exporter.metrics.bpf_worker.clone()); + host_scanner.start(task_set); + Ok((exporter, rx)) +} diff --git a/fact/src/metrics/exporter.rs b/fact/src/metrics/exporter.rs index a3a4b772..25ec66b0 100644 --- a/fact/src/metrics/exporter.rs +++ b/fact/src/metrics/exporter.rs @@ -12,14 +12,15 @@ use super::{Metrics, kernel_metrics::KernelMetrics}; pub struct Exporter { registry: Arc, pub metrics: Arc, - kernel_metrics: Arc, + kernel_metrics: Option>, } impl Exporter { - pub fn new(kernel_metrics: PerCpuArray) -> Self { + pub fn new(kernel_metrics: Option>) -> Self { let mut registry = Registry::with_prefix("stackrox_fact"); let metrics = Arc::new(Metrics::new(&mut registry)); - let kernel_metrics = Arc::new(KernelMetrics::new(&mut registry, kernel_metrics)); + let kernel_metrics = + kernel_metrics.map(|km| Arc::new(KernelMetrics::new(&mut registry, km))); let registry = Arc::new(registry); Exporter { registry, @@ -30,7 +31,9 @@ impl Exporter { pub fn encode(&self) -> anyhow::Result { let mut buf = String::new(); - if let Err(e) = self.kernel_metrics.collect() { + if let Some(ref km) = self.kernel_metrics + && let Err(e) = km.collect() + { warn!("Failed to collect kernel metrics: {e}"); } encode(&mut buf, &self.registry)?; diff --git a/fact/src/output/mod.rs b/fact/src/output/mod.rs index 756c8eec..94df6995 100644 --- a/fact/src/output/mod.rs +++ b/fact/src/output/mod.rs @@ -4,7 +4,7 @@ use anyhow::bail; use log::{debug, warn}; use tokio::{ sync::{broadcast, mpsc, watch}, - task::JoinHandle, + task::JoinSet, }; use crate::{config::GrpcConfig, event::Event, metrics::OutputMetrics}; @@ -17,12 +17,13 @@ mod stdout; /// Each task is responsible for managing its lifetime, handling /// incoming events and reloading configuration. pub fn start( + task_set: &mut JoinSet>, mut rx: mpsc::Receiver, running: watch::Receiver, metrics: OutputMetrics, config: watch::Receiver, stdout_enabled: bool, -) -> JoinHandle> { +) { let (broad_tx, broad_rx) = broadcast::channel(100); let mut run = running.clone(); @@ -46,7 +47,7 @@ pub fn start( let mut grpc_handle = grpc_client.start(); - tokio::spawn(async move { + task_set.spawn(async move { debug!("Starting output component..."); loop { tokio::select! { @@ -71,5 +72,5 @@ pub fn start( } debug!("Stopping output component..."); Ok(()) - }) + }); } diff --git a/fact/src/rate_limiter.rs b/fact/src/rate_limiter.rs index 84894d76..8746808f 100644 --- a/fact/src/rate_limiter.rs +++ b/fact/src/rate_limiter.rs @@ -5,8 +5,10 @@ use governor::{ }; use log::warn; use std::num::NonZeroU32; -use tokio::sync::{mpsc, watch}; -use tokio::task::JoinHandle; +use tokio::{ + sync::{mpsc, watch}, + task::JoinSet, +}; use crate::event::Event; use crate::metrics::EventCounter; @@ -63,8 +65,8 @@ impl RateLimiter { Ok(()) } - pub fn start(mut self) -> JoinHandle> { - tokio::spawn(async move { + pub fn start(mut self, task_set: &mut JoinSet>) { + task_set.spawn(async move { loop { tokio::select! { event = self.rx.recv() => { @@ -92,6 +94,6 @@ impl RateLimiter { } } Ok(()) - }) + }); } } diff --git a/fact/src/replay.rs b/fact/src/replay.rs new file mode 100644 index 00000000..71bb7fa2 --- /dev/null +++ b/fact/src/replay.rs @@ -0,0 +1,61 @@ +use std::path::Path; + +use anyhow::Context; +use log::{info, warn}; +use tokio::{ + io::{AsyncBufReadExt, BufReader}, + sync::{mpsc, watch}, + task::JoinSet, +}; + +use crate::event::Event; + +pub fn start( + task_set: &mut JoinSet>, + path: &Path, + running: watch::Receiver, +) -> anyhow::Result> { + anyhow::ensure!( + path.exists(), + "Replay file does not exist: {}", + path.display() + ); + let (tx, rx) = mpsc::channel(100); + let path = path.to_owned(); + + task_set.spawn(async move { + let file = tokio::fs::File::open(&path) + .await + .with_context(|| format!("Failed to open replay file: {}", path.display()))?; + let reader = BufReader::new(file); + let mut lines = reader.lines(); + + info!("Replaying events from {}", path.display()); + while let Some(line) = lines.next_line().await? { + if !*running.borrow() { + break; + } + + let value: serde_json::Value = match serde_json::from_str(&line) { + Ok(v) => v, + Err(e) => { + warn!("Failed to parse JSON: {e}"); + continue; + } + }; + match serde_json::from_value::(value) { + Ok(event) => { + if tx.send(event).await.is_err() { + break; + } + } + Err(e) => warn!("Failed to deserialize event: {e}"), + } + } + + info!("Replay finished"); + Ok(()) + }); + + Ok(rx) +} From bd88a246b08b909db1537dae9c698b6e67554008 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Tue, 7 Jul 2026 18:14:01 +0200 Subject: [PATCH 2/2] chore: add CHANGELOG line --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca27cddb..9e67e6cf 100644 --- a/CHANGELOG.md +++ b/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(grpc): add exponential backoff for reconnection attempts (#789) * feat: run integration tests on more platforms (#760) * ROX-34502: reload mTLS certificates on each gRPC connection attempt (#788)