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) 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) +}