feat(hot-reload): reload paths on the fly (#128)

This patch makes it so detaching and reattaching programs is no longer
needed for updating the list of monitored paths. This is achieved by
first loading all the new paths into the BPF trie map and then removing
the entries that are not part of the new configuration.

An integration test was added to ensure adding new paths does not remove
existing paths accidentally.

Author: Molter73

fact-ebpf/src/lib.rs

@@ -54,6 +54,12 @@ impl From<path_prefix_t> for lpm_trie::Key<[c_char; LPM_SIZE_MAX as usize]> {
     }
 }
 
+impl PartialEq for path_prefix_t {
+    fn eq(&self, other: &Self) -> bool {
+        self.bit_len == other.bit_len && self.path == other.path
+    }
+}
+
 unsafe impl Pod for path_prefix_t {}
 
 impl metrics_by_hook_t {

fact/src/bpf.rs

@@ -3,7 +3,7 @@ use std::{io, path::PathBuf, sync::Arc};
 use anyhow::{bail, Context};
 use aya::{
     maps::{Array, LpmTrie, MapData, PerCpuArray, RingBuf},
-    programs::{lsm::LsmLink, Lsm},
+    programs::Lsm,
     Btf, Ebpf,
 };
 use libc::c_char;
@@ -27,7 +27,6 @@ pub struct Bpf {
 
     paths: Vec<path_prefix_t>,
     paths_config: watch::Receiver<Vec<PathBuf>>,
-    links: Vec<LsmLink>,
 }
 
 impl Bpf {
@@ -44,15 +43,13 @@ impl Bpf {
             .set_max_entries(RINGBUFFER_NAME, ringbuf_size * 1024)
             .load(fact_ebpf::EBPF_OBJ)?;
 
-        let links = Vec::new();
         let paths = Vec::new();
         let (tx, _) = broadcast::channel(100);
         let mut bpf = Bpf {
             obj,
             tx,
             paths,
             paths_config,
-            links,
         };
 
         bpf.load_paths()?;
@@ -113,19 +110,21 @@ impl Bpf {
         let mut path_prefix: LpmTrie<&mut MapData, [c_char; LPM_SIZE_MAX as usize], c_char> =
             LpmTrie::try_from(path_prefix)?;
 
-        // Remove old prefixes
-        for p in &self.paths {
-            path_prefix.remove(&(*p).into())?;
-        }
-        self.paths.clear();
-
-        // Add the new ones
+        // Add the new prefixes
+        let mut new_paths = Vec::with_capacity(paths_config.len());
         for p in paths_config.iter() {
             let prefix = path_prefix_t::try_from(p)?;
             path_prefix.insert(&prefix.into(), 0, 0)?;
-            self.paths.push(prefix);
+            new_paths.push(prefix);
         }
 
+        // Remove old prefixes
+        for p in self.paths.iter().filter(|p| !new_paths.contains(p)) {
+            path_prefix.remove(&(*p).into())?;
+        }
+
+        self.paths = new_paths;
+
         Ok(())
     }
 
@@ -147,16 +146,11 @@ impl Bpf {
     fn attach_progs(&mut self) -> anyhow::Result<()> {
         for (_, prog) in self.obj.programs_mut() {
             let prog: &mut Lsm = prog.try_into()?;
-            let id = prog.attach()?;
-            self.links.push(prog.take_link(id)?);
+            prog.attach()?;
         }
         Ok(())
     }
 
-    fn detach_progs(&mut self) {
-        self.links.clear();
-    }
-
     // Gather events from the ring buffer and print them out.
     pub fn start(
         mut self,
@@ -199,11 +193,7 @@ impl Bpf {
                         guard.clear_ready();
                     },
                     _ = self.paths_config.changed() => {
-                        info!("Reloading path configuration...");
-                        self.detach_progs();
                         self.load_paths().context("Failed to load paths")?;
-                        self.attach_progs().context("Failed to re-attach eBPF programs")?;
-                        info!("Done reloading path configuration");
                     },
                     _ = running.changed() => {
                         if !*running.borrow() {

tests/test_config_hotreload.py

@@ -160,3 +160,45 @@ def test_paths(fact, fact_config, monitored_dir, ignored_dir, server):
     print(f'Ignoring: {ignored_event}')
 
     server.wait_events([e], ignored=[ignored_event])
+
+
+def test_paths_addition(fact, fact_config, monitored_dir, ignored_dir, server):
+    p = Process()
+
+    # Ignored file, must not show up in the server
+    ignored_file = os.path.join(ignored_dir, 'test.txt')
+    with open(ignored_file, 'w') as f:
+        f.write('This is to be ignored')
+
+    ignored_event = Event(
+        process=p, event_type=EventType.CREATION, file=ignored_file)
+    print(f'Ignoring: {ignored_event}')
+
+    # File Under Test
+    fut = os.path.join(monitored_dir, 'test.txt')
+    with open(fut, 'w') as f:
+        f.write('This is a test')
+
+    e = Event(process=p, event_type=EventType.CREATION, file=fut)
+    print(f'Waiting for event: {e}')
+
+    server.wait_events([e], ignored=[ignored_event])
+
+    config, config_file = fact_config
+    config['paths'] = [monitored_dir, ignored_dir]
+    reload_config(fact, config, config_file, delay=0.5)
+
+    # At this point, the event in the ignored directory should show up
+    # alongside the regular event
+    with open(ignored_file, 'w') as f:
+        f.write('This is another test')
+    with open(fut, 'w') as f:
+        f.write('This is one final event')
+
+    events = [
+        Event(process=p, event_type=EventType.OPEN, file=ignored_file),
+        Event(process=p, event_type=EventType.OPEN, file=fut)
+    ]
+    print(f'Waiting for events: {events}')
+
+    server.wait_events(events)