sipmon 0.1.5

Passive SIP/RTP signaling & media quality monitoring (pcap/live, TUI + JSONL export)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

use std::collections::HashMap;

use crate::model::stats::MetricSet;

/// Two-dimensional time×key aggregation for reliability/quality heatmap.
pub struct Heatmap {
    bucket_secs: u64,
    /// bucket_index_us -> key -> metrics.
    cells: HashMap<u64, HashMap<String, MetricSet>>,
}

impl Heatmap {
    pub fn new(bucket_secs: u64) -> Self {
        Self {
            bucket_secs,
            cells: HashMap::new(),
        }
    }

    pub fn bucket_secs(&self) -> u64 {
        self.bucket_secs
    }

    fn bucket_of(&self, ts_us: u64) -> u64 {
        let secs = ts_us / 1_000_000;
        let b = secs / self.bucket_secs.max(1);
        b * self.bucket_secs * 1_000_000
    }

    /// Record a terminated call's contributions into its (bucket, key) cell.
    #[allow(clippy::too_many_arguments)]
    pub fn record_call(
        &mut self,
        ts_us: u64,
        key: String,
        answered: bool,
        failed: bool,
        pdd_ms: Option<f64>,
        jitter_ms: Option<f64>,
        loss_pct: Option<f64>,
        rtt_ms: Option<f64>,
        mos: Option<f64>,
    ) {
        let bucket = self.bucket_of(ts_us);
        let cell = self
            .cells
            .entry(bucket)
            .or_default()
            .entry(key)
            .or_default();
        cell.calls += 1;
        if answered {
            cell.answered += 1;
        }
        if failed {
            cell.failed += 1;
        }
        if let Some(p) = pdd_ms {
            cell.pdd_sum_ms += p;
            cell.pdd_n += 1;
        }
        if let Some(j) = jitter_ms {
            cell.jitter_sum_ms += j;
            cell.jitter_n += 1;
        }
        if let Some(l) = loss_pct {
            cell.loss_sum_pct += l;
            cell.loss_n += 1;
        }
        if let Some(r) = rtt_ms {
            cell.rtt_sum_ms += r;
            cell.rtt_n += 1;
        }
        if let Some(m) = mos {
            cell.mos_sum += m;
            cell.mos_n += 1;
        }
    }

    #[allow(dead_code)]
    pub fn cells(&self) -> &HashMap<u64, HashMap<String, MetricSet>> {
        &self.cells
    }

    pub fn flat(&self) -> Vec<(u64, String, MetricSet)> {
        let mut out = Vec::new();
        for (b, m) in &self.cells {
            for (k, v) in m {
                out.push((*b, k.clone(), v.clone()));
            }
        }
        out.sort_by_key(|(b, _, _)| *b);
        out
    }

    /// Drop buckets whose window ended before `cutoff_us` (bounds memory on
    /// long-running sessions / many distinct keys).
    pub fn prune_older_than(&mut self, cutoff_us: u64) {
        self.cells.retain(|bucket, _| *bucket >= cutoff_us);
    }

    /// Number of retained cells (observable for tests).
    #[cfg(test)]
    pub fn cell_count(&self) -> usize {
        self.cells.values().map(|m| m.len()).sum()
    }
}