From b957e4eb7a74a906192f09173aee6552fd45e621 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Thu, 23 Jul 2026 13:39:11 +0200 Subject: [PATCH 1/2] feat(endpoints): add inodes introspection endpoint Adds an "/inodes" endpoint that returns the contents of the userspace inode map for inspection. This will be useful in tests that need to look into the internal state after operations that don't emit events or during development to validate what is currently being tracked. The new endpoint is disabled by default and can be enabled with the new `endpoint.introspection` configuration value, that should gate any future introspection endpoints we decide to add in the future. This configuration value is not hotreloadable, it can only be set at start up, this is meant to make it harder for the endpoint to come online on production environments. --- fact/src/config/mod.rs | 20 ++++++++++ fact/src/config/reloader/tests.rs | 2 + fact/src/config/tests.rs | 17 +++++++++ fact/src/endpoints.rs | 45 +++++++++++++++++++---- fact/src/host_scanner.rs | 61 +++++++++++++++++++++++++++++-- fact/src/lib.rs | 17 +++++++-- 6 files changed, 148 insertions(+), 14 deletions(-) diff --git a/fact/src/config/mod.rs b/fact/src/config/mod.rs index ab8e9d6a..229bd0f9 100644 --- a/fact/src/config/mod.rs +++ b/fact/src/config/mod.rs @@ -276,6 +276,7 @@ pub struct EndpointConfig { address: Option, expose_metrics: Option, health_check: Option, + introspection: Option, } impl EndpointConfig { @@ -291,6 +292,10 @@ impl EndpointConfig { if let Some(health_check) = from.health_check { self.health_check = Some(health_check); } + + if let Some(introspection) = from.introspection { + self.introspection = Some(introspection); + } } pub fn address(&self) -> SocketAddr { @@ -305,6 +310,10 @@ impl EndpointConfig { pub fn health_check(&self) -> bool { self.health_check.unwrap_or(false) } + + pub fn introspection(&self) -> bool { + self.introspection.unwrap_or(false) + } } impl TryFrom<&yaml::Hash> for EndpointConfig { @@ -792,6 +801,16 @@ pub struct FactCli { #[arg(long, overrides_with = "health_check", hide(true))] no_health_check: bool, + /// Whether the introspection endpoints should be enabled + #[arg( + long, + overrides_with("no_introspection"), + env = "FACT_ENDPOINT_INTROSPECTION" + )] + introspection: bool, + #[arg(long, overrides_with = "introspection", hide(true))] + no_introspection: bool, + /// Whether to perform a pre flight check #[arg( long, @@ -880,6 +899,7 @@ impl FactCli { address: self.address, expose_metrics: resolve_bool_arg(self.expose_metrics, self.no_expose_metrics), health_check: resolve_bool_arg(self.health_check, self.no_health_check), + introspection: resolve_bool_arg(self.introspection, self.no_introspection), }, bpf: BpfConfig { ringbuf_size: self.ringbuf_size, diff --git a/fact/src/config/reloader/tests.rs b/fact/src/config/reloader/tests.rs index c7d67a23..a5e128b5 100644 --- a/fact/src/config/reloader/tests.rs +++ b/fact/src/config/reloader/tests.rs @@ -255,6 +255,7 @@ generate_endpoint_test! { address: Some(ENDPOINT_ADDRESS_DEFAULT), expose_metrics: Some(true), health_check: Some(true), + introspection: Some(true), }, ..Default::default() }, @@ -263,6 +264,7 @@ generate_endpoint_test! { address: Some(ENDPOINT_ADDRESS_DEFAULT), expose_metrics: Some(true), health_check: Some(true), + introspection: Some(true), }, paths: Some(vec!["/etc".into()]), ..Default::default() diff --git a/fact/src/config/tests.rs b/fact/src/config/tests.rs index f00bae2c..14adda96 100644 --- a/fact/src/config/tests.rs +++ b/fact/src/config/tests.rs @@ -514,6 +514,7 @@ fn parsing() { address: Some(SocketAddr::from(([0, 0, 0, 0], 8080))), expose_metrics: Some(true), health_check: Some(true), + introspection: None, }, skip_pre_flight: Some(false), json: Some(false), @@ -1863,6 +1864,7 @@ fn update() { address: Some(SocketAddr::from(([0, 0, 0, 0], 9000))), expose_metrics: Some(false), health_check: Some(false), + introspection: None, }, skip_pre_flight: Some(true), json: Some(true), @@ -1901,6 +1903,7 @@ fn update() { address: Some(SocketAddr::from(([127, 0, 0, 1], 8080))), expose_metrics: Some(true), health_check: Some(true), + introspection: None, }, skip_pre_flight: Some(false), json: Some(false), @@ -1952,6 +1955,7 @@ fn defaults() { ); assert!(!config.endpoint.expose_metrics()); assert!(!config.endpoint.health_check()); + assert!(!config.endpoint.introspection()); assert!(!config.skip_pre_flight()); assert!(!config.json()); assert_eq!(config.bpf.ringbuf_size(), 8192); @@ -2313,6 +2317,19 @@ fn env_vars() { ..Default::default() }, ), + ( + EnvVar { + name: "FACT_ENDPOINT_INTROSPECTION", + value: "true", + }, + FactConfig { + endpoint: EndpointConfig { + introspection: Some(true), + ..Default::default() + }, + ..Default::default() + }, + ), ( EnvVar { name: "FACT_SKIP_PRE_FLIGHT", diff --git a/fact/src/endpoints.rs b/fact/src/endpoints.rs index cc97415a..947076b6 100644 --- a/fact/src/endpoints.rs +++ b/fact/src/endpoints.rs @@ -9,7 +9,11 @@ use hyper::{ }; use hyper_util::rt::TokioIo; use log::{info, warn}; -use tokio::{net::TcpListener, sync::watch, task::JoinHandle}; +use tokio::{ + net::TcpListener, + sync::{mpsc, oneshot, watch}, + task::JoinHandle, +}; use crate::{config::EndpointConfig, metrics::exporter::Exporter}; @@ -18,6 +22,8 @@ pub struct Server { metrics: Exporter, config: watch::Receiver, running: watch::Receiver, + + host_scanner_intro: mpsc::Sender>>, } impl Server { @@ -25,11 +31,13 @@ impl Server { metrics: Exporter, config: watch::Receiver, running: watch::Receiver, + host_scanner_intro: mpsc::Sender>>, ) -> Self { Server { metrics, config, running, + host_scanner_intro, } } @@ -95,7 +103,7 @@ impl Server { /// Check if there are active endpoints to serve. fn is_active(&self) -> bool { let config = self.config.borrow(); - config.health_check() || config.expose_metrics() + config.health_check() || config.expose_metrics() || config.introspection() } fn health_check_is_active(&self) -> bool { @@ -108,17 +116,17 @@ impl Server { fn make_response( res: StatusCode, - body: String, + body: impl Into, ) -> Result>, anyhow::Error> { Ok(Response::builder() .status(res) - .body(Full::new(Bytes::from(body))) + .body(Full::new(body.into())) .unwrap()) } fn handle_metrics(&self) -> Result>, anyhow::Error> { if !self.metrics_is_active() { - return Server::make_response(StatusCode::SERVICE_UNAVAILABLE, String::new()); + return Server::make_response(StatusCode::SERVICE_UNAVAILABLE, ""); } self.metrics.encode().map(|buf| { @@ -139,7 +147,29 @@ impl Server { } else { StatusCode::SERVICE_UNAVAILABLE }; - Server::make_response(res, String::new()) + Server::make_response(res, "") + } + + async fn handle_inodes(&self) -> anyhow::Result>> { + if !self.config.borrow().introspection() { + return Server::make_response(StatusCode::SERVICE_UNAVAILABLE, ""); + } + + let (tx, rx) = oneshot::channel(); + if let Err(e) = self.host_scanner_intro.send(tx).await { + return Server::make_response(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()); + } + match rx.await { + Ok(Ok(b)) => Response::builder() + .header( + hyper::header::CONTENT_TYPE, + "application/json; charset=utf-8", + ) + .body(Full::new(Bytes::from(b))) + .map_err(anyhow::Error::new), + Ok(Err(e)) => Server::make_response(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()), + Err(e) => Server::make_response(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()), + } } } @@ -154,7 +184,8 @@ impl Service> for Server { match (req.method(), req.uri().path()) { (&Method::GET, "/metrics") => s.handle_metrics(), (&Method::GET, "/health_check") => s.handle_health_check(), - _ => Server::make_response(StatusCode::NOT_FOUND, String::new()), + (&Method::GET, "/inodes") => s.handle_inodes().await, + _ => Server::make_response(StatusCode::NOT_FOUND, ""), } }) } diff --git a/fact/src/host_scanner.rs b/fact/src/host_scanner.rs index 9e46eef1..04dff52a 100644 --- a/fact/src/host_scanner.rs +++ b/fact/src/host_scanner.rs @@ -20,7 +20,9 @@ use std::{ cell::RefCell, + collections::HashMap, io, + ops::{Deref, DerefMut}, os::linux::fs::MetadataExt, path::{Path, PathBuf}, sync::Arc, @@ -35,8 +37,9 @@ use aya::{ use fact_ebpf::{inode_key_t, inode_value_t, monitored_t}; use globset::{Glob, GlobSet, GlobSetBuilder}; use log::{debug, info, warn}; +use serde::{Serialize, ser::SerializeMap}; use tokio::{ - sync::{Notify, mpsc, watch}, + sync::{Notify, mpsc, oneshot, watch}, task::JoinSet, }; @@ -47,15 +50,55 @@ use crate::{ metrics::host_scanner::{HostScannerMetrics, ScanLabels}, }; +struct InodeMap(HashMap); + +impl InodeMap { + fn new() -> Self { + InodeMap(HashMap::new()) + } +} + +impl Deref for InodeMap { + type Target = HashMap; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl DerefMut for InodeMap { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 + } +} + +impl Serialize for InodeMap { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + let mut map = serializer.serialize_map(Some(self.len()))?; + for (k, v) in &self.0 { + // In order to be able to serialize InodeMap to JSON, we use + // a string key. This enables us to send this type over HTTP + // as part of the "/inodes" introspection endpoint, while + // keeping the existing inode_key_t Serialize implementation. + map.serialize_entry(&format!("{}:{}", k.dev, k.inode), v)?; + } + map.end() + } +} + pub struct HostScanner { kernel_inode_map: RefCell>, - inode_map: RefCell>, + inode_map: RefCell, paths: watch::Receiver>, scan_interval: watch::Receiver, rx: mpsc::Receiver, tx: mpsc::Sender, + introspection: mpsc::Receiver>>, metrics: HostScannerMetrics, @@ -69,9 +112,10 @@ impl HostScanner { paths: watch::Receiver>, scan_interval: watch::Receiver, metrics: HostScannerMetrics, + introspection: mpsc::Receiver>>, ) -> anyhow::Result<(Self, mpsc::Receiver)> { let kernel_inode_map = RefCell::new(bpf.take_inode_map()?); - let inode_map = RefCell::new(std::collections::HashMap::new()); + let inode_map = RefCell::new(InodeMap::new()); let (tx, output) = mpsc::channel(100); let paths_globset = HostScanner::build_globset(paths.borrow().as_slice())?; @@ -82,6 +126,7 @@ impl HostScanner { scan_interval, rx, tx, + introspection, metrics, paths_globset, }; @@ -473,6 +518,16 @@ You can increase this limit with: warn!("Failed to send event: {e}"); } }, + req = self.introspection.recv() => { + let Some(req) = req else { + continue; + }; + + let resp = serde_json::to_string(&*self.inode_map.borrow()); + if let Err(e) = req.send(resp) { + warn!("Failed to reply introspection query: {e:?}"); + } + } _ = scan_trigger.notified() => self.scan()?, _ = self.paths.changed() => { self.paths_globset = HostScanner::build_globset(self.paths.borrow().as_slice())?; diff --git a/fact/src/lib.rs b/fact/src/lib.rs index 08e02846..0969615b 100644 --- a/fact/src/lib.rs +++ b/fact/src/lib.rs @@ -9,7 +9,7 @@ use metrics::exporter::Exporter; use rate_limiter::RateLimiter; use tokio::{ signal::unix::{SignalKind, signal}, - sync::{mpsc, watch}, + sync::{mpsc, oneshot, watch}, task::JoinSet, time::timeout, }; @@ -102,6 +102,7 @@ struct SetupArgs<'a> { // BPF mode bpf_config: BpfConfig, + host_scanner_intro: mpsc::Receiver>>, } pub async fn run(config: FactConfig) -> anyhow::Result<()> { @@ -110,6 +111,7 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { log_system_information(); let (running_pipeline_tx, running_pipeline_rx) = watch::channel(true); let (running_helpers, _) = watch::channel(true); + let (host_scanner_intro_tx, host_scanner_intro_rx) = mpsc::channel(10); let stdout_enabled = config.json(); let skip_pre_flight = config.skip_pre_flight(); @@ -119,6 +121,7 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { let metrics_userspace = Metrics::new(); let mut task_set = JoinSet::new(); let reloader = config::reloader::Reloader::from(config); + let config_trigger = reloader.get_trigger(); let setup_args = SetupArgs { skip_pre_flight, @@ -128,10 +131,9 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { metrics: &metrics_userspace, replay, bpf_config, + host_scanner_intro: host_scanner_intro_rx, }; - let config_trigger = reloader.get_trigger(); - let (metrics_kernelspace, rx) = setup_input(setup_args)?; let (rate_limiter, rx) = RateLimiter::new( rx, @@ -150,7 +152,13 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { rate_limiter.start(&mut task_set); let exporter = Exporter::new(&metrics_userspace, metrics_kernelspace); - endpoints::Server::new(exporter, reloader.endpoint(), running_helpers.subscribe()).start(); + endpoints::Server::new( + exporter, + reloader.endpoint(), + running_helpers.subscribe(), + host_scanner_intro_tx, + ) + .start(); reloader.start(running_helpers.subscribe()); let mut sigterm = signal(SignalKind::terminate())?; @@ -218,6 +226,7 @@ fn bpf_input(args: SetupArgs) -> anyhow::Result<(Option, mpsc::Re args.reloader.paths(), args.reloader.scan_interval(), args.metrics.host_scanner.clone(), + args.host_scanner_intro, )?; bpf.start(args.task_set); From 91987cbf48ccfa1385e9a0b6338c86cbb3e91bbf Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Thu, 23 Jul 2026 18:29:03 +0200 Subject: [PATCH 2/2] chore: add CHANGELOG line --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a75a9bc1..9807023f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ possible include a PR number for easier tracking. ## Next +* feat(endpoints): add inodes introspection endpoint (#1273) * feat: add --replay mode for JSONL event replay without eBPF (#1010) * feat(config): configurable loaded BPF programs (#1086) * feat(output): add basic opentelemetry output (#971)