use crate::stats::CallStats;
use anyhow::Result;
use chrono::Local;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tokio::fs::OpenOptions;
use tokio::io::AsyncWriteExt;
use tokio::time::{Duration, interval};
use tracing::{error, info};
/// CSV Statistics Recorder for load testing
pub struct CsvStatsRecorder {
stats: Arc<CallStats>,
file_path: String,
interval_secs: u64,
start_time: std::time::Instant,
}
impl CsvStatsRecorder {
pub fn new(stats: Arc<CallStats>, file_path: impl Into<String>, interval_secs: u64) -> Self {
Self {
stats,
file_path: file_path.into(),
interval_secs,
start_time: std::time::Instant::now(),
}
}
/// Write CSV header
async fn write_header(&self, file: &mut tokio::fs::File) -> Result<()> {
let header = "timestamp,elapsed_time_ms,total_planned,total_calls,current_calls,finished_calls,c200_count,c180_count,c183_count,c3xx_count,c4xx_count,c5xx_count,c6xx_count,avg_duration_ms,avg_setup_latency_ms,avg_rtcp_rtt_ms,tx_packets,rx_packets,tx_bytes,rx_bytes,rx_lost_packets,packet_loss_rate_pct,nack_sent,nack_recv,nack_recovered,audio_quality_silence_frames,audio_quality_clipping_frames,audio_quality_total_frames,audio_quality_shrill_count,audio_quality_muffled_count,rx_dtmf_events,tx_dtmf_events,seq_gap_events,seq_gap_total,seq_gap_max,seq_reorder_events,ts_jump_events,ts_jump_ms_total,ts_jump_ms_max,stream_switch_events\n";
file.write_all(header.as_bytes()).await?;
file.flush().await?;
Ok(())
}
/// Format current statistics as CSV row
fn format_csv_row(&self) -> String {
let timestamp = Local::now().format("%Y-%m-%dT%H:%M:%S%.3f");
let elapsed_ms = self.start_time.elapsed().as_millis();
let total_planned = self.stats.total_planned_calls.load(Ordering::Relaxed);
let total_calls = self.stats.total_calls.load(Ordering::Relaxed);
let current = self.stats.current_calls.load(Ordering::Relaxed);
let finished = self.stats.finished_calls.load(Ordering::Relaxed);
// SIP response code counts
let c200 = self.stats.status_codes[200].load(Ordering::Relaxed);
let c180 = self.stats.status_codes[180].load(Ordering::Relaxed);
let c183 = self.stats.status_codes[183].load(Ordering::Relaxed);
let mut c3xx = 0u32;
let mut c4xx = 0u32;
let mut c5xx = 0u32;
let mut c6xx = 0u32;
for code in 300..700 {
let count = self.stats.status_codes[code].load(Ordering::Relaxed);
if count == 0 {
continue;
}
match code {
300..=399 => c3xx += count,
400..=499 => c4xx += count,
500..=599 => c5xx += count,
600..=699 => c6xx += count,
_ => {}
}
}
// Duration and latency
let total_duration_ms = self.stats.total_duration.load(Ordering::Relaxed);
let avg_duration = if finished > 0 {
total_duration_ms as f64 / finished as f64
} else {
0.0
};
let setup_samples = self.stats.setup_samples.load(Ordering::Relaxed);
let total_setup_ms = self.stats.total_setup_latency_ms.load(Ordering::Relaxed);
let avg_setup_ms = if setup_samples > 0 {
total_setup_ms as f64 / setup_samples as f64
} else {
0.0
};
// RTCP RTT
let rtcp_samples = self.stats.rtcp_rtt_samples.load(Ordering::Relaxed);
let total_rtcp_us = self.stats.total_rtcp_rtt_us.load(Ordering::Relaxed);
let avg_rtcp_ms = if rtcp_samples > 0 {
(total_rtcp_us as f64 / rtcp_samples as f64) / 1000.0
} else {
0.0
};
// RTP stats
let tx_packets = self.stats.tx_packets.load(Ordering::Relaxed);
let rx_packets = self.stats.rx_packets.load(Ordering::Relaxed);
let tx_bytes = self.stats.tx_bytes.load(Ordering::Relaxed);
let rx_bytes = self.stats.rx_bytes.load(Ordering::Relaxed);
let rx_lost = self.stats.rx_lost_packets.load(Ordering::Relaxed);
let loss_rate = self.stats.average_loss_rate();
// NACK stats
let nack_sent = self.stats.nack_sent.load(Ordering::Relaxed);
let nack_recv = self.stats.nack_recv.load(Ordering::Relaxed);
let nack_recovered = self.stats.nack_recovered.load(Ordering::Relaxed);
// Audio quality + DTMF stats
let aq_silence = self.stats.audio_quality_silence_frames.load(Ordering::Relaxed);
let aq_clipping = self.stats.audio_quality_clipping_frames.load(Ordering::Relaxed);
let aq_total = self.stats.audio_quality_total_frames.load(Ordering::Relaxed);
let aq_shrill = self.stats.audio_quality_shrill_count.load(Ordering::Relaxed);
let aq_muffled = self.stats.audio_quality_muffled_count.load(Ordering::Relaxed);
let rx_dtmf = self.stats.rx_dtmf_events.load(Ordering::Relaxed);
let tx_dtmf = self.stats.tx_dtmf_events.load(Ordering::Relaxed);
// Media-continuity (seq/ts jump) stats
let seq_gap_events = self.stats.seq_gap_events.load(Ordering::Relaxed);
let seq_gap_total = self.stats.seq_gap_total.load(Ordering::Relaxed);
let seq_gap_max = self.stats.seq_gap_max.load(Ordering::Relaxed);
let seq_reorder = self.stats.seq_reorder_events.load(Ordering::Relaxed);
let ts_jump_events = self.stats.ts_jump_events.load(Ordering::Relaxed);
let ts_jump_ms_total = self.stats.ts_jump_ms_total.load(Ordering::Relaxed);
let ts_jump_ms_max = self.stats.ts_jump_ms_max.load(Ordering::Relaxed);
let stream_switch = self.stats.stream_switch_events.load(Ordering::Relaxed);
format!(
"{},{},{},{},{},{},{},{},{},{},{},{},{},{:.2},{:.2},{:.2},{},{},{},{},{},{:.2},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{}\n",
timestamp,
elapsed_ms,
total_planned,
total_calls,
current,
finished,
c200,
c180,
c183,
c3xx,
c4xx,
c5xx,
c6xx,
avg_duration,
avg_setup_ms,
avg_rtcp_ms,
tx_packets,
rx_packets,
tx_bytes,
rx_bytes,
rx_lost,
loss_rate,
nack_sent,
nack_recv,
nack_recovered,
aq_silence,
aq_clipping,
aq_total,
aq_shrill,
aq_muffled,
rx_dtmf,
tx_dtmf,
seq_gap_events,
seq_gap_total,
seq_gap_max,
seq_reorder,
ts_jump_events,
ts_jump_ms_total,
ts_jump_ms_max,
stream_switch
)
}
/// Start the CSV recording loop
pub async fn start(&self) -> Result<()> {
let path = Path::new(&self.file_path);
// Create parent directory if needed
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent).await.ok();
}
// Create/truncate file and write header
let mut file = OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(path)
.await?;
self.write_header(&mut file).await?;
info!("[CSV] Started recording stats to: {}", self.file_path);
// Record initial state
let initial_row = self.format_csv_row();
file.write_all(initial_row.as_bytes()).await?;
file.flush().await?;
// Start periodic recording
let mut ticker = interval(Duration::from_secs(self.interval_secs));
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
ticker.tick().await;
let row = self.format_csv_row();
if let Err(e) = file.write_all(row.as_bytes()).await {
error!("[CSV] Failed to write stats: {}", e);
continue;
}
if let Err(e) = file.flush().await {
error!("[CSV] Failed to flush stats: {}", e);
continue;
}
}
}
/// Start the recorder in a background task
pub fn spawn(self: Arc<Self>) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
if let Err(e) = self.start().await {
error!("[CSV] Recorder failed: {}", e);
}
})
}
}
/// Write final summary to a separate file
pub async fn write_final_summary(stats: &CallStats, file_path: impl AsRef<Path>) -> Result<()> {
let path = file_path.as_ref();
let mut file = OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(path)
.await?;
let finished = stats.finished_calls.load(Ordering::Relaxed);
let total = stats.total_planned_calls.load(Ordering::Relaxed);
let current = stats.current_calls.load(Ordering::Relaxed);
let total_duration_ms = stats.total_duration.load(Ordering::Relaxed);
let avg_duration = if finished > 0 {
total_duration_ms as f64 / finished as f64 / 1000.0
} else {
0.0
};
let setup_samples = stats.setup_samples.load(Ordering::Relaxed);
let total_setup_ms = stats.total_setup_latency_ms.load(Ordering::Relaxed);
let avg_setup_ms = if setup_samples > 0 {
total_setup_ms as f64 / setup_samples as f64
} else {
0.0
};
let avg_rtcp_ms = stats.average_rtcp_rtt_ms();
let summary = format!(
r#"Sipbot Load Test Summary
==========================
Timestamp: {}
Call Statistics
---------------
Total Planned: {}
Total Finished: {}
Current Active: {}
Success Rate: {:.2}%
Timing
------
Average Call Duration: {:.2} seconds
Average Setup Latency: {:.2} ms
Average RTCP RTT: {:.2} ms
SIP Response Codes
------------------
"#,
Local::now().format("%Y-%m-%d %H:%M:%S"),
total,
finished,
current,
if total > 0 {
finished as f64 * 100.0 / total as f64
} else {
0.0
},
avg_duration,
avg_setup_ms,
avg_rtcp_ms
);
file.write_all(summary.as_bytes()).await?;
// Write response code counts
for code in 100..700 {
let count = stats.status_codes[code].load(Ordering::Relaxed);
if count > 0 {
let line = format!(" {}: {}\n", code, count);
file.write_all(line.as_bytes()).await?;
}
}
// RTP Stats
let tx_packets = stats.tx_packets.load(Ordering::Relaxed);
let rx_packets = stats.rx_packets.load(Ordering::Relaxed);
let tx_bytes = stats.tx_bytes.load(Ordering::Relaxed);
let rx_bytes = stats.rx_bytes.load(Ordering::Relaxed);
let rx_lost = stats.rx_lost_packets.load(Ordering::Relaxed);
let loss_rate = stats.average_loss_rate();
let rtp_summary = format!(
r#"
RTP Statistics
--------------
TX Packets: {}
RX Packets: {}
TX Bytes: {}
RX Bytes: {}
RX Lost Packets: {}
Packet Loss Rate: {:.2}%
NACK Statistics
---------------
NACK Sent: {}
NACK Received: {}
NACK Recovered: {}
Audio Quality
-------------
Total Frames: {}
Silence Frames: {}
Clipping Frames: {}
Shrill Count: {}
Muffled Count: {}
DTMF Events
-----------
RX DTMF Events: {}
TX DTMF Events: {}
Media Continuity (seq/ts jumps — audio-glitch warnings)
-------------------------------------------------------
Seq Gap Events: {}
Seq Gap Total (lost pkts): {}
Seq Gap Max: {}
Seq Reorder Events: {}
TS Jump Events: {}
TS Jump Total (ms): {}
TS Jump Max (ms): {}
Stream Switch Events: {}
"#,
tx_packets,
rx_packets,
tx_bytes,
rx_bytes,
rx_lost,
loss_rate,
stats.nack_sent.load(Ordering::Relaxed),
stats.nack_recv.load(Ordering::Relaxed),
stats.nack_recovered.load(Ordering::Relaxed),
stats.audio_quality_total_frames.load(Ordering::Relaxed),
stats.audio_quality_silence_frames.load(Ordering::Relaxed),
stats.audio_quality_clipping_frames.load(Ordering::Relaxed),
stats.audio_quality_shrill_count.load(Ordering::Relaxed),
stats.audio_quality_muffled_count.load(Ordering::Relaxed),
stats.rx_dtmf_events.load(Ordering::Relaxed),
stats.tx_dtmf_events.load(Ordering::Relaxed),
stats.seq_gap_events.load(Ordering::Relaxed),
stats.seq_gap_total.load(Ordering::Relaxed),
stats.seq_gap_max.load(Ordering::Relaxed),
stats.seq_reorder_events.load(Ordering::Relaxed),
stats.ts_jump_events.load(Ordering::Relaxed),
stats.ts_jump_ms_total.load(Ordering::Relaxed),
stats.ts_jump_ms_max.load(Ordering::Relaxed),
stats.stream_switch_events.load(Ordering::Relaxed)
);
file.write_all(rtp_summary.as_bytes()).await?;
file.flush().await?;
info!("[CSV] Final summary written to: {}", path.display());
Ok(())
}