1use std::sync::{Arc, Mutex};
2
3use super::{
4 counter::{MetricLabel, PaddedCounter, PaddedCounterFamily},
5 labels::{
6 RtcDatagramDropReason, RtcDatagramRoutePath, RtcKeyframeRequestOutcome, RtcNackDirection,
7 RtcOutputBudgetLimit, RtcRelayEnqueueResult, RtcRemoteControlDropKind,
8 RtcRemotePacketGateConvergence, RtcRouteControlOutcome,
9 },
10};
11
12const RTC_DATAGRAM_ROUTE_PATH_COUNT: usize = <RtcDatagramRoutePath as MetricLabel>::COUNT;
13const RTC_DATAGRAM_DROP_REASON_COUNT: usize = <RtcDatagramDropReason as MetricLabel>::COUNT;
14const RTC_NACK_DIRECTION_COUNT: usize = <RtcNackDirection as MetricLabel>::COUNT;
15const RTC_OUTPUT_BUDGET_LIMIT_COUNT: usize = <RtcOutputBudgetLimit as MetricLabel>::COUNT;
16const RTC_ROUTE_CONTROL_OUTCOME_COUNT: usize = <RtcRouteControlOutcome as MetricLabel>::COUNT;
17const RTC_KEYFRAME_REQUEST_OUTCOME_COUNT: usize = <RtcKeyframeRequestOutcome as MetricLabel>::COUNT;
18const RTC_RELAY_ENQUEUE_RESULT_COUNT: usize = <RtcRelayEnqueueResult as MetricLabel>::COUNT;
19const RTC_REMOTE_CONTROL_DROP_KIND_COUNT: usize = <RtcRemoteControlDropKind as MetricLabel>::COUNT;
20const RTC_REMOTE_PACKET_GATE_CONVERGENCE_COUNT: usize =
21 <RtcRemotePacketGateConvergence as MetricLabel>::COUNT;
22
23#[derive(Debug, Default)]
29pub struct RtcMetricsRecorder {
30 datagram_routes: PaddedCounterFamily<RtcDatagramRoutePath>,
31 datagram_drops: PaddedCounterFamily<RtcDatagramDropReason>,
32 datagram_fallback_scans: PaddedCounter,
33 datagram_scan_users: PaddedCounter,
34 nacks: PaddedCounterFamily<RtcNackDirection>,
35 rtx_packets_received_from_publisher: PaddedCounter,
36 rtx_payload_bytes_received_from_publisher: PaddedCounter,
37 rtcp_ingress_budget_drops: PaddedCounter,
38 output_budget_exhaustions: PaddedCounterFamily<RtcOutputBudgetLimit>,
39 output_budget_session_closes: PaddedCounter,
40 route_control: PaddedCounterFamily<RtcRouteControlOutcome>,
41 keyframe_requests: PaddedCounterFamily<RtcKeyframeRequestOutcome>,
42 relay_enqueues: PaddedCounterFamily<RtcRelayEnqueueResult>,
43 relay_mailbox_depth_samples: PaddedCounter,
44 relay_mailbox_depth_total: PaddedCounter,
45 relay_drain_batches: PaddedCounter,
46 relay_drained_packets: PaddedCounter,
47 relay_drain_cap_hits: PaddedCounter,
48 remote_control_drops: PaddedCounterFamily<RtcRemoteControlDropKind>,
49 remote_packet_gate_convergence: PaddedCounterFamily<RtcRemotePacketGateConvergence>,
50}
51
52impl RtcMetricsRecorder {
53 pub fn record_rtc_datagram_route(&self, path: RtcDatagramRoutePath) {
54 self.datagram_routes.increment(path);
55 }
56
57 pub fn record_rtc_datagram_drop(&self, reason: RtcDatagramDropReason) {
58 self.datagram_drops.increment(reason);
59 }
60
61 pub fn record_rtc_datagram_fallback_scan(&self, examined_sessions: usize) {
62 self.datagram_fallback_scans.increment();
63 self.datagram_scan_users.add(examined_sessions);
64 }
65
66 pub fn record_rtc_nacks(&self, direction: RtcNackDirection, count: u64) {
67 self.nacks.add_u64(direction, count);
68 }
69
70 pub fn record_rtc_rtx_received_from_publisher(&self, payload_bytes: usize) {
71 self.rtx_packets_received_from_publisher.increment();
72 self.rtx_payload_bytes_received_from_publisher
73 .add(payload_bytes);
74 }
75
76 pub fn record_rtc_rtcp_ingress_budget_drop(&self) {
77 self.rtcp_ingress_budget_drops.increment();
78 }
79
80 pub fn record_rtc_output_budget_exhaustion(&self, limit: RtcOutputBudgetLimit) {
81 self.output_budget_exhaustions.increment(limit);
82 }
83
84 pub fn record_rtc_output_budget_session_close(&self) {
85 self.output_budget_session_closes.increment();
86 }
87
88 pub fn record_rtc_route_control(&self, outcome: RtcRouteControlOutcome) {
89 self.route_control.increment(outcome);
90 }
91
92 pub fn record_rtc_keyframe_request(&self, outcome: RtcKeyframeRequestOutcome) {
93 self.keyframe_requests.increment(outcome);
94 }
95
96 pub fn record_rtc_relay_enqueue(&self, result: RtcRelayEnqueueResult) {
97 self.relay_enqueues.increment(result);
98 }
99
100 pub fn record_rtc_relay_mailbox_depth(&self, depth: usize) {
101 self.relay_mailbox_depth_samples.increment();
102 self.relay_mailbox_depth_total.add(depth);
103 }
104
105 pub fn record_rtc_relay_drain_batch(&self, drained_packets: usize, cap_hit: bool) {
106 if drained_packets == 0 {
107 return;
108 }
109 self.relay_drain_batches.increment();
110 self.relay_drained_packets.add(drained_packets);
111 if cap_hit {
112 self.relay_drain_cap_hits.increment();
113 }
114 }
115
116 pub fn record_rtc_remote_control_drop(&self, kind: RtcRemoteControlDropKind) {
117 self.remote_control_drops.increment(kind);
118 }
119
120 pub fn record_rtc_remote_packet_gate_convergence(
121 &self,
122 outcome: RtcRemotePacketGateConvergence,
123 ) {
124 self.remote_packet_gate_convergence.increment(outcome);
125 }
126}
127
128#[derive(Debug, Default)]
129pub(super) struct RtcMetrics {
130 worker_recorders: Mutex<Vec<Arc<RtcMetricsRecorder>>>,
131}
132
133impl RtcMetrics {
134 pub(super) fn register_worker(&self) -> Arc<RtcMetricsRecorder> {
135 let recorder = Arc::new(RtcMetricsRecorder::default());
136 {
137 let mut workers = match self.worker_recorders.lock() {
138 Ok(workers) => workers,
139 Err(poisoned) => poisoned.into_inner(),
140 };
141 workers.push(Arc::clone(&recorder));
142 }
143 recorder
144 }
145
146 pub(super) fn snapshot(&self) -> RtcMetricsSnapshot {
147 let mut snapshot = RtcMetricsSnapshot::default();
148 {
149 let workers = match self.worker_recorders.lock() {
150 Ok(workers) => workers,
151 Err(poisoned) => poisoned.into_inner(),
152 };
153 for recorder in workers.iter() {
154 snapshot.add_recorder(recorder);
155 }
156 }
157 snapshot
158 }
159}
160
161#[derive(Debug, Default)]
162pub(super) struct RtcMetricsSnapshot {
163 datagram_routes: [u64; RTC_DATAGRAM_ROUTE_PATH_COUNT],
164 datagram_drops: [u64; RTC_DATAGRAM_DROP_REASON_COUNT],
165 datagram_fallback_scans: u64,
166 datagram_scan_users: u64,
167 nacks: [u64; RTC_NACK_DIRECTION_COUNT],
168 rtx_packets_received_from_publisher: u64,
169 rtx_payload_bytes_received_from_publisher: u64,
170 rtcp_ingress_budget_drops: u64,
171 output_budget_exhaustions: [u64; RTC_OUTPUT_BUDGET_LIMIT_COUNT],
172 output_budget_session_closes: u64,
173 route_control: [u64; RTC_ROUTE_CONTROL_OUTCOME_COUNT],
174 keyframe_requests: [u64; RTC_KEYFRAME_REQUEST_OUTCOME_COUNT],
175 relay_enqueues: [u64; RTC_RELAY_ENQUEUE_RESULT_COUNT],
176 relay_mailbox_depth_samples: u64,
177 relay_mailbox_depth_total: u64,
178 relay_drain_batches: u64,
179 relay_drained_packets: u64,
180 relay_drain_cap_hits: u64,
181 remote_control_drops: [u64; RTC_REMOTE_CONTROL_DROP_KIND_COUNT],
182 remote_packet_gate_convergence: [u64; RTC_REMOTE_PACKET_GATE_CONVERGENCE_COUNT],
183}
184
185impl RtcMetricsSnapshot {
186 pub(super) fn datagram_routes(&self, path: RtcDatagramRoutePath) -> u64 {
187 self.datagram_routes
188 .get(path.as_index())
189 .copied()
190 .unwrap_or(0)
191 }
192
193 pub(super) fn datagram_drops(&self, reason: RtcDatagramDropReason) -> u64 {
194 self.datagram_drops
195 .get(reason.as_index())
196 .copied()
197 .unwrap_or(0)
198 }
199
200 pub(super) const fn datagram_fallback_scans(&self) -> u64 {
201 self.datagram_fallback_scans
202 }
203
204 pub(super) const fn datagram_scan_users(&self) -> u64 {
205 self.datagram_scan_users
206 }
207
208 pub(super) fn nacks(&self, direction: RtcNackDirection) -> u64 {
209 self.nacks.get(direction.as_index()).copied().unwrap_or(0)
210 }
211
212 pub(super) const fn rtx_packets_received_from_publisher(&self) -> u64 {
213 self.rtx_packets_received_from_publisher
214 }
215
216 pub(super) const fn rtx_payload_bytes_received_from_publisher(&self) -> u64 {
217 self.rtx_payload_bytes_received_from_publisher
218 }
219
220 pub(super) const fn rtcp_ingress_budget_drops(&self) -> u64 {
221 self.rtcp_ingress_budget_drops
222 }
223
224 pub(super) fn output_budget_exhaustions(&self, limit: RtcOutputBudgetLimit) -> u64 {
225 self.output_budget_exhaustions
226 .get(limit.as_index())
227 .copied()
228 .unwrap_or(0)
229 }
230
231 pub(super) const fn output_budget_session_closes(&self) -> u64 {
232 self.output_budget_session_closes
233 }
234
235 pub(super) fn route_control(&self, outcome: RtcRouteControlOutcome) -> u64 {
236 self.route_control
237 .get(outcome.as_index())
238 .copied()
239 .unwrap_or(0)
240 }
241
242 pub(super) fn keyframe_requests(&self, outcome: RtcKeyframeRequestOutcome) -> u64 {
243 self.keyframe_requests
244 .get(outcome.as_index())
245 .copied()
246 .unwrap_or(0)
247 }
248
249 pub(super) fn relay_enqueues(&self, result: RtcRelayEnqueueResult) -> u64 {
250 self.relay_enqueues
251 .get(result.as_index())
252 .copied()
253 .unwrap_or(0)
254 }
255
256 pub(super) const fn relay_mailbox_depth_samples(&self) -> u64 {
257 self.relay_mailbox_depth_samples
258 }
259
260 pub(super) const fn relay_mailbox_depth_total(&self) -> u64 {
261 self.relay_mailbox_depth_total
262 }
263
264 pub(super) const fn relay_drain_batches(&self) -> u64 {
265 self.relay_drain_batches
266 }
267
268 pub(super) const fn relay_drained_packets(&self) -> u64 {
269 self.relay_drained_packets
270 }
271
272 pub(super) const fn relay_drain_cap_hits(&self) -> u64 {
273 self.relay_drain_cap_hits
274 }
275
276 pub(super) fn remote_control_drops(&self, kind: RtcRemoteControlDropKind) -> u64 {
277 self.remote_control_drops
278 .get(kind.as_index())
279 .copied()
280 .unwrap_or(0)
281 }
282
283 pub(super) fn remote_packet_gate_convergence(
284 &self,
285 outcome: RtcRemotePacketGateConvergence,
286 ) -> u64 {
287 self.remote_packet_gate_convergence
288 .get(outcome.as_index())
289 .copied()
290 .unwrap_or(0)
291 }
292
293 fn add_recorder(&mut self, recorder: &RtcMetricsRecorder) {
294 for path in <RtcDatagramRoutePath as MetricLabel>::VARIANTS {
295 self.add_datagram_route(*path, recorder.datagram_routes.load(*path));
296 }
297 for reason in <RtcDatagramDropReason as MetricLabel>::VARIANTS {
298 self.add_datagram_drop(*reason, recorder.datagram_drops.load(*reason));
299 }
300 self.datagram_fallback_scans = self
301 .datagram_fallback_scans
302 .saturating_add(recorder.datagram_fallback_scans.load());
303 self.datagram_scan_users = self
304 .datagram_scan_users
305 .saturating_add(recorder.datagram_scan_users.load());
306 for direction in <RtcNackDirection as MetricLabel>::VARIANTS {
307 self.add_nack(*direction, recorder.nacks.load(*direction));
308 }
309 self.rtx_packets_received_from_publisher = self
310 .rtx_packets_received_from_publisher
311 .saturating_add(recorder.rtx_packets_received_from_publisher.load());
312 self.rtx_payload_bytes_received_from_publisher = self
313 .rtx_payload_bytes_received_from_publisher
314 .saturating_add(recorder.rtx_payload_bytes_received_from_publisher.load());
315 self.rtcp_ingress_budget_drops = self
316 .rtcp_ingress_budget_drops
317 .saturating_add(recorder.rtcp_ingress_budget_drops.load());
318 for limit in <RtcOutputBudgetLimit as MetricLabel>::VARIANTS {
319 self.add_output_budget_exhaustion(
320 *limit,
321 recorder.output_budget_exhaustions.load(*limit),
322 );
323 }
324 self.output_budget_session_closes = self
325 .output_budget_session_closes
326 .saturating_add(recorder.output_budget_session_closes.load());
327 for outcome in <RtcRouteControlOutcome as MetricLabel>::VARIANTS {
328 self.add_route_control(*outcome, recorder.route_control.load(*outcome));
329 }
330 for outcome in <RtcKeyframeRequestOutcome as MetricLabel>::VARIANTS {
331 self.add_keyframe_request(*outcome, recorder.keyframe_requests.load(*outcome));
332 }
333 for result in <RtcRelayEnqueueResult as MetricLabel>::VARIANTS {
334 self.add_relay_enqueue(*result, recorder.relay_enqueues.load(*result));
335 }
336 self.relay_mailbox_depth_samples = self
337 .relay_mailbox_depth_samples
338 .saturating_add(recorder.relay_mailbox_depth_samples.load());
339 self.relay_mailbox_depth_total = self
340 .relay_mailbox_depth_total
341 .saturating_add(recorder.relay_mailbox_depth_total.load());
342 self.relay_drain_batches = self
343 .relay_drain_batches
344 .saturating_add(recorder.relay_drain_batches.load());
345 self.relay_drained_packets = self
346 .relay_drained_packets
347 .saturating_add(recorder.relay_drained_packets.load());
348 self.relay_drain_cap_hits = self
349 .relay_drain_cap_hits
350 .saturating_add(recorder.relay_drain_cap_hits.load());
351 for kind in <RtcRemoteControlDropKind as MetricLabel>::VARIANTS {
352 self.add_remote_control_drop(*kind, recorder.remote_control_drops.load(*kind));
353 }
354 for outcome in <RtcRemotePacketGateConvergence as MetricLabel>::VARIANTS {
355 self.add_remote_packet_gate_convergence(
356 *outcome,
357 recorder.remote_packet_gate_convergence.load(*outcome),
358 );
359 }
360 }
361
362 fn add_datagram_route(&mut self, path: RtcDatagramRoutePath, count: u64) {
363 if let Some(counter) = self.datagram_routes.get_mut(path.as_index()) {
364 *counter = counter.saturating_add(count);
365 }
366 }
367
368 fn add_datagram_drop(&mut self, reason: RtcDatagramDropReason, count: u64) {
369 if let Some(counter) = self.datagram_drops.get_mut(reason.as_index()) {
370 *counter = counter.saturating_add(count);
371 }
372 }
373
374 fn add_nack(&mut self, direction: RtcNackDirection, count: u64) {
375 if let Some(counter) = self.nacks.get_mut(direction.as_index()) {
376 *counter = counter.saturating_add(count);
377 }
378 }
379
380 fn add_output_budget_exhaustion(&mut self, limit: RtcOutputBudgetLimit, count: u64) {
381 if let Some(counter) = self.output_budget_exhaustions.get_mut(limit.as_index()) {
382 *counter = counter.saturating_add(count);
383 }
384 }
385
386 fn add_route_control(&mut self, outcome: RtcRouteControlOutcome, count: u64) {
387 if let Some(counter) = self.route_control.get_mut(outcome.as_index()) {
388 *counter = counter.saturating_add(count);
389 }
390 }
391
392 fn add_keyframe_request(&mut self, outcome: RtcKeyframeRequestOutcome, count: u64) {
393 if let Some(counter) = self.keyframe_requests.get_mut(outcome.as_index()) {
394 *counter = counter.saturating_add(count);
395 }
396 }
397
398 fn add_relay_enqueue(&mut self, result: RtcRelayEnqueueResult, count: u64) {
399 if let Some(counter) = self.relay_enqueues.get_mut(result.as_index()) {
400 *counter = counter.saturating_add(count);
401 }
402 }
403
404 fn add_remote_control_drop(&mut self, kind: RtcRemoteControlDropKind, count: u64) {
405 if let Some(counter) = self.remote_control_drops.get_mut(kind.as_index()) {
406 *counter = counter.saturating_add(count);
407 }
408 }
409
410 fn add_remote_packet_gate_convergence(
411 &mut self,
412 outcome: RtcRemotePacketGateConvergence,
413 count: u64,
414 ) {
415 if let Some(counter) = self
416 .remote_packet_gate_convergence
417 .get_mut(outcome.as_index())
418 {
419 *counter = counter.saturating_add(count);
420 }
421 }
422}