refactor(metrics): untangle metrics from exporter

fact/src/bpf/mod.rs

@@ -36,12 +36,17 @@ pub struct Bpf {
     paths_globset: GlobSet,
 
     links: Vec<LsmLink>,
+
+    running: watch::Receiver<bool>,
+    metrics: EventCounter,
 }
 
 impl Bpf {
     pub fn new(
         paths_config: watch::Receiver<Vec<PathBuf>>,
         bpf_config: &BpfConfig,
+        running: watch::Receiver<bool>,
+        metrics: EventCounter,
     ) -> anyhow::Result<(Self, mpsc::Receiver<Event>)> {
         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<anyhow::Result<()>>,
-        mut running: watch::Receiver<bool>,
-        event_counter: EventCounter,
-    ) {
+    pub fn start(mut self, task_set: &mut JoinSet<anyhow::Result<()>>) {
         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;
 

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<anyhow::Result<()>>,
     reloader: &config::reloader::Reloader,
+    metrics: &Metrics,
     running: watch::Receiver<bool>,
-) -> anyhow::Result<(Exporter, mpsc::Receiver<Event>)> {
+) -> anyhow::Result<(Option<KernelMetrics>, mpsc::Receiver<Event>)> {
     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<anyhow::Result<()>>,
     reloader: &config::reloader::Reloader,
     running: watch::Receiver<bool>,
-) -> anyhow::Result<(Exporter, mpsc::Receiver<Event>)> {
-    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<KernelMetrics>, mpsc::Receiver<Event>)> {
+    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))
 }

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<Registry>,
-    pub metrics: Arc<Metrics>,
     kernel_metrics: Option<Arc<KernelMetrics>>,
 }
 
 impl Exporter {
-    pub fn new(kernel_metrics: Option<PerCpuArray<MapData, metrics_t>>) -> Self {
+    pub fn new(metrics_user: &Metrics, metrics_kernel: Option<KernelMetrics>) -> 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,
         }
     }

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<MapData, metrics_t>) -> Self {
+            pub fn new(kernel_metrics: PerCpuArray<MapData, metrics_t>) -> 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

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