DaLaw2/NetGuardia
1
0
Fork 0
mirror of https://github.com/DaLaw2/NetGuardia.git synced 2026-08-24 14:10:28 +09:00
Code Issues Packages Projects Releases Wiki Activity

perf(soar,audit,suricata): unblock tokio worker on hot paths

Browse Source 
Three sources of executor stall under load:

* `audit_logger::handle_audit_event` did synchronous `insert_audit_log`
  inside the broadcast subscriber loop. Every fused-threat emission lands
  here, so the rusqlite WAL write blocks the worker thread that drains
  the channel and cascades back-pressure into broadcast Lagged drops.
  Both AuditEvent and DriftDetectedEvent paths now offload the insert
  via `spawn_blocking`.

* `soar::actions::action_block_ip` made 3-4 synchronous DB calls
  (`get_setting` x2, `commit_soar_block_to_db`, optional
  `insert_pending_unblock`) directly inside the SOAR async task body.
  With r2d2 pool max_size=6 and tokio::spawn-per-event in the SOAR loop,
  a burst of fused detections starves every other async task. The two
  setting reads now batch into a single `spawn_blocking` hop, and the
  commit/pending-unblock writes each run on the blocking pool.

* `soar::rate_limit_owner` ran the synchronous `adjust` /
  `restore_if_expired` bodies (rusqlite + eBPF map writes) directly
  inside the owner async task. The owner's serialization guarantee is
  preserved — only one command runs at a time — but each command body
  now executes on the blocking pool so it can't block the runtime
  thread that's also draining the command channel.

* `suricata_monitor::handle_line` parsed every eve.json line into
  `serde_json::Value` just to test event_type=="alert", but flow / stats
  / dns / http / fileinfo lines vastly outnumber alerts on a busy
  Suricata. Added a cheap substring pre-filter that drops non-alerts
  before the full parse — the strict event_type check still runs after
  the parse, so false positives go through without misclassification.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
DaLaw2 2026-04-19 14:09:39 +08:00
parent 9e5377dab3
commit 80bea35010
5 changed files with 152 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
net-guardia/src/core/soar/actions.rs
View File

