From 196e22afae2ca103f252f69c388c037f87bdb146 Mon Sep 17 00:00:00 2001 From: ParrotXray Date: Mon, 25 May 2026 03:19:44 +0000 Subject: [PATCH] feat: Add live system log WebSocket API and frontend log viewer --- mantis-frontend | 2 +- mantis/src/core/app_state.rs | 2 + .../core/infrastructure/log_broadcaster.rs | 53 +++++++++++++++ mantis/src/core/infrastructure/mod.rs | 9 ++- mantis/src/core/system.rs | 14 +++- mantis/src/utils/logging.rs | 8 ++- mantis/src/utils/mod.rs | 1 + mantis/src/utils/tracing_layer.rs | 55 ++++++++++++++++ mantis/src/web/api/logs.rs | 25 ++++++++ mantis/src/web/api/mod.rs | 1 + mantis/src/web/websocket/log_websocket.rs | 64 +++++++++++++++++++ mantis/src/web/websocket/mod.rs | 1 + 12 files changed, 229 insertions(+), 6 deletions(-) create mode 100644 mantis/src/core/infrastructure/log_broadcaster.rs create mode 100644 mantis/src/utils/tracing_layer.rs create mode 100644 mantis/src/web/api/logs.rs create mode 100644 mantis/src/web/websocket/log_websocket.rs diff --git a/mantis-frontend b/mantis-frontend index fd1ca15..2b9c85d 160000 --- a/mantis-frontend +++ b/mantis-frontend @@ -1 +1 @@ -Subproject commit fd1ca155e745e1e2121b4a0c2add3a06a71aec42 +Subproject commit 2b9c85de173c2810a8ffca183ddddcc751f4e227 diff --git a/mantis/src/core/app_state.rs b/mantis/src/core/app_state.rs index 4be2a92..e580104 100644 --- a/mantis/src/core/app_state.rs +++ b/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, pub detection_alert: Arc, pub app_db: Option>, + pub log_broadcaster: Arc, } diff --git a/mantis/src/core/infrastructure/log_broadcaster.rs b/mantis/src/core/infrastructure/log_broadcaster.rs new file mode 100644 index 0000000..a8eaedb --- /dev/null +++ b/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, + recent: Mutex>, +} + +impl LogBroadcaster { + pub fn new() -> Arc { + 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 { + self.tx.subscribe() + } + + pub fn recent_logs(&self) -> Vec { + self.recent + .lock() + .map(|buf| buf.iter().cloned().collect()) + .unwrap_or_default() + } +} diff --git a/mantis/src/core/infrastructure/mod.rs b/mantis/src/core/infrastructure/mod.rs index 6a62c73..c24fe84 100644 --- a/mantis/src/core/infrastructure/mod.rs +++ b/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, pub suricata_engine: Option>, pub app_db: Option>, + pub log_broadcaster: Arc, shutdowns: SegQueue>, } impl AppServices { - pub fn new(app_config: Arc, inference_config: Arc) -> Result { + pub fn new( + app_config: Arc, + inference_config: Arc, + log_broadcaster: Arc, + ) -> Result { 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(), }) } diff --git a/mantis/src/core/system.rs b/mantis/src/core/system.rs index 5c9ee9d..f8ee8b9 100644 --- a/mantis/src/core/system.rs +++ b/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, @@ -40,7 +41,8 @@ pub struct System { impl System { pub async fn new() -> Result { - 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) diff --git a/mantis/src/utils/logging.rs b/mantis/src/utils/logging.rs index 18b0904..5c967dc 100644 --- a/mantis/src/utils/logging.rs +++ b/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) -> 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(); diff --git a/mantis/src/utils/mod.rs b/mantis/src/utils/mod.rs index b3ee894..6e26a27 100644 --- a/mantis/src/utils/mod.rs +++ b/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; diff --git a/mantis/src/utils/tracing_layer.rs b/mantis/src/utils/tracing_layer.rs new file mode 100644 index 0000000..19d08b6 --- /dev/null +++ b/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, +} + +impl BroadcastLayer { + pub fn new(broadcaster: Arc) -> 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 Layer 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); + } +} diff --git a/mantis/src/web/api/logs.rs b/mantis/src/web/api/logs.rs new file mode 100644 index 0000000..716066d --- /dev/null +++ b/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 { + Router::new() + .route("/recent", get(recent_logs)) + .route("/websocket", get(log_websocket)) +} + +async fn recent_logs(State(state): State) -> impl IntoResponse { + let logs = state.log_broadcaster.recent_logs(); + axum::Json(logs) +} + +async fn log_websocket( + ws: WebSocketUpgrade, + State(state): State, +) -> impl IntoResponse { + ws.on_upgrade(move |socket| handle_log_stream(socket, state.log_broadcaster)) +} diff --git a/mantis/src/web/api/mod.rs b/mantis/src/web/api/mod.rs index 37af96a..c6341a5 100644 --- a/mantis/src/web/api/mod.rs +++ b/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; diff --git a/mantis/src/web/websocket/log_websocket.rs b/mantis/src/web/websocket/log_websocket.rs new file mode 100644 index 0000000..9048ca9 --- /dev/null +++ b/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) { + 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, + } + } + } + } +} diff --git a/mantis/src/web/websocket/mod.rs b/mantis/src/web/websocket/mod.rs index 8ae1755..84b14cc 100644 --- a/mantis/src/web/websocket/mod.rs +++ b/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;