Add NIDS-specific pipeline health counters

Introduce PipelineCounters (Arc<AtomicU64>) shared across all pipeline
stages to replace log-only error reporting with queryable counters:

- xsk_rx_packets / xsk_tx_packets: AF_XDP throughput per queue
- xsk_dropped_forward: ForwardChannelFull drops (ML/Suricata miss)
- xsk_fill_starvations: UMEM pool empty events (NIC drop risk)
- xsk_no_tx_frames: TX throttle events
- anomaly_events_emitted: TCP anomaly ring buffer events consumed
- suricata_dropped: Suricata mirror channel full drops
- alerts_emitted / alerts_dropped: detection alert broadcast counters

Wire counters through EbpfServices, AppServices, and AppState.
Expose as GET /health/pipeline returning a PipelineSnapshot JSON.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138PxtKH73hqxv7h1oaoSdS

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<DetectionAlert>,
     pub app_db: Option<Arc<AppDB>>,
     pub log_broadcaster: Arc<LogBroadcaster>,
+    pub pipeline_counters: Arc<PipelineCounters>,
 }

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<AppConfig>, ingress_ebpf: &mut Ebpf, egress_ebpf: &mut Ebpf) -> Result<Self, Error> {
-        let xsk_manager = XskManager::new(app_config.clone(), ingress_ebpf, egress_ebpf)?;
+    pub fn new(
+        app_config: Arc<AppConfig>,
+        ingress_ebpf: &mut Ebpf,
+        egress_ebpf: &mut Ebpf,
+        counters: Arc<PipelineCounters>,
+    ) -> Result<Self, Error> {
+        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),

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),
+        }
+    }
+}

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<MapData>,
     egress_ring_buf: RingBuf<MapData>,
+    counters: Arc<PipelineCounters>,
 }
 
 impl TcpAnomalyTracker {
-    pub fn new(ingress_ebpf: &mut aya::Ebpf, egress_ebpf: &mut aya::Ebpf) -> Result<Self, Error> {
+    pub fn new(
+        ingress_ebpf: &mut aya::Ebpf,
+        egress_ebpf: &mut aya::Ebpf,
+        counters: Arc<PipelineCounters>,
+    ) -> Result<Self, Error> {
         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<oneshot::Sender<()>, 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::<TcpAnomalyEvent>() {
             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));
         }

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<AppConfig>,
     xsk_map: Mutex<XskMap<MapData>>,
     egress_xsk_map: Mutex<XskMap<MapData>>,
+    counters: Arc<PipelineCounters>,
 }
 
 impl XskManager {
-    pub fn new(app_config: Arc<AppConfig>, ingress_ebpf: &mut Ebpf, egress_ebpf: &mut Ebpf) -> Result<Self, Error> {
+    pub fn new(
+        app_config: Arc<AppConfig>,
+        ingress_ebpf: &mut Ebpf,
+        egress_ebpf: &mut Ebpf,
+        counters: Arc<PipelineCounters>,
+    ) -> Result<Self, Error> {
         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<Mutex<Vec<FrameDesc>>>,
     engine: Option<Arc<Engine>>,
     suricata_engine: Option<Arc<SuricataEngine>>,
+    counters: Arc<PipelineCounters>,
 }
 
 impl XskPair {
@@ -137,6 +148,7 @@ impl XskPair {
         direction: Direction,
         engine: Option<Arc<Engine>>,
         suricata_engine: Option<Arc<SuricataEngine>>,
+        counters: Arc<PipelineCounters>,
     ) -> Result<Self, Error> {
         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 {

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<UnifiedAlert>,
+    counters: Arc<PipelineCounters>,
 }
 
 impl DetectionAlert {
-    pub fn new() -> Self {
+    pub fn new(counters: Arc<PipelineCounters>) -> Self {
         let (broadcast_tx, _) = broadcast::channel(256);
-        DetectionAlert { broadcast_tx }
+        DetectionAlert { broadcast_tx, counters }
     }
 
     pub fn subscribe(&self) -> broadcast::Receiver<UnifiedAlert> {
@@ -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<DetectionAlert>;

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<AppConfig>,
         inference_config: Arc<InferenceConfig>,
         log_broadcaster: Arc<crate::core::infrastructure::log_broadcaster::LogBroadcaster>,
+        counters: Arc<PipelineCounters>,
     ) -> Result<Self, Error> {
         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

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<InferenceConfig>,
     pub ebpf_services: Arc<EbpfServices>,
     pub app_services: Arc<AppServices>,
+    pub pipeline_counters: Arc<PipelineCounters>,
     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()

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<Bytes>,
     child: std::sync::Mutex<Child>,
+    counters: Arc<PipelineCounters>,
 }
 
 impl SuricataEngine {
@@ -33,6 +35,7 @@ impl SuricataEngine {
         rule_path: &Path,
         eve_socket: &Path,
         fusion: Arc<FusionEngine>,
+        counters: Arc<PipelineCounters>,
     ) -> Result<Arc<Self>, 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(_)) => {}
         }

mantis/src/web/api/health.rs

@@ -10,6 +10,7 @@ pub fn router() -> Router<AppState> {
     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<AppState>) -> impl IntoResponse
     Json(state.health.get_current_metrics().await)
 }
 
+async fn get_pipeline_metrics(State(state): State<AppState>) -> impl IntoResponse {
+    Json(state.pipeline_counters.snapshot())
+}
+
 async fn get_health_status(State(state): State<AppState>) -> impl IntoResponse {
     Json(state.health.is_system_healthy().await)
 }