feat(output): add basic opentelemetry output

fact/Cargo.toml

@@ -20,6 +20,9 @@ hyper-util = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
 native-tls = { workspace = true }
+opentelemetry = { version = "0.32.0", optional = true }
+opentelemetry-otlp = { version = "0.32.0", optional = true }
+opentelemetry_sdk = { version = "0.32.1", features = [ "logs"], optional = true }
 openssl = { workspace = true }
 tonic = { workspace = true }
 tokio = { workspace = true }
@@ -51,3 +54,4 @@ path = "src/main.rs"
 
 [features]
 bpf-test = []
+otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-otlp"]

fact/src/config/mod.rs

@@ -34,6 +34,7 @@ fn yaml_to_duration_secs(v: &Yaml) -> Option<Duration> {
 pub struct FactConfig {
     paths: Option<Vec<PathBuf>>,
     pub grpc: GrpcConfig,
+    pub otel: OTelConfig,
     pub endpoint: EndpointConfig,
     pub bpf: BpfConfig,
     skip_pre_flight: Option<bool>,
@@ -72,7 +73,7 @@ impl FactConfig {
             )?;
 
         // Once file configuration is handled, apply CLI arguments
-        static CLI_ARGS: LazyLock<FactConfig> = LazyLock::new(|| FactCli::parse().to_config());
+        static CLI_ARGS: LazyLock<FactConfig> = LazyLock::new(|| FactCli::parse().into_config());
         config.update(&CLI_ARGS);
 
         Ok(config)
@@ -84,6 +85,7 @@ impl FactConfig {
         }
 
         self.grpc.update(&from.grpc);
+        self.otel.update(&from.otel);
         self.endpoint.update(&from.endpoint);
         self.bpf.update(&from.bpf);
 
@@ -196,6 +198,10 @@ impl TryFrom<Vec<Yaml>> for FactConfig {
                     let grpc = v.as_hash().unwrap();
                     config.grpc = GrpcConfig::try_from(grpc)?;
                 }
+                "otel" if v.is_hash() => {
+                    let otel = v.as_hash().unwrap();
+                    config.otel = OTelConfig::try_from(otel)?;
+                }
                 "endpoint" if v.is_hash() => {
                     let endpoint = v.as_hash().unwrap();
                     config.endpoint = EndpointConfig::try_from(endpoint)?;
@@ -492,6 +498,48 @@ impl TryFrom<&yaml::Hash> for GrpcConfig {
     }
 }
 
+#[derive(Debug, Default, PartialEq, Eq, Clone)]
+pub struct OTelConfig {
+    endpoint: Option<String>,
+}
+
+impl OTelConfig {
+    fn update(&mut self, from: &OTelConfig) {
+        if let Some(endpoint) = from.endpoint.as_deref() {
+            self.endpoint = Some(endpoint.to_owned());
+        }
+    }
+
+    pub fn endpoint(&self) -> Option<&str> {
+        self.endpoint.as_deref()
+    }
+}
+
+impl TryFrom<&yaml::Hash> for OTelConfig {
+    type Error = anyhow::Error;
+
+    fn try_from(value: &yaml::Hash) -> Result<Self, Self::Error> {
+        let mut otel = OTelConfig::default();
+        for (k, v) in value.iter() {
+            let Some(k) = k.as_str() else {
+                bail!("key is not string: {k:?}");
+            };
+
+            match k {
+                "endpoint" => {
+                    let Some(endpoint) = v.as_str() else {
+                        bail!("otel.endpoint field has incorrect type: {v:?}")
+                    };
+                    otel.endpoint = Some(endpoint.to_owned());
+                }
+                name => bail!("Invalid field 'otel.{name}' with value: {v:?}"),
+            }
+        }
+
+        Ok(otel)
+    }
+}
+
 #[derive(Debug, Default, PartialEq, Eq, Clone)]
 pub struct BpfConfig {
     ringbuf_size: Option<u32>,
@@ -627,6 +675,10 @@ pub struct FactCli {
     #[arg(long, env = "FACT_GRPC_BACKOFF_RETRIES_MAX")]
     backoff_retries_max: Option<u64>,
 
+    /// OpenTelemetry endpoint to push logs into
+    #[arg(long, env = "FACT_OTEL_ENDPOINT")]
+    otel_endpoint: Option<String>,
+
     /// The port to bind for all exposed endpoints
     #[arg(long, short, env = "FACT_ENDPOINT_ADDRESS")]
     address: Option<SocketAddr>,
@@ -711,12 +763,12 @@ pub struct FactCli {
 }
 
 impl FactCli {
-    fn to_config(&self) -> FactConfig {
+    fn into_config(self) -> FactConfig {
         FactConfig {
-            paths: self.paths.clone(),
+            paths: self.paths,
             grpc: GrpcConfig {
-                url: self.url.clone(),
-                certs: self.certs.clone(),
+                url: self.url,
+                certs: self.certs,
                 backoff: BackoffConfig {
                     initial: self.backoff_initial,
                     max: self.backoff_max,
@@ -725,6 +777,9 @@ impl FactCli {
                     retries_max: self.backoff_retries_max,
                 },
             },
+            otel: OTelConfig {
+                endpoint: self.otel_endpoint,
+            },
             endpoint: EndpointConfig {
                 address: self.address,
                 expose_metrics: resolve_bool_arg(self.expose_metrics, self.no_expose_metrics),

fact/src/config/reloader.rs

@@ -9,12 +9,15 @@ use tokio::{
     time::interval,
 };
 
+use crate::config::OTelConfig;
+
 use super::{CONFIG_FILES, EndpointConfig, FactConfig, GrpcConfig};
 
 pub struct Reloader {
     config: FactConfig,
     endpoint: watch::Sender<EndpointConfig>,
     grpc: watch::Sender<GrpcConfig>,
+    otel: watch::Sender<OTelConfig>,
     paths: watch::Sender<Vec<PathBuf>>,
     files: HashMap<&'static str, i64>,
     scan_interval: watch::Sender<Duration>,
@@ -71,6 +74,12 @@ impl Reloader {
         self.grpc.subscribe()
     }
 
+    /// Subscribe to get notifications when otel configuration is
+    /// changed.
+    pub fn otel(&self) -> watch::Receiver<OTelConfig> {
+        self.otel.subscribe()
+    }
+
     /// Subscribe to get notifications when paths configuration is
     /// changed.
     pub fn paths(&self) -> watch::Receiver<Vec<PathBuf>> {
@@ -174,6 +183,16 @@ impl Reloader {
             }
         });
 
+        self.otel.send_if_modified(|old| {
+            if *old != new.otel {
+                debug!("Sending new OTel configuration...");
+                *old = new.otel.clone();
+                true
+            } else {
+                false
+            }
+        });
+
         self.paths.send_if_modified(|old| {
             let new = new.paths();
             if *old != new {
@@ -238,6 +257,7 @@ impl From<FactConfig> for Reloader {
             .collect();
         let (endpoint, _) = watch::channel(config.endpoint.clone());
         let (grpc, _) = watch::channel(config.grpc.clone());
+        let (otel, _) = watch::channel(config.otel.clone());
         let (paths, _) = watch::channel(config.paths().to_vec());
         let (scan_interval, _) = watch::channel(config.scan_interval());
         let (rate_limit, _) = watch::channel(config.rate_limit());
@@ -247,6 +267,7 @@ impl From<FactConfig> for Reloader {
             config,
             endpoint,
             grpc,
+            otel,
             paths,
             scan_interval,
             rate_limit,