diff --git a/mantis/src/core/app_state.rs b/mantis/src/core/app_state.rs index e580104..64b8582 100644 --- a/mantis/src/core/app_state.rs +++ b/mantis/src/core/app_state.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use crate::core::ebpf::access_control::AccessControl; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::core::ebpf::statistics::Statistics; use crate::core::infrastructure::app_config::AppConfig; use crate::core::infrastructure::app_db::AppDB; @@ -19,4 +20,5 @@ pub struct AppState { pub detection_alert: Arc, pub app_db: Option>, pub log_broadcaster: Arc, + pub pipeline_counters: Arc, } diff --git a/mantis/src/core/ebpf/mod.rs b/mantis/src/core/ebpf/mod.rs index 61a5dc6..9e45a9d 100644 --- a/mantis/src/core/ebpf/mod.rs +++ b/mantis/src/core/ebpf/mod.rs @@ -1,4 +1,5 @@ pub mod access_control; +pub mod pipeline_counters; pub mod statistics; pub mod tcp_anomaly_tracker; pub mod xsk_manager; @@ -12,6 +13,7 @@ use macros::log; use tokio::sync::oneshot; use crate::core::ebpf::access_control::AccessControl; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::core::ebpf::statistics::Statistics; use crate::core::ebpf::tcp_anomaly_tracker::TcpAnomalyTracker; use crate::core::ebpf::xsk_manager::XskManager; @@ -31,11 +33,16 @@ pub struct EbpfServices { } impl EbpfServices { - pub fn new(app_config: Arc, ingress_ebpf: &mut Ebpf, egress_ebpf: &mut Ebpf) -> Result { - let xsk_manager = XskManager::new(app_config.clone(), ingress_ebpf, egress_ebpf)?; + pub fn new( + app_config: Arc, + ingress_ebpf: &mut Ebpf, + egress_ebpf: &mut Ebpf, + counters: Arc, + ) -> Result { + let xsk_manager = XskManager::new(app_config.clone(), ingress_ebpf, egress_ebpf, counters.clone())?; let access_control = AccessControl::new(ingress_ebpf, egress_ebpf)?; let statistics = Statistics::new(app_config.clone(), ingress_ebpf, egress_ebpf)?; - let tcp_anomaly_tracker = TcpAnomalyTracker::new(ingress_ebpf, egress_ebpf)?; + let tcp_anomaly_tracker = TcpAnomalyTracker::new(ingress_ebpf, egress_ebpf, counters)?; let ebpf_services = Self { xsk_manager: Arc::new(xsk_manager), access_control: Arc::new(access_control), diff --git a/mantis/src/core/ebpf/pipeline_counters.rs b/mantis/src/core/ebpf/pipeline_counters.rs new file mode 100644 index 0000000..409d106 --- /dev/null +++ b/mantis/src/core/ebpf/pipeline_counters.rs @@ -0,0 +1,69 @@ +use std::sync::atomic::{AtomicU64, Ordering}; + +use serde::Serialize; + +pub struct PipelineCounters { + /// Packets received from NIC via AF_XDP RX queue. + pub xsk_rx_packets: AtomicU64, + /// Packets forwarded out via AF_XDP TX queue. + pub xsk_tx_packets: AtomicU64, + /// Packets dropped because the ML/Suricata forward channel was full. + /// Non-zero means detectors are missing traffic. + pub xsk_dropped_forward: AtomicU64, + /// Times the UMEM frame pool had no free frames to replenish the fill queue. + /// Non-zero means the NIC may be dropping inbound packets. + pub xsk_fill_starvations: AtomicU64, + /// Times TX had no frames available (TX throttled). + pub xsk_no_tx_frames: AtomicU64, + /// TCP anomaly events consumed from the ring buffer and turned into alerts. + pub anomaly_events_emitted: AtomicU64, + /// Packets dropped by the Suricata mirror channel (channel full). + pub suricata_dropped: AtomicU64, + /// Alerts successfully broadcast to at least one subscriber. + pub alerts_emitted: AtomicU64, + /// Alerts dropped because no subscriber was listening. + pub alerts_dropped: AtomicU64, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PipelineSnapshot { + pub xsk_rx_packets: u64, + pub xsk_tx_packets: u64, + pub xsk_dropped_forward: u64, + pub xsk_fill_starvations: u64, + pub xsk_no_tx_frames: u64, + pub anomaly_events_emitted: u64, + pub suricata_dropped: u64, + pub alerts_emitted: u64, + pub alerts_dropped: u64, +} + +impl PipelineCounters { + pub fn new() -> Self { + Self { + xsk_rx_packets: AtomicU64::new(0), + xsk_tx_packets: AtomicU64::new(0), + xsk_dropped_forward: AtomicU64::new(0), + xsk_fill_starvations: AtomicU64::new(0), + xsk_no_tx_frames: AtomicU64::new(0), + anomaly_events_emitted: AtomicU64::new(0), + suricata_dropped: AtomicU64::new(0), + alerts_emitted: AtomicU64::new(0), + alerts_dropped: AtomicU64::new(0), + } + } + + pub fn snapshot(&self) -> PipelineSnapshot { + PipelineSnapshot { + xsk_rx_packets: self.xsk_rx_packets.load(Ordering::Relaxed), + xsk_tx_packets: self.xsk_tx_packets.load(Ordering::Relaxed), + xsk_dropped_forward: self.xsk_dropped_forward.load(Ordering::Relaxed), + xsk_fill_starvations: self.xsk_fill_starvations.load(Ordering::Relaxed), + xsk_no_tx_frames: self.xsk_no_tx_frames.load(Ordering::Relaxed), + anomaly_events_emitted: self.anomaly_events_emitted.load(Ordering::Relaxed), + suricata_dropped: self.suricata_dropped.load(Ordering::Relaxed), + alerts_emitted: self.alerts_emitted.load(Ordering::Relaxed), + alerts_dropped: self.alerts_dropped.load(Ordering::Relaxed), + } + } +} diff --git a/mantis/src/core/ebpf/tcp_anomaly_tracker.rs b/mantis/src/core/ebpf/tcp_anomaly_tracker.rs index 8e8ae34..fa2c307 100644 --- a/mantis/src/core/ebpf/tcp_anomaly_tracker.rs +++ b/mantis/src/core/ebpf/tcp_anomaly_tracker.rs @@ -1,8 +1,12 @@ +use std::sync::Arc; +use std::sync::atomic::Ordering; + use aya::maps::{MapData, RingBuf}; use common::model::tcp_anomaly::TcpAnomalyEvent; use tokio::io::unix::AsyncFd; use tokio::sync::oneshot; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::core::infrastructure::detection_alert::SharedDetectionAlert; use crate::model::error::Error; use crate::model::error::ebpf::EbpfError; @@ -11,10 +15,15 @@ use crate::model::alert::UnifiedAlert; pub struct TcpAnomalyTracker { ingress_ring_buf: RingBuf, egress_ring_buf: RingBuf, + counters: Arc, } impl TcpAnomalyTracker { - pub fn new(ingress_ebpf: &mut aya::Ebpf, egress_ebpf: &mut aya::Ebpf) -> Result { + pub fn new( + ingress_ebpf: &mut aya::Ebpf, + egress_ebpf: &mut aya::Ebpf, + counters: Arc, + ) -> Result { let ingress_map = ingress_ebpf .take_map("TCP_ANOMALY_EVENTS") .ok_or(EbpfError::MapNotFound)?; @@ -25,7 +34,7 @@ impl TcpAnomalyTracker { RingBuf::try_from(ingress_map).map_err(EbpfError::MapOperationError)?; let egress_ring_buf = RingBuf::try_from(egress_map).map_err(EbpfError::MapOperationError)?; - Ok(Self { ingress_ring_buf, egress_ring_buf }) + Ok(Self { ingress_ring_buf, egress_ring_buf, counters }) } pub fn run(self, detection_alert: SharedDetectionAlert) -> Result, Error> { @@ -35,6 +44,7 @@ impl TcpAnomalyTracker { AsyncFd::new(self.ingress_ring_buf).map_err(EbpfError::MapOperationError)?; let egress_fd = AsyncFd::new(self.egress_ring_buf).map_err(EbpfError::MapOperationError)?; + let counters = self.counters; tokio::spawn(async move { let mut rx = rx; @@ -48,7 +58,7 @@ impl TcpAnomalyTracker { if let Ok(mut guard) = result { let rb = guard.get_inner_mut(); while let Some(item) = rb.next() { - Self::process_event(&*item, &detection_alert); + Self::process_event(&*item, &detection_alert, &counters); } guard.clear_ready(); } @@ -57,7 +67,7 @@ impl TcpAnomalyTracker { if let Ok(mut guard) = result { let rb = guard.get_inner_mut(); while let Some(item) = rb.next() { - Self::process_event(&*item, &detection_alert); + Self::process_event(&*item, &detection_alert, &counters); } guard.clear_ready(); } @@ -69,11 +79,12 @@ impl TcpAnomalyTracker { Ok(tx) } - fn process_event(item: &[u8], detection_alert: &SharedDetectionAlert) { + fn process_event(item: &[u8], detection_alert: &SharedDetectionAlert, counters: &PipelineCounters) { if item.len() < core::mem::size_of::() { return; } let event = unsafe { &*(item.as_ptr() as *const TcpAnomalyEvent) }; + counters.anomaly_events_emitted.fetch_add(1, Ordering::Relaxed); if detection_alert.has_subscribers() { detection_alert.broadcast(UnifiedAlert::from_tcp_anomaly(event)); } diff --git a/mantis/src/core/ebpf/xsk_manager.rs b/mantis/src/core/ebpf/xsk_manager.rs index 7197476..a690056 100644 --- a/mantis/src/core/ebpf/xsk_manager.rs +++ b/mantis/src/core/ebpf/xsk_manager.rs @@ -18,6 +18,7 @@ use tokio::sync::oneshot; use xsk_rs::config::{BindFlags, FrameSize, Interface, LibxdpFlags, QueueSize, SocketConfig, UmemConfig}; use xsk_rs::{CompQueue, FillQueue, FrameDesc, RxQueue, Socket, TxQueue, Umem}; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::core::infrastructure::app_config::AppConfig; use crate::detection::ml::engine::Engine; use crate::detection::suricata::SuricataEngine; @@ -33,10 +34,16 @@ pub struct XskManager { app_config: Arc, xsk_map: Mutex>, egress_xsk_map: Mutex>, + counters: Arc, } impl XskManager { - pub fn new(app_config: Arc, ingress_ebpf: &mut Ebpf, egress_ebpf: &mut Ebpf) -> Result { + pub fn new( + app_config: Arc, + ingress_ebpf: &mut Ebpf, + egress_ebpf: &mut Ebpf, + counters: Arc, + ) -> Result { let map = ingress_ebpf .take_map("INGRESS_XSKS_MAP") .ok_or(EbpfError::MapNotFound)?; @@ -49,6 +56,7 @@ impl XskManager { app_config, xsk_map: Mutex::new(xsk_map), egress_xsk_map: Mutex::new(egress_xsk_map), + counters, }) } @@ -72,6 +80,7 @@ impl XskManager { Direction::Ingress, engine.clone(), suricata_engine.clone(), + self.counters.clone(), )?; let egress_xsk = XskPair::new( @@ -81,6 +90,7 @@ impl XskManager { Direction::Egress, engine.clone(), suricata_engine.clone(), + self.counters.clone(), )?; // Extract fds before run() consumes the XskPair structs. @@ -127,6 +137,7 @@ pub struct XskPair { frame_pool: Arc>>, engine: Option>, suricata_engine: Option>, + counters: Arc, } impl XskPair { @@ -137,6 +148,7 @@ impl XskPair { direction: Direction, engine: Option>, suricata_engine: Option>, + counters: Arc, ) -> Result { let rx_ifname_c = CString::new(rx_ifname).map_err(|_| SystemError::UnknownError)?; @@ -252,6 +264,7 @@ impl XskPair { frame_pool: Arc::new(Mutex::new(pool_frames)), engine, suricata_engine, + counters, }; Ok(xsk_pair) @@ -369,6 +382,7 @@ impl XskPair { let rx_count = unsafe { self.rx.consume(&mut rx_descs) }; if rx_count > 0 { + self.counters.xsk_rx_packets.fetch_add(rx_count as u64, std::sync::atomic::Ordering::Relaxed); for rx_desc in rx_descs.iter().take(rx_count) { let packet_len = rx_desc.lengths().data() as usize; let data = unsafe { self.umem.data(rx_desc) }; @@ -388,6 +402,7 @@ impl XskPair { match e { crossbeam::channel::TrySendError::Full(_) => { log!(EbpfLog::ForwardChannelFull); + self.counters.xsk_dropped_forward.fetch_add(1, std::sync::atomic::Ordering::Relaxed); } crossbeam::channel::TrySendError::Disconnected(_) => { log!(EbpfLog::ForwardChannelDisconnected); @@ -430,6 +445,7 @@ impl XskPair { let mut pool = self.frame_pool.lock(); let available = pool.len().saturating_sub(RESERVED_FOR_TX); if available == 0 { + self.counters.xsk_fill_starvations.fetch_add(1, std::sync::atomic::Ordering::Relaxed); return; } let start = pool.len() - available; @@ -470,6 +486,7 @@ impl XskPair { }; if pool_size == 0 { + self.counters.xsk_no_tx_frames.fetch_add(1, std::sync::atomic::Ordering::Relaxed); return Ok(0); } @@ -515,6 +532,7 @@ impl XskPair { } let nb_submitted = unsafe { self.tx.produce(&frames) }; + self.counters.xsk_tx_packets.fetch_add(nb_submitted as u64, std::sync::atomic::Ordering::Relaxed); if let Err(e) = self.tx.wakeup() { if e.kind() != std::io::ErrorKind::WouldBlock { diff --git a/mantis/src/core/infrastructure/detection_alert.rs b/mantis/src/core/infrastructure/detection_alert.rs index d4ddfa0..1701fc5 100644 --- a/mantis/src/core/infrastructure/detection_alert.rs +++ b/mantis/src/core/infrastructure/detection_alert.rs @@ -1,17 +1,20 @@ use std::sync::Arc; +use std::sync::atomic::Ordering; use tokio::sync::broadcast; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::model::alert::UnifiedAlert; pub struct DetectionAlert { broadcast_tx: broadcast::Sender, + counters: Arc, } impl DetectionAlert { - pub fn new() -> Self { + pub fn new(counters: Arc) -> Self { let (broadcast_tx, _) = broadcast::channel(256); - DetectionAlert { broadcast_tx } + DetectionAlert { broadcast_tx, counters } } pub fn subscribe(&self) -> broadcast::Receiver { @@ -20,7 +23,12 @@ impl DetectionAlert { pub fn broadcast(&self, alert: UnifiedAlert) { if self.broadcast_tx.receiver_count() > 0 { - let _ = self.broadcast_tx.send(alert); + match self.broadcast_tx.send(alert) { + Ok(_) => { self.counters.alerts_emitted.fetch_add(1, Ordering::Relaxed); } + Err(_) => { self.counters.alerts_dropped.fetch_add(1, Ordering::Relaxed); } + } + } else { + self.counters.alerts_dropped.fetch_add(1, Ordering::Relaxed); } } @@ -29,10 +37,4 @@ impl DetectionAlert { } } -impl Default for DetectionAlert { - fn default() -> Self { - Self::new() - } -} - pub type SharedDetectionAlert = Arc; diff --git a/mantis/src/core/infrastructure/mod.rs b/mantis/src/core/infrastructure/mod.rs index ae65b75..c3d781b 100644 --- a/mantis/src/core/infrastructure/mod.rs +++ b/mantis/src/core/infrastructure/mod.rs @@ -14,6 +14,7 @@ use crossbeam::queue::SegQueue; use macros::log; use tokio::sync::oneshot; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::core::infrastructure::app_config::AppConfig; use crate::core::infrastructure::app_db::AppDB; use crate::core::infrastructure::detection_alert::DetectionAlert; @@ -46,10 +47,11 @@ impl AppServices { app_config: Arc, inference_config: Arc, log_broadcaster: Arc, + counters: Arc, ) -> Result { let health = SystemHealth::new(app_config.clone())?; - let detection_alert = Arc::new(DetectionAlert::new()); + let detection_alert = Arc::new(DetectionAlert::new(counters.clone())); let mode = FusionMode::from_str(&app_config.fusion_mode); let fusion_engine = Arc::new(FusionEngine::new( mode, @@ -95,6 +97,7 @@ impl AppServices { &rule_path, &eve_socket, fusion_engine.clone(), + counters, )?) } else { None diff --git a/mantis/src/core/system.rs b/mantis/src/core/system.rs index be4a4c1..4951fd2 100644 --- a/mantis/src/core/system.rs +++ b/mantis/src/core/system.rs @@ -12,6 +12,7 @@ use tower_http::cors::CorsLayer; use crate::core::app_state::AppState; use crate::core::ebpf::EbpfServices; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::core::infrastructure::AppServices; use crate::core::infrastructure::app_config::AppConfig; use crate::core::infrastructure::log_broadcaster::LogBroadcaster; @@ -31,6 +32,7 @@ pub struct System { pub inference_config: Arc, pub ebpf_services: Arc, pub app_services: Arc, + pub pipeline_counters: Arc, pub ingress_ebpf: Ebpf, pub egress_ebpf: Ebpf, #[allow(dead_code)] @@ -55,16 +57,20 @@ impl System { &app_config.ae_threshold_method, )?); + let pipeline_counters = Arc::new(PipelineCounters::new()); + let ebpf_services = Arc::new(EbpfServices::new( app_config.clone(), &mut ingress_ebpf, &mut egress_ebpf, + pipeline_counters.clone(), )?); let app_services = Arc::new(AppServices::new( app_config.clone(), inference_config.clone(), log_broadcaster, + pipeline_counters.clone(), )?); let system = System { @@ -72,6 +78,7 @@ impl System { inference_config, ebpf_services, app_services, + pipeline_counters, ingress_ebpf, egress_ebpf, ingress_program_array, @@ -165,6 +172,7 @@ impl System { detection_alert: self.app_services.detection_alert.clone(), app_db: self.app_services.app_db.clone(), log_broadcaster: self.app_services.log_broadcaster.clone(), + pipeline_counters: self.pipeline_counters.clone(), }; let app = Router::new() diff --git a/mantis/src/detection/suricata/engine.rs b/mantis/src/detection/suricata/engine.rs index 0d06b68..ba124ad 100644 --- a/mantis/src/detection/suricata/engine.rs +++ b/mantis/src/detection/suricata/engine.rs @@ -11,6 +11,7 @@ use crossbeam::channel::{Sender, bounded}; use macros::log; use super::output; +use crate::core::ebpf::pipeline_counters::PipelineCounters; use crate::detection::fusion::FusionEngine; use crate::model::config::SuricataConfig; use crate::model::error::suricata::SuricataError; @@ -25,6 +26,7 @@ const CHANNEL_CAP: usize = 4096; pub struct SuricataEngine { tx: Sender, child: std::sync::Mutex, + counters: Arc, } impl SuricataEngine { @@ -33,6 +35,7 @@ impl SuricataEngine { rule_path: &Path, eve_socket: &Path, fusion: Arc, + counters: Arc, ) -> Result, SuricataError> { Self::kill_existing(); Self::setup_veth()?; @@ -179,6 +182,7 @@ impl SuricataEngine { Ok(Arc::new(Self { tx, child: std::sync::Mutex::new(child), + counters, })) } @@ -188,6 +192,7 @@ impl SuricataEngine { Ok(()) => {} Err(crossbeam::channel::TrySendError::Full(_)) => { log!(SuricataLog::ChannelFull); + self.counters.suricata_dropped.fetch_add(1, std::sync::atomic::Ordering::Relaxed); } Err(crossbeam::channel::TrySendError::Disconnected(_)) => {} } diff --git a/mantis/src/web/api/health.rs b/mantis/src/web/api/health.rs index 7872464..fcf2701 100644 --- a/mantis/src/web/api/health.rs +++ b/mantis/src/web/api/health.rs @@ -10,6 +10,7 @@ pub fn router() -> Router { Router::new() .route("/metrics", get(get_current_metrics)) .route("/status", get(get_health_status)) + .route("/pipeline", get(get_pipeline_metrics)) .route("/websocket/metrics", get(websocket_metrics)) } @@ -17,6 +18,10 @@ async fn get_current_metrics(State(state): State) -> impl IntoResponse Json(state.health.get_current_metrics().await) } +async fn get_pipeline_metrics(State(state): State) -> impl IntoResponse { + Json(state.pipeline_counters.snapshot()) +} + async fn get_health_status(State(state): State) -> impl IntoResponse { Json(state.health.is_system_healthy().await) }