diff --git a/fact/src/bpf/mod.rs b/fact/src/bpf/mod.rs index 6c81a577..44f48974 100644 --- a/fact/src/bpf/mod.rs +++ b/fact/src/bpf/mod.rs @@ -36,12 +36,17 @@ pub struct Bpf { paths_globset: GlobSet, links: Vec, + + running: watch::Receiver, + metrics: EventCounter, } impl Bpf { pub fn new( paths_config: watch::Receiver>, bpf_config: &BpfConfig, + running: watch::Receiver, + metrics: EventCounter, ) -> anyhow::Result<(Self, mpsc::Receiver)> { Bpf::bump_memlock_rlimit()?; @@ -71,6 +76,8 @@ impl Bpf { paths_config, paths_globset: GlobSet::empty(), links: Vec::new(), + running, + metrics, }; bpf.load_progs(&btf)?; @@ -223,12 +230,7 @@ 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, - ) { + pub fn start(mut self, task_set: &mut JoinSet>) { info!("Starting BPF worker..."); task_set.spawn(async move { @@ -251,19 +253,19 @@ impl Bpf { // decision there. if !event.is_monitored_by_parent() && event.is_ignored(&self.paths_globset) { - event_counter.dropped(); + self.metrics.dropped(); continue; } event }, Err(e) => { error!("Failed to parse event: '{e}'"); - event_counter.dropped(); + self.metrics.dropped(); continue; } }; - event_counter.added(); + self.metrics.added(); if self.tx.send(event).await.is_err() { info!("No BPF consumers left, stopping..."); break; @@ -274,8 +276,8 @@ impl Bpf { _ = self.paths_config.changed() => { self.load_paths().context("Failed to load paths")?; }, - _ = running.changed() => { - if !*running.borrow() { + _ = self.running.changed() => { + if !*self.running.borrow() { info!("Stopping BPF worker..."); break; } @@ -299,7 +301,7 @@ mod bpf_tests { config::{FactConfig, reloader::Reloader}, event::{EventTestData, process::Process}, host_info, - metrics::exporter::Exporter, + metrics::Metrics, }; use super::*; @@ -319,14 +321,18 @@ mod bpf_tests { let mut config = FactConfig::default(); config.set_paths(paths); let reloader = Reloader::from(config); - let (mut bpf, mut rx) = - Bpf::new(reloader.paths(), &reloader.config().bpf).expect("Failed to load BPF code"); + let metrics = Metrics::new(); let (run_tx, run_rx) = watch::channel(true); - // Create a metrics exporter, but don't start it - let exporter = Exporter::new(Some(bpf.take_metrics().unwrap())); + let (bpf, mut rx) = Bpf::new( + reloader.paths(), + &reloader.config().bpf, + run_rx, + metrics.bpf_worker.clone(), + ) + .expect("Failed to load BPF code"); let mut task_set = JoinSet::new(); - bpf.start(&mut task_set, run_rx, exporter.metrics.bpf_worker.clone()); + bpf.start(&mut task_set); tokio::time::sleep(Duration::from_millis(500)).await; diff --git a/fact/src/lib.rs b/fact/src/lib.rs index 29103e12..463c0c8f 100644 --- a/fact/src/lib.rs +++ b/fact/src/lib.rs @@ -29,7 +29,10 @@ mod replay; use config::FactConfig; use pre_flight::pre_flight; -use crate::event::Event; +use crate::{ + event::Event, + metrics::{Metrics, kernel_metrics::KernelMetrics}, +}; pub fn init_log() -> anyhow::Result<()> { let log_level = std::env::var("FACT_LOGLEVEL").unwrap_or("info".to_owned()); @@ -95,30 +98,32 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { let reloader = config::reloader::Reloader::from(config); let config_trigger = reloader.get_trigger(); let mut task_set = JoinSet::new(); + let metrics_userspace = Metrics::new(); - let (exporter, rx) = setup_input(&mut task_set, &reloader, running_pipeline_rx)?; + let (metrics_kernelspace, rx) = setup_input( + &mut task_set, + &reloader, + &metrics_userspace, + running_pipeline_rx, + )?; let (rate_limiter, rx) = RateLimiter::new( rx, reloader.rate_limit(), - exporter.metrics.rate_limiter.clone(), + metrics_userspace.rate_limiter.clone(), )?; output::start( &mut task_set, rx, - exporter.metrics.output.clone(), + metrics_userspace.output.clone(), reloader.grpc(), reloader.otel(), reloader.config().json(), ); rate_limiter.start(&mut task_set); - endpoints::Server::new( - exporter.clone(), - reloader.endpoint(), - running_helpers.subscribe(), - ) - .start(); + let exporter = Exporter::new(&metrics_userspace, metrics_kernelspace); + endpoints::Server::new(exporter, reloader.endpoint(), running_helpers.subscribe()).start(); reloader.start(running_helpers.subscribe()); let mut sigterm = signal(SignalKind::terminate())?; @@ -155,12 +160,13 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { fn setup_input( task_set: &mut JoinSet>, reloader: &config::reloader::Reloader, + metrics: &Metrics, running: watch::Receiver, -) -> anyhow::Result<(Exporter, mpsc::Receiver)> { +) -> anyhow::Result<(Option, mpsc::Receiver)> { match reloader.config().replay() { Some(replay_file) => { let rx = replay::start(task_set, replay_file, running)?; - Ok((Exporter::new(None), rx)) + Ok((None, rx)) } None => { if !reloader.config().skip_pre_flight() { @@ -170,7 +176,7 @@ fn setup_input( debug!("Skipping pre-flight checks"); } - bpf_input(task_set, reloader, running) + bpf_input(task_set, reloader, running, metrics) } } } @@ -179,19 +185,25 @@ 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()?)); + metrics_userspace: &Metrics, +) -> anyhow::Result<(Option, mpsc::Receiver)> { + let (mut bpf, rx) = Bpf::new( + reloader.paths(), + &reloader.config().bpf, + running.clone(), + metrics_userspace.bpf_worker.clone(), + )?; + let metrics_kernelspace = KernelMetrics::new(bpf.take_metrics()?); let (host_scanner, rx) = HostScanner::new( &mut bpf, rx, reloader.paths(), reloader.scan_interval(), - exporter.metrics.host_scanner.clone(), + metrics_userspace.host_scanner.clone(), )?; - bpf.start(task_set, running, exporter.metrics.bpf_worker.clone()); + bpf.start(task_set); host_scanner.start(task_set); - Ok((exporter, rx)) + Ok((Some(metrics_kernelspace), rx)) } diff --git a/fact/src/metrics/exporter.rs b/fact/src/metrics/exporter.rs index 25ec66b0..a2abdcc5 100644 --- a/fact/src/metrics/exporter.rs +++ b/fact/src/metrics/exporter.rs @@ -1,30 +1,27 @@ use std::sync::Arc; -use aya::maps::{MapData, PerCpuArray}; use log::warn; use prometheus_client::{encoding::text::encode, registry::Registry}; -use fact_ebpf::metrics_t; - use super::{Metrics, kernel_metrics::KernelMetrics}; #[derive(Clone)] pub struct Exporter { registry: Arc, - pub metrics: Arc, kernel_metrics: Option>, } impl Exporter { - pub fn new(kernel_metrics: Option>) -> Self { + pub fn new(metrics_user: &Metrics, metrics_kernel: Option) -> Self { let mut registry = Registry::with_prefix("stackrox_fact"); - let metrics = Arc::new(Metrics::new(&mut registry)); - let kernel_metrics = - kernel_metrics.map(|km| Arc::new(KernelMetrics::new(&mut registry, km))); + metrics_user.register(&mut registry); + if let Some(metrics_kernel) = &metrics_kernel { + metrics_kernel.register(&mut registry); + } + let kernel_metrics = metrics_kernel.map(Arc::new); let registry = Arc::new(registry); Exporter { registry, - metrics, kernel_metrics, } } diff --git a/fact/src/metrics/kernel_metrics.rs b/fact/src/metrics/kernel_metrics.rs index d1e77bfb..2039d7d7 100644 --- a/fact/src/metrics/kernel_metrics.rs +++ b/fact/src/metrics/kernel_metrics.rs @@ -15,14 +15,13 @@ macro_rules! define_kernel_metrics { } impl KernelMetrics { - pub fn new(reg: &mut Registry, kernel_metrics: PerCpuArray) -> Self { + pub fn new(kernel_metrics: PerCpuArray) -> Self { $( let $hook = EventCounter::new( concat!("kernel_", stringify!($hook), "_events"), concat!("Events processed by the ", stringify!($hook), " LSM hook"), &[], ); - $hook.register(reg); )+ KernelMetrics { @@ -31,6 +30,10 @@ macro_rules! define_kernel_metrics { } } + pub fn register(&self, reg: &mut Registry) { + $(self.$hook.register(reg);)+ + } + pub fn collect(&self) -> anyhow::Result<()> { let metrics = self .map diff --git a/fact/src/metrics/mod.rs b/fact/src/metrics/mod.rs index af8cdebc..14308b08 100644 --- a/fact/src/metrics/mod.rs +++ b/fact/src/metrics/mod.rs @@ -8,7 +8,7 @@ use host_scanner::HostScannerMetrics; pub mod exporter; pub mod host_scanner; -mod kernel_metrics; +pub mod kernel_metrics; #[derive(Clone, Hash, Eq, Debug, PartialEq, EncodeLabelValue, Copy)] enum LabelValues { @@ -145,7 +145,7 @@ pub struct Metrics { } impl Metrics { - fn new(registry: &mut Registry) -> Self { + pub fn new() -> Self { let bpf_worker = EventCounter::new( "bpf_events", "Events processed by the BPF worker", @@ -155,26 +155,25 @@ impl Metrics { LabelValues::Ignored, ], ); - bpf_worker.register(registry); let rate_limiter = EventCounter::new( "rate_limiter_events", "Events processed by the rate limiter", &[LabelValues::Added, LabelValues::Dropped, LabelValues::Error], ); - rate_limiter.register(registry); - - let output_metrics = OutputMetrics::new(); - output_metrics.register(registry); - - let host_scanner = HostScannerMetrics::new(); - host_scanner.register(registry); Metrics { bpf_worker, rate_limiter, - output: output_metrics, - host_scanner, + output: OutputMetrics::new(), + host_scanner: HostScannerMetrics::new(), } } + + fn register(&self, reg: &mut Registry) { + self.bpf_worker.register(reg); + self.rate_limiter.register(reg); + self.output.register(reg); + self.host_scanner.register(reg); + } }