libdd_telemetry/
metrics.rs

1// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5    collections::HashMap,
6    sync::{Arc, Mutex, MutexGuard},
7};
8// `time::SystemTime::UNIX_EPOCH.elapsed()` panics on wasm32-unknown-unknown
9// (the host has no `std::time` backend); `web_time` proxies to `Date.now()`.
10use web_time as time;
11
12use libdd_common::tag::Tag;
13use libdd_ddsketch::DDSketch;
14use serde::{Deserialize, Serialize};
15
16use crate::data::{self, metrics};
17
18fn unix_timestamp_now() -> u64 {
19    time::SystemTime::UNIX_EPOCH
20        .elapsed()
21        .map_or(0, |d| d.as_secs())
22}
23
24#[derive(Debug)]
25struct MetricBucket {
26    aggr: MetricAggr,
27}
28
29#[derive(Debug)]
30enum MetricAggr {
31    Count { count: f64 },
32    Gauge { value: f64 },
33}
34
35impl MetricBucket {
36    fn add_point(&mut self, point: f64) {
37        match &mut self.aggr {
38            MetricAggr::Count { count } => *count += point,
39            MetricAggr::Gauge { value } => *value = point,
40        }
41    }
42
43    fn value(&self) -> f64 {
44        match self.aggr {
45            MetricAggr::Count { count } => count,
46            MetricAggr::Gauge { value } => value,
47        }
48    }
49}
50
51#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
52#[repr(C)]
53pub struct ContextKey(u32, metrics::MetricType);
54
55impl ContextKey {
56    /// The index into the metric-context store.
57    pub(crate) fn index(self) -> u32 {
58        self.0
59    }
60
61    /// The metric type this context was registered with.
62    pub(crate) fn metric_type(self) -> metrics::MetricType {
63        self.1
64    }
65
66    /// Reconstruct a key from its parts (used by the metric ring buffer, which encodes the key
67    /// into an atomic slot). Only valid for indices/types produced by [`MetricContexts`].
68    pub(crate) fn from_parts(index: u32, metric_type: metrics::MetricType) -> Self {
69        ContextKey(index, metric_type)
70    }
71}
72
73#[derive(Debug, PartialEq, Eq, Hash)]
74struct BucketKey {
75    context_key: ContextKey,
76    extra_tags: Vec<Tag>,
77}
78
79#[derive(Debug, Default)]
80pub struct MetricBuckets {
81    buckets: HashMap<BucketKey, MetricBucket>,
82    series: HashMap<BucketKey, Vec<(u64, f64)>>,
83    distributions: HashMap<BucketKey, DDSketch>,
84}
85
86#[derive(Debug, Default, Serialize, Deserialize)]
87pub struct MetricBucketStats {
88    pub buckets: u32,
89    pub series: u32,
90    pub series_points: u32,
91    pub distributions: u32,
92    pub distributions_points: u32,
93}
94
95impl MetricBuckets {
96    pub const METRICS_FLUSH_INTERVAL: time::Duration = time::Duration::from_secs(10);
97
98    pub fn flush_aggregates(&mut self) {
99        let timestamp = unix_timestamp_now();
100        for (key, bucket) in self.buckets.drain() {
101            self.series
102                .entry(key)
103                .or_default()
104                .push((timestamp, bucket.value()))
105        }
106    }
107
108    pub fn flush_series(
109        &mut self,
110    ) -> impl Iterator<Item = (ContextKey, Vec<Tag>, Vec<(u64, f64)>)> + '_ {
111        self.series.drain().map(
112            |(
113                BucketKey {
114                    context_key,
115                    extra_tags,
116                },
117                points,
118            )| (context_key, extra_tags, points),
119        )
120    }
121
122    pub fn flush_distributions(
123        &mut self,
124    ) -> impl Iterator<Item = (ContextKey, Vec<Tag>, DDSketch)> + '_ {
125        self.distributions.drain().map(
126            |(
127                BucketKey {
128                    context_key,
129                    extra_tags,
130                },
131                points,
132            )| (context_key, extra_tags, points),
133        )
134    }
135
136    pub fn add_point(&mut self, context_key: ContextKey, point: f64, extra_tags: Vec<Tag>) {
137        let bucket_key = BucketKey {
138            context_key,
139            extra_tags,
140        };
141        match context_key.1 {
142            metrics::MetricType::Count | metrics::MetricType::Rate => self
143                .buckets
144                .entry(bucket_key)
145                .or_insert_with(|| MetricBucket {
146                    aggr: MetricAggr::Count { count: 0.0 },
147                })
148                .add_point(point),
149            metrics::MetricType::Gauge => self
150                .buckets
151                .entry(bucket_key)
152                .or_insert_with(|| MetricBucket {
153                    aggr: MetricAggr::Gauge { value: 0.0 },
154                })
155                .add_point(point),
156            metrics::MetricType::Distribution => {
157                let _ = self.distributions.entry(bucket_key).or_default().add(point);
158            }
159        }
160    }
161
162    pub fn stats(&self) -> MetricBucketStats {
163        MetricBucketStats {
164            buckets: self.buckets.len() as u32,
165            series: self.series.len() as u32,
166            series_points: self.series.values().map(|v| v.len() as u32).sum(),
167            distributions: self.distributions.len() as u32,
168            distributions_points: self
169                .distributions
170                .values()
171                .flat_map(|sketch| {
172                    sketch
173                        .ordered_bins()
174                        .into_iter()
175                        .map(|(_, weight)| weight as u32)
176                })
177                .sum(),
178        }
179    }
180}
181
182#[derive(Clone, Debug, Serialize, Deserialize)]
183pub struct MetricContext {
184    pub namespace: data::metrics::MetricNamespace,
185    pub name: String,
186    pub tags: Vec<Tag>,
187    pub metric_type: data::metrics::MetricType,
188    pub common: bool,
189}
190
191pub struct MetricContextGuard<'a> {
192    guard: MutexGuard<'a, InnerMetricContexts>,
193}
194
195impl MetricContextGuard<'_> {
196    pub fn read(&self, key: ContextKey) -> Option<&MetricContext> {
197        self.guard.store.get(key.0 as usize)
198    }
199
200    pub fn is_empty(&self) -> bool {
201        self.guard.store.is_empty()
202    }
203
204    pub fn len(&self) -> usize {
205        self.guard.store.len()
206    }
207}
208
209#[derive(Debug, Default)]
210struct InnerMetricContexts {
211    store: Vec<MetricContext>,
212}
213
214#[derive(Debug, Clone, Default)]
215pub struct MetricContexts {
216    inner: Arc<Mutex<InnerMetricContexts>>,
217}
218
219impl MetricContexts {
220    pub fn register_metric_context(
221        &self,
222        name: String,
223        tags: Vec<Tag>,
224        metric_type: data::metrics::MetricType,
225        common: bool,
226        namespace: data::metrics::MetricNamespace,
227    ) -> ContextKey {
228        #[allow(clippy::unwrap_used)]
229        let mut contexts = self.inner.lock().unwrap();
230        let key = ContextKey(contexts.store.len() as u32, metric_type);
231        contexts.store.push(MetricContext {
232            name,
233            tags,
234            metric_type,
235            common,
236            namespace,
237        });
238        key
239    }
240
241    pub fn lock(&self) -> MetricContextGuard<'_> {
242        #[allow(clippy::unwrap_used)]
243        MetricContextGuard {
244            guard: self.inner.as_ref().lock().unwrap(),
245        }
246    }
247}
248
249#[cfg(test)]
250mod tests {
251    use libdd_common::tag;
252    use std::fmt::Debug;
253
254    use super::*;
255    use crate::data::metrics::{MetricNamespace, MetricType};
256
257    /// Check if a and b are approximately equal with the given precision or 1.0e-6 by default
258    macro_rules! assert_approx_eq {
259        ($a:expr, $b:expr) => {{
260            let (a, b) = (&$a, &$b);
261            assert!(
262                (*a - *b).abs() < 1.0e-6,
263                "{} is not approximately equal to {}",
264                *a,
265                *b
266            );
267        }};
268        ($a:expr, $b:expr, $precision:expr) => {{
269            let (a, b) = (&$a, &$b);
270            assert!(
271                (*a - *b).abs() < $precision,
272                "{} is not approximately equal to {}",
273                *a,
274                *b
275            );
276        }};
277    }
278
279    // Test util used to run assertions against an unsorted list
280    fn check_iter<'a, U: 'a + Debug, T: Iterator<Item = &'a U>>(
281        elements: T,
282        assertions: &[&dyn Fn(&U) -> bool],
283    ) {
284        let mut used = vec![false; assertions.len()];
285        for e in elements {
286            let mut found = false;
287            for (i, &a) in assertions.iter().enumerate() {
288                if a(e) {
289                    if used[i] {
290                        panic!("Assertion {i} has been used multiple times");
291                    }
292                    found = true;
293                    used[i] = true;
294                    break;
295                }
296            }
297            if !found {
298                panic!("No assertion found for elem {e:?}")
299            }
300        }
301    }
302
303    #[test]
304    fn test_bucket_flushes() {
305        let mut buckets = MetricBuckets::default();
306        let contexts = MetricContexts::default();
307
308        let context_key_1 = contexts.register_metric_context(
309            "metric1".into(),
310            Vec::new(),
311            MetricType::Gauge,
312            false,
313            MetricNamespace::Tracers,
314        );
315        let context_key_2 = contexts.register_metric_context(
316            "metric2".into(),
317            Vec::new(),
318            MetricType::Gauge,
319            false,
320            MetricNamespace::Tracers,
321        );
322        let extra_tags = vec![tag!("service", "foobar")];
323
324        buckets.add_point(context_key_1, 0.1, Vec::new());
325        buckets.add_point(context_key_1, 0.2, Vec::new());
326        assert_eq!(buckets.buckets.len(), 1);
327
328        buckets.add_point(context_key_2, 0.3, Vec::new());
329        assert_eq!(buckets.buckets.len(), 2);
330
331        buckets.add_point(context_key_2, 0.4, extra_tags.clone());
332        assert_eq!(buckets.buckets.len(), 3);
333
334        buckets.flush_aggregates();
335        assert_eq!(buckets.buckets.len(), 0);
336        assert_eq!(buckets.series.len(), 3);
337
338        buckets.add_point(context_key_1, 0.5, Vec::new());
339        buckets.add_point(context_key_2, 0.6, extra_tags);
340        assert_eq!(buckets.buckets.len(), 2);
341
342        buckets.flush_aggregates();
343        assert_eq!(buckets.buckets.len(), 0);
344        assert_eq!(buckets.series.len(), 3);
345
346        let series: Vec<_> = buckets.flush_series().collect();
347        assert_eq!(buckets.buckets.len(), 0);
348        assert_eq!(buckets.series.len(), 0);
349        assert_eq!(series.len(), 3);
350
351        check_iter(
352            series.iter(),
353            &[
354                &|(c, t, points)| {
355                    if !(c == &context_key_1 && t.is_empty()) {
356                        return false;
357                    }
358                    assert_eq!(points.len(), 2);
359                    assert_approx_eq!(points[0].1, 0.2);
360                    assert_approx_eq!(points[1].1, 0.5);
361                    true
362                },
363                &|(c, t, points)| {
364                    if !(c == &context_key_2 && t.is_empty()) {
365                        return false;
366                    }
367                    assert_eq!(points.len(), 1);
368                    assert_approx_eq!(points[0].1, 0.3);
369                    true
370                },
371                &|(c, t, points)| {
372                    if !(c == &context_key_2 && !t.is_empty()) {
373                        return false;
374                    }
375                    assert_eq!(points.len(), 2);
376                    assert_approx_eq!(points[0].1, 0.4);
377                    assert_approx_eq!(points[1].1, 0.6);
378                    true
379                },
380            ],
381        );
382    }
383
384    #[test]
385    fn test_distributions() {
386        let mut buckets = MetricBuckets::default();
387        let contexts = MetricContexts::default();
388
389        let context_key_distribution = contexts.register_metric_context(
390            "metric_distribution".into(),
391            Vec::new(),
392            MetricType::Distribution,
393            false,
394            MetricNamespace::Tracers,
395        );
396        let context_key_distribution_2 = contexts.register_metric_context(
397            "metric_distribution_2".into(),
398            Vec::new(),
399            MetricType::Distribution,
400            false,
401            MetricNamespace::Tracers,
402        );
403        let extra_tags = vec![tag!("service", "foo")];
404
405        // Create 2 distributions with 2 and 3 points
406        buckets.add_point(context_key_distribution, 1.0, Vec::new());
407        buckets.add_point(context_key_distribution, 1.0, Vec::new());
408        buckets.add_point(context_key_distribution, 100.0, Vec::new());
409        buckets.add_point(context_key_distribution, 1000.0, Vec::new());
410
411        buckets.add_point(context_key_distribution_2, 2.0, Vec::new());
412        buckets.add_point(context_key_distribution_2, 200.0, Vec::new());
413
414        buckets.add_point(context_key_distribution_2, 3.0, extra_tags.clone());
415        buckets.add_point(context_key_distribution_2, 300.0, extra_tags.clone());
416
417        let distributions: Vec<_> = buckets.flush_distributions().collect();
418
419        check_iter(
420            distributions.iter(),
421            &[
422                &|(c, t, points)| {
423                    if !(c == &context_key_distribution && t.is_empty()) {
424                        return false;
425                    }
426                    let bins: Vec<_> = points
427                        .ordered_bins()
428                        .into_iter()
429                        .filter(|(_, w)| *w != 0.0)
430                        .collect();
431                    assert_eq!(bins.len(), 3);
432                    // The precision is quite low since it is up to the ddsketch implementation to
433                    // test the precision
434                    assert_approx_eq!(bins[0].0, 1.0, 1.0e-1);
435                    assert_approx_eq!(bins[0].1, 2.0);
436                    assert_approx_eq!(bins[1].0, 100.0, 1.0);
437                    assert_approx_eq!(bins[1].1, 1.0);
438                    assert_approx_eq!(bins[2].0, 1000.0, 10.0);
439                    assert_approx_eq!(bins[2].1, 1.0);
440                    true
441                },
442                &|(c, t, points)| {
443                    if !(c == &context_key_distribution_2 && t.is_empty()) {
444                        return false;
445                    }
446                    let bins: Vec<_> = points
447                        .ordered_bins()
448                        .into_iter()
449                        .filter(|(_, w)| *w != 0.0)
450                        .collect();
451                    assert_eq!(bins.len(), 2);
452                    assert_approx_eq!(bins[0].0, 2.0, 1.0e-1);
453                    assert_approx_eq!(bins[0].1, 1.0);
454                    assert_approx_eq!(bins[1].0, 200.0, 1.0);
455                    assert_approx_eq!(bins[1].1, 1.0);
456                    true
457                },
458                &|(c, t, points)| {
459                    if !(c == &context_key_distribution_2 && !t.is_empty()) {
460                        return false;
461                    }
462                    let bins: Vec<_> = points
463                        .ordered_bins()
464                        .into_iter()
465                        .filter(|(_, w)| *w != 0.0)
466                        .collect();
467                    assert_eq!(bins.len(), 2);
468                    assert_approx_eq!(bins[0].0, 3.0, 1.0e-1);
469                    assert_approx_eq!(bins[0].1, 1.0);
470                    assert_approx_eq!(bins[1].0, 300.0, 1.0);
471                    assert_approx_eq!(bins[1].1, 1.0);
472                    true
473                },
474            ],
475        )
476    }
477
478    #[test]
479    fn test_stats() {
480        let mut buckets = MetricBuckets::default();
481        let contexts = MetricContexts::default();
482
483        let context_key_1 = contexts.register_metric_context(
484            "metric1".into(),
485            Vec::new(),
486            MetricType::Count,
487            false,
488            MetricNamespace::Tracers,
489        );
490
491        let context_key_2 = contexts.register_metric_context(
492            "metric2".into(),
493            Vec::new(),
494            MetricType::Gauge,
495            false,
496            MetricNamespace::Tracers,
497        );
498
499        let context_key_distribution = contexts.register_metric_context(
500            "metric_distribution".into(),
501            Vec::new(),
502            MetricType::Distribution,
503            false,
504            MetricNamespace::Tracers,
505        );
506
507        let context_key_distribution_2 = contexts.register_metric_context(
508            "metric_distribution_2".into(),
509            Vec::new(),
510            MetricType::Distribution,
511            false,
512            MetricNamespace::Tracers,
513        );
514
515        // Create 2 series with 2 and 3 points
516        buckets.add_point(context_key_1, 1.0, Vec::new());
517        buckets.add_point(context_key_2, 2.0, Vec::new());
518        buckets.flush_aggregates();
519
520        buckets.add_point(context_key_1, 1.0, Vec::new());
521        buckets.add_point(context_key_2, 2.0, Vec::new());
522        buckets.flush_aggregates();
523
524        buckets.add_point(context_key_1, 1.1, Vec::new());
525        buckets.add_point(context_key_1, 2.1, Vec::new());
526        buckets.flush_aggregates();
527
528        // Create 2 buckets
529        buckets.add_point(context_key_1, 1.0, Vec::new());
530        buckets.add_point(context_key_2, 2.0, Vec::new());
531
532        // Create 2 distributions with 2 and 3 points
533        buckets.add_point(context_key_distribution, 1.0, Vec::new());
534        buckets.add_point(context_key_distribution, 1.1, Vec::new());
535        buckets.add_point(context_key_distribution, 1.2, Vec::new());
536
537        buckets.add_point(context_key_distribution_2, 2.0, Vec::new());
538        buckets.add_point(context_key_distribution_2, 2.1, Vec::new());
539
540        let stats = buckets.stats();
541
542        assert_eq!(stats.buckets, 2);
543        assert_eq!(stats.series, 2);
544        assert_eq!(stats.series_points, 5);
545        assert_eq!(stats.distributions, 2);
546        assert_eq!(stats.distributions_points, 5);
547    }
548}
libdd_telemetry/metrics.rs

libdd_telemetry/
metrics.rs