@ -130,26 +130,15 @@ impl SoarEngine {
) -> Result<String, Error> {
let ttl_secs = action.params.get("ttl_secs").and_then(|v| v.as_u64()).unwrap_or(1800);
// Validate TTL (runtime-configurable via DB)
let max_ttl: u64 = self
.db
.get_setting("soar_max_ttl_secs")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(86400);
// Two settings read together off the tokio worker thread — r2d2's
// pool.get() and rusqlite are blocking, so back-to-back calls inside
// an async fn can stall the executor under burst load.
let (max_ttl, max_cap) = read_block_caps(self.db.clone()).await?;
if ttl_secs > max_ttl {
Err(SoarError::InvalidTtl(ttl_secs, max_ttl))?;
}
// Atomically check cap and reserve a slot using CAS loop (runtime-configurable via DB)
let max_cap: u32 = self
.db
.get_setting("soar_max_auto_block_cap")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(100);
// Atomically check cap and reserve a slot using CAS loop.
loop {
let current_count = self.active_block_count.load(Ordering::SeqCst);
if current_count >= max_cap {
@ -175,13 +164,19 @@ impl SoarEngine {
let expires_at = Utc::now() + ChronoDuration::seconds(ttl_secs as i64);
let expires_str = expires_at.format("%Y-%m-%d %H:%M:%S").to_string();
// tx-1 (R2 mitigation): atomically write soar_block_rules + acl_rules.
// Either both commit or both roll back — no half-state possible.
// Atomically write soar_block_rules + acl_rules. Either both commit
// or both roll back — no half-state possible. Offloaded so the
// SQLite WAL fsync can't block the tokio worker.
let ip_version = ip_version_from_str(&event.source_ip);
if let Err(e) = self
.db
.commit_soar_block_to_db(&event.source_ip, ip_version, playbook_id, &expires_str)
{
let commit_result = commit_block_blocking(
self.db.clone(),
event.source_ip.clone(),
ip_version,
playbook_id,
expires_str.clone(),
)
.await;
if let Err(e) = commit_result {
// DB tx rolled back both rows; now roll back the eBPF block.
if let Err(unblock_err) = self.access_control.unblock_ip(&event.source_ip) {
log!(SoarLog::EventHandlingFailed(format!(
@ -189,7 +184,7 @@ impl SoarEngine {
event.source_ip, unblock_err
)));
// Write to pending_unblock table so recovery can retry later
if let Err(pend_err) = self.db.insert_pending_unblock(&event.source_ip) {
if let Err(pend_err) = insert_pending_unblock_blocking(self.db.clone(), event.source_ip.clone()).await {
log!(SoarLog::EventHandlingFailed(format!(
"CRITICAL: Failed to queue pending unblock for IP {}: {}",
event.source_ip, pend_err
@ -220,13 +215,7 @@ impl SoarEngine {
Err(SoarError::InvalidRateLimitFactor(factor))?;
}
let max_ttl: u64 = self
.db
.get_setting("soar_max_ttl_secs")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(86400);
let max_ttl = read_max_ttl(self.db.clone()).await;
if ttl_secs > max_ttl {
Err(SoarError::InvalidTtl(ttl_secs, max_ttl))?;
}
@ -449,3 +438,60 @@ impl SoarEngine {
Ok(())
}
}
use std::sync::Arc;
use crate::interface::port::app_repo::AppRepo;
/// Read both block-related caps in a single offloaded blocking call so the
/// async caller pays one spawn_blocking hop instead of two.
async fn read_block_caps(db: Arc<dyn AppRepo>) -> Result<(u64, u32), Error> {
spawn_blocking(move || {
let max_ttl: u64 = db
.get_setting("soar_max_ttl_secs")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(86400);
let max_cap: u32 = db
.get_setting("soar_max_auto_block_cap")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(100);
Ok::<_, Error>((max_ttl, max_cap))
})
.await
.map_err(|e| SoarError::ActionFailed("read_block_caps", e))?
}
async fn read_max_ttl(db: Arc<dyn AppRepo>) -> u64 {
spawn_blocking(move || {
db.get_setting("soar_max_ttl_secs")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(86400)
})
.await
.unwrap_or(86400)
}
async fn commit_block_blocking(
db: Arc<dyn AppRepo>,
source_ip: String,
ip_version: u8,
playbook_id: i64,
expires_str: String,
) -> Result<(), Error> {
spawn_blocking(move || db.commit_soar_block_to_db(&source_ip, ip_version, playbook_id, &expires_str))
.await
.map_err(|e| SoarError::ActionFailed("commit_soar_block_to_db", e))??;
Ok(())
}
async fn insert_pending_unblock_blocking(db: Arc<dyn AppRepo>, source_ip: String) -> Result<i64, Error> {
spawn_blocking(move || db.insert_pending_unblock(&source_ip))
.await
.map_err(|e| SoarError::ActionFailed("insert_pending_unblock", e))?
}

24
net-guardia/src/core/soar/rate_limit_owner.rs
View File

@ -53,6 +53,11 @@ impl RateLimitOwnerHandle {
/// Spawn the owner task on the current tokio runtime. The owner holds
/// the only mutating references to the DB rate-limit settings and the
/// eBPF rate-limit map for the duration of an Adjust / Restore batch.
///
/// Each command body runs inside `spawn_blocking` because both halves
/// (rusqlite via r2d2 and eBPF map writes) are synchronous I/O — running
/// them directly inside the async owner loop would block the tokio
/// worker thread for the entire adjust/restore batch.
pub fn spawn(db: Arc<dyn AppRepo>, rate_limit: Arc<dyn RateLimitPort>) -> Self {
let (tx, mut rx) = mpsc::channel::<RateLimitCmd>(RATE_LIMIT_CMD_CHANNEL_CAPACITY);
tokio::spawn(async move {
@ -65,11 +70,26 @@ impl RateLimitOwnerHandle {
attack_type,
reply,
} => {
let result = adjust(&db, rate_limit.as_ref(), factor, ttl_secs, &source_ip, &attack_type);
let db_c = db.clone();
let rl_c = rate_limit.clone();
let join = tokio::task::spawn_blocking(move || {
adjust(&db_c, rl_c.as_ref(), factor, ttl_secs, &source_ip, &attack_type)
})
.await;
let result = match join {
Ok(r) => r,
Err(e) => Err(SoarError::RateLimitOwnerJoinFailed(e.to_string()).into()),
};
let _ = reply.send(result);
}
RateLimitCmd::RestoreIfExpired { reply } => {
let result = restore_if_expired(&db, rate_limit.as_ref());
let db_c = db.clone();
let rl_c = rate_limit.clone();
let join = tokio::task::spawn_blocking(move || restore_if_expired(&db_c, rl_c.as_ref())).await;
let result = match join {
Ok(r) => r,
Err(e) => Err(SoarError::RateLimitOwnerJoinFailed(e.to_string()).into()),
};
let _ = reply.send(result);
}
}

55
net-guardia/src/infrastructure/audit_logger.rs
View File

@ -29,7 +29,7 @@ impl AuditLogger {
loop {
match rx.recv().await {
Ok(event) => {
this.handle_audit_event(&event);
this.handle_audit_event(event).await;
}
Err(RecvError::Lagged(n)) => {
log!(AuditLog::AuditLagged(n));
@ -52,7 +52,7 @@ impl AuditLogger {
loop {
match rx.recv().await {
Ok(event) => {
this.handle_drift_event(&event);
this.handle_drift_event(event).await;
}
Err(RecvError::Lagged(n)) => {
log!(AuditLog::AuditLagged(n));
@ -69,21 +69,36 @@ impl AuditLogger {
}
}
fn handle_audit_event(&self, event: &AuditEvent) {
async fn handle_audit_event(&self, event: AuditEvent) {
// Always emit a structured log line
log!(AuditLog::AuditEvent(event.actor.clone(), event.action.clone(),));
// Attempt DB insert; on failure, log a warning but do not panic
if let Err(e) = self.db.insert_audit_log(&event.actor, &event.action, &event.detail) {
log!(AuditLog::AuditDbWriteFailed(
e.to_string(),
event.actor.clone(),
event.action.clone(),
));
// SQLite insert via r2d2 is blocking; offload so it can't stall the
// tokio worker that drains the broadcast channel. The actor/action
// strings are cloned for the log call above; the event itself moves
// into the blocking task.
let db = self.db.clone();
let join = tokio::task::spawn_blocking(move || {
db.insert_audit_log(&event.actor, &event.action, &event.detail)
.map_err(|e| (e.to_string(), event.actor, event.action))
})
.await;
match join {
Ok(Ok(())) => {}
Ok(Err((err, actor, action))) => {
log!(AuditLog::AuditDbWriteFailed(err, actor, action));
}
Err(join_err) => {
log!(AuditLog::AuditDbWriteFailed(
format!("blocking task join failed: {join_err}"),
"<lost>".to_string(),
"<lost>".to_string(),
));
}
}
}
fn handle_drift_event(&self, event: &DriftDetectedEvent) {
async fn handle_drift_event(&self, event: DriftDetectedEvent) {
let detail = serde_json::json!({
"drifted_features": event.drifted_features,
"max_deviation": event.max_deviation,
@ -92,8 +107,22 @@ impl AuditLogger {
log!(AuditLog::AuditDriftEvent(event.drifted_features.len()));
if let Err(e) = self.db.insert_audit_log("system", "ml_drift_detected", &detail) {
log!(AuditLog::AuditDriftDbWriteFailed(e.to_string()));
let db = self.db.clone();
let join = tokio::task::spawn_blocking(move || {
db.insert_audit_log("system", "ml_drift_detected", &detail)
.map_err(|e| e.to_string())
})
.await;
match join {
Ok(Ok(())) => {}
Ok(Err(err)) => {
log!(AuditLog::AuditDriftDbWriteFailed(err));
}
Err(join_err) => {
log!(AuditLog::AuditDriftDbWriteFailed(format!(
"blocking task join failed: {join_err}"
)));
}
}
}
}

8
net-guardia/src/infrastructure/suricata_monitor.rs
View File

@ -115,6 +115,14 @@ impl SuricataMonitor {
if raw.is_empty() {
return;
}
// Cheap pre-filter: a busy Suricata writes flow/stats/dns/http/fileinfo
// lines that vastly outnumber alerts. Parsing every line into
// serde_json::Value just to drop it dominates CPU on this tail. The
// substring match is a sound over-approximation — false positives go
// through the full parse + the strict event_type=="alert" check below.
if !raw.contains("\"event_type\":\"alert\"") {
return;
}
let v: serde_json::Value = match serde_json::from_str(raw) {
Ok(v) => v,
Err(_) => return,

4
net-guardia/src/model/error/soar.rs
View File

@ -59,5 +59,9 @@ traceable! {
#[no_source]
#[error("Rate-limit owner task is unavailable (channel closed)")]
RateLimitOwnerUnavailable => tracing::Level::ERROR,
#[no_source]
#[error("Rate-limit owner blocking task panicked or was cancelled: {detail}")]
RateLimitOwnerJoinFailed { detail: String } => tracing::Level::ERROR,
}
}