From ebe666c2d63058c47f471330cb638f04fc0f8244 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Wed, 1 Jul 2026 17:48:20 +0200 Subject: [PATCH 1/8] feat(output): add basic opentelemetry output This output allows exporting file activity events as opentelemetry logs via otlp to be collected by compatible systems. TODO: * Add feature flag * Add configuration * Add integration tests * Add docs on how to use this output --- Cargo.lock | 442 ++++++++++++++++++++++++++++++++++++ fact/Cargo.toml | 3 + fact/src/event/mod.rs | 117 ++++++++++ fact/src/event/process.rs | 52 ++++- fact/src/output/mod.rs | 4 + fact/src/output/otel/mod.rs | 59 +++++ 6 files changed, 676 insertions(+), 1 deletion(-) create mode 100644 fact/src/output/otel/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 6930c0bd..91fdc4e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -414,6 +414,17 @@ dependencies = [ "thiserror", ] +[[package]] +name = "displaydoc" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "dtoa" version = "1.0.11" @@ -493,6 +504,9 @@ dependencies = [ "log", "native-tls", "openssl", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", "prometheus-client", "prost", "prost-types", @@ -585,6 +599,15 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + [[package]] name = "futures-channel" version = "0.3.32" @@ -592,6 +615,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -600,6 +624,34 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -625,8 +677,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", + "futures-io", + "futures-macro", "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -871,13 +926,16 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64", "bytes", "futures-channel", "futures-util", "http", "http-body", "hyper", + "ipnet", "libc", + "percent-encoding", "pin-project-lite", "socket2", "tokio", @@ -885,6 +943,109 @@ dependencies = [ "tracing", ] +[[package]] +name = "icu_collections" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2984d1cd16c883d7935b9e07e44071dca8d917fd52ecc02c04d5fa0b5a3f191c" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92219b62b3e2b4d88ac5119f8904c10f8f61bf7e95b640d25ba3075e6cac2c29" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c56e5ee99d6e3d33bd91c5d85458b6005a22140021cc324cea84dd0e72cff3b4" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da3be0ae77ea334f4da67c12f149704f19f81d1adf7c51cf482943e84a2bad38" + +[[package]] +name = "icu_properties" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bee3b67d0ea5c2cca5003417989af8996f8604e34fb9ddf96208a033901e70de" +dependencies = [ + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e2bbb201e0c04f7b4b3e14382af113e17ba4f63e2c9d2ee626b720cbce54a14" + +[[package]] +name = "icu_provider" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "139c4cf31c8b5f33d7e199446eff9c1e02decfc2f0eec2c8d71f65befa45b421" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + [[package]] name = "indexmap" version = "2.14.0" @@ -895,6 +1056,12 @@ dependencies = [ "hashbrown 0.17.1", ] +[[package]] +name = "ipnet" +version = "2.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -983,6 +1150,12 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "litemap" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" + [[package]] name = "lock_api" version = "0.4.14" @@ -1139,6 +1312,76 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror", + "tracing", +] + +[[package]] +name = "opentelemetry-http" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" +dependencies = [ + "async-trait", + "bytes", + "http", + "opentelemetry", + "reqwest", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest", + "thiserror", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.4", + "thiserror", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -1226,6 +1469,15 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "potential_utf" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0103b1cef7ec0cf76490e969665504990193874ea05c85ff9bab8b911d0a0564" +dependencies = [ + "zerovec", +] + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1501,6 +1753,37 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "reqwest" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" +dependencies = [ + "base64", + "bytes", + "futures-channel", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -1660,6 +1943,12 @@ dependencies = [ "lock_api", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "strsim" version = "0.11.1" @@ -1682,6 +1971,20 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] + +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] [[package]] name = "tempfile" @@ -1716,6 +2019,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tinystr" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8323304221c2a851516f22236c5722a72eaa19749016521d6dff0824447d96d" +dependencies = [ + "displaydoc", + "zerovec", +] + [[package]] name = "tokio" version = "1.52.3" @@ -1865,6 +2178,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags 2.13.0", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -1926,6 +2257,24 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + [[package]] name = "utf8parse" version = "0.2.2" @@ -1986,6 +2335,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.76" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.126" @@ -2081,6 +2440,12 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "writeable" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" + [[package]] name = "yaml-rust2" version = "0.11.0" @@ -2092,6 +2457,29 @@ dependencies = [ "hashlink", ] +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "synstructure", +] + [[package]] name = "zerocopy" version = "0.8.52" @@ -2112,6 +2500,60 @@ dependencies = [ "syn", ] +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "synstructure", +] + +[[package]] +name = "zerotrie" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f9152d31db0792fa83f70fb2f83148effb5c1f5b8c7686c3459e361d9bc20bf" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90f911cbc359ab6af17377d242225f4d75119aec87ea711a880987b18cd7b239" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "zmij" version = "1.0.21" diff --git a/fact/Cargo.toml b/fact/Cargo.toml index 8bba7aff..3501995d 100644 --- a/fact/Cargo.toml +++ b/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" } +opentelemetry-otlp = { version = "0.32.0" } +opentelemetry_sdk = { version = "0.32.1", features = [ "logs"] } openssl = { workspace = true } tonic = { workspace = true } tokio = { workspace = true } diff --git a/fact/src/event/mod.rs b/fact/src/event/mod.rs index a91c5f00..c863a14b 100644 --- a/fact/src/event/mod.rs +++ b/fact/src/event/mod.rs @@ -1,12 +1,14 @@ #[cfg(all(test, feature = "bpf-test"))] use std::time::{SystemTime, UNIX_EPOCH}; use std::{ + collections::HashMap, ffi::{CStr, OsStr}, os::{raw::c_char, unix::ffi::OsStrExt}, path::{Path, PathBuf}, }; use globset::GlobSet; +use opentelemetry::logs::AnyValue; use serde::Serialize; use fact_ebpf::{ @@ -357,6 +359,17 @@ impl From for fact_api::FileActivity { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: Event) -> Self { + AnyValue::Map(Box::new(HashMap::from([ + ("file".into(), value.file.into()), + ("timestamp".into(), AnyValue::Int(value.timestamp as i64)), + ("process".into(), value.process.into()), + ("hostname".into(), value.hostname.into()), + ]))) + } +} + #[cfg(test)] impl PartialEq for Event { fn eq(&self, other: &Self) -> bool { @@ -444,6 +457,21 @@ impl FileData { Ok(file) } + + fn event_type(&self) -> &'static str { + match self { + FileData::Open(_) => "open", + FileData::Creation(_) => "creation", + FileData::MkDir(_) => "mkdir", + FileData::RmDir(_) => "rmdir", + FileData::Unlink(_) => "unlink", + FileData::Chmod(_) => "chmod", + FileData::Chown(_) => "chown", + FileData::Rename(_) => "rename", + FileData::SetXattr(_) => "setxattr", + FileData::RemoveXattr(_) => "removexattr", + } + } } impl From for fact_api::file_activity::File { @@ -494,6 +522,29 @@ impl From for fact_api::file_activity::File { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: FileData) -> Self { + let event_type = value.event_type(); + let AnyValue::Map(mut map) = (match value { + FileData::Open(data) + | FileData::Creation(data) + | FileData::MkDir(data) + | FileData::RmDir(data) + | FileData::Unlink(data) => AnyValue::from(data), + FileData::Chmod(data) => AnyValue::from(data), + FileData::Chown(data) => AnyValue::from(data), + FileData::Rename(data) => AnyValue::from(data), + FileData::SetXattr(data) | FileData::RemoveXattr(data) => AnyValue::from(data), + }) else { + unreachable!("event data did not serialize to map"); + }; + + map.insert("event_type".into(), event_type.into()); + + AnyValue::Map(map) + } +} + #[cfg(test)] impl PartialEq for FileData { fn eq(&self, other: &Self) -> bool { @@ -554,6 +605,21 @@ impl From for fact_api::FileActivityBase { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: BaseFileData) -> Self { + AnyValue::Map(Box::new(HashMap::from([ + ( + "filename".into(), + value.filename.to_string_lossy().to_string().into(), + ), + ( + "host_path".into(), + value.host_file.to_string_lossy().to_string().into(), + ), + ]))) + } +} + #[derive(Debug, Clone, Serialize)] pub struct ChmodFileData { inner: BaseFileData, @@ -576,6 +642,20 @@ impl From for fact_api::FilePermissionChange { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: ChmodFileData) -> Self { + let AnyValue::Map(mut map) = value.inner.into() else { + unreachable!("inner value did not serialize to map"); + }; + map.extend([ + ("new_mode".into(), value.new_mode.into()), + ("old_mode".into(), value.old_mode.into()), + ]); + + AnyValue::Map(map) + } +} + #[cfg(test)] impl PartialEq for ChmodFileData { fn eq(&self, other: &Self) -> bool { @@ -613,6 +693,22 @@ impl From for fact_api::FileOwnershipChange { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: ChownFileData) -> Self { + let AnyValue::Map(mut map) = value.inner.into() else { + unreachable!("inner value did not serialize to map"); + }; + map.extend([ + ("new_uid".into(), value.new_uid.into()), + ("old_uid".into(), value.old_uid.into()), + ("new_gid".into(), value.new_gid.into()), + ("old_gid".into(), value.old_gid.into()), + ]); + + AnyValue::Map(map) + } +} + #[derive(Debug, Clone, Serialize)] pub struct RenameFileData { new: BaseFileData, @@ -630,6 +726,16 @@ impl From for fact_api::FileRename { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: RenameFileData) -> Self { + let AnyValue::Map(mut map) = value.new.into() else { + unreachable!("new value did not serialize to map"); + }; + map.insert("old".into(), value.old.into()); + AnyValue::Map(map) + } +} + #[cfg(test)] impl PartialEq for RenameFileData { fn eq(&self, other: &Self) -> bool { @@ -653,6 +759,17 @@ impl From for fact_api::FileXattrChange { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: XattrFileData) -> Self { + let AnyValue::Map(mut map) = value.inner.into() else { + unreachable!("inner value did not serialize to map"); + }; + map.insert("xattr_name".into(), value.xattr_name.into()); + + AnyValue::Map(map) + } +} + #[cfg(test)] impl PartialEq for XattrFileData { fn eq(&self, other: &Self) -> bool { diff --git a/fact/src/event/process.rs b/fact/src/event/process.rs index f295ac66..048b9a44 100644 --- a/fact/src/event/process.rs +++ b/fact/src/event/process.rs @@ -1,6 +1,7 @@ -use std::{ffi::CStr, path::PathBuf}; +use std::{collections::HashMap, ffi::CStr, path::PathBuf}; use fact_ebpf::{lineage_t, process_t}; +use opentelemetry::logs::AnyValue; use serde::Serialize; use uuid::Uuid; @@ -38,6 +39,18 @@ impl From for fact_api::process_signal::LineageInfo { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: Lineage) -> Self { + AnyValue::Map(Box::new(HashMap::from([ + ("uid".into(), value.uid.into()), + ( + "exec_path".into(), + value.exe_path.to_string_lossy().to_string().into(), + ), + ]))) + } +} + #[derive(Debug, Clone, Default, Serialize)] pub struct Process { comm: String, @@ -220,6 +233,43 @@ impl From for fact_api::ProcessSignal { } } +impl From for opentelemetry::logs::AnyValue { + fn from(value: Process) -> Self { + let args = value + .args + .into_iter() + .map(AnyValue::from) + .collect::>(); + + let lineage = value + .lineage + .into_iter() + .map(AnyValue::from) + .collect::>(); + + let mut map = HashMap::from([ + ("comm".into(), value.comm.into()), + ("args".into(), AnyValue::ListAny(Box::new(args))), + ( + "exe_path".into(), + value.exe_path.to_string_lossy().to_string().into(), + ), + ("uid".into(), value.uid.into()), + ("gid".into(), value.gid.into()), + ("login_uid".into(), value.login_uid.into()), + ("username".into(), value.username.into()), + ("in_root_mount_ns".into(), value.in_root_mount_ns.into()), + ("lineage".into(), AnyValue::ListAny(Box::new(lineage))), + ]); + + if let Some(container_id) = value.container_id { + map.insert("container_id".into(), container_id.into()); + } + + AnyValue::Map(Box::new(map)) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/fact/src/output/mod.rs b/fact/src/output/mod.rs index 756c8eec..492fc4be 100644 --- a/fact/src/output/mod.rs +++ b/fact/src/output/mod.rs @@ -10,6 +10,7 @@ use tokio::{ use crate::{config::GrpcConfig, event::Event, metrics::OutputMetrics}; mod grpc; +mod otel; mod stdout; /// Starts all the output tasks. @@ -33,6 +34,8 @@ pub fn start( config.clone(), ); + let otel_client = otel::Client::new(broad_rx.resubscribe()); + // JSON client will only start if explicitly enabled or no other // output is active at startup if !grpc_client.is_enabled() || stdout_enabled { @@ -45,6 +48,7 @@ pub fn start( } let mut grpc_handle = grpc_client.start(); + otel_client.start(); tokio::spawn(async move { debug!("Starting output component..."); diff --git a/fact/src/output/otel/mod.rs b/fact/src/output/otel/mod.rs new file mode 100644 index 00000000..de55f589 --- /dev/null +++ b/fact/src/output/otel/mod.rs @@ -0,0 +1,59 @@ +use std::sync::Arc; + +use log::warn; +use opentelemetry::logs::{AnyValue, LogRecord, Logger, LoggerProvider, Severity}; +use opentelemetry_otlp::{LogExporter, WithExportConfig}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::logs::SdkLoggerProvider; +use tokio::sync::broadcast::{self, error::RecvError}; + +use crate::event::Event; + +pub(super) struct Client { + rx: broadcast::Receiver>, +} + +impl Client { + pub(super) fn new(rx: broadcast::Receiver>) -> Self { + Client { rx } + } + + pub(super) fn start(mut self) { + tokio::spawn(async move { + let exporter_otlp = LogExporter::builder() + .with_http() + .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) + .with_endpoint("http://127.0.0.1:4318/v1/logs") + .build() + .expect("Failed to create log exporter"); + + let logger_provider = SdkLoggerProvider::builder() + .with_batch_exporter(exporter_otlp) + .with_resource(Resource::builder().with_service_name("fact").build()) + .build(); + let logger = logger_provider.logger("fact"); + + loop { + match self.rx.recv().await { + Ok(event) => { + let event = Arc::unwrap_or_clone(event).into(); + let mut record = logger.create_log_record(); + record.set_severity_number(Severity::Info); + if let AnyValue::Map(map) = event { + for (k, v) in *map { + record.add_attribute(k, v); + } + } + logger.emit(record); + } + Err(RecvError::Closed) => break, + Err(RecvError::Lagged(n)) => { + warn!("oTel stream lagged, dropped {n} events"); + } + } + } + + let _ = logger_provider.shutdown(); + }); + } +} From c2d0d32cd870c952fbda74d428169fa27bcc0fca Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Fri, 3 Jul 2026 14:56:14 +0200 Subject: [PATCH 2/8] feat(otel): add basic configuration --- fact/src/config/mod.rs | 65 ++++++- fact/src/config/reloader.rs | 21 +++ fact/src/config/tests.rs | 358 +++++++++++++++++++++++------------- fact/src/lib.rs | 1 + fact/src/metrics/mod.rs | 8 + fact/src/output/mod.rs | 45 +++-- fact/src/output/otel.rs | 123 +++++++++++++ fact/src/output/otel/mod.rs | 59 ------ 8 files changed, 472 insertions(+), 208 deletions(-) create mode 100644 fact/src/output/otel.rs delete mode 100644 fact/src/output/otel/mod.rs diff --git a/fact/src/config/mod.rs b/fact/src/config/mod.rs index ff829eb5..88daa519 100644 --- a/fact/src/config/mod.rs +++ b/fact/src/config/mod.rs @@ -34,6 +34,7 @@ fn yaml_to_duration_secs(v: &Yaml) -> Option { pub struct FactConfig { paths: Option>, pub grpc: GrpcConfig, + pub otel: OTelConfig, pub endpoint: EndpointConfig, pub bpf: BpfConfig, skip_pre_flight: Option, @@ -72,7 +73,7 @@ impl FactConfig { )?; // Once file configuration is handled, apply CLI arguments - static CLI_ARGS: LazyLock = LazyLock::new(|| FactCli::parse().to_config()); + static CLI_ARGS: LazyLock = 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> 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, +} + +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 { + 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, @@ -627,6 +675,10 @@ pub struct FactCli { #[arg(long, env = "FACT_GRPC_BACKOFF_RETRIES_MAX")] backoff_retries_max: Option, + /// OpenTelemetry endpoint to push logs into + #[arg(long, env = "FACT_OTEL_ENDPOINT")] + otel_endpoint: Option, + /// The port to bind for all exposed endpoints #[arg(long, short, env = "FACT_ENDPOINT_ADDRESS")] address: Option, @@ -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), diff --git a/fact/src/config/reloader.rs b/fact/src/config/reloader.rs index 2ac14742..48baecb7 100644 --- a/fact/src/config/reloader.rs +++ b/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, grpc: watch::Sender, + otel: watch::Sender, paths: watch::Sender>, files: HashMap<&'static str, i64>, scan_interval: watch::Sender, @@ -71,6 +74,12 @@ impl Reloader { self.grpc.subscribe() } + /// Subscribe to get notifications when otel configuration is + /// changed. + pub fn otel(&self) -> watch::Receiver { + self.otel.subscribe() + } + /// Subscribe to get notifications when paths configuration is /// changed. pub fn paths(&self) -> watch::Receiver> { @@ -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 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 for Reloader { config, endpoint, grpc, + otel, paths, scan_interval, rate_limit, diff --git a/fact/src/config/tests.rs b/fact/src/config/tests.rs index 01edbbbf..57f4d401 100644 --- a/fact/src/config/tests.rs +++ b/fact/src/config/tests.rs @@ -49,6 +49,144 @@ fn parsing() { ..Default::default() }, ), + ( + r#" + grpc: + backoff: + initial: 2 + "#, + FactConfig { + grpc: GrpcConfig { + backoff: BackoffConfig { + initial: Some(Duration::from_secs(2)), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }, + ), + ( + r#" + grpc: + backoff: + max: 30 + "#, + FactConfig { + grpc: GrpcConfig { + backoff: BackoffConfig { + max: Some(Duration::from_secs(30)), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }, + ), + ( + r#" + grpc: + backoff: + jitter: false + "#, + FactConfig { + grpc: GrpcConfig { + backoff: BackoffConfig { + jitter: Some(false), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }, + ), + ( + r#" + grpc: + backoff: + multiplier: 2 + "#, + FactConfig { + grpc: GrpcConfig { + backoff: BackoffConfig { + multiplier: Some(2.0), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }, + ), + ( + r#" + grpc: + backoff: + multiplier: 3.5 + "#, + FactConfig { + grpc: GrpcConfig { + backoff: BackoffConfig { + multiplier: Some(3.5), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }, + ), + ( + r#" + grpc: + backoff: + retries: 5 + "#, + FactConfig { + grpc: GrpcConfig { + backoff: BackoffConfig { + retries_max: Some(5), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }, + ), + ( + r#" + grpc: + backoff: + initial: 0.5 + max: 120 + jitter: false + multiplier: 2 + retries: 5 + "#, + FactConfig { + grpc: GrpcConfig { + backoff: BackoffConfig { + initial: Some(Duration::from_secs_f64(0.5)), + max: Some(Duration::from_secs(120)), + jitter: Some(false), + multiplier: Some(2.0), + retries_max: Some(5), + }, + ..Default::default() + }, + ..Default::default() + }, + ), + ( + r#" + otel: + endpoint: http://localhost:4317 + "#, + FactConfig { + otel: OTelConfig { + endpoint: Some("http://localhost:4317".into()), + }, + ..Default::default() + }, + ), ( r#" endpoint: @@ -255,136 +393,12 @@ fn parsing() { ..Default::default() }, ), - ( - r#" - grpc: - backoff: - initial: 2 - "#, - FactConfig { - grpc: GrpcConfig { - backoff: BackoffConfig { - initial: Some(Duration::from_secs(2)), - ..Default::default() - }, - ..Default::default() - }, - ..Default::default() - }, - ), - ( - r#" - grpc: - backoff: - max: 30 - "#, - FactConfig { - grpc: GrpcConfig { - backoff: BackoffConfig { - max: Some(Duration::from_secs(30)), - ..Default::default() - }, - ..Default::default() - }, - ..Default::default() - }, - ), - ( - r#" - grpc: - backoff: - jitter: false - "#, - FactConfig { - grpc: GrpcConfig { - backoff: BackoffConfig { - jitter: Some(false), - ..Default::default() - }, - ..Default::default() - }, - ..Default::default() - }, - ), - ( - r#" - grpc: - backoff: - multiplier: 2 - "#, - FactConfig { - grpc: GrpcConfig { - backoff: BackoffConfig { - multiplier: Some(2.0), - ..Default::default() - }, - ..Default::default() - }, - ..Default::default() - }, - ), - ( - r#" - grpc: - backoff: - multiplier: 3.5 - "#, - FactConfig { - grpc: GrpcConfig { - backoff: BackoffConfig { - multiplier: Some(3.5), - ..Default::default() - }, - ..Default::default() - }, - ..Default::default() - }, - ), - ( - r#" - grpc: - backoff: - retries: 5 - "#, - FactConfig { - grpc: GrpcConfig { - backoff: BackoffConfig { - retries_max: Some(5), - ..Default::default() - }, - ..Default::default() - }, - ..Default::default() - }, - ), - ( - r#" - grpc: - backoff: - initial: 0.5 - max: 120 - jitter: false - multiplier: 2 - retries: 5 - "#, - FactConfig { - grpc: GrpcConfig { - backoff: BackoffConfig { - initial: Some(Duration::from_secs_f64(0.5)), - max: Some(Duration::from_secs(120)), - jitter: Some(false), - multiplier: Some(2.0), - retries_max: Some(5), - }, - ..Default::default() - }, - ..Default::default() - }, - ), ( r#" paths: - /etc + otel: + endpoint: 'http://localhost:4317' grpc: url: 'https://svc.sensor.stackrox:9090' certs: /etc/stackrox/certs @@ -419,6 +433,9 @@ fn parsing() { retries_max: Some(5), }, }, + otel: OTelConfig { + endpoint: Some("http://localhost:4317".into()), + }, endpoint: EndpointConfig { address: Some(SocketAddr::from(([0, 0, 0, 0], 8080))), expose_metrics: Some(true), @@ -596,6 +613,26 @@ paths: "#, "Invalid field 'grpc.backoff.unknown' with value: Integer(4)", ), + ( + r#" + otel: 5 + "#, + "Invalid field 'otel' with value: Integer(5)", + ), + ( + r#" + otel: + something: true + "#, + "Invalid field 'otel.something' with value: Boolean(true)", + ), + ( + r#" + otel: + endpoint: false + "#, + "otel.endpoint field has incorrect type: Boolean(false)", + ), ( "endpoint: true", "Invalid field 'endpoint' with value: Boolean(true)", @@ -1148,6 +1185,37 @@ fn update() { ..Default::default() }, ), + ( + r#" + otel: + endpoint: 'http://localhost:4317' + "#, + FactConfig::default(), + FactConfig { + otel: OTelConfig { + endpoint: Some(String::from("http://localhost:4317")), + }, + ..Default::default() + }, + ), + ( + r#" + otel: + endpoint: 'http://localhost:4317' + "#, + FactConfig { + otel: OTelConfig { + endpoint: Some(String::from("http://localhost:1234")), + }, + ..Default::default() + }, + FactConfig { + otel: OTelConfig { + endpoint: Some(String::from("http://localhost:4317")), + }, + ..Default::default() + }, + ), ( r#" endpoint: @@ -1516,6 +1584,8 @@ fn update() { jitter: false multiplier: 3.0 retries: 5 + otel: + endpoint: 'http://localhost:4317' endpoint: address: 127.0.0.1:8080 expose_metrics: true @@ -1541,6 +1611,9 @@ fn update() { retries_max: Some(20), }, }, + otel: OTelConfig { + endpoint: Some(String::from("http://localhost:1234")), + }, endpoint: EndpointConfig { address: Some(SocketAddr::from(([0, 0, 0, 0], 9000))), expose_metrics: Some(false), @@ -1569,6 +1642,9 @@ fn update() { retries_max: Some(5), }, }, + otel: OTelConfig { + endpoint: Some(String::from("http://localhost:4317")), + }, endpoint: EndpointConfig { address: Some(SocketAddr::from(([127, 0, 0, 1], 8080))), expose_metrics: Some(true), @@ -1678,7 +1754,7 @@ impl Display for EnvVar { fn with_env_var(env: EnvVar) -> Result { let _guard = env.set(); - FactCli::try_parse_from(["fact"]).map(|cli| cli.to_config()) + FactCli::try_parse_from(["fact"]).map(|cli| cli.into_config()) } #[test] @@ -1866,6 +1942,18 @@ fn env_vars() { ..Default::default() }, ), + ( + EnvVar { + name: "FACT_OTEL_ENDPOINT", + value: "http://localhost:4317", + }, + FactConfig { + otel: OTelConfig { + endpoint: Some(String::from("http://localhost:4317")), + }, + ..Default::default() + }, + ), ( EnvVar { name: "FACT_ENDPOINT_ADDRESS", @@ -2011,6 +2099,22 @@ fn env_vars_override_yaml() { ..Default::default() }, ), + ( + EnvVar { + name: "FACT_OTEL_ENDPOINT", + value: "http://localhost:4317", + }, + r#" + otel: + endpoint: 'http://localhost:1234' + "#, + FactConfig { + otel: OTelConfig { + endpoint: Some(String::from("http://localhost:4317")), + }, + ..Default::default() + }, + ), ( EnvVar { name: "FACT_PATHS", diff --git a/fact/src/lib.rs b/fact/src/lib.rs index 8ec187f6..94fe1cc4 100644 --- a/fact/src/lib.rs +++ b/fact/src/lib.rs @@ -118,6 +118,7 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { running.subscribe(), exporter.metrics.output.clone(), reloader.grpc(), + reloader.otel(), reloader.config().json(), ); let mut host_scanner_handle = host_scanner.start(); diff --git a/fact/src/metrics/mod.rs b/fact/src/metrics/mod.rs index 98297329..af8cdebc 100644 --- a/fact/src/metrics/mod.rs +++ b/fact/src/metrics/mod.rs @@ -101,6 +101,7 @@ impl EventCounter { pub struct OutputMetrics { pub stdout: EventCounter, pub grpc: EventCounter, + pub otel: EventCounter, } impl OutputMetrics { @@ -116,16 +117,23 @@ impl OutputMetrics { "Events processed by the grpc output component", &labels, ); + let otel_counter = EventCounter::new( + "output_otel_events", + "Events processed by the otel output component", + &labels, + ); OutputMetrics { stdout: stdout_counter, grpc: grpc_counter, + otel: otel_counter, } } fn register(&self, reg: &mut Registry) { self.stdout.register(reg); self.grpc.register(reg); + self.otel.register(reg); } } diff --git a/fact/src/output/mod.rs b/fact/src/output/mod.rs index 492fc4be..734c04f0 100644 --- a/fact/src/output/mod.rs +++ b/fact/src/output/mod.rs @@ -1,13 +1,17 @@ use std::{borrow::BorrowMut, sync::Arc}; -use anyhow::bail; use log::{debug, warn}; use tokio::{ sync::{broadcast, mpsc, watch}, task::JoinHandle, }; -use crate::{config::GrpcConfig, event::Event, metrics::OutputMetrics}; +use crate::{ + config::{GrpcConfig, OTelConfig}, + event::Event, + flatten_task_result, + metrics::OutputMetrics, +}; mod grpc; mod otel; @@ -21,7 +25,8 @@ pub fn start( mut rx: mpsc::Receiver, running: watch::Receiver, metrics: OutputMetrics, - config: watch::Receiver, + grpc_config: watch::Receiver, + otel_config: watch::Receiver, stdout_enabled: bool, ) -> JoinHandle> { let (broad_tx, broad_rx) = broadcast::channel(100); @@ -31,14 +36,19 @@ pub fn start( broad_rx.resubscribe(), running.clone(), metrics.grpc.clone(), - config.clone(), + grpc_config, ); - let otel_client = otel::Client::new(broad_rx.resubscribe()); + let otel_client = otel::Client::new( + broad_rx.resubscribe(), + running.clone(), + metrics.otel.clone(), + otel_config, + ); // JSON client will only start if explicitly enabled or no other // output is active at startup - if !grpc_client.is_enabled() || stdout_enabled { + if (!grpc_client.is_enabled() && !otel_client.is_enabled()) || stdout_enabled { stdout::Client::new( broad_rx.resubscribe(), running.clone(), @@ -48,15 +58,17 @@ pub fn start( } let mut grpc_handle = grpc_client.start(); - otel_client.start(); + let mut otel_handle = otel_client.start(); tokio::spawn(async move { debug!("Starting output component..."); - loop { + let res = loop { tokio::select! { event = rx.recv() => { let Some(event) = event else { - break; + // Channel has been closed and no more messages + // are present. + break Ok(()); }; if let Err(e) = broad_tx.send(Arc::new(event)) { @@ -64,16 +76,15 @@ pub fn start( } } res = grpc_handle.borrow_mut() => { - match res { - Ok(Ok(_)) => break, - Ok(Err(e)) => bail!("gRPC worker errored out: {e:?}"), - Err(e) => bail!("gRPC task errored out: {e:?}"), - } + break flatten_task_result("gRPC", res); + } + res = otel_handle.borrow_mut() => { + break flatten_task_result("oTel", res); } - _ = run.changed() => if !*run.borrow() { break; } + _ = run.changed() => if !*run.borrow() { break Ok(()); } } - } + }; debug!("Stopping output component..."); - Ok(()) + res }) } diff --git a/fact/src/output/otel.rs b/fact/src/output/otel.rs new file mode 100644 index 00000000..6d1d4c31 --- /dev/null +++ b/fact/src/output/otel.rs @@ -0,0 +1,123 @@ +use std::sync::Arc; + +use anyhow::bail; +use log::{debug, info, warn}; +use opentelemetry::logs::{AnyValue, LogRecord, Logger, LoggerProvider, Severity}; +use opentelemetry_otlp::{LogExporter, WithExportConfig}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::logs::SdkLoggerProvider; +use tokio::{ + sync::{ + broadcast::{self, error::RecvError}, + watch, + }, + task::JoinHandle, +}; + +use crate::{config::OTelConfig, event::Event, metrics::EventCounter}; + +pub(super) struct Client { + rx: broadcast::Receiver>, + running: watch::Receiver, + config: watch::Receiver, + metrics: EventCounter, +} + +impl Client { + pub(super) fn new( + rx: broadcast::Receiver>, + running: watch::Receiver, + metrics: EventCounter, + config: watch::Receiver, + ) -> Self { + Client { + rx, + running, + config, + metrics, + } + } + + pub(super) fn start(mut self) -> JoinHandle> { + tokio::spawn(async move { + loop { + let res = if self.is_enabled() { + self.run().await + } else { + self.idle().await + }; + + match res { + Ok(true) => info!("Reloading oTel configuration..."), + Ok(false) => { + info!("Stopping oTel output..."); + break; + } + Err(e) => bail!("oTel error: {e:?}"), + } + } + Ok(()) + }) + } + + async fn run(&mut self) -> anyhow::Result { + let Some(endpoint) = self.config.borrow().endpoint().map(|e| e.to_string()) else { + bail!("Attempted to unwrap empty endpoint"); + }; + debug!("oTel: forwarding events to {endpoint}"); + let exporter_otlp = LogExporter::builder() + .with_http() + .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) + .with_endpoint(endpoint) + .build() + .expect("Failed to create log exporter"); + + let logger_provider = SdkLoggerProvider::builder() + .with_batch_exporter(exporter_otlp) + .with_resource(Resource::builder().with_service_name("fact").build()) + .build(); + let logger = logger_provider.logger("fact"); + + let res = loop { + tokio::select! { + event = self.rx.recv() => { + match event { + Ok(event) => { + self.metrics.added(); + let event = Arc::unwrap_or_clone(event).into(); + let mut record = logger.create_log_record(); + record.set_severity_number(Severity::Info); + if let AnyValue::Map(map) = event { + for (k, v) in *map { + record.add_attribute(k, v); + } + } + logger.emit(record); + } + Err(RecvError::Closed) => break Err(anyhow::anyhow!("oTel: event stream closed")), + Err(RecvError::Lagged(n)) => { + warn!("oTel stream lagged, dropped {n} events"); + self.metrics.dropped_n(n); + } + } + } + _ = self.config.changed() => break Ok(true), + _ = self.running.changed() => break Ok(*self.running.borrow()), + } + }; + + logger_provider.shutdown()?; + res + } + + pub(super) fn is_enabled(&self) -> bool { + self.config.borrow().endpoint().is_some() + } + + async fn idle(&mut self) -> anyhow::Result { + tokio::select! { + _ = self.config.changed() => Ok(true), + _ = self.running.changed() => Ok(*self.running.borrow()), + } + } +} diff --git a/fact/src/output/otel/mod.rs b/fact/src/output/otel/mod.rs deleted file mode 100644 index de55f589..00000000 --- a/fact/src/output/otel/mod.rs +++ /dev/null @@ -1,59 +0,0 @@ -use std::sync::Arc; - -use log::warn; -use opentelemetry::logs::{AnyValue, LogRecord, Logger, LoggerProvider, Severity}; -use opentelemetry_otlp::{LogExporter, WithExportConfig}; -use opentelemetry_sdk::Resource; -use opentelemetry_sdk::logs::SdkLoggerProvider; -use tokio::sync::broadcast::{self, error::RecvError}; - -use crate::event::Event; - -pub(super) struct Client { - rx: broadcast::Receiver>, -} - -impl Client { - pub(super) fn new(rx: broadcast::Receiver>) -> Self { - Client { rx } - } - - pub(super) fn start(mut self) { - tokio::spawn(async move { - let exporter_otlp = LogExporter::builder() - .with_http() - .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) - .with_endpoint("http://127.0.0.1:4318/v1/logs") - .build() - .expect("Failed to create log exporter"); - - let logger_provider = SdkLoggerProvider::builder() - .with_batch_exporter(exporter_otlp) - .with_resource(Resource::builder().with_service_name("fact").build()) - .build(); - let logger = logger_provider.logger("fact"); - - loop { - match self.rx.recv().await { - Ok(event) => { - let event = Arc::unwrap_or_clone(event).into(); - let mut record = logger.create_log_record(); - record.set_severity_number(Severity::Info); - if let AnyValue::Map(map) = event { - for (k, v) in *map { - record.add_attribute(k, v); - } - } - logger.emit(record); - } - Err(RecvError::Closed) => break, - Err(RecvError::Lagged(n)) => { - warn!("oTel stream lagged, dropped {n} events"); - } - } - } - - let _ = logger_provider.shutdown(); - }); - } -} From 3c2a9dd11b05b4e1057a21aef139126d72e16319 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Fri, 3 Jul 2026 16:32:45 +0200 Subject: [PATCH 3/8] feat(output): change resubscription method The new way for output components to resubscribe the broadcast channel distributing events is to send a oneshot sender back to the main task which will resubscribe from the broadcast sender and forward the receiver back. While this approach might seem overly complicated, it ensures there will be no lingering receivers laying around, which in turn allows for Arc::unwrap_or_clone to properly unwrap all messages when only one output is in use. --- fact/src/output/grpc.rs | 38 ++++++++++++++++++++------------------ fact/src/output/mod.rs | 23 ++++++++++++++--------- fact/src/output/otel.rs | 19 ++++++++++--------- fact/src/output/stdout.rs | 20 +++++++++----------- 4 files changed, 53 insertions(+), 47 deletions(-) diff --git a/fact/src/output/grpc.rs b/fact/src/output/grpc.rs index 969a15fd..af547ec1 100644 --- a/fact/src/output/grpc.rs +++ b/fact/src/output/grpc.rs @@ -9,7 +9,7 @@ use native_tls::{Certificate, Identity}; use openssl::{ec::EcKey, pkey::PKey}; use tokio::{ fs, - sync::{broadcast, watch}, + sync::{mpsc, oneshot, watch}, task::JoinHandle, time::sleep, }; @@ -21,8 +21,8 @@ use tonic::transport::Channel; use crate::{ config::{BackoffConfig, GrpcConfig}, - event::Event, metrics::EventCounter, + output::EventReceiver, }; struct Backoff { @@ -104,7 +104,7 @@ impl From<&BackoffConfig> for Backoff { } pub struct Client { - rx: broadcast::Receiver>, + subscriber: mpsc::Sender>, running: watch::Receiver, config: watch::Receiver, metrics: EventCounter, @@ -112,13 +112,13 @@ pub struct Client { impl Client { pub fn new( - rx: broadcast::Receiver>, + subscriber: mpsc::Sender>, running: watch::Receiver, metrics: EventCounter, config: watch::Receiver, ) -> Self { Client { - rx, + subscriber, running, config, metrics, @@ -230,19 +230,21 @@ impl Client { let mut client = FileActivityServiceClient::new(channel); let metrics = self.metrics.clone(); - let rx = - BroadcastStream::new(self.rx.resubscribe()).filter_map(move |event| match event { - Ok(event) => { - metrics.added(); - let event = Arc::unwrap_or_clone(event); - Some(event.into()) - } - Err(BroadcastStreamRecvError::Lagged(n)) => { - warn!("gRPC stream lagged, dropped {n} events"); - metrics.dropped_n(n); - None - } - }); + let (tx, rx) = oneshot::channel(); + self.subscriber.send(tx).await?; + let rx = rx.await?; + let rx = BroadcastStream::new(rx).filter_map(move |event| match event { + Ok(event) => { + metrics.added(); + let event = Arc::unwrap_or_clone(event); + Some(event.into()) + } + Err(BroadcastStreamRecvError::Lagged(n)) => { + warn!("gRPC stream lagged, dropped {n} events"); + metrics.dropped_n(n); + None + } + }); tokio::select! { res = client.communicate(rx) => { diff --git a/fact/src/output/mod.rs b/fact/src/output/mod.rs index 734c04f0..ecbc48d3 100644 --- a/fact/src/output/mod.rs +++ b/fact/src/output/mod.rs @@ -17,6 +17,8 @@ mod grpc; mod otel; mod stdout; +type EventReceiver = broadcast::Receiver>; + /// Starts all the output tasks. /// /// Each task is responsible for managing its lifetime, handling @@ -29,18 +31,19 @@ pub fn start( otel_config: watch::Receiver, stdout_enabled: bool, ) -> JoinHandle> { - let (broad_tx, broad_rx) = broadcast::channel(100); + let (broad_tx, _) = broadcast::channel(100); + let (subs_req, mut subs_rx) = mpsc::channel(10); let mut run = running.clone(); let grpc_client = grpc::Client::new( - broad_rx.resubscribe(), + subs_req.clone(), running.clone(), metrics.grpc.clone(), grpc_config, ); let otel_client = otel::Client::new( - broad_rx.resubscribe(), + subs_req.clone(), running.clone(), metrics.otel.clone(), otel_config, @@ -49,12 +52,7 @@ pub fn start( // JSON client will only start if explicitly enabled or no other // output is active at startup if (!grpc_client.is_enabled() && !otel_client.is_enabled()) || stdout_enabled { - stdout::Client::new( - broad_rx.resubscribe(), - running.clone(), - metrics.stdout.clone(), - ) - .start(); + stdout::Client::new(subs_req, running.clone(), metrics.stdout.clone()).start(); } let mut grpc_handle = grpc_client.start(); @@ -75,6 +73,13 @@ pub fn start( warn!("Failed to forward output event: {e}"); } } + req = subs_rx.recv() => { + let Some(req) = req else { continue; }; + let rx = broad_tx.subscribe(); + if let Err(e) = req.send(rx) { + break Err(anyhow::anyhow!("Failed to subscribe worker: {e:?}")); + } + } res = grpc_handle.borrow_mut() => { break flatten_task_result("gRPC", res); } diff --git a/fact/src/output/otel.rs b/fact/src/output/otel.rs index 6d1d4c31..91acfa56 100644 --- a/fact/src/output/otel.rs +++ b/fact/src/output/otel.rs @@ -7,17 +7,14 @@ use opentelemetry_otlp::{LogExporter, WithExportConfig}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::logs::SdkLoggerProvider; use tokio::{ - sync::{ - broadcast::{self, error::RecvError}, - watch, - }, + sync::{broadcast::error::RecvError, mpsc, oneshot, watch}, task::JoinHandle, }; -use crate::{config::OTelConfig, event::Event, metrics::EventCounter}; +use crate::{config::OTelConfig, metrics::EventCounter, output::EventReceiver}; pub(super) struct Client { - rx: broadcast::Receiver>, + subscriber: mpsc::Sender>, running: watch::Receiver, config: watch::Receiver, metrics: EventCounter, @@ -25,13 +22,13 @@ pub(super) struct Client { impl Client { pub(super) fn new( - rx: broadcast::Receiver>, + subscriber: mpsc::Sender>, running: watch::Receiver, metrics: EventCounter, config: watch::Receiver, ) -> Self { Client { - rx, + subscriber, running, config, metrics, @@ -78,9 +75,13 @@ impl Client { .build(); let logger = logger_provider.logger("fact"); + let (tx, rx) = oneshot::channel(); + self.subscriber.send(tx).await?; + let mut rx = rx.await?; + let res = loop { tokio::select! { - event = self.rx.recv() => { + event = rx.recv() => { match event { Ok(event) => { self.metrics.added(); diff --git a/fact/src/output/stdout.rs b/fact/src/output/stdout.rs index debc57fd..7b23e82e 100644 --- a/fact/src/output/stdout.rs +++ b/fact/src/output/stdout.rs @@ -1,27 +1,22 @@ -use std::sync::Arc; - use log::{info, warn}; -use tokio::sync::{ - broadcast::{self, error::RecvError}, - watch, -}; +use tokio::sync::{broadcast::error::RecvError, mpsc, oneshot, watch}; -use crate::{event::Event, metrics::EventCounter}; +use crate::{metrics::EventCounter, output::EventReceiver}; pub struct Client { - rx: broadcast::Receiver>, + subscriber: mpsc::Sender>, running: watch::Receiver, metrics: EventCounter, } impl Client { pub fn new( - rx: broadcast::Receiver>, + subscriber: mpsc::Sender>, running: watch::Receiver, metrics: EventCounter, ) -> Self { Client { - rx, + subscriber, running, metrics, } @@ -29,9 +24,12 @@ impl Client { pub fn start(mut self) { tokio::spawn(async move { + let (tx, rx) = oneshot::channel(); + self.subscriber.send(tx).await.unwrap(); + let mut rx = rx.await.unwrap(); loop { tokio::select! { - event = self.rx.recv() => { + event = rx.recv() => { let event = match event { Ok(event) => event, Err(RecvError::Closed) => { From 3803d50b3cc80fb659a44289a31e73cf194a77fb Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Fri, 3 Jul 2026 17:15:38 +0200 Subject: [PATCH 4/8] cleanup(otel): put otel behind a feature This allows regular fact builds to keep using just the gRPC and stodout outputs, no additional dependencies. --- fact/Cargo.toml | 7 +++--- fact/src/event/mod.rs | 12 +++++++++- fact/src/event/process.rs | 7 +++++- fact/src/output/grpc.rs | 8 +++---- fact/src/output/mod.rs | 48 +++++++++++++++++++++++---------------- fact/src/output/otel.rs | 8 +++---- 6 files changed, 58 insertions(+), 32 deletions(-) diff --git a/fact/Cargo.toml b/fact/Cargo.toml index 3501995d..4e649315 100644 --- a/fact/Cargo.toml +++ b/fact/Cargo.toml @@ -20,9 +20,9 @@ hyper-util = { workspace = true } libc = { workspace = true } log = { workspace = true } native-tls = { workspace = true } -opentelemetry = { version = "0.32.0" } -opentelemetry-otlp = { version = "0.32.0" } -opentelemetry_sdk = { version = "0.32.1", features = [ "logs"] } +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 } @@ -54,3 +54,4 @@ path = "src/main.rs" [features] bpf-test = [] +otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-otlp"] diff --git a/fact/src/event/mod.rs b/fact/src/event/mod.rs index c863a14b..49a50229 100644 --- a/fact/src/event/mod.rs +++ b/fact/src/event/mod.rs @@ -1,13 +1,15 @@ +#[cfg(feature = "otel")] +use std::collections::HashMap; #[cfg(all(test, feature = "bpf-test"))] use std::time::{SystemTime, UNIX_EPOCH}; use std::{ - collections::HashMap, ffi::{CStr, OsStr}, os::{raw::c_char, unix::ffi::OsStrExt}, path::{Path, PathBuf}, }; use globset::GlobSet; +#[cfg(feature = "otel")] use opentelemetry::logs::AnyValue; use serde::Serialize; @@ -359,6 +361,7 @@ impl From for fact_api::FileActivity { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: Event) -> Self { AnyValue::Map(Box::new(HashMap::from([ @@ -458,6 +461,7 @@ impl FileData { Ok(file) } + #[cfg(feature = "otel")] fn event_type(&self) -> &'static str { match self { FileData::Open(_) => "open", @@ -522,6 +526,7 @@ impl From for fact_api::file_activity::File { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: FileData) -> Self { let event_type = value.event_type(); @@ -605,6 +610,7 @@ impl From for fact_api::FileActivityBase { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: BaseFileData) -> Self { AnyValue::Map(Box::new(HashMap::from([ @@ -642,6 +648,7 @@ impl From for fact_api::FilePermissionChange { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: ChmodFileData) -> Self { let AnyValue::Map(mut map) = value.inner.into() else { @@ -693,6 +700,7 @@ impl From for fact_api::FileOwnershipChange { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: ChownFileData) -> Self { let AnyValue::Map(mut map) = value.inner.into() else { @@ -726,6 +734,7 @@ impl From for fact_api::FileRename { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: RenameFileData) -> Self { let AnyValue::Map(mut map) = value.new.into() else { @@ -759,6 +768,7 @@ impl From for fact_api::FileXattrChange { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: XattrFileData) -> Self { let AnyValue::Map(mut map) = value.inner.into() else { diff --git a/fact/src/event/process.rs b/fact/src/event/process.rs index 048b9a44..f18d4e42 100644 --- a/fact/src/event/process.rs +++ b/fact/src/event/process.rs @@ -1,6 +1,9 @@ -use std::{collections::HashMap, ffi::CStr, path::PathBuf}; +#[cfg(feature = "otel")] +use std::collections::HashMap; +use std::{ffi::CStr, path::PathBuf}; use fact_ebpf::{lineage_t, process_t}; +#[cfg(feature = "otel")] use opentelemetry::logs::AnyValue; use serde::Serialize; use uuid::Uuid; @@ -39,6 +42,7 @@ impl From for fact_api::process_signal::LineageInfo { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: Lineage) -> Self { AnyValue::Map(Box::new(HashMap::from([ @@ -233,6 +237,7 @@ impl From for fact_api::ProcessSignal { } } +#[cfg(feature = "otel")] impl From for opentelemetry::logs::AnyValue { fn from(value: Process) -> Self { let args = value diff --git a/fact/src/output/grpc.rs b/fact/src/output/grpc.rs index af547ec1..b643b363 100644 --- a/fact/src/output/grpc.rs +++ b/fact/src/output/grpc.rs @@ -10,7 +10,7 @@ use openssl::{ec::EcKey, pkey::PKey}; use tokio::{ fs, sync::{mpsc, oneshot, watch}, - task::JoinHandle, + task::JoinSet, time::sleep, }; use tokio_stream::{ @@ -125,8 +125,8 @@ impl Client { } } - pub fn start(mut self) -> JoinHandle> { - tokio::spawn(async move { + pub fn start(mut self, set: &mut JoinSet>) { + set.spawn(async move { loop { let res = if self.is_enabled() { self.run().await @@ -144,7 +144,7 @@ impl Client { } } Ok(()) - }) + }); } async fn get_connector(&self) -> anyhow::Result>> { diff --git a/fact/src/output/mod.rs b/fact/src/output/mod.rs index ecbc48d3..373ff9e2 100644 --- a/fact/src/output/mod.rs +++ b/fact/src/output/mod.rs @@ -1,19 +1,19 @@ -use std::{borrow::BorrowMut, sync::Arc}; +use std::sync::Arc; use log::{debug, warn}; use tokio::{ sync::{broadcast, mpsc, watch}, - task::JoinHandle, + task::{JoinHandle, JoinSet}, }; use crate::{ config::{GrpcConfig, OTelConfig}, event::Event, - flatten_task_result, metrics::OutputMetrics, }; mod grpc; +#[cfg(feature = "otel")] mod otel; mod stdout; @@ -28,12 +28,14 @@ pub fn start( running: watch::Receiver, metrics: OutputMetrics, grpc_config: watch::Receiver, - otel_config: watch::Receiver, + #[allow(unused)] otel_config: watch::Receiver, stdout_enabled: bool, ) -> JoinHandle> { let (broad_tx, _) = broadcast::channel(100); let (subs_req, mut subs_rx) = mpsc::channel(10); let mut run = running.clone(); + let mut handles = JoinSet::new(); + let mut enabled_outputs = Vec::new(); let grpc_client = grpc::Client::new( subs_req.clone(), @@ -41,23 +43,27 @@ pub fn start( metrics.grpc.clone(), grpc_config, ); + enabled_outputs.push(grpc_client.is_enabled()); + grpc_client.start(&mut handles); - let otel_client = otel::Client::new( - subs_req.clone(), - running.clone(), - metrics.otel.clone(), - otel_config, - ); + #[cfg(feature = "otel")] + { + let otel_client = otel::Client::new( + subs_req.clone(), + running.clone(), + metrics.otel.clone(), + otel_config, + ); + enabled_outputs.push(otel_client.is_enabled()); + otel_client.start(&mut handles); + } // JSON client will only start if explicitly enabled or no other // output is active at startup - if (!grpc_client.is_enabled() && !otel_client.is_enabled()) || stdout_enabled { + if stdout_enabled || enabled_outputs.iter().all(|enabled| !enabled) { stdout::Client::new(subs_req, running.clone(), metrics.stdout.clone()).start(); } - let mut grpc_handle = grpc_client.start(); - let mut otel_handle = otel_client.start(); - tokio::spawn(async move { debug!("Starting output component..."); let res = loop { @@ -80,11 +86,15 @@ pub fn start( break Err(anyhow::anyhow!("Failed to subscribe worker: {e:?}")); } } - res = grpc_handle.borrow_mut() => { - break flatten_task_result("gRPC", res); - } - res = otel_handle.borrow_mut() => { - break flatten_task_result("oTel", res); + res = handles.join_next() => { + let Some(res) = res else { + unreachable!("output handles should always have a task"); + }; + match res { + Ok(Ok(_)) => break Ok(()), + Ok(Err(e)) => break Err(e), + Err(e) => break Err(e.into()), + } } _ = run.changed() => if !*run.borrow() { break Ok(()); } } diff --git a/fact/src/output/otel.rs b/fact/src/output/otel.rs index 91acfa56..8a33ab3b 100644 --- a/fact/src/output/otel.rs +++ b/fact/src/output/otel.rs @@ -8,7 +8,7 @@ use opentelemetry_sdk::Resource; use opentelemetry_sdk::logs::SdkLoggerProvider; use tokio::{ sync::{broadcast::error::RecvError, mpsc, oneshot, watch}, - task::JoinHandle, + task::JoinSet, }; use crate::{config::OTelConfig, metrics::EventCounter, output::EventReceiver}; @@ -35,8 +35,8 @@ impl Client { } } - pub(super) fn start(mut self) -> JoinHandle> { - tokio::spawn(async move { + pub(super) fn start(mut self, set: &mut JoinSet>) { + set.spawn(async move { loop { let res = if self.is_enabled() { self.run().await @@ -54,7 +54,7 @@ impl Client { } } Ok(()) - }) + }); } async fn run(&mut self) -> anyhow::Result { From 86b1a62baa2dd71987de7dd48bd514a38813f90f Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Mon, 6 Jul 2026 12:22:21 +0200 Subject: [PATCH 5/8] cleanup(otel): minor consistency fixes * Make event type names match the protobuf definitions. * Send pid on otel output. --- fact/src/event/mod.rs | 8 ++++---- fact/src/event/process.rs | 1 + 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/fact/src/event/mod.rs b/fact/src/event/mod.rs index 49a50229..95ecdc77 100644 --- a/fact/src/event/mod.rs +++ b/fact/src/event/mod.rs @@ -469,11 +469,11 @@ impl FileData { FileData::MkDir(_) => "mkdir", FileData::RmDir(_) => "rmdir", FileData::Unlink(_) => "unlink", - FileData::Chmod(_) => "chmod", - FileData::Chown(_) => "chown", + FileData::Chmod(_) => "permission", + FileData::Chown(_) => "ownership", FileData::Rename(_) => "rename", - FileData::SetXattr(_) => "setxattr", - FileData::RemoveXattr(_) => "removexattr", + FileData::SetXattr(_) => "xattr_set", + FileData::RemoveXattr(_) => "xattr_remove", } } } diff --git a/fact/src/event/process.rs b/fact/src/event/process.rs index f18d4e42..db3ff2ec 100644 --- a/fact/src/event/process.rs +++ b/fact/src/event/process.rs @@ -259,6 +259,7 @@ impl From for opentelemetry::logs::AnyValue { "exe_path".into(), value.exe_path.to_string_lossy().to_string().into(), ), + ("pid".into(), value.pid.into()), ("uid".into(), value.uid.into()), ("gid".into(), value.gid.into()), ("login_uid".into(), value.login_uid.into()), From aa1e62057191a04a4b06f8a2f70148a7b29c6ed3 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Mon, 6 Jul 2026 12:24:06 +0200 Subject: [PATCH 6/8] test: add OTLP output support to integration tests Extend the test infrastructure to support both gRPC and OTLP output modes. Both servers translate their native message format into the test's Event/Process types, keeping protocol-specific code isolated. Changes: - Add EventServer base class with shared queue/wait logic - Add GrpcServer translating FileActivity protobufs into Events - Add OtlpServer receiving OTLP/HTTP binary protobuf log exports - Refactor Event.diff() and Process.diff() to compare Event vs Event - Add --output pytest option (grpc, otlp, all; default: grpc) - Parameterize server fixture so tests run per output mode - Add pytest-otlp and pytest-all Makefile targets - Fix rust_style_quote to match Rust shlex backslash handling - Add opentelemetry-proto dependency - Add CARGO_ARGS build arg to Containerfile - Add image-otel Makefile target Assisted-by: claude-opus-4-6@default --- Containerfile | 3 +- Makefile | 11 +- tests/Makefile | 10 +- tests/conftest.py | 53 ++++- tests/event.py | 102 ++++----- tests/requirements.txt | 1 + tests/server.py | 356 ++++++++++++++++++++++++++------ tests/test_config_hotreload.py | 21 +- tests/test_editors/test_nvim.py | 10 +- tests/test_editors/test_sed.py | 6 +- tests/test_editors/test_vi.py | 10 +- tests/test_editors/test_vim.py | 10 +- tests/test_file_open.py | 18 +- tests/test_misc.py | 4 +- tests/test_path_chmod.py | 16 +- tests/test_path_chown.py | 10 +- tests/test_path_mkdir.py | 6 +- tests/test_path_rename.py | 18 +- tests/test_path_rmdir.py | 10 +- tests/test_path_unlink.py | 16 +- tests/test_rate_limit.py | 6 +- tests/test_wildcard.py | 12 +- tests/test_xattr.py | 16 +- tests/utils.py | 10 +- 24 files changed, 496 insertions(+), 239 deletions(-) diff --git a/Containerfile b/Containerfile index 68869141..e296dc64 100644 --- a/Containerfile +++ b/Containerfile @@ -40,9 +40,10 @@ COPY . . FROM builder AS build ARG FACT_VERSION +ARG CARGO_ARGS="" RUN --mount=type=cache,target=/root/.cargo/registry \ --mount=type=cache,target=/app/target \ - cargo build --release && \ + cargo build --release $CARGO_ARGS && \ cp target/release/fact fact FROM ubi-micro-base diff --git a/Makefile b/Makefile index f1739a18..366692c3 100644 --- a/Makefile +++ b/Makefile @@ -20,6 +20,15 @@ image: -t $(FACT_IMAGE_NAME) \ $(CURDIR) +image-otel: + $(DOCKER) build \ + -f Containerfile \ + --build-arg FACT_VERSION=$(FACT_VERSION) \ + --build-arg RUST_VERSION=$(RUST_VERSION) \ + --build-arg CARGO_ARGS="--features otel" \ + -t $(FACT_IMAGE_NAME)-otel \ + $(CURDIR) + licenses:THIRD_PARTY_LICENSES.html THIRD_PARTY_LICENSES.html:Cargo.lock @@ -54,4 +63,4 @@ format: make -C fact-ebpf format ruff format tests/ -.PHONY: tag mock-server integration-tests image image-name licenses coverage lint clean +.PHONY: tag mock-server integration-tests image image-otel image-name licenses coverage lint clean diff --git a/tests/Makefile b/tests/Makefile index e69a13a1..d0923b49 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -3,7 +3,13 @@ include ../constants.mk all: pytest pytest: grpc-gen - pytest --image="${FACT_IMAGE_NAME}" --junit-xml=results.xml + pytest --image="${FACT_IMAGE_NAME}" --output=grpc --junit-xml=results.xml + +pytest-otlp: grpc-gen + pytest --image="${FACT_IMAGE_NAME}" --output=otlp --junit-xml=results-otlp.xml + +pytest-all: grpc-gen + pytest --image="${FACT_IMAGE_NAME}-otel" --output=all --junit-xml=results-all.xml PYOUT = $(CURDIR) @@ -29,4 +35,4 @@ clean: rm -rf logs.tar.gz rm -f results.xml -.PHONY: all pytest grpc-gen lint clean +.PHONY: all pytest pytest-otlp pytest-all grpc-gen lint clean diff --git a/tests/conftest.py b/tests/conftest.py index e366f9cf..40e27665 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,7 +12,7 @@ import requests import yaml -from server import FileActivityService +from server import EventServer, GrpcServer, OtlpServer # Declare files holding fixtures pytest_plugins = ['test_editors.commons'] @@ -67,12 +67,34 @@ def docker_client(): return docker.from_env() +def _get_output_modes(config: pytest.Config) -> list[str]: + output = config.getoption('--output') + assert isinstance(output, str) + if output == 'all': + return ['grpc', 'otlp'] + return [output] + + +def pytest_generate_tests(metafunc: pytest.Metafunc): + if 'server' in metafunc.fixturenames: + modes = _get_output_modes(metafunc.config) + metafunc.parametrize('server', modes, indirect=True) + + @pytest.fixture -def server(): +def server(request: pytest.FixtureRequest): """ - Fixture to start and stop the FileActivityService. + Start and stop an event server. + + Parameterised via --output to create either a GrpcServer or an + OtlpServer. When --output=all, every test that uses this fixture + runs once per output mode. """ - s = FileActivityService() + mode = request.param + if mode == 'otlp': + s: EventServer = OtlpServer() + else: + s = GrpcServer() s.serve() yield s s.stop() @@ -114,18 +136,16 @@ def fact_config( request: pytest.FixtureRequest, monitored_dir: str, logs_dir: str, + server: EventServer, ): cwd = os.getcwd() - config = { + config: dict = { 'paths': [ f'{monitored_dir}', f'{monitored_dir}/**/*', '/mounted/**/*', '/container-dir/**/*', ], - 'grpc': { - 'url': 'http://127.0.0.1:9999', - }, 'endpoint': { 'address': '127.0.0.1:9000', 'expose_metrics': True, @@ -134,6 +154,12 @@ def fact_config( 'json': True, 'scan_interval': 0, } + + if server.output_mode == 'otlp': + config['otel'] = {'endpoint': 'http://127.0.0.1:4318/v1/logs'} + else: + config['grpc'] = {'url': 'http://127.0.0.1:9999'} + config_file = NamedTemporaryFile( # noqa: SIM115 prefix='fact-config-', suffix='.yml', @@ -190,7 +216,7 @@ def fact( request: pytest.FixtureRequest, docker_client: docker.DockerClient, fact_config: tuple[dict, str], - server: FileActivityService, + server: EventServer, logs_dir: str, test_file: str, ): @@ -206,6 +232,8 @@ def fact( environment={ 'FACT_LOGLEVEL': 'debug', 'FACT_HOST_MOUNT': '/host', + 'OTEL_BLRP_SCHEDULE_DELAY': '100', + 'OTEL_BLRP_MAX_EXPORT_BATCH_SIZE': '1', }, name='fact', network_mode='host', @@ -264,3 +292,10 @@ def pytest_addoption(parser: pytest.Parser): default='quay.io/stackrox-io/fact:latest', help='The image to be used for testing', ) + parser.addoption( + '--output', + action='store', + default='grpc', + choices=['grpc', 'otlp', 'all'], + help='Output mode to test: grpc, otlp, or all (default: grpc)', + ) diff --git a/tests/event.py b/tests/event.py index 2993b396..aab4ac87 100644 --- a/tests/event.py +++ b/tests/event.py @@ -15,8 +15,6 @@ def override(func): # type: ignore[reportMissingParameterType] import utils -from internalapi.sensor.collector_pb2 import ProcessSignal -from internalapi.sensor.sfa_pb2 import FileActivity def extract_container_id(cgroup: str) -> str: @@ -177,25 +175,26 @@ def container_id(self) -> str: def loginuid(self) -> int: return self._loginuid - def diff(self, other: ProcessSignal) -> dict | None: + def diff(self, other: Process) -> dict | None: """ - Compare this Process with a ProcessSignal protobuf message. + Compare this Process with another Process instance. + + PID comparison is skipped if self.pid is None. Args: - other: ProcessSignal protobuf message to compare against + other: Process instance to compare against. Returns: - None if identical, dict of differences if not matching + None if identical, dict of differences if not matching. """ diff = {} - # Compare each field if self.pid is not None: Event._diff_field(diff, 'pid', self.pid, other.pid) Event._diff_field(diff, 'uid', self.uid, other.uid) Event._diff_field(diff, 'gid', self.gid, other.gid) - Event._diff_field(diff, 'exe_path', self.exe_path, other.exec_file_path) + Event._diff_field(diff, 'exe_path', self.exe_path, other.exe_path) Event._diff_field(diff, 'args', self.args, other.args) Event._diff_field(diff, 'name', self.name, other.name) Event._diff_field( @@ -204,7 +203,7 @@ def diff(self, other: ProcessSignal) -> dict | None: self.container_id, other.container_id, ) - Event._diff_field(diff, 'loginuid', self.loginuid, other.login_uid) + Event._diff_field(diff, 'loginuid', self.loginuid, other.loginuid) return diff if diff else None @@ -302,106 +301,77 @@ def _diff_path( diff: dict, name: str, expected: str | Pattern[str] | None, - actual: str, + actual: str | Pattern[str] | None, ): """ Compare paths with regex pattern support. + + When expected is a compiled regex pattern, actual must be a + string that matches it. Otherwise a simple equality check is + performed. """ if isinstance(expected, Pattern): - if not expected.match(actual): + if not isinstance(actual, str) or not expected.match(actual): diff[name] = {'expected': f'{expected}', 'actual': actual} elif expected != actual: diff[name] = {'expected': expected, 'actual': actual} - def diff(self, other: FileActivity) -> dict | None: + def diff(self, other: Event) -> dict | None: """ - Compare this Event with a FileActivity protobuf message. + Compare this Event with another Event instance. + + Both gRPC and OTLP servers translate their native messages + into Event objects, so this method provides a single + protocol-agnostic comparison path. Args: - other: FileActivity protobuf message to compare against + other: Event instance to compare against. Returns: - None if identical, dict of differences if not matching + None if identical, dict of differences if not matching. """ diff = {} - # Check process differences first process_diff = self.process.diff(other.process) if process_diff is not None: diff['process'] = process_diff - # Check event type - event_type_expected = self.event_type.name.lower() - event_type_actual = other.WhichOneof('file') - Event._diff_field( diff, 'event_type', - event_type_expected, - event_type_actual, + self.event_type, + other.event_type, ) if diff: return diff - # Get the appropriate event field based on type - event_field = getattr(other, event_type_expected) - # Rename handling is a bit different to the rest, since it has # new and old paths. - if self.event_type == EventType.RENAME: - Event._diff_path(diff, 'new_file', self.file, event_field.new.path) + if self.event_type != EventType.RENAME: + Event._diff_path(diff, 'file', self.file, other.file) + Event._diff_path(diff, 'host_path', self.host_path, other.host_path) + else: + Event._diff_path(diff, 'new_file', self.file, other.file) Event._diff_path( - diff, - 'new_host_path', - self.host_path, - event_field.new.host_path, + diff, 'new_host_path', self.host_path, other.host_path ) + Event._diff_path(diff, 'old_file', self.old_file, other.old_file) Event._diff_path( - diff, - 'old_file', - self.old_file, - event_field.old.path, + diff, 'old_host_path', self.old_host_path, other.old_host_path ) - Event._diff_path( - diff, - 'old_host_path', - self.old_host_path, - event_field.old.host_path, - ) - return diff if diff else None - - # Compare file and host_path (common to all event types) - # All event types have .activity.path and .activity.host_path - # accessed differently - Event._diff_path(diff, 'file', self.file, event_field.activity.path) - Event._diff_path( - diff, - 'host_path', - self.host_path, - event_field.activity.host_path, - ) if self.event_type == EventType.PERMISSION: - Event._diff_field(diff, 'mode', self.mode, event_field.mode) + Event._diff_field(diff, 'mode', self.mode, other.mode) elif self.event_type == EventType.OWNERSHIP: Event._diff_field( - diff, - 'owner_uid', - self.owner_uid, - event_field.uid, + diff, 'owner_uid', self.owner_uid, other.owner_uid ) Event._diff_field( - diff, - 'owner_gid', - self.owner_gid, - event_field.gid, + diff, 'owner_gid', self.owner_gid, other.owner_gid ) elif self.event_type in (EventType.XATTR_SET, EventType.XATTR_REMOVE): Event._diff_field( - diff, - 'xattr_name', - self.xattr_name, - event_field.xattr_name, + diff, 'xattr_name', self.xattr_name, other.xattr_name ) return diff if diff else None diff --git a/tests/requirements.txt b/tests/requirements.txt index 61679796..2e47e4fd 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -1,6 +1,7 @@ docker==7.1.0 grpcio==1.76.0 grpcio-tools==1.76.0 +opentelemetry-proto==1.43.0 pytest==8.4.1 requests==2.32.4 pyyaml==6.0.3 diff --git a/tests/server.py b/tests/server.py index 89835cd6..e2ae9325 100644 --- a/tests/server.py +++ b/tests/server.py @@ -1,91 +1,76 @@ from __future__ import annotations import json +from abc import ABC, abstractmethod from collections import deque +from collections.abc import Iterable from concurrent import futures +from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Event as ThreadingEvent +from threading import Thread from time import sleep from typing import TYPE_CHECKING, Any import grpc +from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ( + ExportLogsServiceRequest, + ExportLogsServiceResponse, +) +import utils +from event import Event, EventType, Process from internalapi.sensor import sfa_iservice_pb2_grpc +from internalapi.sensor.sfa_pb2 import FileActivity if TYPE_CHECKING: - from event import Event + from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue + from opentelemetry.proto.logs.v1.logs_pb2 import LogRecord +EVENT_TYPE_MAP: dict[str, EventType] = { + 'open': EventType.OPEN, + 'creation': EventType.CREATION, + 'unlink': EventType.UNLINK, + 'permission': EventType.PERMISSION, + 'ownership': EventType.OWNERSHIP, + 'rename': EventType.RENAME, + 'xattr_set': EventType.XATTR_SET, + 'xattr_remove': EventType.XATTR_REMOVE, +} -class FileActivityService(sfa_iservice_pb2_grpc.FileActivityServiceServicer): - """ - GRPC server for the File Activity Service. - This service allows clients to communicate file activity events. - """ + +class EventServer(ABC): + """Base class for event-receiving test servers.""" def __init__(self): - """ - Initializes the FileActivityService. - Sets up the GRPC server, a queue for incoming requests, and an - event for other threads to know when the server stops. - """ - super().__init__() - self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=2)) - self.queue = deque() + self.queue: deque[Event] = deque() self.running = ThreadingEvent() self.executor = futures.ThreadPoolExecutor(max_workers=2) - def Communicate(self, request_iterator: Any, context: Any): - """ - GRPC method to receive a stream of file activity requests. - Appends each incoming request to an internal queue. - """ - for req in request_iterator: - self.queue.append(req) + @property + @abstractmethod + def output_mode(self) -> str: ... - def serve(self, addr: str = '0.0.0.0:9999'): - """ - Starts the GRPC server. - Sets the running event once the server starts. - """ - sfa_iservice_pb2_grpc.add_FileActivityServiceServicer_to_server( - self, - self.server, - ) - self.server.add_insecure_port(addr) - self.server.start() - self.running.set() + @abstractmethod + def serve(self) -> None: ... - def stop(self): - """ - Stops the GRPC server. - Marks the server as not running once it is done. - """ - self.server.stop(1) - self.running.clear() + @abstractmethod + def stop(self) -> None: ... - def get_next(self): + def get_next(self) -> Event | None: """ - Retrieves and removes the next file activity event from the - queue. + Retrieve and remove the next event from the queue. Returns None if the queue is empty. """ if self.is_empty(): return None return self.queue.popleft() - def is_empty(self): - """ - Checks if the internal queue of file activity events is empty. - Returns: - bool: True if the queue is empty, False otherwise. - """ + def is_empty(self) -> bool: + """Check if the internal queue of events is empty.""" return len(self.queue) == 0 - def is_running(self): - """ - Checks if the GRPC server is currently running. - Returns: - bool: True if the server is running, False otherwise. - """ + def is_running(self) -> bool: + """Check if the server is currently running.""" return self.running.is_set() def _wait_events( @@ -103,20 +88,19 @@ def _wait_events( print(f'Got event: {msg}') - if skip_xattr and msg.WhichOneof('file') in ( - 'xattr_set', - 'xattr_remove', + if skip_xattr and msg.event_type in ( + EventType.XATTR_SET, + EventType.XATTR_REMOVE, ): continue - # Check if msg matches the next expected event diff = events[0].diff(msg) if diff is None: events.pop(0) if len(events) == 0: return elif strict: - raise ValueError(json.dumps(diff, indent=4)) + raise ValueError(json.dumps(diff, indent=4, default=str)) def wait_events( self, @@ -129,8 +113,9 @@ def wait_events( specified events are found. Args: - events (list['Event']): The events to search for. - strict (bool): Fail if an unexpected event is detected. + events: The events to search for. + strict: Fail if an unexpected event is detected. + skip_xattr: Skip xattr events when waiting. Raises: TimeoutError: If the required events are not found in 5 seconds. @@ -150,3 +135,250 @@ def wait_events( raise finally: cancel.set() + + +class GrpcServer( + EventServer, sfa_iservice_pb2_grpc.FileActivityServiceServicer +): + """gRPC server for the File Activity Service.""" + + def __init__(self): + super().__init__() + self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=2)) + + @property + def output_mode(self) -> str: + return 'grpc' + + @staticmethod + def _translate(msg: FileActivity) -> Event | None: + """Translate a FileActivity protobuf message into an Event.""" + oneof = msg.WhichOneof('file') + if oneof is None or oneof not in EVENT_TYPE_MAP: + return None + + event_type = EVENT_TYPE_MAP[oneof] + field = getattr(msg, oneof) + + proc = msg.process + process = Process( + pid=proc.pid, + uid=proc.uid, + gid=proc.gid, + exe_path=proc.exec_file_path, + args=proc.args, + name=proc.name, + container_id=proc.container_id, + loginuid=proc.login_uid, + ) + + if event_type != EventType.RENAME: + file_path = field.activity.path + host_path = field.activity.host_path + else: + file_path = field.new.path + host_path = field.new.host_path + + kwargs: dict[str, Any] = { + 'file': file_path, + 'host_path': host_path, + } + + if event_type == EventType.RENAME: + kwargs['old_file'] = field.old.path + kwargs['old_host_path'] = field.old.host_path + elif event_type == EventType.PERMISSION: + kwargs['mode'] = field.mode + elif event_type == EventType.OWNERSHIP: + kwargs['owner_uid'] = field.uid + kwargs['owner_gid'] = field.gid + elif event_type in (EventType.XATTR_SET, EventType.XATTR_REMOVE): + kwargs['xattr_name'] = field.xattr_name + + return Event( + process=process, + event_type=event_type, + **kwargs, + ) + + def Communicate(self, request_iterator: Any, context: Any): + """ + gRPC method to receive a stream of file activity requests. + Translates each FileActivity protobuf into an Event and + appends it to the queue. + """ + for req in request_iterator: + event = self._translate(req) + if event is not None: + self.queue.append(event) + + def serve(self, addr: str = '0.0.0.0:9999'): + """Start the gRPC server on the given address.""" + sfa_iservice_pb2_grpc.add_FileActivityServiceServicer_to_server( + self, + self.server, + ) + self.server.add_insecure_port(addr) + self.server.start() + self.running.set() + + def stop(self): + """Stop the gRPC server.""" + self.server.stop(1) + self.running.clear() + + +class OtlpServer(EventServer): + """HTTP server that receives OTLP/HTTP binary protobuf log exports.""" + + def __init__(self): + super().__init__() + self._httpd: HTTPServer | None = None + self._thread: Thread | None = None + + @property + def output_mode(self) -> str: + return 'otlp' + + @staticmethod + def _anyvalue_to_python(value: AnyValue) -> Any: + """Convert an OTLP AnyValue protobuf to a native Python type.""" + kind = value.WhichOneof('value') + if kind == 'string_value': + return value.string_value + if kind == 'int_value': + return value.int_value + if kind == 'bool_value': + return value.bool_value + if kind == 'double_value': + return value.double_value + if kind == 'kvlist_value': + return OtlpServer._kvlist_to_dict(value.kvlist_value.values) + if kind == 'array_value': + return [ + OtlpServer._anyvalue_to_python(v) + for v in value.array_value.values + ] + if kind == 'bytes_value': + return value.bytes_value + return None + + @staticmethod + def _kvlist_to_dict(kvs: Iterable[KeyValue]) -> dict[str, Any]: + """Convert a list of OTLP KeyValue pairs to a Python dict.""" + return {kv.key: OtlpServer._anyvalue_to_python(kv.value) for kv in kvs} + + @staticmethod + def _translate(record: LogRecord) -> Event | None: + """ + Translate an OTLP LogRecord into an Event. + + Returns None for event types that only exist in the OTLP + schema (mkdir, rmdir) or are otherwise unrecognised. + """ + attrs = OtlpServer._kvlist_to_dict(record.attributes) + + file_data = attrs.get('file', {}) + event_type_str = file_data.get('event_type', '') + + if event_type_str not in EVENT_TYPE_MAP: + return None + + event_type = EVENT_TYPE_MAP[event_type_str] + + proc_data = attrs.get('process', {}) + args_list = proc_data.get('args', []) + args = ( + utils.rust_style_join(args_list) + if isinstance(args_list, list) + else str(args_list) + ) + + process = Process( + pid=proc_data.get('pid', 0), + uid=proc_data.get('uid', 0), + gid=proc_data.get('gid', 0), + exe_path=proc_data.get('exe_path', ''), + args=args, + name=proc_data.get('comm', ''), + container_id=proc_data.get('container_id', ''), + loginuid=proc_data.get('login_uid', 0), + ) + + kwargs: dict[str, Any] = { + 'file': file_data.get('filename', ''), + 'host_path': file_data.get('host_path', ''), + } + if event_type == EventType.RENAME: + old = file_data.get('old', {}) + kwargs['old_file'] = old.get('filename', '') + kwargs['old_host_path'] = old.get('host_path', '') + elif event_type == EventType.PERMISSION: + kwargs['mode'] = file_data.get('new_mode') + elif event_type == EventType.OWNERSHIP: + kwargs['owner_uid'] = file_data.get('new_uid') + kwargs['owner_gid'] = file_data.get('new_gid') + elif event_type in (EventType.XATTR_SET, EventType.XATTR_REMOVE): + kwargs['xattr_name'] = file_data.get('xattr_name') + + return Event( + process=process, + event_type=event_type, + **kwargs, + ) + + def serve(self, addr: str = '0.0.0.0', port: int = 4318): + """ + Start the HTTP server on the given address and port. + + Handles POST /v1/logs with OTLP binary protobuf payloads. + Each log record is translated into an Event and appended + to the queue. Events with types not present in the gRPC + schema (mkdir, rmdir) are silently dropped. + """ + parent = self + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + if self.path != '/v1/logs': + self.send_error(404) + return + + length = int(self.headers.get('Content-Length', 0)) + body = self.rfile.read(length) + + request = ExportLogsServiceRequest() + request.ParseFromString(body) + + for resource_logs in request.resource_logs: + for scope_logs in resource_logs.scope_logs: + for record in scope_logs.log_records: + event = OtlpServer._translate(record) + if event is not None: + parent.queue.append(event) + + response = ExportLogsServiceResponse() + response_bytes = response.SerializeToString() + + self.send_response(200) + self.send_header('Content-Type', 'application/x-protobuf') + self.send_header('Content-Length', str(len(response_bytes))) + self.end_headers() + self.wfile.write(response_bytes) + + def log_message(self, format: str, *args: Any): + pass + + self._httpd = HTTPServer((addr, port), Handler) + self._thread = Thread(target=self._httpd.serve_forever, daemon=True) + self._thread.start() + self.running.set() + + def stop(self): + """Stop the HTTP server and close its socket.""" + if self._httpd is not None: + self._httpd.shutdown() + self._httpd.server_close() + if self._thread is not None: + self._thread.join(timeout=5) + self.running.clear() diff --git a/tests/test_config_hotreload.py b/tests/test_config_hotreload.py index f9aa3d91..7ef6a8f5 100644 --- a/tests/test_config_hotreload.py +++ b/tests/test_config_hotreload.py @@ -10,7 +10,7 @@ import yaml from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer, GrpcServer DEFAULT_URL = 'http://127.0.0.1:9000' @@ -96,10 +96,10 @@ def test_endpoint_address_change( @pytest.fixture def alternate_server(): """ - Fixture to start and stop a FileActivityService on an alternate + Fixture to start and stop a GrpcServer on an alternate address. """ - s = FileActivityService() + s = GrpcServer() s.serve(f'0.0.0.0:{ALTERNATE_PORT}') yield s s.stop() @@ -109,13 +109,16 @@ def test_output_grpc_address_change( fact: docker.models.containers.Container, fact_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, - alternate_server: FileActivityService, + server: EventServer, + alternate_server: EventServer, ): """ Tests we can receive events on a new endpoint after a configuration change. """ + if server.output_mode != 'grpc': + pytest.skip('gRPC-specific test') + # File Under Test fut = os.path.join(monitored_dir, 'test2.txt') with open(fut, 'w') as f: @@ -154,7 +157,7 @@ def test_paths( fact_config: tuple[dict, str], monitored_dir: str, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): p = Process.from_proc() @@ -199,7 +202,7 @@ def test_no_paths_then_add( fact: docker.models.containers.Container, fact_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Start with no paths configured, verify no events are produced, @@ -238,7 +241,7 @@ def test_paths_then_remove( fact: docker.models.containers.Container, fact_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Start with paths configured, verify events are produced, @@ -274,7 +277,7 @@ def test_paths_addition( fact_config: tuple[dict, str], monitored_dir: str, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): p = Process.from_proc() diff --git a/tests/test_editors/test_nvim.py b/tests/test_editors/test_nvim.py index 0bb269dd..d2217440 100644 --- a/tests/test_editors/test_nvim.py +++ b/tests/test_editors/test_nvim.py @@ -3,13 +3,13 @@ import docker.models.containers from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from test_editors.commons import get_vi_test_file def test_new_file( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/mounted/test.txt' @@ -37,7 +37,7 @@ def test_new_file( def test_open_file( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/mounted/test.txt' @@ -125,7 +125,7 @@ def test_open_file( def test_new_file_ovfs( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/container-dir/test.txt' @@ -153,7 +153,7 @@ def test_new_file_ovfs( def test_open_file_ovfs( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/container-dir/test.txt' diff --git a/tests/test_editors/test_sed.py b/tests/test_editors/test_sed.py index cb239ec8..7f823974 100644 --- a/tests/test_editors/test_sed.py +++ b/tests/test_editors/test_sed.py @@ -5,12 +5,12 @@ import docker.models.containers from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer def test_sed( vi_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert vi_container.id is not None # File Under Test @@ -73,7 +73,7 @@ def test_sed( def test_sed_ovfs( vi_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert vi_container.id is not None # File Under Test diff --git a/tests/test_editors/test_vi.py b/tests/test_editors/test_vi.py index 27f6d1bf..5cd9239c 100644 --- a/tests/test_editors/test_vi.py +++ b/tests/test_editors/test_vi.py @@ -3,13 +3,13 @@ import docker.models.containers from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from test_editors.commons import get_vi_test_file def test_new_file( vi_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert vi_container.id is not None fut = '/mounted/test.txt' @@ -78,7 +78,7 @@ def test_new_file( def test_new_file_ovfs( vi_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert vi_container.id is not None fut = '/container-dir/test.txt' @@ -147,7 +147,7 @@ def test_new_file_ovfs( def test_open_file( vi_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert vi_container.id is not None fut = '/mounted/test.txt' @@ -281,7 +281,7 @@ def test_open_file( def test_open_file_ovfs( vi_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert vi_container.id is not None fut = '/container-dir/test.txt' diff --git a/tests/test_editors/test_vim.py b/tests/test_editors/test_vim.py index e9ff6f21..4dabb8b5 100644 --- a/tests/test_editors/test_vim.py +++ b/tests/test_editors/test_vim.py @@ -3,13 +3,13 @@ import docker.models.containers from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from test_editors.commons import get_vi_test_file def test_new_file( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/mounted/test.txt' @@ -76,7 +76,7 @@ def test_new_file( def test_new_file_ovfs( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/container-dir/test.txt' @@ -143,7 +143,7 @@ def test_new_file_ovfs( def test_open_file( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/mounted/test.txt' @@ -276,7 +276,7 @@ def test_open_file( def test_open_file_ovfs( editor_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert editor_container.id is not None fut = '/container-dir/test.txt' diff --git a/tests/test_file_open.py b/tests/test_file_open.py index 706b6473..b1ef5719 100644 --- a/tests/test_file_open.py +++ b/tests/test_file_open.py @@ -8,7 +8,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from utils import join_path_with_filename, path_to_string @@ -25,7 +25,7 @@ ) def test_open( monitored_dir: str, - server: FileActivityService, + server: EventServer, filename: str | bytes, ): """ @@ -56,7 +56,7 @@ def test_open( server.wait_events([e]) -def test_multiple(monitored_dir: str, server: FileActivityService): +def test_multiple(monitored_dir: str, server: EventServer): """ Tests the opening of multiple files and verifies that the corresponding events are captured by the server. @@ -86,7 +86,7 @@ def test_multiple(monitored_dir: str, server: FileActivityService): server.wait_events(events) -def test_multiple_access(test_file: str, server: FileActivityService): +def test_multiple_access(test_file: str, server: EventServer): """ Tests multiple opening of a file and verifies that the corresponding events are captured by the server. @@ -112,7 +112,7 @@ def test_multiple_access(test_file: str, server: FileActivityService): server.wait_events(events) -def test_ignored(test_file: str, ignored_dir: str, server: FileActivityService): +def test_ignored(test_file: str, ignored_dir: str, server: EventServer): """ Tests that open events on ignored files are not captured by the server. @@ -153,7 +153,7 @@ def do_test(fut: str, stop_event: MpEvent): stop_event.wait() -def test_external_process(monitored_dir: str, server: FileActivityService): +def test_external_process(monitored_dir: str, server: EventServer): """ Tests the opening of a file by an external process and verifies that the corresponding event is captured by the server. @@ -192,7 +192,7 @@ def test_external_process(monitored_dir: str, server: FileActivityService): def test_overlay( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test @@ -222,7 +222,7 @@ def test_overlay( def test_mounted_dir( test_container: docker.models.containers.Container, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test @@ -251,7 +251,7 @@ def test_mounted_dir( def test_unmonitored_mounted_dir( test_container: docker.models.containers.Container, test_file: str, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test diff --git a/tests/test_misc.py b/tests/test_misc.py index 7605d94b..ce275239 100644 --- a/tests/test_misc.py +++ b/tests/test_misc.py @@ -9,7 +9,7 @@ from conftest import dump_logs from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer @pytest.fixture @@ -54,7 +54,7 @@ def run_self_deleter( def test_d_path_sanitization( monitored_dir: str, - server: FileActivityService, + server: EventServer, run_self_deleter: docker.models.containers.Container, docker_client: docker.DockerClient, ): diff --git a/tests/test_path_chmod.py b/tests/test_path_chmod.py index be775a7c..d1d5dd7c 100644 --- a/tests/test_path_chmod.py +++ b/tests/test_path_chmod.py @@ -8,7 +8,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from utils import join_path_with_filename, path_to_string @@ -25,7 +25,7 @@ ) def test_chmod( monitored_dir: str, - server: FileActivityService, + server: EventServer, filename: str | bytes, ): """ @@ -70,7 +70,7 @@ def test_chmod( server.wait_events(events) -def test_multiple(monitored_dir: str, server: FileActivityService): +def test_multiple(monitored_dir: str, server: EventServer): """ Tests modifying permissions on multiple files. @@ -109,7 +109,7 @@ def test_multiple(monitored_dir: str, server: FileActivityService): server.wait_events(events) -def test_ignored(test_file: str, ignored_dir: str, server: FileActivityService): +def test_ignored(test_file: str, ignored_dir: str, server: EventServer): """ Tests that permission events on ignored files are not captured. @@ -150,7 +150,7 @@ def do_test(fut: str, mode: int, stop_event: MpEvent): stop_event.wait() -def test_external_process(monitored_dir: str, server: FileActivityService): +def test_external_process(monitored_dir: str, server: EventServer): """ Tests permission change of a file by an external process and verifies that the corresponding event is captured by the server. @@ -193,7 +193,7 @@ def test_external_process(monitored_dir: str, server: FileActivityService): def test_overlay( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): """ Test permission changes on an overlayfs file (inside a container) @@ -245,7 +245,7 @@ def test_overlay( def test_mounted_dir( test_container: docker.models.containers.Container, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Test permission changes on a file bind mounted into a container @@ -300,7 +300,7 @@ def test_mounted_dir( def test_unmonitored_mounted_dir( test_container: docker.models.containers.Container, test_file: str, - server: FileActivityService, + server: EventServer, ): """ Test permission changes on a file bind mounted to a container and diff --git a/tests/test_path_chown.py b/tests/test_path_chown.py index f86d749c..27fd25c0 100644 --- a/tests/test_path_chown.py +++ b/tests/test_path_chown.py @@ -4,7 +4,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from utils import path_to_string, rust_style_quote # Tests here have to use a container to do 'chown', @@ -28,7 +28,7 @@ ) def test_chown( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, filename: str | bytes, ): """ @@ -88,7 +88,7 @@ def test_chown( def test_multiple( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): """ Tests ownership operations on multiple files and verifies the corresponding @@ -146,7 +146,7 @@ def test_multiple( def test_ignored( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): """ Tests that ownership events on ignored files are not captured by the @@ -204,7 +204,7 @@ def test_ignored( def test_no_change( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): """ Tests that chown to the same UID/GID triggers events for all calls. diff --git a/tests/test_path_mkdir.py b/tests/test_path_mkdir.py index d18efe9f..6c45dee5 100644 --- a/tests/test_path_mkdir.py +++ b/tests/test_path_mkdir.py @@ -5,7 +5,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer @pytest.mark.parametrize( @@ -19,7 +19,7 @@ ) def test_mkdir_nested( monitored_dir: str, - server: FileActivityService, + server: EventServer, dirname: str, ): """ @@ -58,7 +58,7 @@ def test_mkdir_nested( def test_mkdir_ignored( monitored_dir: str, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Tests that directories created outside monitored paths are ignored. diff --git a/tests/test_path_rename.py b/tests/test_path_rename.py index 5445c76e..00987509 100644 --- a/tests/test_path_rename.py +++ b/tests/test_path_rename.py @@ -6,7 +6,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from utils import join_path_with_filename, path_to_string @@ -23,7 +23,7 @@ ) def test_rename( monitored_dir: str, - server: FileActivityService, + server: EventServer, filename: str | bytes, ): """ @@ -79,7 +79,7 @@ def test_rename( def test_ignored( monitored_dir: str, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Tests that rename events on ignored files are not captured by the @@ -134,9 +134,7 @@ def test_ignored( ) -def test_rename_dir( - monitored_dir: str, ignored_dir: str, server: FileActivityService -): +def test_rename_dir(monitored_dir: str, ignored_dir: str, server: EventServer): """ Test renaming a directory is caught @@ -249,7 +247,7 @@ def test_rename_overwrite( to_monitored: bool, monitored_dir: str, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): events = [] p = Process.from_proc() @@ -314,7 +312,7 @@ def test_rename_overwrite( def test_overlay( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test @@ -360,7 +358,7 @@ def test_overlay( def test_mounted_dir( test_container: docker.models.containers.Container, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test @@ -407,7 +405,7 @@ def test_mounted_dir( def test_cross_mountpoints( test_container: docker.models.containers.Container, monitored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Attempt to rename files/directories across mountpoints diff --git a/tests/test_path_rmdir.py b/tests/test_path_rmdir.py index 0dba5780..fcefa185 100644 --- a/tests/test_path_rmdir.py +++ b/tests/test_path_rmdir.py @@ -7,7 +7,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from utils import get_metric_value @@ -71,7 +71,7 @@ def get_kernel_rmdir_processed(fact_config: tuple[dict, str]): ) def test_rmdir_empty( monitored_dir: str, - server: FileActivityService, + server: EventServer, fact_config: tuple[dict, str], dirname: str, ): @@ -161,7 +161,7 @@ def test_rmdir_empty( ) def test_rmdir_recursive( monitored_dir: str, - server: FileActivityService, + server: EventServer, fact_config: tuple[dict, str], ): """ @@ -278,7 +278,7 @@ def test_rmdir_recursive( def test_rmdir_ignored( monitored_dir: str, ignored_dir: str, - server: FileActivityService, + server: EventServer, fact_config: tuple[dict, str], ): """ @@ -372,7 +372,7 @@ def test_rmdir_ignored( def test_rmdir_with_parent_inode( monitored_dir: str, - server: FileActivityService, + server: EventServer, fact_config: tuple[dict, str], ): """ diff --git a/tests/test_path_unlink.py b/tests/test_path_unlink.py index af3fda98..cce47557 100644 --- a/tests/test_path_unlink.py +++ b/tests/test_path_unlink.py @@ -8,7 +8,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from utils import join_path_with_filename, path_to_string @@ -25,7 +25,7 @@ ) def test_remove( monitored_dir: str, - server: FileActivityService, + server: EventServer, filename: str | bytes, ): """ @@ -73,7 +73,7 @@ def test_remove( server.wait_events(events) -def test_multiple(monitored_dir: str, server: FileActivityService): +def test_multiple(monitored_dir: str, server: EventServer): """ Tests the removal of multiple files and verifies the corresponding events are captured by the server. @@ -112,7 +112,7 @@ def test_multiple(monitored_dir: str, server: FileActivityService): server.wait_events(events) -def test_ignored(test_file: str, ignored_dir: str, server: FileActivityService): +def test_ignored(test_file: str, ignored_dir: str, server: EventServer): """ Tests that unlink events on ignored files are not captured by the server. @@ -152,7 +152,7 @@ def do_test(fut: str, stop_event: MpEvent): stop_event.wait() -def test_external_process(monitored_dir: str, server: FileActivityService): +def test_external_process(monitored_dir: str, server: EventServer): """ Tests the removal of a file by an external process and verifies that the corresponding event is captured by the server. @@ -192,7 +192,7 @@ def test_external_process(monitored_dir: str, server: FileActivityService): def test_overlay( test_container: docker.models.containers.Container, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test @@ -235,7 +235,7 @@ def test_overlay( def test_mounted_dir( test_container: docker.models.containers.Container, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test @@ -279,7 +279,7 @@ def test_mounted_dir( def test_unmonitored_mounted_dir( test_container: docker.models.containers.Container, test_file: str, - server: FileActivityService, + server: EventServer, ): assert test_container.id is not None # File Under Test diff --git a/tests/test_rate_limit.py b/tests/test_rate_limit.py index af15eea0..53bb16f5 100644 --- a/tests/test_rate_limit.py +++ b/tests/test_rate_limit.py @@ -10,7 +10,7 @@ import yaml from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer @pytest.fixture @@ -36,7 +36,7 @@ def rate_limited_config( def test_rate_limit_drops_events( rate_limited_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Test that the rate limiter drops events when the rate limit is exceeded. @@ -98,7 +98,7 @@ def test_rate_limit_drops_events( def test_rate_limit_unlimited( monitored_dir: str, - server: FileActivityService, + server: EventServer, fact_config: tuple[dict, str], ): """ diff --git a/tests/test_wildcard.py b/tests/test_wildcard.py index cce40875..c2a2f01c 100644 --- a/tests/test_wildcard.py +++ b/tests/test_wildcard.py @@ -8,7 +8,7 @@ import yaml from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer @pytest.fixture @@ -35,7 +35,7 @@ def wildcard_config( def test_extension_wildcard( wildcard_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): process = Process.from_proc() @@ -64,7 +64,7 @@ def test_extension_wildcard( def test_prefix_wildcard( wildcard_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): process = Process.from_proc() @@ -93,7 +93,7 @@ def test_prefix_wildcard( def test_recursive_wildcard( wildcard_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): process = Process.from_proc() @@ -137,7 +137,7 @@ def test_recursive_wildcard( def test_nonrecursive_wildcard( wildcard_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): process = Process.from_proc() @@ -161,7 +161,7 @@ def test_nonrecursive_wildcard( def test_multiple_patterns( wildcard_config: tuple[dict, str], monitored_dir: str, - server: FileActivityService, + server: EventServer, ): process = Process.from_proc() diff --git a/tests/test_xattr.py b/tests/test_xattr.py index eb2b1971..bdaf2cea 100644 --- a/tests/test_xattr.py +++ b/tests/test_xattr.py @@ -7,7 +7,7 @@ import pytest from event import Event, EventType, Process -from server import FileActivityService +from server import EventServer from utils import join_path_with_filename, path_to_string @@ -38,7 +38,7 @@ def _xattr_supported() -> bool: def test_setxattr( test_file: str, - server: FileActivityService, + server: EventServer, ): """ Tests that setting a user xattr on a monitored file generates @@ -71,7 +71,7 @@ def test_setxattr( def test_xattr_set_and_remove( test_file: str, - server: FileActivityService, + server: EventServer, ): """ Tests that setting and then removing a user xattr from a monitored @@ -109,7 +109,7 @@ def test_xattr_set_and_remove( def test_xattr_multiple( test_file: str, - server: FileActivityService, + server: EventServer, ): """ Tests that setting and removing multiple xattrs on a monitored file @@ -180,7 +180,7 @@ def test_xattr_multiple( def test_xattr_ignored( test_file: str, ignored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Tests that xattr changes on unmonitored files are not tracked, @@ -233,7 +233,7 @@ def test_xattr_ignored( def test_xattr_new_file( monitored_dir: str, - server: FileActivityService, + server: EventServer, ): """ Tests that xattr tracking works for files created while fact is @@ -301,7 +301,7 @@ def test_xattr_new_file( ) def test_xattr_utf8_filenames( monitored_dir: str, - server: FileActivityService, + server: EventServer, filename: str | bytes, ): """ @@ -371,7 +371,7 @@ def test_xattr_utf8_filenames( ) def test_xattr_utf8_names( test_file: str, - server: FileActivityService, + server: EventServer, xattr_name: str, ): """ diff --git a/tests/utils.py b/tests/utils.py index 1026c1ea..85eb176a 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -58,10 +58,12 @@ def rust_style_quote(s: str): """ if not s: return "''" - if re.search(r'[^a-zA-Z0-9_.:/-]', s): - # Try to match the behavior of shlex.try_join() - if "'" in s and '"' not in s: - return f'"{s}"' + if re.search(r'[^a-zA-Z0-9_.:\-/+@\]]', s): + # Backslash and single-quote cannot appear in single-quoted + # strings in Rust's shlex, use double-quoting instead. + if ('\\' in s or "'" in s) and '`' not in s and '$' not in s: + escaped = s.replace('\\', '\\\\').replace('"', '\\"') + return f'"{escaped}"' escaped = s.replace("'", "\\'") return f"'{escaped}'" return s From f50505e29d4df6438e37ad58cddab6aca7ecb234 Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Mon, 6 Jul 2026 12:51:32 +0200 Subject: [PATCH 7/8] feat(ci): build otel images This enables building images with the opentelemetry output enabled, making it easier to run integration tests with it. --- .github/workflows/ci.yml | 130 +++----------------------- .github/workflows/container-build.yml | 125 +++++++++++++++++++++++++ Makefile | 11 +-- 3 files changed, 142 insertions(+), 124 deletions(-) create mode 100644 .github/workflows/container-build.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ff25db7f..8b92e520 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -128,113 +128,16 @@ jobs: pip install ruff make format-check - vars: - runs-on: ubuntu-24.04 - outputs: - tag: ${{ steps.vars.outputs.tag }} - stackrox-io-image: ${{ steps.vars.outputs.stackrox-io-image }} - rhacs-eng-image: ${{ steps.vars.outputs.rhacs-eng-image }} - steps: - - uses: actions/checkout@v4 - with: - submodules: true - fetch-depth: 0 - - - id: vars - run: | - cat << EOF >> "$GITHUB_OUTPUT" - tag=$(make tag) - stackrox-io-image=$(make image-name) - rhacs-eng-image=$(FACT_REGISTRY="quay.io/rhacs-eng/fact" make image-name) - EOF - - container: - needs: - - vars - strategy: - fail-fast: false - matrix: - arch: - - amd64 - - arm64 - runs-on: ${{ (matrix.arch == 'arm64' && 'ubuntu-24.04-arm') || 'ubuntu-24.04' }} - env: - FACT_TAG: ${{ needs.vars.outputs.tag }}-${{ matrix.arch }} - FACT_VERSION: ${{ needs.vars.outputs.tag }} - STACKROX_IO_IMAGE: ${{ needs.vars.outputs.stackrox-io-image }}-${{ matrix.arch }} - RHACS_ENG_IMAGE: ${{ needs.vars.outputs.rhacs-eng-image }}-${{ matrix.arch }} - steps: - - uses: actions/checkout@v4 - with: - submodules: true - - - name: Build image - run: DOCKER=podman make image - - - name: Login to quay.io/stackrox-io - uses: docker/login-action@v3 - with: - registry: quay.io - username: ${{ secrets.QUAY_STACKROX_IO_RW_USERNAME }} - password: ${{ secrets.QUAY_STACKROX_IO_RW_PASSWORD }} - - - run: podman push "${STACKROX_IO_IMAGE}" - - - name: Login to quay.io/rhacs-eng - uses: docker/login-action@v3 - with: - registry: quay.io - username: ${{ secrets.QUAY_RHACS_ENG_RW_USERNAME }} - password: ${{ secrets.QUAY_RHACS_ENG_RW_PASSWORD }} - - - name: Tag and push to rhacs-eng - run: | - podman tag "${STACKROX_IO_IMAGE}" "${RHACS_ENG_IMAGE}" - podman push "${RHACS_ENG_IMAGE}" - - manifest-stackrox-io: - runs-on: ubuntu-24.04 - needs: - - container - - vars - env: - ARCHS: amd64 arm64 - steps: - - uses: redhat-actions/podman-login@v1 - with: - registry: quay.io - username: ${{ secrets.QUAY_STACKROX_IO_RW_USERNAME }} - password: ${{ secrets.QUAY_STACKROX_IO_RW_PASSWORD }} - - - name: Create multiarch manifest - run: | - podman manifest create ${{ needs.vars.outputs.stackrox-io-image }} \ - ${{ needs.vars.outputs.stackrox-io-image }}-amd64 \ - ${{ needs.vars.outputs.stackrox-io-image }}-arm64 - podman manifest inspect ${{ needs.vars.outputs.stackrox-io-image }} - podman manifest push ${{ needs.vars.outputs.stackrox-io-image }} - - manifest-rhacs-eng: - runs-on: ubuntu-24.04 - needs: - - container - - vars - env: - ARCHS: amd64 arm64 - steps: - - uses: redhat-actions/podman-login@v1 - with: - registry: quay.io - username: ${{ secrets.QUAY_RHACS_ENG_RW_USERNAME }} - password: ${{ secrets.QUAY_RHACS_ENG_RW_PASSWORD }} + container-build: + uses: ./.github/workflows/container-build.yml + secrets: inherit - - name: Create multiarch manifest - run: | - podman manifest create ${{ needs.vars.outputs.rhacs-eng-image }} \ - ${{ needs.vars.outputs.rhacs-eng-image }}-amd64 \ - ${{ needs.vars.outputs.rhacs-eng-image }}-arm64 - podman manifest inspect ${{ needs.vars.outputs.rhacs-eng-image }} - podman manifest push ${{ needs.vars.outputs.rhacs-eng-image }} + container-build-otel: + uses: ./.github/workflows/container-build.yml + secrets: inherit + with: + tag-suffix: '-otel' + make-target: 'image-otel' unit-tests: uses: ./.github/workflows/unit-tests.yml @@ -242,22 +145,17 @@ jobs: integration-tests: needs: - - vars - - manifest-stackrox-io - - manifest-rhacs-eng + - container-build uses: ./.github/workflows/integration-tests.yml with: - tag: ${{ needs.vars.outputs.tag }} + tag: ${{ needs.container-build.outputs.tag }} secrets: inherit performance-tests: if: github.event_name == 'schedule' needs: - - vars - - manifest-stackrox-io - - manifest-rhacs-eng - - integration-tests - uses: ./.github/workflows/performance-tests.yml + - container-build + uses: ./.github/workflows/integration-tests.yml with: - tag: ${{ needs.vars.outputs.tag }} + tag: ${{ needs.container-build.outputs.tag }} secrets: inherit diff --git a/.github/workflows/container-build.yml b/.github/workflows/container-build.yml new file mode 100644 index 00000000..80d69a94 --- /dev/null +++ b/.github/workflows/container-build.yml @@ -0,0 +1,125 @@ +name: Build container +on: + workflow_call: + inputs: + tag-suffix: + required: false + default: '' + type: string + make-target: + required: false + default: 'image' + type: string + outputs: + tag: + value: ${{ jobs.vars.outputs.tag }} + +jobs: + vars: + runs-on: ubuntu-24.04 + outputs: + tag: ${{ steps.vars.outputs.tag }} + stackrox-io-image: ${{ steps.vars.outputs.stackrox-io-image }} + rhacs-eng-image: ${{ steps.vars.outputs.rhacs-eng-image }} + steps: + - uses: actions/checkout@v4 + with: + submodules: true + fetch-depth: 0 + + - id: vars + run: | + cat << EOF >> "$GITHUB_OUTPUT" + tag=$(make tag)${{ inputs.tag-suffix }} + stackrox-io-image=$(make image-name)${{ inputs.tag-suffix }} + rhacs-eng-image=$(FACT_REGISTRY="quay.io/rhacs-eng/fact" make image-name)${{ inputs.tag-suffix }} + EOF + + container: + needs: + - vars + strategy: + fail-fast: false + matrix: + arch: + - amd64 + - arm64 + runs-on: ${{ (matrix.arch == 'arm64' && 'ubuntu-24.04-arm') || 'ubuntu-24.04' }} + env: + FACT_TAG: ${{ needs.vars.outputs.tag }}-${{ matrix.arch }} + FACT_VERSION: ${{ needs.vars.outputs.tag }} + STACKROX_IO_IMAGE: ${{ needs.vars.outputs.stackrox-io-image }}-${{ matrix.arch }} + RHACS_ENG_IMAGE: ${{ needs.vars.outputs.rhacs-eng-image }}-${{ matrix.arch }} + steps: + - uses: actions/checkout@v4 + with: + submodules: true + + - name: Build image + run: DOCKER=podman make ${{ inputs.make-target }} + + - name: Login to quay.io/stackrox-io + uses: docker/login-action@v3 + with: + registry: quay.io + username: ${{ secrets.QUAY_STACKROX_IO_RW_USERNAME }} + password: ${{ secrets.QUAY_STACKROX_IO_RW_PASSWORD }} + + - run: podman push "${STACKROX_IO_IMAGE}" + + - name: Login to quay.io/rhacs-eng + uses: docker/login-action@v3 + with: + registry: quay.io + username: ${{ secrets.QUAY_RHACS_ENG_RW_USERNAME }} + password: ${{ secrets.QUAY_RHACS_ENG_RW_PASSWORD }} + + - name: Tag and push to rhacs-eng + run: | + podman tag "${STACKROX_IO_IMAGE}" "${RHACS_ENG_IMAGE}" + podman push "${RHACS_ENG_IMAGE}" + + manifest-stackrox-io: + runs-on: ubuntu-24.04 + needs: + - container + - vars + env: + ARCHS: amd64 arm64 + steps: + - uses: redhat-actions/podman-login@v1 + with: + registry: quay.io + username: ${{ secrets.QUAY_STACKROX_IO_RW_USERNAME }} + password: ${{ secrets.QUAY_STACKROX_IO_RW_PASSWORD }} + + - name: Create multiarch manifest + run: | + podman manifest create ${{ needs.vars.outputs.stackrox-io-image }} \ + ${{ needs.vars.outputs.stackrox-io-image }}-amd64 \ + ${{ needs.vars.outputs.stackrox-io-image }}-arm64 + podman manifest inspect ${{ needs.vars.outputs.stackrox-io-image }} + podman manifest push ${{ needs.vars.outputs.stackrox-io-image }} + + manifest-rhacs-eng: + runs-on: ubuntu-24.04 + needs: + - container + - vars + env: + ARCHS: amd64 arm64 + steps: + - uses: redhat-actions/podman-login@v1 + with: + registry: quay.io + username: ${{ secrets.QUAY_RHACS_ENG_RW_USERNAME }} + password: ${{ secrets.QUAY_RHACS_ENG_RW_PASSWORD }} + + - name: Create multiarch manifest + run: | + podman manifest create ${{ needs.vars.outputs.rhacs-eng-image }} \ + ${{ needs.vars.outputs.rhacs-eng-image }}-amd64 \ + ${{ needs.vars.outputs.rhacs-eng-image }}-arm64 + podman manifest inspect ${{ needs.vars.outputs.rhacs-eng-image }} + podman manifest push ${{ needs.vars.outputs.rhacs-eng-image }} + diff --git a/Makefile b/Makefile index 366692c3..ef94eb3d 100644 --- a/Makefile +++ b/Makefile @@ -17,17 +17,12 @@ image: -f Containerfile \ --build-arg FACT_VERSION=$(FACT_VERSION) \ --build-arg RUST_VERSION=$(RUST_VERSION) \ + --build-arg CARGO_ARGS="$(CARGO_ARGS)" \ -t $(FACT_IMAGE_NAME) \ $(CURDIR) -image-otel: - $(DOCKER) build \ - -f Containerfile \ - --build-arg FACT_VERSION=$(FACT_VERSION) \ - --build-arg RUST_VERSION=$(RUST_VERSION) \ - --build-arg CARGO_ARGS="--features otel" \ - -t $(FACT_IMAGE_NAME)-otel \ - $(CURDIR) +image-otel: CARGO_ARGS = --features otel +image-otel: image licenses:THIRD_PARTY_LICENSES.html From 229cd7de8427304d885d19492c73816927dc936c Mon Sep 17 00:00:00 2001 From: Mauro Ezequiel Moltrasio Date: Mon, 6 Jul 2026 13:06:57 +0200 Subject: [PATCH 8/8] feat(ci): run integration tests with otel output Since the grpc output already tests all kernels we support, these tests are only run on fedora-coreos. --- .github/workflows/ci.yml | 13 +++++++++++++ .github/workflows/integration-tests.yml | 19 ++++++++++++------- .github/workflows/konflux-tests.yml | 1 + ansible/run-tests.yml | 6 +++--- 4 files changed, 29 insertions(+), 10 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8b92e520..3d51dbb1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -149,6 +149,18 @@ jobs: uses: ./.github/workflows/integration-tests.yml with: tag: ${{ needs.container-build.outputs.tag }} + vms: '["fedora-coreos", "fcarm", "rhel", "rhel-arm64", "rhcos", "rhcos-arm64"]' + secrets: inherit + + integration-tests-otel: + needs: + - container-build-otel + uses: ./.github/workflows/integration-tests.yml + with: + tag: ${{ needs.container-build-otel.outputs.tag }} + job-tag: otel + vms: '["fedora-coreos"]' + output-type: otlp secrets: inherit performance-tests: @@ -158,4 +170,5 @@ jobs: uses: ./.github/workflows/integration-tests.yml with: tag: ${{ needs.container-build.outputs.tag }} + vms: '["fedora-coreos", "fcarm", "rhel", "rhel-arm64", "rhcos", "rhcos-arm64"]' secrets: inherit diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 20002076..cebf0150 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -17,6 +17,15 @@ on: description: Additional tag to prevent collision on GCP VM naming type: string default: '' + vms: + description: JSON encoded list of vms to run tests on + type: string + required: true + output-type: + description: The output type to be used in the tests + type: string + required: false + default: grpc jobs: integration-tests: @@ -24,13 +33,7 @@ jobs: strategy: fail-fast: false matrix: - vm: - - fedora-coreos - - fcarm - - rhel - - rhel-arm64 - - rhcos - - rhcos-arm64 + vm: ${{ fromJson(inputs.vms) }} steps: - uses: actions/checkout@v4 @@ -73,6 +76,7 @@ jobs: FACT_VERSION: ${{ inputs.version }} FACT_REGISTRY: ${{ inputs.registry }} FACT_TAG: ${{ inputs.tag }} + OUTPUT_TYPE: ${{ inputs.output-type }} run: | FACT_IMAGE_NAME="$(make -sC "${GITHUB_WORKSPACE}/fact" image-name)" cat << EOF > vars.yml @@ -81,6 +85,7 @@ jobs: fact: image: ${FACT_IMAGE_NAME} version: ${FACT_VERSION} + output_type: ${OUTPUT_TYPE} quay: username: ${{ secrets.QUAY_RHACS_ENG_RO_USERNAME }} password: ${{ secrets.QUAY_RHACS_ENG_RO_PASSWORD }} diff --git a/.github/workflows/konflux-tests.yml b/.github/workflows/konflux-tests.yml index f3415578..bed02ef0 100644 --- a/.github/workflows/konflux-tests.yml +++ b/.github/workflows/konflux-tests.yml @@ -50,4 +50,5 @@ jobs: registry: quay.io/rhacs-eng/release-fact tag: ${{ needs.init.outputs.fact-tag }} job-tag: konf + vms: '["fedora-coreos", "fcarm", "rhel", "rhel-arm64", "rhcos", "rhcos-arm64"]' secrets: inherit diff --git a/ansible/run-tests.yml b/ansible/run-tests.yml index 446326f2..9d5ff91f 100644 --- a/ansible/run-tests.yml +++ b/ansible/run-tests.yml @@ -1,8 +1,6 @@ --- - name: Run integration tests hosts: "platform_*:&job_id_{{ job_id }}" - environment: - FACT_IMAGE_NAME: "{{ fact.image | default(None) }}" tasks: - name: Clone the repo @@ -44,6 +42,8 @@ become: true environment: DOCKER_HOST: "{{ runtime_host }}" + FACT_IMAGE_NAME: "{{ fact.image | default(None) }}" + OUTPUT_TYPE: "{{ output_type | default('grpc') }}" ansible.builtin.shell: cmd: | set -euo pipefail @@ -65,7 +65,7 @@ ../third_party/stackrox/proto/internalapi/sensor/sfa_iservice.proto # Run the tests - pytest --image="${FACT_IMAGE_NAME}" --junit-xml=results.xml + pytest --image="${FACT_IMAGE_NAME}" --junit-xml=results.xml --output="${OUTPUT_TYPE}" always: - name: Make logs directories