Skip to content

Commit 68ec0c4

Browse files
committed
feat: add --replay mode for JSONL event replay without eBPF
Add a --replay <file> 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` <noreply@opencode.ai>
1 parent 17d94b4 commit 68ec0c4

13 files changed

Lines changed: 267 additions & 98 deletions

File tree

fact-ebpf/src/lib.rs

Lines changed: 69 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,11 @@ use std::{error::Error, ffi::c_char, fmt::Display, hash::Hash, path::PathBuf};
44

55
use aya::{maps::lpm_trie, Pod};
66
use libc::memcpy;
7-
use serde::{ser::SerializeStruct, Serialize};
7+
use serde::{
8+
de::{self, MapAccess, Visitor},
9+
ser::SerializeStruct,
10+
Deserialize, Serialize,
11+
};
812

913
include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
1014

@@ -101,6 +105,51 @@ impl Serialize for inode_key_t {
101105
}
102106
}
103107

108+
impl<'de> Deserialize<'de> for inode_key_t {
109+
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
110+
where
111+
D: de::Deserializer<'de>,
112+
{
113+
struct InodeKeyVisitor;
114+
115+
impl<'de> Visitor<'de> for InodeKeyVisitor {
116+
type Value = inode_key_t;
117+
118+
fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
119+
f.write_str("a struct with inode and dev fields")
120+
}
121+
122+
fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
123+
where
124+
A: MapAccess<'de>,
125+
{
126+
let mut inode = None;
127+
let mut dev = None;
128+
while let Some(key) = map.next_key::<String>()? {
129+
match key.as_str() {
130+
"inode" => inode = Some(map.next_value()?),
131+
"dev" => dev = Some(map.next_value()?),
132+
_ => {
133+
map.next_value::<de::IgnoredAny>()?;
134+
}
135+
}
136+
}
137+
138+
let Some(inode) = inode else {
139+
return Err(de::Error::missing_field("inode"));
140+
};
141+
let Some(dev) = dev else {
142+
return Err(de::Error::missing_field("dev"));
143+
};
144+
145+
Ok(inode_key_t { inode, dev })
146+
}
147+
}
148+
149+
deserializer.deserialize_struct("inode_key_t", &["inode", "dev"], InodeKeyVisitor)
150+
}
151+
}
152+
104153
unsafe impl Pod for inode_key_t {}
105154

106155
impl Default for monitored_t {
@@ -124,6 +173,25 @@ impl Serialize for monitored_t {
124173
}
125174
}
126175

176+
impl<'de> Deserialize<'de> for monitored_t {
177+
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
178+
where
179+
D: de::Deserializer<'de>,
180+
{
181+
let s = String::deserialize(deserializer)?;
182+
match s.as_str() {
183+
"not monitored" => Ok(monitored_t::NOT_MONITORED),
184+
"by inode" => Ok(monitored_t::MONITORED_BY_INODE),
185+
"by path" => Ok(monitored_t::MONITORED_BY_PATH),
186+
"by parent" => Ok(monitored_t::MONITORED_BY_PARENT),
187+
_ => Err(de::Error::unknown_variant(
188+
&s,
189+
&["not monitored", "by inode", "by path", "by parent"],
190+
)),
191+
}
192+
}
193+
}
194+
127195
impl metrics_by_hook_t {
128196
fn accumulate(mut self, other: &metrics_by_hook_t) -> metrics_by_hook_t {
129197
self.total += other.total;

fact/src/bpf/mod.rs

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ use log::{error, info, warn};
1313
use tokio::{
1414
io::unix::AsyncFd,
1515
sync::{mpsc, watch},
16-
task::JoinHandle,
16+
task::JoinSet,
1717
};
1818

1919
use crate::{config::BpfConfig, event::Event, host_info, metrics::EventCounter};
@@ -225,12 +225,13 @@ impl Bpf {
225225
// Gather events from the ring buffer and print them out.
226226
pub fn start(
227227
mut self,
228+
task_set: &mut JoinSet<anyhow::Result<()>>,
228229
mut running: watch::Receiver<bool>,
229230
event_counter: EventCounter,
230-
) -> JoinHandle<anyhow::Result<()>> {
231+
) {
231232
info!("Starting BPF worker...");
232233

233-
tokio::spawn(async move {
234+
task_set.spawn(async move {
234235
let rb = self.take_ringbuffer()?;
235236
let mut fd = AsyncFd::new(rb)?;
236237

@@ -283,7 +284,7 @@ impl Bpf {
283284
}
284285

285286
Ok(())
286-
})
287+
});
287288
}
288289
}
289290

@@ -322,9 +323,10 @@ mod bpf_tests {
322323
Bpf::new(reloader.paths(), &reloader.config().bpf).expect("Failed to load BPF code");
323324
let (run_tx, run_rx) = watch::channel(true);
324325
// Create a metrics exporter, but don't start it
325-
let exporter = Exporter::new(bpf.take_metrics().unwrap());
326+
let exporter = Exporter::new(Some(bpf.take_metrics().unwrap()));
327+
let mut task_set = JoinSet::new();
326328

327-
let handle = bpf.start(run_rx, exporter.metrics.bpf_worker.clone());
329+
bpf.start(&mut task_set, run_rx, exporter.metrics.bpf_worker.clone());
328330

329331
tokio::time::sleep(Duration::from_millis(500)).await;
330332

@@ -412,7 +414,7 @@ mod bpf_tests {
412414

413415
tokio::select! {
414416
res = wait => res.unwrap(),
415-
res = handle => res.unwrap().unwrap(),
417+
res = task_set.join_next() => res.unwrap().unwrap().unwrap(),
416418
}
417419

418420
run_tx.send(false).unwrap();

fact/src/config/mod.rs

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ pub struct FactConfig {
4141
hotreload: Option<bool>,
4242
scan_interval: Option<Duration>,
4343
rate_limit: Option<u64>,
44+
replay: Option<PathBuf>,
4445
}
4546

4647
impl FactConfig {
@@ -106,6 +107,10 @@ impl FactConfig {
106107
if let Some(rate_limit) = from.rate_limit {
107108
self.rate_limit = Some(rate_limit);
108109
}
110+
111+
if let Some(ref replay) = from.replay {
112+
self.replay = Some(replay.clone());
113+
}
109114
}
110115

111116
pub fn paths(&self) -> &[PathBuf] {
@@ -132,6 +137,10 @@ impl FactConfig {
132137
self.rate_limit.unwrap_or(0)
133138
}
134139

140+
pub fn replay(&self) -> Option<&Path> {
141+
self.replay.as_deref()
142+
}
143+
135144
#[cfg(test)]
136145
pub fn set_paths(&mut self, paths: Vec<PathBuf>) {
137146
self.paths = Some(paths);
@@ -708,6 +717,13 @@ pub struct FactCli {
708717
/// Default value is 0 (unlimited)
709718
#[arg(long, short = 'l', env = "FACT_RATE_LIMIT")]
710719
rate_limit: Option<u64>,
720+
721+
/// Replay events from a JSONL file instead of loading eBPF programs.
722+
///
723+
/// This mode skips BPF loading and HostScanner, reading pre-recorded
724+
/// events for profiling purposes (e.g., valgrind DHAT).
725+
#[arg(long, env = "FACT_REPLAY")]
726+
replay: Option<PathBuf>,
711727
}
712728

713729
impl FactCli {
@@ -739,6 +755,7 @@ impl FactCli {
739755
hotreload: resolve_bool_arg(self.hotreload, self.no_hotreload),
740756
scan_interval: self.scan_interval,
741757
rate_limit: self.rate_limit,
758+
replay: self.replay.clone(),
742759
}
743760
}
744761
}

fact/src/config/reloader.rs

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ use std::{
55
use log::{debug, info, warn};
66
use tokio::{
77
sync::{Notify, watch},
8-
task::JoinHandle,
8+
task::JoinSet,
99
time::interval,
1010
};
1111

@@ -31,13 +31,17 @@ impl Reloader {
3131
///
3232
/// If hotreload is disabled on startup the task will not be
3333
/// spawned.
34-
pub fn start(mut self, mut running: watch::Receiver<bool>) -> Option<JoinHandle<()>> {
34+
pub fn start(
35+
mut self,
36+
task_set: &mut JoinSet<anyhow::Result<()>>,
37+
mut running: watch::Receiver<bool>,
38+
) {
3539
if !self.config.hotreload() {
3640
info!("Configuration hotreload is disabled, changes will require a restart.");
37-
return None;
41+
return;
3842
}
3943

40-
let handle = tokio::spawn(async move {
44+
task_set.spawn(async move {
4145
let mut ticker = interval(Duration::from_secs(10));
4246
loop {
4347
tokio::select! {
@@ -46,13 +50,12 @@ impl Reloader {
4650
_ = running.changed() => {
4751
if !*running.borrow() {
4852
info!("Stopping config reloader...");
49-
return;
53+
return Ok(());
5054
}
5155
}
5256
}
5357
}
5458
});
55-
Some(handle)
5659
}
5760

5861
pub fn config(&self) -> &FactConfig {

fact/src/config/tests.rs

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -433,6 +433,7 @@ fn parsing() {
433433
hotreload: Some(false),
434434
scan_interval: Some(Duration::from_secs(60)),
435435
rate_limit: None,
436+
replay: None,
436437
},
437438
),
438439
];
@@ -1555,6 +1556,7 @@ fn update() {
15551556
hotreload: Some(true),
15561557
scan_interval: Some(Duration::from_secs(30)),
15571558
rate_limit: None,
1559+
replay: None,
15581560
},
15591561
FactConfig {
15601562
paths: Some(vec![PathBuf::from("/etc")]),
@@ -1583,6 +1585,7 @@ fn update() {
15831585
hotreload: Some(false),
15841586
scan_interval: Some(Duration::from_secs(60)),
15851587
rate_limit: None,
1588+
replay: None,
15861589
},
15871590
),
15881591
];

fact/src/event/mod.rs

Lines changed: 13 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ use std::{
77
};
88

99
use globset::GlobSet;
10-
use serde::Serialize;
10+
use serde::{Deserialize, Serialize};
1111

1212
use fact_ebpf::{
1313
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 {
7070
Rename(PathBuf),
7171
}
7272

73-
#[derive(Debug, Clone, Serialize)]
73+
#[derive(Debug, Clone, Serialize, Deserialize)]
7474
pub struct Event {
7575
timestamp: u64,
76+
#[serde(skip_deserializing)]
7677
hostname: &'static str,
7778
process: Process,
7879
file: FileData,
@@ -370,7 +371,7 @@ impl PartialEq for Event {
370371
}
371372
}
372373

373-
#[derive(Debug, Clone, Serialize)]
374+
#[derive(Debug, Clone, Serialize, Deserialize)]
374375
pub enum FileData {
375376
Open(BaseFileData),
376377
Creation(BaseFileData),
@@ -544,7 +545,7 @@ impl PartialEq for FileData {
544545
}
545546
}
546547

547-
#[derive(Debug, Clone, Serialize, Default)]
548+
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
548549
pub struct BaseFileData {
549550
pub filename: PathBuf,
550551
host_file: PathBuf,
@@ -586,7 +587,7 @@ impl From<BaseFileData> for fact_api::FileActivityBase {
586587
}
587588
}
588589

589-
#[derive(Debug, Clone, Serialize)]
590+
#[derive(Debug, Clone, Serialize, Deserialize)]
590591
pub struct ChmodFileData {
591592
inner: BaseFileData,
592593
new_mode: u16,
@@ -617,7 +618,7 @@ impl PartialEq for ChmodFileData {
617618
}
618619
}
619620

620-
#[derive(Debug, Clone, Serialize)]
621+
#[derive(Debug, Clone, Serialize, Deserialize)]
621622
pub struct ChownFileData {
622623
inner: BaseFileData,
623624
new_uid: u32,
@@ -656,13 +657,13 @@ impl From<ChownFileData> for fact_api::FileOwnershipChange {
656657
}
657658
}
658659

659-
#[derive(Debug, Clone, Serialize)]
660+
#[derive(Debug, Clone, Serialize, Deserialize)]
660661
pub struct RenameFileData {
661662
new: BaseFileData,
662663
old: BaseFileData,
663664
}
664665

665-
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
666+
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
666667
pub enum AclTag {
667668
UserObj,
668669
User,
@@ -694,7 +695,7 @@ impl AclTag {
694695
}
695696
}
696697

697-
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
698+
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
698699
pub struct AclEntry {
699700
tag: AclTag,
700701
perm: u16,
@@ -713,13 +714,13 @@ impl AclEntry {
713714
}
714715
}
715716

716-
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
717+
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
717718
pub enum AclType {
718719
Access,
719720
Default,
720721
}
721722

722-
#[derive(Debug, Clone, Serialize)]
723+
#[derive(Debug, Clone, Serialize, Deserialize)]
723724
pub struct AclSetFileData {
724725
inner: BaseFileData,
725726
acl_type: AclType,
@@ -788,7 +789,7 @@ impl PartialEq for RenameFileData {
788789
}
789790
}
790791

791-
#[derive(Debug, Clone, Serialize)]
792+
#[derive(Debug, Clone, Serialize, Deserialize)]
792793
pub struct XattrFileData {
793794
inner: BaseFileData,
794795
xattr_name: String,

0 commit comments

Comments
 (0)