diff --git a/net-guardia/src/adapter/http/fusion.rs b/net-guardia/src/adapter/http/fusion.rs new file mode 100644 index 0000000..6eb953a --- /dev/null +++ b/net-guardia/src/adapter/http/fusion.rs @@ -0,0 +1,21 @@ +//! HTTP surface for fusion-layer observability. Reads shared atomic +//! counters maintained by the detection orchestrator — handlers here +//! never touch orchestrator state, so a hung dashboard cannot stall +//! the detection pipeline. + +use std::sync::Arc; + +use actix_web::{HttpResponse, Responder, Scope, web}; + +use crate::core::detection::metrics::FusionMetrics; + +pub fn initialize() -> Scope { + web::scope("/fusion").route("/metrics", web::get().to(get_metrics)) +} + +/// `GET /api/fusion/metrics` — lock-free snapshot of fusion counters and +/// derived rates. Drives the operator dashboard's "how well is fusion +/// working on my network?" view. +async fn get_metrics(metrics: web::Data>) -> impl Responder { + HttpResponse::Ok().json(metrics.snapshot()) +} diff --git a/net-guardia/src/adapter/http/mod.rs b/net-guardia/src/adapter/http/mod.rs index 4c87808..a152164 100644 --- a/net-guardia/src/adapter/http/mod.rs +++ b/net-guardia/src/adapter/http/mod.rs @@ -4,6 +4,7 @@ pub mod audit; pub mod auth; pub mod default; pub mod filter; +pub mod fusion; pub mod health; pub mod logs; pub mod ml; diff --git a/net-guardia/src/core/detection/metrics.rs b/net-guardia/src/core/detection/metrics.rs new file mode 100644 index 0000000..59fdd2e --- /dev/null +++ b/net-guardia/src/core/detection/metrics.rs @@ -0,0 +1,223 @@ +//! Lock-free fusion observability counters. The orchestrator bumps these +//! on ingress / emit / eviction; HTTP handlers (and eventually the admin +//! dashboard) read atomic snapshots without touching orchestrator state. + +use std::sync::atomic::{AtomicU64, Ordering}; + +use serde::Serialize; + +use crate::model::event::DetectionSource; + +/// Relaxed ordering is enough for counters: readers tolerate arbitrary +/// interleaving, and no counter's value gates access to another memory +/// location. Anything stronger would just waste fence instructions on the +/// hot packet path without adding any real invariant. +const ORDER: Ordering = Ordering::Relaxed; + +/// Atomic counters maintained by the detection orchestrator. Shared via +/// `Arc` with the HTTP metrics handler so dashboards can read without +/// blocking the orchestrator task. +#[derive(Debug, Default)] +pub struct FusionMetrics { + /// Total fused emits (single-source + multi-source combined). + total_emits: AtomicU64, + /// Fused emits whose final `active_source_count` was ≥ 2 (i.e. fusion + /// actually fired across multiple sources rather than single-source solo). + multi_source_emits: AtomicU64, + /// Fusion windows evicted under LRU pressure before they could emit. + /// Kept separate from shutdown-drain drops — those are legitimate. + windows_evicted: AtomicU64, + /// Per-source event count. Increments once per ingress detection + /// regardless of whether the event fires a fused emit downstream. + ml_fires: AtomicU64, + suricata_fires: AtomicU64, + beaconing_fires: AtomicU64, + correlation_fires: AtomicU64, +} + +impl FusionMetrics { + pub fn new() -> Self { + Self::default() + } + + /// Record an ingress detection from the given source. Called before + /// fusion-window bookkeeping so per-source counters reflect raw + /// volume, not what survives dedup. + pub fn record_fire(&self, source: DetectionSource) { + let counter = match source { + DetectionSource::ML => &self.ml_fires, + DetectionSource::Suricata => &self.suricata_fires, + DetectionSource::Beaconing => &self.beaconing_fires, + DetectionSource::Correlation => &self.correlation_fires, + }; + counter.fetch_add(1, ORDER); + } + + /// Record a fused emit. `source_count` is the number of distinct + /// sources that contributed to this emit — 1 for single-source, + /// 2..=4 when fusion actually agreed. + pub fn record_emit(&self, source_count: usize) { + self.total_emits.fetch_add(1, ORDER); + if source_count >= 2 { + self.multi_source_emits.fetch_add(1, ORDER); + } + } + + /// Record a fusion-window eviction that happened before the window + /// could emit. Shutdown-drain drops are not counted here. + pub fn record_eviction(&self) { + self.windows_evicted.fetch_add(1, ORDER); + } + + /// Take an atomic snapshot of every counter and derive the three + /// rate figures the dashboard surfaces. + pub fn snapshot(&self) -> FusionMetricsSnapshot { + let total_emits = self.total_emits.load(ORDER); + let multi_source_emits = self.multi_source_emits.load(ORDER); + let windows_evicted = self.windows_evicted.load(ORDER); + + let ml = self.ml_fires.load(ORDER); + let suricata = self.suricata_fires.load(ORDER); + let beaconing = self.beaconing_fires.load(ORDER); + let correlation = self.correlation_fires.load(ORDER); + + let agreed_rate = ratio(multi_source_emits, total_emits); + let drop_denominator = total_emits + windows_evicted; + let window_drop_rate = ratio(windows_evicted, drop_denominator); + + let total_fires = ml + suricata + beaconing + correlation; + let per_source_fire_rate = PerSourceRate { + ml: ratio(ml, total_fires), + suricata: ratio(suricata, total_fires), + beaconing: ratio(beaconing, total_fires), + correlation: ratio(correlation, total_fires), + }; + + FusionMetricsSnapshot { + total_emits, + multi_source_emits, + agreed_rate, + windows_evicted, + window_drop_rate, + per_source_fires: PerSourceCount { + ml, + suricata, + beaconing, + correlation, + }, + per_source_fire_rate, + } + } +} + +/// Return `numerator / denominator` as `f64`, or `0.0` when the +/// denominator is zero. Saves every rate caller from an `if denom == 0` +/// rewrite of the same guard. +fn ratio(numerator: u64, denominator: u64) -> f64 { + if denominator == 0 { + 0.0 + } else { + numerator as f64 / denominator as f64 + } +} + +/// Wire-format snapshot consumed by `GET /api/fusion/metrics`. Derived +/// fields (`agreed_rate`, `window_drop_rate`, `per_source_fire_rate`) +/// are precomputed server-side so the UI doesn't have to re-implement +/// the formulas and drift. +#[derive(Debug, Clone, Serialize)] +pub struct FusionMetricsSnapshot { + pub total_emits: u64, + pub multi_source_emits: u64, + /// `multi_source_emits / total_emits`. + pub agreed_rate: f64, + pub windows_evicted: u64, + /// `windows_evicted / (windows_evicted + total_emits)`. + pub window_drop_rate: f64, + pub per_source_fires: PerSourceCount, + pub per_source_fire_rate: PerSourceRate, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PerSourceCount { + pub ml: u64, + pub suricata: u64, + pub beaconing: u64, + pub correlation: u64, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PerSourceRate { + pub ml: f64, + pub suricata: f64, + pub beaconing: f64, + pub correlation: f64, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn zero_traffic_snapshot_reports_zero_rates() { + let m = FusionMetrics::new(); + let s = m.snapshot(); + assert_eq!(s.total_emits, 0); + assert_eq!(s.agreed_rate, 0.0); + assert_eq!(s.window_drop_rate, 0.0); + assert_eq!(s.per_source_fire_rate.ml, 0.0); + } + + #[test] + fn agreed_rate_reflects_multi_source_ratio() { + let m = FusionMetrics::new(); + m.record_emit(1); // single-source + m.record_emit(2); // multi + m.record_emit(3); // multi + m.record_emit(1); // single + let s = m.snapshot(); + assert_eq!(s.total_emits, 4); + assert_eq!(s.multi_source_emits, 2); + assert!((s.agreed_rate - 0.5).abs() < 1e-9); + } + + #[test] + fn window_drop_rate_isolates_evictions_from_emits() { + let m = FusionMetrics::new(); + for _ in 0..9 { + m.record_emit(1); + } + m.record_eviction(); // 1 evicted / 10 total tracked + let s = m.snapshot(); + assert_eq!(s.windows_evicted, 1); + assert!((s.window_drop_rate - 0.1).abs() < 1e-9); + } + + #[test] + fn per_source_fire_rate_sums_to_one_when_nonzero() { + let m = FusionMetrics::new(); + m.record_fire(DetectionSource::ML); + m.record_fire(DetectionSource::ML); + m.record_fire(DetectionSource::Suricata); + m.record_fire(DetectionSource::Beaconing); + let s = m.snapshot(); + let sum = s.per_source_fire_rate.ml + + s.per_source_fire_rate.suricata + + s.per_source_fire_rate.beaconing + + s.per_source_fire_rate.correlation; + assert!((sum - 1.0).abs() < 1e-9, "per-source rates must sum to 1, got {sum}"); + assert!((s.per_source_fire_rate.ml - 0.5).abs() < 1e-9); + } + + #[test] + fn record_fire_routes_to_correct_source_bucket() { + let m = FusionMetrics::new(); + m.record_fire(DetectionSource::Suricata); + m.record_fire(DetectionSource::Correlation); + let s = m.snapshot(); + assert_eq!(s.per_source_fires.suricata, 1); + assert_eq!(s.per_source_fires.correlation, 1); + assert_eq!(s.per_source_fires.ml, 0); + assert_eq!(s.per_source_fires.beaconing, 0); + } +} diff --git a/net-guardia/src/core/detection/mod.rs b/net-guardia/src/core/detection/mod.rs index d497600..fedc72c 100644 --- a/net-guardia/src/core/detection/mod.rs +++ b/net-guardia/src/core/detection/mod.rs @@ -1,3 +1,4 @@ pub mod beaconing; pub mod fusion_math; +pub mod metrics; pub mod orchestrator; diff --git a/net-guardia/src/core/detection/orchestrator.rs b/net-guardia/src/core/detection/orchestrator.rs index 09ef549..24dc783 100644 --- a/net-guardia/src/core/detection/orchestrator.rs +++ b/net-guardia/src/core/detection/orchestrator.rs @@ -8,6 +8,7 @@ use tokio::sync::mpsc; use tokio::time::interval; use super::fusion_math::{FusionWindowLengths, fused_confidence}; +use super::metrics::FusionMetrics; use crate::infrastructure::communication_manager::CommunicationManager; use crate::infrastructure::geoip::GeoIpService; use crate::model::detection::attack_type::translate; @@ -67,6 +68,7 @@ pub struct DetectionOrchestrator { rx: mpsc::Receiver, comm: Arc, geoip: Option>, + metrics: Arc, // Enrichment state src_ip_counts: lru::LruCache, repeat_tracker: lru::LruCache, @@ -83,11 +85,13 @@ impl DetectionOrchestrator { rx: mpsc::Receiver, comm: Arc, geoip: Option>, + metrics: Arc, ) -> Self { Self { rx, comm, geoip, + metrics, // SAFETY: NonZero::new on a non-zero literal is infallible. src_ip_counts: LruCache::new(NonZero::new(10_000).unwrap()), repeat_tracker: LruCache::new(NonZero::new(5_000).unwrap()), @@ -122,6 +126,10 @@ impl DetectionOrchestrator { } async fn handle_detection(&mut self, mut event: DetectionEvent) { + // Count every ingress event per-source before dedup — this is the + // raw firing rate, independent of whether the event survives to emit. + self.metrics.record_fire(event.source); + // Canonicalize the raw attack_type so Suricata's "brute-force" // classtype and ML's "Brute Force" class name land on the same dedup // key — the precondition for cross-source fusion. Keep the original @@ -187,6 +195,22 @@ impl DetectionOrchestrator { // Path B: brand-new key (or expired dedup). Emit single-source, // open a fusion window sized by this source. let fusion_window = Duration::from_secs(self.fusion_windows.for_source(event.source)); + + // Detect LRU-pressure eviction: if the dedup map is already at capacity + // and this key wasn't present, inserting will evict the least-recently- + // used entry silently. That's a real lost-signal event; count it and + // warn so the operator sees sustained-attack saturation. + let cap = self.dedup.cap().get(); + let was_full = self.dedup.len() >= cap; + let key_was_absent = self.dedup.peek(&key).is_none(); + if was_full && key_was_absent { + self.metrics.record_eviction(); + log!(DetectionLog::FusionWindowEvicted { + key_src: key.0.clone(), + key_type: key.1.clone(), + }); + } + self.dedup.put( key.clone(), DedupEntry { @@ -233,6 +257,8 @@ impl DetectionOrchestrator { threat_event.active_source_count, )); + self.metrics.record_emit(threat_event.active_source_count); + self.publish_fusion_audit(trigger_event, fused, &per_source_samples) .await; diff --git a/net-guardia/src/infrastructure/app_services.rs b/net-guardia/src/infrastructure/app_services.rs index ffc1ed2..3aefe74 100644 --- a/net-guardia/src/infrastructure/app_services.rs +++ b/net-guardia/src/infrastructure/app_services.rs @@ -6,6 +6,7 @@ use crossbeam::queue::SegQueue; use macros::log; use tokio::sync::oneshot; +use crate::core::detection::metrics::FusionMetrics; use crate::core::ml::adapter::ModelSourceState; use crate::core::ml::alert::MLAlert; use crate::core::ml::drift_detector::DriftDetector; @@ -38,6 +39,7 @@ pub struct AppServices { pub ml_inference: Arc, pub ml_engine: Arc, pub flow_statistics: Arc, + pub fusion_metrics: Arc, shutdowns: SegQueue>, } @@ -123,6 +125,7 @@ impl AppServices { )); let flow_statistics = Arc::new(FlowStatistics::new(ml_engine.clone())); + let fusion_metrics = Arc::new(FusionMetrics::new()); Ok(Self { health: Arc::new(health), @@ -130,6 +133,7 @@ impl AppServices { ml_inference, ml_engine, flow_statistics, + fusion_metrics, shutdowns: SegQueue::new(), }) } diff --git a/net-guardia/src/infrastructure/http_server.rs b/net-guardia/src/infrastructure/http_server.rs index dff8392..c3252ed 100644 --- a/net-guardia/src/infrastructure/http_server.rs +++ b/net-guardia/src/infrastructure/http_server.rs @@ -10,7 +10,7 @@ use macros::log; use crate::adapter::ebpf::EbpfServices; use crate::adapter::http::{ - acl, api_keys, audit as audit_api, auth, default, filter, health as health_api, logs as logs_api, ml, + acl, api_keys, audit as audit_api, auth, default, filter, fusion, health as health_api, logs as logs_api, ml, notification as notification_api, rate_limit as rate_limit_api, report as report_api, setup as setup_api, soar, stats, system as system_api, }; @@ -209,6 +209,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> { let ml_alert = params.app_services.ml_alert.clone(); let ml_engine = params.app_services.ml_engine.clone(); let ml_inference = params.app_services.ml_inference.clone(); + let fusion_metrics = params.app_services.fusion_metrics.clone(); let flow_statistics = params.app_services.flow_statistics.clone(); let drop_monitor = params.ebpf_services.drop_monitor.clone(); let app_config = params.app_config; @@ -248,6 +249,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> { .app_data(web::Data::from(ml_alert.clone())) .app_data(web::Data::from(ml_engine.clone())) .app_data(web::Data::from(ml_inference.clone())) + .app_data(web::Data::from(fusion_metrics.clone())) .app_data(web::Data::from(flow_statistics.clone())) .app_data(web::Data::from(drop_monitor.clone())) .app_data(web::Data::from(db.clone() as Arc)) @@ -277,6 +279,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> { .service(stats::initialize()) .service(health_api::initialize()) .service(ml::initialize()) + .service(fusion::initialize()) .service(system_api::initialize()) .service(soar::initialize()) .service(notification_api::initialize()) diff --git a/net-guardia/src/infrastructure/system.rs b/net-guardia/src/infrastructure/system.rs index 578da0f..b3b5ea5 100644 --- a/net-guardia/src/infrastructure/system.rs +++ b/net-guardia/src/infrastructure/system.rs @@ -246,7 +246,12 @@ impl System { // Start detection orchestrator (dedup + enrichment + source attribution) let (detection_tx, detection_rx) = mpsc::channel::(1024); - let orchestrator = DetectionOrchestrator::new(detection_rx, self.comm.clone(), self.geoip.clone()); + let orchestrator = DetectionOrchestrator::new( + detection_rx, + self.comm.clone(), + self.geoip.clone(), + self.app_services.fusion_metrics.clone(), + ); orchestrator.start(); // Clone detection_tx for correlation engine and beaconing detector