feat(endpoints): add inodes introspection endpoint

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)

fact/src/config/mod.rs

@@ -276,6 +276,7 @@ pub struct EndpointConfig {
     address: Option<SocketAddr>,
     expose_metrics: Option<bool>,
     health_check: Option<bool>,
+    introspection: Option<bool>,
 }
 
 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,

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()

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",

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,18 +22,22 @@ pub struct Server {
     metrics: Exporter,
     config: watch::Receiver<EndpointConfig>,
     running: watch::Receiver<bool>,
+
+    host_scanner_intro: mpsc::Sender<oneshot::Sender<serde_json::Result<String>>>,
 }
 
 impl Server {
     pub fn new(
         metrics: Exporter,
         config: watch::Receiver<EndpointConfig>,
         running: watch::Receiver<bool>,
+        host_scanner_intro: mpsc::Sender<oneshot::Sender<serde_json::Result<String>>>,
     ) -> 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<Bytes>,
     ) -> Result<Response<Full<Bytes>>, anyhow::Error> {
         Ok(Response::builder()
             .status(res)
-            .body(Full::new(Bytes::from(body)))
+            .body(Full::new(body.into()))
             .unwrap())
     }
 
     fn handle_metrics(&self) -> Result<Response<Full<Bytes>>, 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<Response<Full<Bytes>>> {
+        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<Request<Incoming>> 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, ""),
             }
         })
     }

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<inode_key_t, PathBuf>);
+
+impl InodeMap {
+    fn new() -> Self {
+        InodeMap(HashMap::new())
+    }
+}
+
+impl Deref for InodeMap {
+    type Target = HashMap<inode_key_t, PathBuf>;
+
+    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<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    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<aya::maps::HashMap<MapData, inode_key_t, inode_value_t>>,
-    inode_map: RefCell<std::collections::HashMap<inode_key_t, PathBuf>>,
+    inode_map: RefCell<InodeMap>,
 
     paths: watch::Receiver<Vec<PathBuf>>,
     scan_interval: watch::Receiver<Duration>,
 
     rx: mpsc::Receiver<Event>,
     tx: mpsc::Sender<Event>,
+    introspection: mpsc::Receiver<oneshot::Sender<serde_json::Result<String>>>,
 
     metrics: HostScannerMetrics,
 
@@ -69,9 +112,10 @@ impl HostScanner {
         paths: watch::Receiver<Vec<PathBuf>>,
         scan_interval: watch::Receiver<Duration>,
         metrics: HostScannerMetrics,
+        introspection: mpsc::Receiver<oneshot::Sender<serde_json::Result<String>>>,
     ) -> anyhow::Result<(Self, mpsc::Receiver<Event>)> {
         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())?;