From 2f2e639dce61ce949d5f365a4bff9d905b564b8c Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Tue, 7 Jul 2026 16:02:19 +0200 Subject: [PATCH 1/4] 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 | 23 ++++++ fact/src/config/reloader.rs | 8 +- fact/src/config/tests.rs | 35 ++++++++ fact/src/event/mod.rs | 25 +++--- fact/src/event/process.rs | 7 +- fact/src/host_scanner.rs | 24 +++--- fact/src/lib.rs | 151 ++++++++++++++++++++++------------- fact/src/metrics/exporter.rs | 11 ++- fact/src/output/grpc.rs | 4 + fact/src/output/mod.rs | 50 ++++++++---- fact/src/output/otel.rs | 5 +- fact/src/output/stdout.rs | 13 +-- fact/src/rate_limiter.rs | 26 +++--- fact/src/replay.rs | 61 ++++++++++++++ 16 files changed, 390 insertions(+), 139 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 88daa519..00d2252a 100644 --- a/fact/src/config/mod.rs +++ b/fact/src/config/mod.rs @@ -42,6 +42,7 @@ pub struct FactConfig { hotreload: Option, scan_interval: Option, rate_limit: Option, + replay: Option, } 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) { self.paths = Some(paths); @@ -247,6 +256,12 @@ impl TryFrom> 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, + + /// 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 { @@ -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(), } } } diff --git a/fact/src/config/reloader.rs b/fact/src/config/reloader.rs index 48baecb7..362f337e 100644 --- a/fact/src/config/reloader.rs +++ b/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) -> Option> { + pub fn start(mut self, 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 { + 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 { diff --git a/fact/src/config/tests.rs b/fact/src/config/tests.rs index bd47960e..dacd4a25 100644 --- a/fact/src/config/tests.rs +++ b/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(()); diff --git a/fact/src/event/mod.rs b/fact/src/event/mod.rs index 744a61b5..98808af8 100644 --- a/fact/src/event/mod.rs +++ b/fact/src/event/mod.rs @@ -11,7 +11,7 @@ use std::{ use globset::GlobSet; #[cfg(feature = "otel")] use opentelemetry::logs::AnyValue; -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, @@ -74,9 +74,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, @@ -391,7 +392,7 @@ impl PartialEq for Event { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum FileData { Open(BaseFileData), Creation(BaseFileData), @@ -607,7 +608,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, @@ -665,7 +666,7 @@ impl From for opentelemetry::logs::AnyValue { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChmodFileData { inner: BaseFileData, new_mode: u16, @@ -711,7 +712,7 @@ impl PartialEq for ChmodFileData { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChownFileData { inner: BaseFileData, new_uid: u32, @@ -767,7 +768,7 @@ impl From for opentelemetry::logs::AnyValue { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct RenameFileData { new: BaseFileData, old: BaseFileData, @@ -795,7 +796,7 @@ impl From for opentelemetry::logs::AnyValue { } } -#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] pub enum AclTag { UserObj, User, @@ -842,7 +843,7 @@ impl AclTag { } } -#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct AclEntry { tag: AclTag, perm: u16, @@ -876,7 +877,7 @@ impl From for opentelemetry::logs::AnyValue { } } -#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] pub enum AclType { Access, Default, @@ -892,7 +893,7 @@ impl From for opentelemetry::logs::AnyValue { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct AclSetFileData { inner: BaseFileData, acl_type: AclType, @@ -968,7 +969,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 db3ff2ec..6f0733e0 100644 --- a/fact/src/event/process.rs +++ b/fact/src/event/process.rs @@ -5,14 +5,14 @@ use std::{ffi::CStr, path::PathBuf}; use fact_ebpf::{lineage_t, process_t}; #[cfg(feature = "otel")] use opentelemetry::logs::AnyValue; -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, @@ -55,13 +55,14 @@ impl From for opentelemetry::logs::AnyValue { } } -#[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..9e46eef1 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::{ @@ -53,7 +53,6 @@ pub struct HostScanner { paths: watch::Receiver>, scan_interval: watch::Receiver, - running: watch::Receiver, rx: mpsc::Receiver, tx: mpsc::Sender, @@ -69,7 +68,6 @@ impl HostScanner { rx: mpsc::Receiver, paths: watch::Receiver>, scan_interval: watch::Receiver, - running: watch::Receiver, metrics: HostScannerMetrics, ) -> anyhow::Result<(Self, mpsc::Receiver)> { let kernel_inode_map = RefCell::new(bpf.take_inode_map()?); @@ -82,7 +80,6 @@ impl HostScanner { inode_map, paths, scan_interval, - running, rx, tx, metrics, @@ -392,8 +389,7 @@ You can increase this limit with: /// `tokio::select` with other events that trigger more often, the /// tick will never happen. This way we have a separate task that /// will reliably send a notification to the main one. - fn start_scan_notifier(&self, scan_trigger: Arc) { - let mut running = self.running.clone(); + fn start_scan_notifier(&self, scan_trigger: Arc, mut running: watch::Receiver) { let mut scan_interval = self.scan_interval.clone(); tokio::spawn(async move { while *running.borrow() { @@ -409,17 +405,18 @@ 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()); + let (running, running_rx) = watch::channel(true); if scan_interval_value.is_zero() { warn!("Host scanner periodic scans permanently disabled (scan_interval is 0)"); } else { - self.start_scan_notifier(scan_trigger.clone()); + self.start_scan_notifier(scan_trigger.clone(), running_rx); } - tokio::spawn(async move { + task_set.spawn(async move { info!("Starting host scanner..."); loop { @@ -481,16 +478,13 @@ You can increase this limit with: self.paths_globset = HostScanner::build_globset(self.paths.borrow().as_slice())?; self.scan()?; } - _ = self.running.changed() => { - if !*self.running.borrow() { - break; - } - } } } + let _ = running.send(false); + info!("Stopping host scanner"); Ok(()) - }) + }); } } diff --git a/fact/src/lib.rs b/fact/src/lib.rs index 94fe1cc4..f3b7aa7a 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, time::Duration}; -use anyhow::{Context, bail}; +use anyhow::{Context, Result}; use bpf::Bpf; use host_info::{SystemInfo, get_distro, get_hostname}; use host_scanner::HostScanner; @@ -9,8 +9,9 @@ use metrics::exporter::Exporter; use rate_limiter::RateLimiter; use tokio::{ signal::unix::{SignalKind, signal}, - sync::watch, - task::JoinError, + sync::{mpsc, watch}, + task::JoinSet, + time::timeout, }; mod bpf; @@ -23,10 +24,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)?; @@ -66,91 +70,130 @@ pub fn log_system_information() { } fn flatten_task_result( - component: &str, - res: Result, JoinError>, + task_res: Result, impl Into>, ) -> anyhow::Result<()> { - match res { + match task_res { Ok(Ok(_)) => Ok(()), - Ok(Err(e)) => { - bail!("{component} worker errored out: {e:?}"); - } - Err(e) => bail!("{component} task errored out: {e:?}"), + Ok(Err(e)) => Err(e), + Err(e) => Err(e.into()), } } +async fn join_all_tasks(task_set: JoinSet>) -> anyhow::Result<()> { + task_set + .join_all() + .await + .into_iter() + .collect::, _>>()?; + Ok(()) +} + 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 (running_pipeline_tx, running_pipeline_rx) = watch::channel(true); + let (running_helpers, _) = watch::channel(true); 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 (host_scanner, rx) = HostScanner::new( - &mut bpf, - rx, - reloader.paths(), - reloader.scan_interval(), - running.subscribe(), - exporter.metrics.host_scanner.clone(), - )?; - + let (exporter, rx) = setup_input(&mut task_set, &reloader, running_pipeline_rx)?; let (rate_limiter, rx) = RateLimiter::new( rx, reloader.rate_limit(), - running.subscribe(), 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.otel(), reloader.config().json(), ); - let mut host_scanner_handle = host_scanner.start(); - let mut rate_limiter_handle = rate_limiter.start(); - 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()); + + rate_limiter.start(&mut task_set); + endpoints::Server::new( + exporter.clone(), + reloader.endpoint(), + running_helpers.subscribe(), + ) + .start(); + reloader.start(running_helpers.subscribe()); let mut sigterm = signal(SignalKind::terminate())?; let mut sighup = signal(SignalKind::hangup())?; - let res = loop { + let mut res = loop { tokio::select! { _ = 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"); + }; + break flatten_task_result(task_res); } } }; - running.send(false)?; - info!("Exiting..."); + // Stop the input of the pipeline, this will cascade as the sender + // ends of the channels used for communication are dropped, causing + // elements in the pipeline to be stopped as they empty their + // receivers. + let _ = running_pipeline_tx.send(false); + if res.is_ok() { + let join_res = timeout(Duration::from_secs(5), join_all_tasks(task_set)).await; + res = flatten_task_result(join_res); + } + let _ = running_helpers.send(false); + info!("Exiting..."); res } + +fn setup_input( + task_set: &mut JoinSet>, + reloader: &config::reloader::Reloader, + running: watch::Receiver, +) -> anyhow::Result<(Exporter, mpsc::Receiver)> { + match reloader.config().replay() { + Some(replay_file) => { + let rx = replay::start(task_set, replay_file, running)?; + Ok((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"); + } + + bpf_input(task_set, reloader, running) + } + } +} + +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(), + 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/grpc.rs b/fact/src/output/grpc.rs index b643b363..33da7146 100644 --- a/fact/src/output/grpc.rs +++ b/fact/src/output/grpc.rs @@ -250,6 +250,10 @@ impl Client { res = client.communicate(rx) => { match res { Ok(_) => info!("gRPC stream ended"), + Err(_) if self.subscriber.is_closed() => { + info!("Channel closed, stopping gRPC output..."); + return Ok(false); + } Err(e) => warn!("gRPC stream error: {e:?}"), } } diff --git a/fact/src/output/mod.rs b/fact/src/output/mod.rs index 8879b431..049f6e9c 100644 --- a/fact/src/output/mod.rs +++ b/fact/src/output/mod.rs @@ -3,12 +3,13 @@ use std::sync::Arc; use log::{debug, warn}; use tokio::{ sync::{broadcast, mpsc, watch}, - task::{JoinHandle, JoinSet}, + task::JoinSet, }; use crate::{ config::{GrpcConfig, OTelConfig}, event::Event, + flatten_task_result, join_all_tasks, metrics::OutputMetrics, }; @@ -24,21 +25,21 @@ type EventReceiver = broadcast::Receiver>; /// 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, grpc_config: watch::Receiver, #[allow(unused)] otel_config: watch::Receiver, stdout_enabled: bool, -) -> JoinHandle> { +) { let (broad_tx, _) = broadcast::channel(100); let (subs_req, mut subs_rx) = mpsc::channel(10); - let mut run = running.clone(); + let (running, _) = watch::channel(true); let mut handles = JoinSet::new(); let grpc_client = grpc::Client::new( subs_req.clone(), - running.clone(), + running.subscribe(), metrics.grpc.clone(), grpc_config, ); @@ -50,7 +51,7 @@ pub fn start( { let otel_client = otel::Client::new( subs_req.clone(), - running.clone(), + running.subscribe(), metrics.otel.clone(), otel_config, ); @@ -63,13 +64,13 @@ pub fn start( if stdout_enabled || !non_stdout_enabled { stdout::Client::new( broad_tx.subscribe(), - running.clone(), + running.subscribe(), metrics.stdout.clone(), ) - .start(); + .start(&mut handles); } - tokio::spawn(async move { + task_set.spawn(async move { debug!("Starting output component..."); let res = loop { tokio::select! { @@ -95,16 +96,31 @@ pub fn start( let Some(res) = res else { unreachable!("output handles should always have a task"); }; - match res { - Ok(Ok(_)) => break Ok(()), - Ok(Err(e)) => break Err(e), - Err(e) => break Err(e.into()), - } + break flatten_task_result(res); } - _ = run.changed() => if !*run.borrow() { break Ok(()); } } }; debug!("Stopping output component..."); - res - }) + + if res.is_ok() { + // Wait for outputs to empty their channels before exiting + // ourselves. + let receiver_count = broad_tx.receiver_count(); + drop(subs_rx); + drop(broad_tx); + + for _ in 0..receiver_count { + let Some(task_res) = handles.join_next().await else { + break; + }; + flatten_task_result(task_res)?; + } + + // Force idle outputs to stop + let _ = running.send(false); + join_all_tasks(handles).await + } else { + res + } + }); } diff --git a/fact/src/output/otel.rs b/fact/src/output/otel.rs index fda0abe2..3af202df 100644 --- a/fact/src/output/otel.rs +++ b/fact/src/output/otel.rs @@ -99,7 +99,10 @@ impl Client { } logger.emit(record); } - Err(RecvError::Closed) => break Err(anyhow::anyhow!("oTel: event stream closed")), + Err(RecvError::Closed) => { + info!("Channel closed, stopping oTel output..."); + break Ok(false); + } Err(RecvError::Lagged(n)) => { warn!("oTel stream lagged, dropped {n} events"); self.metrics.dropped_n(n); diff --git a/fact/src/output/stdout.rs b/fact/src/output/stdout.rs index 020c5534..395132a8 100644 --- a/fact/src/output/stdout.rs +++ b/fact/src/output/stdout.rs @@ -1,5 +1,8 @@ use log::{info, warn}; -use tokio::sync::{broadcast::error::RecvError, watch}; +use tokio::{ + sync::{broadcast::error::RecvError, watch}, + task::JoinSet, +}; use crate::{metrics::EventCounter, output::EventReceiver}; @@ -18,8 +21,8 @@ impl Client { } } - pub fn start(mut self) { - 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() => { @@ -27,7 +30,7 @@ impl Client { Ok(event) => event, Err(RecvError::Closed) => { info!("Channel closed, stopping stdout output..."); - return; + return Ok(()); } Err(RecvError::Lagged(n)) => { self.metrics.dropped_n(n); @@ -49,7 +52,7 @@ impl Client { _ = self.running.changed() => { if !*self.running.borrow() { info!("Stopping stdout output..."); - return; + return Ok(()); } } } diff --git a/fact/src/rate_limiter.rs b/fact/src/rate_limiter.rs index 84894d76..e4bd9978 100644 --- a/fact/src/rate_limiter.rs +++ b/fact/src/rate_limiter.rs @@ -3,10 +3,12 @@ use governor::{ clock::DefaultClock, state::{InMemoryState, NotKeyed}, }; -use log::warn; +use log::{debug, 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; @@ -20,7 +22,6 @@ pub struct RateLimiter { rx: mpsc::Receiver, tx: mpsc::Sender, rate_config: watch::Receiver, - running: watch::Receiver, metrics: EventCounter, } @@ -28,7 +29,6 @@ impl RateLimiter { pub fn new( rx: mpsc::Receiver, rate_config: watch::Receiver, - running: watch::Receiver, metrics: EventCounter, ) -> anyhow::Result<(Self, mpsc::Receiver)> { let limiter = Self::build_limiter(*rate_config.borrow()); @@ -39,7 +39,6 @@ impl RateLimiter { rx, tx, rate_config, - running, metrics, }; @@ -63,12 +62,13 @@ 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 { + debug!("Starting rate limiter..."); loop { tokio::select! { event = self.rx.recv() => { - let Some(event) = event else { break;}; + let Some(event) = event else { break; }; if let Some(limiter) = &self.limiter && limiter.check().is_err() { self.metrics.dropped(); @@ -84,14 +84,10 @@ impl RateLimiter { _ = self.rate_config.changed() => { self.reload_limiter()?; }, - _ = self.running.changed() => { - if !*self.running.borrow() { - break; - } - }, } } + debug!("Stopping rate limiter..."); 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 49232f391fec3b83e77eef6005674b88101a4c07 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Tue, 7 Jul 2026 18:14:01 +0200 Subject: [PATCH 2/4] chore: add CHANGELOG line --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index aeb1de72..6d9a811a 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(output): add basic opentelemetry output (#971) * feat(grpc): add exponential backoff for reconnection attempts (#789) * feat: run integration tests on more platforms (#760) From 1a5976158e9fed7d27cb997f1d822b3e245597f5 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Wed, 15 Jul 2026 16:49:09 +0200 Subject: [PATCH 3/4] fix(grpc): check output is running in loop Also add `RUST_BACKTRACE=1` to the integration tests to get backtraces when fact errors out. --- fact/src/output/grpc.rs | 5 +++++ tests/conftest.py | 1 + 2 files changed, 6 insertions(+) diff --git a/fact/src/output/grpc.rs b/fact/src/output/grpc.rs index 33da7146..e91eb897 100644 --- a/fact/src/output/grpc.rs +++ b/fact/src/output/grpc.rs @@ -207,6 +207,11 @@ impl Client { async fn run(&mut self) -> anyhow::Result { let mut backoff = Backoff::from(&self.config.borrow().backoff); loop { + if self.subscriber.is_closed() { + info!("Channel closed, stopping gRPC output..."); + return Ok(false); + } + // Re-read certs on each connection attempt so rotated certificates // on disk are picked up on the next reconnect. let connector = self.get_connector().await?; diff --git a/tests/conftest.py b/tests/conftest.py index e5f30b92..b66581c8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -246,6 +246,7 @@ def fact( 'FACT_HOST_MOUNT': '/host', 'OTEL_BLRP_SCHEDULE_DELAY': '100', 'OTEL_BLRP_MAX_EXPORT_BATCH_SIZE': '1', + 'RUST_BACKTRACE': '1', }, name='fact', network_mode='host', From a6af5d582bee13850d3336ec35120f7e0ff9ad52 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Wed, 15 Jul 2026 16:53:02 +0200 Subject: [PATCH 4/4] fix: join all tasks with `join_next` This prevents panicking in case a `JoinError` occurs. --- fact/src/lib.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/fact/src/lib.rs b/fact/src/lib.rs index f3b7aa7a..29103e12 100644 --- a/fact/src/lib.rs +++ b/fact/src/lib.rs @@ -79,12 +79,10 @@ fn flatten_task_result( } } -async fn join_all_tasks(task_set: JoinSet>) -> anyhow::Result<()> { - task_set - .join_all() - .await - .into_iter() - .collect::, _>>()?; +async fn join_all_tasks(mut task_set: JoinSet>) -> anyhow::Result<()> { + while let Some(task_res) = task_set.join_next().await { + flatten_task_result(task_res)?; + } Ok(()) }