o_sfu_telemetry/metrics/
rtp.rs

1use std::{
2    collections::BTreeMap,
3    sync::{Arc, Mutex},
4};
5
6use super::{
7    counter::{MetricLabel, PaddedCounterFamily},
8    labels::{RtpDecoderRefreshScope, RtpFlowDirection, RtpForwardDestinationKind},
9};
10
11const RTP_FLOW_DIRECTION_COUNT: usize = <RtpFlowDirection as MetricLabel>::COUNT;
12const RTP_FORWARD_DESTINATION_COUNT: usize = <RtpForwardDestinationKind as MetricLabel>::COUNT;
13const RTP_DECODER_REFRESH_SCOPE_COUNT: usize = <RtpDecoderRefreshScope as MetricLabel>::COUNT;
14
15/// Worker-local RTP packet metric recorder.
16///
17/// Packet loops keep one recorder for their full worker lifetime. Updates touch
18/// only this worker's padded atomics while `RuntimeMetrics` aggregates all
19/// registered recorders during scrape capture.
20#[derive(Debug, Default)]
21pub struct RtpMetricsRecorder {
22    packets: PaddedCounterFamily<RtpFlowDirection>,
23    payload_bytes: PaddedCounterFamily<RtpFlowDirection>,
24    forwarded_packets: PaddedCounterFamily<RtpForwardDestinationKind>,
25    forwarded_payload_bytes: PaddedCounterFamily<RtpForwardDestinationKind>,
26    decoder_refreshes: PaddedCounterFamily<RtpDecoderRefreshScope>,
27}
28
29impl RtpMetricsRecorder {
30    pub fn record_ingress(&self, payload_bytes: usize) {
31        self.packets.increment(RtpFlowDirection::Ingress);
32        self.payload_bytes
33            .add(RtpFlowDirection::Ingress, payload_bytes);
34    }
35
36    pub fn record_egress(&self, payload_bytes: usize) {
37        self.packets.increment(RtpFlowDirection::Egress);
38        self.payload_bytes
39            .add(RtpFlowDirection::Egress, payload_bytes);
40    }
41
42    pub fn record_forwarded(&self, destination: RtpForwardDestinationKind, payload_bytes: usize) {
43        self.forwarded_packets.increment(destination);
44        self.forwarded_payload_bytes.add(destination, payload_bytes);
45    }
46
47    pub fn record_decoder_refresh(&self, scope: RtpDecoderRefreshScope) {
48        self.decoder_refreshes.increment(scope);
49    }
50}
51
52#[derive(Debug, Default)]
53pub(super) struct RtpMetrics {
54    worker_recorders: Mutex<Vec<RtpWorkerMetricsRecorder>>,
55}
56
57impl RtpMetrics {
58    pub(super) fn register_worker(
59        &self,
60        media_worker_id: Option<usize>,
61    ) -> Arc<RtpMetricsRecorder> {
62        let recorder = Arc::new(RtpMetricsRecorder::default());
63        {
64            let mut workers = match self.worker_recorders.lock() {
65                Ok(workers) => workers,
66                Err(poisoned) => poisoned.into_inner(),
67            };
68            workers.push(RtpWorkerMetricsRecorder {
69                media_worker_id,
70                recorder: Arc::clone(&recorder),
71            });
72        }
73        recorder
74    }
75
76    pub(super) fn snapshot(&self) -> RtpMetricsSnapshot {
77        let mut snapshot = RtpMetricsSnapshot::default();
78        {
79            let workers = match self.worker_recorders.lock() {
80                Ok(workers) => workers,
81                Err(poisoned) => poisoned.into_inner(),
82            };
83            let mut worker_snapshots = BTreeMap::<usize, RtpWorkerMetricsSnapshot>::new();
84            for worker in workers.iter() {
85                snapshot.add_recorder(&worker.recorder);
86                if let Some(media_worker_id) = worker.media_worker_id {
87                    worker_snapshots
88                        .entry(media_worker_id)
89                        .or_insert_with(|| RtpWorkerMetricsSnapshot::new(media_worker_id))
90                        .add_recorder(&worker.recorder);
91                }
92            }
93            drop(workers);
94            snapshot.worker_snapshots = worker_snapshots.into_values().collect();
95        }
96        snapshot
97    }
98}
99
100#[derive(Debug)]
101struct RtpWorkerMetricsRecorder {
102    media_worker_id: Option<usize>,
103    recorder: Arc<RtpMetricsRecorder>,
104}
105
106#[derive(Debug, Default)]
107pub(super) struct RtpMetricsSnapshot {
108    packets: [u64; RTP_FLOW_DIRECTION_COUNT],
109    payload_bytes: [u64; RTP_FLOW_DIRECTION_COUNT],
110    forwarded_packets: [u64; RTP_FORWARD_DESTINATION_COUNT],
111    forwarded_payload_bytes: [u64; RTP_FORWARD_DESTINATION_COUNT],
112    decoder_refreshes: [u64; RTP_DECODER_REFRESH_SCOPE_COUNT],
113    worker_snapshots: Vec<RtpWorkerMetricsSnapshot>,
114}
115
116impl RtpMetricsSnapshot {
117    pub(super) fn packets(&self, direction: RtpFlowDirection) -> u64 {
118        self.packets.get(direction.as_index()).copied().unwrap_or(0)
119    }
120
121    pub(super) fn payload_bytes(&self, direction: RtpFlowDirection) -> u64 {
122        self.payload_bytes
123            .get(direction.as_index())
124            .copied()
125            .unwrap_or(0)
126    }
127
128    pub(super) fn forwarded_packets(&self, destination: RtpForwardDestinationKind) -> u64 {
129        self.forwarded_packets
130            .get(destination.as_index())
131            .copied()
132            .unwrap_or(0)
133    }
134
135    pub(super) fn forwarded_payload_bytes(&self, destination: RtpForwardDestinationKind) -> u64 {
136        self.forwarded_payload_bytes
137            .get(destination.as_index())
138            .copied()
139            .unwrap_or(0)
140    }
141
142    pub(super) fn decoder_refreshes(&self, scope: RtpDecoderRefreshScope) -> u64 {
143        self.decoder_refreshes
144            .get(scope.as_index())
145            .copied()
146            .unwrap_or(0)
147    }
148
149    pub(super) fn worker_snapshots(&self) -> &[RtpWorkerMetricsSnapshot] {
150        &self.worker_snapshots
151    }
152
153    fn add_recorder(&mut self, recorder: &RtpMetricsRecorder) {
154        for direction in <RtpFlowDirection as MetricLabel>::VARIANTS {
155            self.add_flow(
156                *direction,
157                recorder.packets.load(*direction),
158                recorder.payload_bytes.load(*direction),
159            );
160        }
161        for destination in <RtpForwardDestinationKind as MetricLabel>::VARIANTS {
162            self.add_forwarded(
163                *destination,
164                recorder.forwarded_packets.load(*destination),
165                recorder.forwarded_payload_bytes.load(*destination),
166            );
167        }
168        for scope in <RtpDecoderRefreshScope as MetricLabel>::VARIANTS {
169            self.add_decoder_refresh(*scope, recorder.decoder_refreshes.load(*scope));
170        }
171    }
172
173    fn add_flow(&mut self, direction: RtpFlowDirection, packets: u64, payload_bytes: u64) {
174        if let Some(counter) = self.packets.get_mut(direction.as_index()) {
175            *counter = counter.saturating_add(packets);
176        }
177        if let Some(counter) = self.payload_bytes.get_mut(direction.as_index()) {
178            *counter = counter.saturating_add(payload_bytes);
179        }
180    }
181
182    fn add_forwarded(
183        &mut self,
184        destination: RtpForwardDestinationKind,
185        packets: u64,
186        payload_bytes: u64,
187    ) {
188        if let Some(counter) = self.forwarded_packets.get_mut(destination.as_index()) {
189            *counter = counter.saturating_add(packets);
190        }
191        if let Some(counter) = self.forwarded_payload_bytes.get_mut(destination.as_index()) {
192            *counter = counter.saturating_add(payload_bytes);
193        }
194    }
195
196    fn add_decoder_refresh(&mut self, scope: RtpDecoderRefreshScope, refreshes: u64) {
197        if let Some(counter) = self.decoder_refreshes.get_mut(scope.as_index()) {
198            *counter = counter.saturating_add(refreshes);
199        }
200    }
201}
202
203#[derive(Debug, Default)]
204pub(super) struct RtpWorkerMetricsSnapshot {
205    media_worker_id: usize,
206    packets: [u64; RTP_FLOW_DIRECTION_COUNT],
207    payload_bytes: [u64; RTP_FLOW_DIRECTION_COUNT],
208    forwarded_packets: [u64; RTP_FORWARD_DESTINATION_COUNT],
209    forwarded_payload_bytes: [u64; RTP_FORWARD_DESTINATION_COUNT],
210}
211
212impl RtpWorkerMetricsSnapshot {
213    fn new(media_worker_id: usize) -> Self {
214        Self {
215            media_worker_id,
216            ..Self::default()
217        }
218    }
219
220    pub(super) const fn media_worker_id(&self) -> usize {
221        self.media_worker_id
222    }
223
224    pub(super) fn packets(&self, direction: RtpFlowDirection) -> u64 {
225        self.packets.get(direction.as_index()).copied().unwrap_or(0)
226    }
227
228    pub(super) fn payload_bytes(&self, direction: RtpFlowDirection) -> u64 {
229        self.payload_bytes
230            .get(direction.as_index())
231            .copied()
232            .unwrap_or(0)
233    }
234
235    pub(super) fn forwarded_packets(&self, destination: RtpForwardDestinationKind) -> u64 {
236        self.forwarded_packets
237            .get(destination.as_index())
238            .copied()
239            .unwrap_or(0)
240    }
241
242    pub(super) fn forwarded_payload_bytes(&self, destination: RtpForwardDestinationKind) -> u64 {
243        self.forwarded_payload_bytes
244            .get(destination.as_index())
245            .copied()
246            .unwrap_or(0)
247    }
248
249    fn add_recorder(&mut self, recorder: &RtpMetricsRecorder) {
250        for direction in <RtpFlowDirection as MetricLabel>::VARIANTS {
251            self.add_flow(
252                *direction,
253                recorder.packets.load(*direction),
254                recorder.payload_bytes.load(*direction),
255            );
256        }
257        for destination in <RtpForwardDestinationKind as MetricLabel>::VARIANTS {
258            self.add_forwarded(
259                *destination,
260                recorder.forwarded_packets.load(*destination),
261                recorder.forwarded_payload_bytes.load(*destination),
262            );
263        }
264    }
265
266    fn add_flow(&mut self, direction: RtpFlowDirection, packets: u64, payload_bytes: u64) {
267        if let Some(counter) = self.packets.get_mut(direction.as_index()) {
268            *counter = counter.saturating_add(packets);
269        }
270        if let Some(counter) = self.payload_bytes.get_mut(direction.as_index()) {
271            *counter = counter.saturating_add(payload_bytes);
272        }
273    }
274
275    fn add_forwarded(
276        &mut self,
277        destination: RtpForwardDestinationKind,
278        packets: u64,
279        payload_bytes: u64,
280    ) {
281        if let Some(counter) = self.forwarded_packets.get_mut(destination.as_index()) {
282            *counter = counter.saturating_add(packets);
283        }
284        if let Some(counter) = self.forwarded_payload_bytes.get_mut(destination.as_index()) {
285            *counter = counter.saturating_add(payload_bytes);
286        }
287    }
288}
o_sfu_telemetry/metrics/rtp.rs

o_sfu_telemetry/metrics/
rtp.rs
