feat: add performance monitoring and logging capabilities to SipFlow[latest]

Author: shenjinti

src/bin/sipflow.rs

@@ -11,13 +11,17 @@ use rustpbx::config::{SipFlowConfig, SipFlowEngine, SipFlowSubdirs};
 use rustpbx::sipflow::{
     SipFlowBackend, SipFlowItem, SipFlowMsgType,
     create_backend,
+    perf::{PerfCounters, PerfDumper},
     protocol::{MsgType, Packet, parse_packet},
     storage::extract_callid,
 };
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::sync::atomic::Ordering;
 use tokio::net::UdpSocket;
+use tracing_appender::non_blocking;
+use tracing_subscriber::{fmt, EnvFilter};
 
 #[derive(Parser, Debug)]
 #[command(author, version, about = "SipFlow - SIP and RTP flow recording server", long_about = None)]
@@ -38,8 +42,8 @@ struct Args {
     #[arg(short, long, default_value = "./config/sipflow")]
     root: String,
 
-    /// Storage engine: "flowdb" (default) or "sqlite"
-    #[arg(long, default_value = "flowdb")]
+    /// Storage engine: "flowdb" or "sqlite" (default)
+    #[arg(long, default_value = "sqlite")]
     engine: String,
 
     /// Channel buffer size
@@ -60,6 +64,16 @@ struct Args {
     #[arg(long, default_value_t = 8192)]
     id_cache_size: usize,
 
+    // ── Logging options ──
+
+    /// Log file path
+    #[arg(long, default_value = "/var/log/sipflow.log")]
+    log_file: String,
+
+    /// Log level (trace, debug, info, warn, error)
+    #[arg(long, default_value = "info")]
+    log_level: String,
+
     // ── FlowDB options ──
 
     /// TTL in seconds for FlowDB records (optional, 0 = no ttl)
@@ -111,15 +125,41 @@ fn convert_packet_to_item(packet: Packet) -> (String, SipFlowItem) {
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    tracing_subscriber::fmt::init();
     let args = Args::parse();
 
+    // Initialize tracing: try log file, fall back to stdout on permission error
+    if let Some(parent) = std::path::Path::new(&args.log_file).parent() {
+        let _ = std::fs::create_dir_all(parent);
+    }
+    let (_guard, writer) = match std::fs::OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(&args.log_file)
+    {
+        Ok(f) => {
+            let (w, g) = non_blocking(f);
+            (g, w)
+        }
+        Err(e) => {
+            eprintln!(
+                "sipflow: cannot open '{}' ({}), falling back to stdout",
+                args.log_file, e
+            );
+            let (w, g) = non_blocking(std::io::stdout());
+            (g, w)
+        }
+    };
+    fmt()
+        .with_env_filter(EnvFilter::new(&args.log_level))
+        .with_writer(writer)
+        .init();
+
     // Ensure data directory exists
     std::fs::create_dir_all(&args.root)?;
 
     let engine = match args.engine.as_str() {
-        "sqlite" => SipFlowEngine::Sqlite,
-        "flowdb" | _ => SipFlowEngine::FlowDb,
+        "flowdb" => SipFlowEngine::FlowDb,
+        "sqlite" | _ => SipFlowEngine::Sqlite,
     };
     let ttl_secs = args.ttl_secs.filter(|&s| s > 0);
 
@@ -137,6 +177,8 @@ async fn main() -> Result<()> {
     };
 
     let backend: Arc<dyn SipFlowBackend> = Arc::from(create_backend(&config)?);
+    let perf_counters = PerfCounters::new_arc();
+
     let app_state = AppState {
         backend: backend.clone(),
     };
@@ -148,12 +190,14 @@ async fn main() -> Result<()> {
     let (tx, mut rx) = tokio::sync::mpsc::channel::<Packet>(args.buffer_size);
 
     // UDP Receiver Task
+    let perf_rx = perf_counters.clone();
     rustpbx::utils::spawn(async move {
         let mut buf = vec![0u8; 65535];
         loop {
             match socket.recv_from(&mut buf).await {
                 Ok((size, _)) => {
                     if let Ok(packet) = parse_packet(&buf[..size]) {
+                        perf_rx.packets_received.fetch_add(1, Ordering::Relaxed);
                         let _ = tx.try_send(packet);
                     }
                 }
@@ -166,6 +210,7 @@ async fn main() -> Result<()> {
 
     // Storage Worker Task
     let storage_backend = backend.clone();
+    let perf_worker = perf_counters.clone();
     rustpbx::utils::spawn(async move {
         let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
         loop {
@@ -174,15 +219,32 @@ async fn main() -> Result<()> {
                     let (call_id, item) = convert_packet_to_item(packet);
                     if !call_id.is_empty() {
                         let _ = storage_backend.record(&call_id, item);
+                        perf_worker.items_recorded.fetch_add(1, Ordering::Relaxed);
+                    } else {
+                        perf_worker.items_dropped.fetch_add(1, Ordering::Relaxed);
                     }
+                    perf_worker.set_pending(rx.len() as i64);
                 }
                 _ = interval.tick() => {
                     let _ = storage_backend.flush().await;
+                    perf_worker.flushes.fetch_add(1, Ordering::Relaxed);
                 }
             }
         }
     });
 
+    // Periodic perf dump (every 10 s, skipped when idle)
+    let perf_dump = perf_counters.clone();
+    rustpbx::utils::spawn(async move {
+        let mut dumper = PerfDumper::new(perf_dump);
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+            if let Some(msg) = dumper.try_dump() {
+                tracing::info!("{msg}");
+            }
+        }
+    });
+
     // HTTP Server
     let app = Router::new()
         .route("/health", get(health_handler))

src/config.rs

@@ -540,11 +540,11 @@ pub enum SipFlowSubdirs {
 /// Storage engine selection for the Local sipflow backend.
 #[derive(Debug, Deserialize, Clone, Copy, Serialize, PartialEq, Eq, Default)]
 pub enum SipFlowEngine {
-    /// Use FlowDB LSM-tree engine (default).
-    #[default]
+    /// Use FlowDB LSM-tree engine.
     #[serde(rename = "flowdb")]
     FlowDb,
-    /// Use the legacy SQLite + raw-file engine.
+    /// Use SQLite + raw-file engine (default).
+    #[default]
     #[serde(rename = "sqlite")]
     Sqlite,
 }
@@ -601,7 +601,7 @@ pub enum SipFlowConfig {
         flush_interval_secs: u64,
         #[serde(default = "default_sipflow_id_cache_size")]
         id_cache_size: usize,
-        /// Storage engine: "flowdb" (default) or "sqlite".
+        /// Storage engine: "flowdb" or "sqlite" (default).
         #[serde(default)]
         engine: SipFlowEngine,
         /// TTL in seconds for FlowDB records (optional). When set,

src/sipflow/backend/local.rs

@@ -9,12 +9,14 @@ use tracing::warn;
 
 use crate::config::SipFlowSubdirs;
 use crate::sipflow::backend::SipFlowBackend;
+use crate::sipflow::perf::{PerfCounters, PerfDumper};
 use crate::sipflow::protocol::{MsgType, Packet};
 use crate::sipflow::storage::{StorageManager, process_packet};
 use crate::sipflow::wav_utils::{
     generate_wav_to_writer,
 };
 use crate::sipflow::{SipFlowItem, SipFlowMediaStats, SipFlowMsgType};
+use std::sync::atomic::Ordering;
 
 enum Command {
     RecordItem {
@@ -50,6 +52,8 @@ impl LocalBackend {
         let root_clone = root.clone();
         let subdirs_clone = subdirs.clone();
         // Spawn background worker task
+        let perf = PerfCounters::new_arc();
+        let perf_dumper = perf.clone();
         crate::utils::spawn(async move {
             let mut storage = StorageManager::new(
                 &PathBuf::from(&root_clone),
@@ -60,6 +64,7 @@ impl LocalBackend {
             );
 
             let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
+            let mut dumper = PerfDumper::new(perf_dumper);
 
             loop {
                 tokio::select! {
@@ -70,6 +75,7 @@ impl LocalBackend {
                     Some(cmd) = rx.recv() => {
                         match cmd {
                             Command::RecordItem { call_id, item } => {
+                                perf.items_recorded.fetch_add(1, Ordering::Relaxed);
                                 // Convert SipFlowItem to Packet for storage
                                 let default_port = if matches!(&item.msg_type, SipFlowMsgType::Sip)
                                 {
@@ -121,12 +127,18 @@ impl LocalBackend {
                             }
                             Command::Flush { done } => {
                                 let _ = storage.force_flush().await;
+                                perf.flushes.fetch_add(1, Ordering::Relaxed);
+                                perf.set_pending(storage.batch_len() as i64);
                                 let _ = done.send(());
                             }
                         }
                     }
                     _ = interval.tick() => {
                         let _ = storage.check_flush().await;
+                        perf.set_pending(storage.batch_len() as i64);
+                        if let Some(msg) = dumper.try_dump() {
+                            tracing::info!("{msg}");
+                        }
                     }
                 }
             }

src/sipflow/mod.rs

@@ -1,6 +1,7 @@
 pub mod backend;
 pub mod flowdb_backend;
 pub mod flowdb_codec;
+pub mod perf;
 pub mod protocol;
 pub mod rtp_stats;
 pub mod sdp_utils;

src/sipflow/perf.rs

@@ -0,0 +1,107 @@
+use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Instant;
+
+/// Shared performance counters for sipflow (atomics, no contention on hot path).
+pub struct PerfCounters {
+    pub packets_received: AtomicU64,
+    pub items_recorded: AtomicU64,
+    pub items_dropped: AtomicU64,
+    pub flushes: AtomicU64,
+    /// Explicitly-set pending/backlog count (e.g. items in channel or batch).
+    /// Set via [`Self::set_pending`] from wherever the backlog is observable.
+    pub pending: AtomicI64,
+}
+
+impl PerfCounters {
+    pub fn new_arc() -> Arc<Self> {
+        Arc::new(Self {
+            packets_received: AtomicU64::new(0),
+            items_recorded: AtomicU64::new(0),
+            items_dropped: AtomicU64::new(0),
+            flushes: AtomicU64::new(0),
+            pending: AtomicI64::new(0),
+        })
+    }
+
+    pub fn set_pending(&self, n: i64) {
+        self.pending.store(n, Ordering::Relaxed);
+    }
+}
+
+/// Perf-dumper that tracks snapshot state.  Call `try_dump()` periodically;
+/// it returns a log message at most once every 10 seconds, skipping periods
+/// with zero activity, and exports counters to `metrics`.
+pub struct PerfDumper {
+    counters: Arc<PerfCounters>,
+    last_time: Instant,
+    last_received: u64,
+    last_recorded: u64,
+    last_dropped: u64,
+    last_flushes: u64,
+}
+
+impl PerfDumper {
+    pub fn new(counters: Arc<PerfCounters>) -> Self {
+        Self {
+            counters,
+            last_time: Instant::now(),
+            last_received: 0,
+            last_recorded: 0,
+            last_dropped: 0,
+            last_flushes: 0,
+        }
+    }
+
+    /// Returns `Some(msg)` every ~10 s if any counter changed, `None` otherwise.
+    pub fn try_dump(&mut self) -> Option<String> {
+        let now = Instant::now();
+        let elapsed = now.duration_since(self.last_time).as_secs_f64();
+        if elapsed < 10.0 {
+            return None;
+        }
+
+        let received = self.counters.packets_received.load(Ordering::Relaxed);
+        let recorded = self.counters.items_recorded.load(Ordering::Relaxed);
+        let dropped = self.counters.items_dropped.load(Ordering::Relaxed);
+        let flushes = self.counters.flushes.load(Ordering::Relaxed);
+        let pending = self.counters.pending.load(Ordering::Relaxed);
+
+        let dr = received - self.last_received;
+        let dw = recorded - self.last_recorded;
+        let dd = dropped - self.last_dropped;
+        let df = flushes - self.last_flushes;
+
+        self.last_received = received;
+        self.last_recorded = recorded;
+        self.last_dropped = dropped;
+        self.last_flushes = flushes;
+        self.last_time = now;
+
+        // ── export to metrics ──
+        metrics::counter!("sipflow_packets_received_total", "component" => "sipflow")
+            .increment(dr);
+        metrics::counter!("sipflow_items_recorded_total", "component" => "sipflow")
+            .increment(dw);
+        metrics::counter!("sipflow_items_dropped_total", "component" => "sipflow")
+            .increment(dd);
+        metrics::counter!("sipflow_flushes_total", "component" => "sipflow")
+            .increment(df);
+        metrics::gauge!("sipflow_pending_items", "component" => "sipflow")
+            .set(pending as f64);
+
+        if dr == 0 && dw == 0 && dd == 0 && df == 0 {
+            return None;
+        }
+
+        Some(format!(
+            "sipflow perf: recv={:.1}/s  record={:.1}/s  drop={:.1}/s  flush={:.1}/s  pending={}  total_recorded={}",
+            dr as f64 / elapsed,
+            dw as f64 / elapsed,
+            dd as f64 / elapsed,
+            df as f64 / elapsed,
+            pending,
+            recorded,
+        ))
+    }
+}

src/sipflow/storage.rs

@@ -237,6 +237,11 @@ impl StorageManager {
         Ok(())
     }
 
+    /// Number of items waiting to be flushed.
+    pub fn batch_len(&self) -> usize {
+        self.batch.len()
+    }
+
     pub async fn check_flush(&mut self) -> Result<()> {
         if !self.batch.is_empty() && self.last_flush.elapsed().as_secs() >= self.flush_interval {
             self.flush_batch().await?;

src/sipflow/tests.rs

@@ -277,9 +277,9 @@ engine = "sqlite"
         }
     }
 
-    /// Default engine (omitted in config) should be FlowDb.
+    /// Default engine (omitted in config) should be Sqlite.
     #[test]
-    fn test_config_default_engine_is_flowdb() {
+    fn test_config_default_engine_is_sqlite() {
         let toml_str = r#"
 type = "local"
 root = "/var/sipflow"
@@ -288,7 +288,7 @@ root = "/var/sipflow"
             toml::from_str(toml_str).expect("should parse default config");
         match cfg {
             crate::config::SipFlowConfig::Local { engine, .. } => {
-                assert_eq!(engine, SipFlowEngine::FlowDb);
+                assert_eq!(engine, SipFlowEngine::Sqlite);
             }
             _ => panic!("expected Local config"),
         }