feat: Add live system log WebSocket API and frontend log viewer

mantis-frontend

@@ -1 +1 @@
-Subproject commit fd1ca155e745e1e2121b4a0c2add3a06a71aec42
+Subproject commit 2b9c85de173c2810a8ffca183ddddcc751f4e227

mantis/src/core/app_state.rs

@@ -6,6 +6,7 @@ use crate::core::infrastructure::app_config::AppConfig;
 use crate::core::infrastructure::app_db::AppDB;
 use crate::core::infrastructure::detection_alert::DetectionAlert;
 use crate::core::infrastructure::health::SystemHealth;
+use crate::core::infrastructure::log_broadcaster::LogBroadcaster;
 use crate::detection::ml::config_loader::InferenceConfig;
 
 #[derive(Clone)]
@@ -17,4 +18,5 @@ pub struct AppState {
     pub health: Arc<SystemHealth>,
     pub detection_alert: Arc<DetectionAlert>,
     pub app_db: Option<Arc<AppDB>>,
+    pub log_broadcaster: Arc<LogBroadcaster>,
 }

mantis/src/core/infrastructure/log_broadcaster.rs

@@ -0,0 +1,53 @@
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
+
+use serde::{Deserialize, Serialize};
+use tokio::sync::broadcast;
+
+const RING_BUFFER_SIZE: usize = 500;
+const CHANNEL_CAPACITY: usize = 512;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct LogRecord {
+    pub timestamp: String,
+    pub level: String,
+    pub message: String,
+}
+
+pub struct LogBroadcaster {
+    tx: broadcast::Sender<LogRecord>,
+    recent: Mutex<VecDeque<LogRecord>>,
+}
+
+impl LogBroadcaster {
+    pub fn new() -> Arc<Self> {
+        let (tx, _) = broadcast::channel(CHANNEL_CAPACITY);
+        Arc::new(Self {
+            tx,
+            recent: Mutex::new(VecDeque::with_capacity(RING_BUFFER_SIZE)),
+        })
+    }
+
+    pub fn emit(&self, record: LogRecord) {
+        if let Ok(mut buf) = self.recent.lock() {
+            if buf.len() >= RING_BUFFER_SIZE {
+                buf.pop_front();
+            }
+            buf.push_back(record.clone());
+        }
+        if self.tx.receiver_count() > 0 {
+            let _ = self.tx.send(record);
+        }
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<LogRecord> {
+        self.tx.subscribe()
+    }
+
+    pub fn recent_logs(&self) -> Vec<LogRecord> {
+        self.recent
+            .lock()
+            .map(|buf| buf.iter().cloned().collect())
+            .unwrap_or_default()
+    }
+}

mantis/src/core/infrastructure/mod.rs

@@ -3,6 +3,7 @@ pub mod app_db;
 pub mod detection_alert;
 pub mod geoip;
 pub mod health;
+pub mod log_broadcaster;
 
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -37,11 +38,16 @@ pub struct AppServices {
     pub ml_engine: Arc<Engine>,
     pub suricata_engine: Option<Arc<SuricataEngine>>,
     pub app_db: Option<Arc<AppDB>>,
+    pub log_broadcaster: Arc<crate::core::infrastructure::log_broadcaster::LogBroadcaster>,
     shutdowns: SegQueue<oneshot::Sender<()>>,
 }
 
 impl AppServices {
-    pub fn new(app_config: Arc<AppConfig>, inference_config: Arc<InferenceConfig>) -> Result<Self, Error> {
+    pub fn new(
+        app_config: Arc<AppConfig>,
+        inference_config: Arc<InferenceConfig>,
+        log_broadcaster: Arc<crate::core::infrastructure::log_broadcaster::LogBroadcaster>,
+    ) -> Result<Self, Error> {
         let health = SystemHealth::new(app_config.clone())?;
         let ml_models = Arc::new(MLModels::load_models(&app_config)?);
 
@@ -118,6 +124,7 @@ impl AppServices {
             ml_engine,
             suricata_engine,
             app_db,
+            log_broadcaster,
             shutdowns: SegQueue::new(),
         })
     }

mantis/src/core/system.rs

@@ -22,8 +22,9 @@ use crate::model::error::misc::MiscError;
 use crate::model::log::ml::MLLog;
 use crate::model::log::system::SystemLog;
 use crate::utils::logging::Logging;
+use crate::core::infrastructure::log_broadcaster::LogBroadcaster;
 use crate::web::api::default::default_route;
-use crate::web::api::{auth, control, detection_alert, health, misc};
+use crate::web::api::{auth, control, detection_alert, health, logs, misc};
 
 pub struct System {
     pub app_config: Arc<AppConfig>,
@@ -40,7 +41,8 @@ pub struct System {
 
 impl System {
     pub async fn new() -> Result<Self, Error> {
-        Logging::initialize()?;
+        let log_broadcaster = LogBroadcaster::new();
+        Logging::initialize(log_broadcaster.clone())?;
 
         log!(SystemLog::Initializing);
 
@@ -59,7 +61,11 @@ impl System {
             &mut egress_ebpf,
         )?);
 
-        let app_services = Arc::new(AppServices::new(app_config.clone(), inference_config.clone())?);
+        let app_services = Arc::new(AppServices::new(
+            app_config.clone(),
+            inference_config.clone(),
+            log_broadcaster,
+        )?);
 
         let system = System {
             app_config,
@@ -154,12 +160,14 @@ impl System {
             health: self.app_services.health.clone(),
             detection_alert: self.app_services.detection_alert.clone(),
             app_db: self.app_services.app_db.clone(),
+            log_broadcaster: self.app_services.log_broadcaster.clone(),
         };
 
         let app = Router::new()
             .nest("/ebpf", control::router())
             .nest("/detection", detection_alert::router())
             .nest("/health", health::router())
+            .nest("/logs", logs::router())
             .nest("/misc", misc::router())
             .nest("/auth", auth::router())
             .fallback(default_route)

mantis/src/utils/logging.rs

@@ -1,4 +1,5 @@
 use std::fs;
+use std::sync::Arc;
 
 use tracing::Level;
 use tracing_appender::rolling::{RollingFileAppender, Rotation};
@@ -6,13 +7,15 @@ use tracing_subscriber::filter::EnvFilter;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
 
+use crate::core::infrastructure::log_broadcaster::LogBroadcaster;
 use crate::model::error::Error;
 use crate::model::error::io::IOError;
+use crate::utils::tracing_layer::BroadcastLayer;
 
 pub struct Logging;
 
 impl Logging {
-    pub fn initialize() -> Result<(), Error> {
+    pub fn initialize(broadcaster: Arc<LogBroadcaster>) -> Result<(), Error> {
         let log_directory = "logs";
         fs::create_dir_all(log_directory).map_err(|err| IOError::CreateDirectoryFailed(log_directory, err))?;
 
@@ -33,6 +36,8 @@ impl Logging {
             .with_ansi(false)
             .with_writer(file_appender);
 
+        let broadcast_layer = BroadcastLayer::new(broadcaster);
+
         let level = if cfg!(debug_assertions) {
             Level::DEBUG
         } else {
@@ -42,6 +47,7 @@ impl Logging {
         tracing_subscriber::registry()
             .with(stdout_layer)
             .with(file_layer)
+            .with(broadcast_layer)
             .with(EnvFilter::from_default_env().add_directive(level.into()))
             .init();
 

mantis/src/utils/mod.rs

@@ -3,5 +3,6 @@ pub mod cpu_affinity;
 pub mod logging;
 pub mod packet_parser;
 pub mod static_files;
+pub mod tracing_layer;
 
 pub mod ip_address;

mantis/src/utils/tracing_layer.rs

@@ -0,0 +1,55 @@
+use std::sync::Arc;
+
+use tracing::Subscriber;
+use tracing_subscriber::Layer;
+use tracing_subscriber::layer::Context;
+
+use crate::core::infrastructure::log_broadcaster::{LogBroadcaster, LogRecord};
+
+pub struct BroadcastLayer {
+    broadcaster: Arc<LogBroadcaster>,
+}
+
+impl BroadcastLayer {
+    pub fn new(broadcaster: Arc<LogBroadcaster>) -> Self {
+        Self { broadcaster }
+    }
+}
+
+struct MessageVisitor {
+    message: String,
+}
+
+impl tracing::field::Visit for MessageVisitor {
+    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+        if field.name() == "message" {
+            self.message = format!("{value:?}");
+        }
+    }
+
+    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+        if field.name() == "message" {
+            self.message = value.to_string();
+        }
+    }
+}
+
+impl<S: Subscriber> Layer<S> for BroadcastLayer {
+    fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
+        let level = event.metadata().level().to_string();
+        let mut visitor = MessageVisitor { message: String::new() };
+        event.record(&mut visitor);
+
+        if visitor.message.is_empty() {
+            return;
+        }
+
+        let record = LogRecord {
+            timestamp: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
+            level,
+            message: visitor.message,
+        };
+
+        self.broadcaster.emit(record);
+    }
+}

mantis/src/web/api/logs.rs

@@ -0,0 +1,25 @@
+use axum::Router;
+use axum::extract::{State, WebSocketUpgrade};
+use axum::response::IntoResponse;
+use axum::routing::get;
+
+use crate::core::app_state::AppState;
+use crate::web::websocket::log_websocket::handle_log_stream;
+
+pub fn router() -> Router<AppState> {
+    Router::new()
+        .route("/recent", get(recent_logs))
+        .route("/websocket", get(log_websocket))
+}
+
+async fn recent_logs(State(state): State<AppState>) -> impl IntoResponse {
+    let logs = state.log_broadcaster.recent_logs();
+    axum::Json(logs)
+}
+
+async fn log_websocket(
+    ws: WebSocketUpgrade,
+    State(state): State<AppState>,
+) -> impl IntoResponse {
+    ws.on_upgrade(move |socket| handle_log_stream(socket, state.log_broadcaster))
+}

mantis/src/web/api/mod.rs

@@ -3,4 +3,5 @@ pub mod control;
 pub mod default;
 pub mod detection_alert;
 pub mod health;
+pub mod logs;
 pub mod misc;

mantis/src/web/websocket/log_websocket.rs

@@ -0,0 +1,64 @@
+use std::sync::Arc;
+
+use axum::extract::ws::{Message, WebSocket};
+use futures_util::{SinkExt, StreamExt};
+use macros::log;
+use tokio::sync::broadcast;
+
+use crate::core::infrastructure::log_broadcaster::LogBroadcaster;
+use crate::model::error::http::HttpError;
+use crate::model::error::misc::MiscError;
+use crate::model::log::http::HttpLog;
+
+pub async fn handle_log_stream(socket: WebSocket, broadcaster: Arc<LogBroadcaster>) {
+    let (mut sender, mut receiver) = socket.split();
+
+    // Send buffered recent logs first
+    let recent = broadcaster.recent_logs();
+    for record in recent {
+        match serde_json::to_string(&record) {
+            Ok(json) => {
+                if sender.send(Message::Text(json.into())).await.is_err() {
+                    return;
+                }
+            }
+            Err(e) => log!(MiscError::SerializeError(e)),
+        }
+    }
+
+    let mut rx = broadcaster.subscribe();
+
+    loop {
+        tokio::select! {
+            msg = receiver.next() => {
+                match msg {
+                    Some(Ok(Message::Ping(data))) => {
+                        if sender.send(Message::Pong(data)).await.is_err() { break; }
+                    }
+                    Some(Ok(Message::Close(_))) | None => break,
+                    Some(Err(e)) => {
+                        log!(HttpError::WebSocketError { msg: e.to_string() });
+                        break;
+                    }
+                    _ => {}
+                }
+            }
+            result = rx.recv() => {
+                match result {
+                    Ok(record) => {
+                        match serde_json::to_string(&record) {
+                            Ok(json) => {
+                                if sender.send(Message::Text(json.into())).await.is_err() { break; }
+                            }
+                            Err(e) => log!(MiscError::SerializeError(e)),
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Lagged(n)) => {
+                        log!(HttpLog::WebSocketLaged { skipped: n });
+                    }
+                    Err(broadcast::error::RecvError::Closed) => break,
+                }
+            }
+        }
+    }
+}

mantis/src/web/websocket/mod.rs

@@ -1,3 +1,4 @@
 pub mod alert_websocket;
 pub mod flow_websocket;
 pub mod health_websocket;
+pub mod log_websocket;