ROX-30836: implement hotreloading configuration for outputs (#119)

* ROX-30836: implement hotreloading configuration for outputs

With this patch, it is possible to reload the configuration used for
gRPC communication. Allowing changes to the certificate directory or
output URL to be done without a full restart being required.

The output code has also been slightly refactored, the module now has a
single start function that will spawn a task for gRPC output and another
for JSON to stdout (if this is required at start up or the gRPC output
is not configured). Both the gRPC and stdout code now live as submodules
of the output module. The stdout output is meant mostly for debugging,
so hotreloading was not implemented for it. Lastly, there used to be a
task that would receive `Arc<Event>` messages from the BPF worker,
translate them to the `FileActivity` type and forward them out a
separate channel that ultimately was the gRPC stream, this whole
workflow has now been replaced by a filter_map operation that does the
translating on the gRPC stream directly, leading to more concise code.

* Implement integration test for gRPC hotreloading

* Rename grpc::Client::is_active to is_enabled

Author: Molter73

fact/src/config/mod.rs

@@ -25,8 +25,7 @@ const CONFIG_FILES: [&str; 4] = [
 #[derive(Debug, Default, PartialEq, Eq, Clone)]
 pub struct FactConfig {
     paths: Option<Vec<PathBuf>>,
-    url: Option<String>,
-    certs: Option<PathBuf>,
+    pub grpc: GrpcConfig,
     pub endpoint: EndpointConfig,
     skip_pre_flight: Option<bool>,
     json: Option<bool>,
@@ -78,14 +77,7 @@ impl FactConfig {
             self.paths = Some(paths.to_owned());
         }
 
-        if let Some(url) = from.url.as_deref() {
-            self.url = Some(url.to_owned());
-        }
-
-        if let Some(certs) = from.certs.as_deref() {
-            self.certs = Some(certs.to_owned());
-        }
-
+        self.grpc.update(&from.grpc);
         self.endpoint.update(&from.endpoint);
 
         if let Some(skip_pre_flight) = from.skip_pre_flight {
@@ -109,14 +101,6 @@ impl FactConfig {
         self.paths.as_ref().map(|v| v.as_ref()).unwrap_or(&[])
     }
 
-    pub fn url(&self) -> Option<&str> {
-        self.url.as_deref()
-    }
-
-    pub fn certs(&self) -> Option<&Path> {
-        self.certs.as_deref()
-    }
-
     pub fn skip_pre_flight(&self) -> bool {
         self.skip_pre_flight.unwrap_or(false)
     }
@@ -193,17 +177,9 @@ impl TryFrom<Vec<Yaml>> for FactConfig {
                 "paths" if v.is_null() => {
                     config.paths = Some(Vec::new());
                 }
-                "url" => {
-                    let Some(url) = v.as_str() else {
-                        bail!("url field has incorrect type: {v:?}");
-                    };
-                    config.url = Some(url.to_owned());
-                }
-                "certs" => {
-                    let Some(certs) = v.as_str() else {
-                        bail!("certs field has incorrect type: {v:?}");
-                    };
-                    config.certs = Some(PathBuf::from(certs));
+                "grpc" if v.is_hash() => {
+                    let grpc = v.as_hash().unwrap();
+                    config.grpc = GrpcConfig::try_from(grpc)?;
                 }
                 "endpoint" if v.is_hash() => {
                     let endpoint = v.as_hash().unwrap();
@@ -325,6 +301,63 @@ impl TryFrom<&yaml::Hash> for EndpointConfig {
     }
 }
 
+#[derive(Debug, Default, PartialEq, Eq, Clone)]
+pub struct GrpcConfig {
+    url: Option<String>,
+    certs: Option<PathBuf>,
+}
+
+impl GrpcConfig {
+    fn update(&mut self, from: &GrpcConfig) {
+        if let Some(url) = from.url.as_deref() {
+            self.url = Some(url.to_owned());
+        }
+
+        if let Some(certs) = from.certs.as_deref() {
+            self.certs = Some(certs.to_owned());
+        }
+    }
+
+    pub fn url(&self) -> Option<&str> {
+        self.url.as_deref()
+    }
+
+    pub fn certs(&self) -> Option<&Path> {
+        self.certs.as_deref()
+    }
+}
+
+impl TryFrom<&yaml::Hash> for GrpcConfig {
+    type Error = anyhow::Error;
+
+    fn try_from(value: &yaml::Hash) -> Result<Self, Self::Error> {
+        let mut grpc = GrpcConfig::default();
+        for (k, v) in value.iter() {
+            let Some(k) = k.as_str() else {
+                bail!("key is not string: {k:?}");
+            };
+
+            match k {
+                "url" => {
+                    let Some(url) = v.as_str() else {
+                        bail!("url field has incorrect type: {v:?}");
+                    };
+                    grpc.url = Some(url.to_owned());
+                }
+                "certs" => {
+                    let Some(certs) = v.as_str() else {
+                        bail!("certs field has incorrect type: {v:?}");
+                    };
+                    grpc.certs = Some(PathBuf::from(certs));
+                }
+                name => bail!("Invalid field 'grpc.{name}' with value: {v:?}"),
+            }
+        }
+
+        Ok(grpc)
+    }
+}
+
 #[derive(Debug, Parser)]
 #[clap(version = crate::version::FACT_VERSION, about)]
 pub struct FactCli {
@@ -402,8 +435,10 @@ impl FactCli {
     fn to_config(&self) -> FactConfig {
         FactConfig {
             paths: self.paths.clone(),
-            url: self.url.clone(),
-            certs: self.certs.clone(),
+            grpc: GrpcConfig {
+                url: self.url.clone(),
+                certs: self.certs.clone(),
+            },
             endpoint: EndpointConfig {
                 address: self.address,
                 expose_metrics: resolve_bool_arg(self.expose_metrics, self.no_expose_metrics),

fact/src/config/reloader.rs

@@ -9,11 +9,12 @@ use tokio::{
     time::interval,
 };
 
-use super::{EndpointConfig, FactConfig, CONFIG_FILES};
+use super::{EndpointConfig, FactConfig, GrpcConfig, CONFIG_FILES};
 
 pub struct Reloader {
     config: FactConfig,
     endpoint: watch::Sender<EndpointConfig>,
+    grpc: watch::Sender<GrpcConfig>,
     files: HashMap<&'static str, i64>,
     trigger: Arc<Notify>,
 }
@@ -51,12 +52,22 @@ impl Reloader {
         Some(handle)
     }
 
+    pub fn config(&self) -> &FactConfig {
+        &self.config
+    }
+
     /// Subscribe to get notifications when endpoint configuration is
     /// changed.
     pub fn endpoint(&self) -> watch::Receiver<EndpointConfig> {
         self.endpoint.subscribe()
     }
 
+    /// Subscribe to get notifications when grpc configuration is
+    /// changed.
+    pub fn grpc(&self) -> watch::Receiver<GrpcConfig> {
+        self.grpc.subscribe()
+    }
+
     /// Get a reference to the internal trigger for manual reloading of
     /// configuration.
     ///
@@ -132,6 +143,16 @@ impl Reloader {
             }
         });
 
+        self.grpc.send_if_modified(|old| {
+            if *old != new.grpc {
+                debug!("Sending new gRPC configuration...");
+                *old = new.grpc.clone();
+                true
+            } else {
+                false
+            }
+        });
+
         if self.config.hotreload() != new.hotreload() {
             warn!("Changes to the hotreload field only take effect on startup");
         }
@@ -162,10 +183,13 @@ impl From<FactConfig> for Reloader {
             })
             .collect();
         let (endpoint, _) = watch::channel(config.endpoint.clone());
+        let (grpc, _) = watch::channel(config.grpc.clone());
         let trigger = Arc::new(Notify::new());
+
         Reloader {
             config,
             endpoint,
+            grpc,
             files,
             trigger,
         }