o_sfu_telemetry/metrics/
rtp.rs

1use std::{
2    collections::BTreeMap,
3    sync::{Arc, Mutex},
4};
5
6use super::{
7    counter::{MetricLabel, PaddedCounter, PaddedCounterFamily},
8    labels::{RtpDecoderRefreshScope, RtpFlowDirection, RtpForwardDestinationKind},
9};
10
11const RTP_FORWARD_DESTINATION_COUNT: usize = <RtpForwardDestinationKind as MetricLabel>::COUNT;
12const RTP_DECODER_REFRESH_SCOPE_COUNT: usize = <RtpDecoderRefreshScope as MetricLabel>::COUNT;
13
14/// Worker-local RTP packet metric recorder.
15///
16/// Packet loops keep one recorder for their full worker lifetime. Updates touch
17/// only this worker's padded atomics while `RuntimeMetrics` aggregates all
18/// registered recorders during scrape capture.
19#[derive(Debug, Default)]
20pub struct RtpMetricsRecorder {
21    ingress_packets: PaddedCounter,
22    ingress_payload_bytes: PaddedCounter,
23    forwarded_packets: PaddedCounterFamily<RtpForwardDestinationKind>,
24    forwarded_payload_bytes: PaddedCounterFamily<RtpForwardDestinationKind>,
25    decoder_refreshes: PaddedCounterFamily<RtpDecoderRefreshScope>,
26}
27
28impl RtpMetricsRecorder {
29    pub fn record_ingress(&self, payload_bytes: usize) {
30        self.ingress_packets.increment();
31        self.ingress_payload_bytes.add(payload_bytes);
32    }
33
34    /// Records one RTP forwarding event supplied by the caller.
35    ///
36    /// `LocalRtc` means successful local RTC queuing and also supplies the
37    /// egress series. Pass the payload length returned by the local send.
38    /// Socket delivery is not counted.
39    /// Updates are additive and safe for concurrent callers.
40    pub fn record_forwarded(&self, destination: RtpForwardDestinationKind, payload_bytes: usize) {
41        self.forwarded_packets.increment(destination);
42        self.forwarded_payload_bytes.add(destination, payload_bytes);
43    }
44
45    pub fn record_decoder_refresh(&self, scope: RtpDecoderRefreshScope) {
46        self.decoder_refreshes.increment(scope);
47    }
48}
49
50#[derive(Debug, Default)]
51pub(super) struct RtpMetrics {
52    worker_recorders: Mutex<Vec<RtpWorkerMetricsRecorder>>,
53}
54
55impl RtpMetrics {
56    pub(super) fn register_worker(
57        &self,
58        media_worker_id: Option<usize>,
59    ) -> Arc<RtpMetricsRecorder> {
60        let recorder = Arc::new(RtpMetricsRecorder::default());
61        {
62            let mut workers = match self.worker_recorders.lock() {
63                Ok(workers) => workers,
64                Err(poisoned) => poisoned.into_inner(),
65            };
66            workers.push(RtpWorkerMetricsRecorder {
67                media_worker_id,
68                recorder: Arc::clone(&recorder),
69            });
70        }
71        recorder
72    }
73
74    pub(super) fn snapshot(&self) -> RtpMetricsSnapshot {
75        let mut snapshot = RtpMetricsSnapshot::default();
76        {
77            let workers = match self.worker_recorders.lock() {
78                Ok(workers) => workers,
79                Err(poisoned) => poisoned.into_inner(),
80            };
81            let mut worker_snapshots = BTreeMap::<usize, RtpWorkerMetricsSnapshot>::new();
82            for worker in workers.iter() {
83                snapshot.add_recorder(&worker.recorder);
84                if let Some(media_worker_id) = worker.media_worker_id {
85                    worker_snapshots
86                        .entry(media_worker_id)
87                        .or_insert_with(|| RtpWorkerMetricsSnapshot::new(media_worker_id))
88                        .traffic
89                        .add_recorder(&worker.recorder);
90                }
91            }
92            drop(workers);
93            snapshot.worker_snapshots = worker_snapshots.into_values().collect();
94        }
95        snapshot
96    }
97}
98
99#[derive(Debug)]
100struct RtpWorkerMetricsRecorder {
101    media_worker_id: Option<usize>,
102    recorder: Arc<RtpMetricsRecorder>,
103}
104
105#[derive(Debug, Default)]
106pub(super) struct RtpMetricsSnapshot {
107    pub(super) traffic: RtpTrafficSnapshot,
108    decoder_refreshes: [u64; RTP_DECODER_REFRESH_SCOPE_COUNT],
109    worker_snapshots: Vec<RtpWorkerMetricsSnapshot>,
110}
111
112impl RtpMetricsSnapshot {
113    pub(super) fn decoder_refreshes(&self, scope: RtpDecoderRefreshScope) -> u64 {
114        self.decoder_refreshes
115            .get(scope.as_index())
116            .copied()
117            .unwrap_or(0)
118    }
119
120    pub(super) fn worker_snapshots(&self) -> &[RtpWorkerMetricsSnapshot] {
121        &self.worker_snapshots
122    }
123
124    fn add_recorder(&mut self, recorder: &RtpMetricsRecorder) {
125        self.traffic.add_recorder(recorder);
126        recorder
127            .decoder_refreshes
128            .accumulate_into(&mut self.decoder_refreshes);
129    }
130}
131
132#[derive(Debug, Default)]
133pub(super) struct RtpWorkerMetricsSnapshot {
134    media_worker_id: usize,
135    pub(super) traffic: RtpTrafficSnapshot,
136}
137
138impl RtpWorkerMetricsSnapshot {
139    fn new(media_worker_id: usize) -> Self {
140        Self {
141            media_worker_id,
142            ..Self::default()
143        }
144    }
145
146    pub(super) const fn media_worker_id(&self) -> usize {
147        self.media_worker_id
148    }
149}
150
151#[derive(Debug, Default)]
152pub(super) struct RtpTrafficSnapshot {
153    ingress_packets: u64,
154    ingress_payload_bytes: u64,
155    forwarded_packets: [u64; RTP_FORWARD_DESTINATION_COUNT],
156    forwarded_payload_bytes: [u64; RTP_FORWARD_DESTINATION_COUNT],
157}
158
159impl RtpTrafficSnapshot {
160    pub(super) fn packets(&self, direction: RtpFlowDirection) -> u64 {
161        match direction {
162            RtpFlowDirection::Ingress => self.ingress_packets,
163            RtpFlowDirection::Egress => self.forwarded_packets(RtpForwardDestinationKind::LocalRtc),
164        }
165    }
166
167    pub(super) fn payload_bytes(&self, direction: RtpFlowDirection) -> u64 {
168        match direction {
169            RtpFlowDirection::Ingress => self.ingress_payload_bytes,
170            RtpFlowDirection::Egress => {
171                self.forwarded_payload_bytes(RtpForwardDestinationKind::LocalRtc)
172            }
173        }
174    }
175
176    pub(super) fn forwarded_packets(&self, destination: RtpForwardDestinationKind) -> u64 {
177        self.forwarded_packets
178            .get(destination.as_index())
179            .copied()
180            .unwrap_or(0)
181    }
182
183    pub(super) fn forwarded_payload_bytes(&self, destination: RtpForwardDestinationKind) -> u64 {
184        self.forwarded_payload_bytes
185            .get(destination.as_index())
186            .copied()
187            .unwrap_or(0)
188    }
189
190    fn add_recorder(&mut self, recorder: &RtpMetricsRecorder) {
191        self.ingress_packets = self
192            .ingress_packets
193            .saturating_add(recorder.ingress_packets.load());
194        self.ingress_payload_bytes = self
195            .ingress_payload_bytes
196            .saturating_add(recorder.ingress_payload_bytes.load());
197        recorder
198            .forwarded_packets
199            .accumulate_into(&mut self.forwarded_packets);
200        recorder
201            .forwarded_payload_bytes
202            .accumulate_into(&mut self.forwarded_payload_bytes);
203    }
204}
o_sfu_telemetry/metrics/rtp.rs

o_sfu_telemetry/metrics/
rtp.rs
