vector_core/event/metric/
mod.rs

1use std::convert::TryFrom;
2use std::{
3    convert::AsRef,
4    fmt::{self, Display, Formatter},
5    num::NonZeroU32,
6};
7
8use chrono::{DateTime, Utc};
9use vector_common::{
10    EventDataEq,
11    byte_size_of::ByteSizeOf,
12    internal_event::{OptionalTag, TaggedEventsSent},
13    json_size::JsonSize,
14    request_metadata::GetEventCountTags,
15};
16use vector_config::configurable_component;
17use vrl::compiler::value::VrlValueConvert;
18
19use super::{
20    BatchNotifier, EventFinalizer, EventFinalizers, EventMetadata, Finalizable, MergeFinalizable,
21    estimated_json_encoded_size_of::EstimatedJsonEncodedSizeOf,
22};
23use crate::config::telemetry;
24
25#[cfg(any(test, feature = "test"))]
26mod arbitrary;
27
28mod data;
29pub use self::data::*;
30
31mod series;
32pub use self::series::*;
33
34mod tags;
35pub use self::tags::*;
36
37mod value;
38pub use self::value::*;
39
40#[macro_export]
41macro_rules! metric_tags {
42    () => { $crate::event::MetricTags::default() };
43
44    ($($key:expr => $value:expr,)+) => { $crate::metric_tags!($($key => $value),+) };
45
46    ($($key:expr => $value:expr),*) => {
47        [
48            $( ($key.into(), $crate::event::metric::TagValue::from($value)), )*
49        ].into_iter().collect::<$crate::event::MetricTags>()
50    };
51}
52
53/// A metric.
54#[configurable_component]
55#[derive(Clone, Debug, PartialEq)]
56pub struct Metric {
57    #[serde(flatten)]
58    pub(super) series: MetricSeries,
59
60    #[serde(flatten)]
61    pub(super) data: MetricData,
62
63    /// Internal event metadata.
64    #[serde(skip, default = "EventMetadata::default")]
65    metadata: EventMetadata,
66}
67
68impl Metric {
69    /// Creates a new `Metric` with the given `name`, `kind`, and `value`.
70    pub fn new<T: Into<String>>(name: T, kind: MetricKind, value: MetricValue) -> Self {
71        Self::new_with_metadata(name, kind, value, EventMetadata::default())
72    }
73
74    /// Creates a new `Metric` with the given `name`, `kind`, `value`, and `metadata`.
75    pub fn new_with_metadata<T: Into<String>>(
76        name: T,
77        kind: MetricKind,
78        value: MetricValue,
79        metadata: EventMetadata,
80    ) -> Self {
81        Self {
82            series: MetricSeries {
83                name: MetricName {
84                    name: name.into(),
85                    namespace: None,
86                },
87                tags: None,
88            },
89            data: MetricData {
90                time: MetricTime {
91                    timestamp: None,
92                    interval_ms: None,
93                },
94                kind,
95                value,
96            },
97            metadata,
98        }
99    }
100
101    /// Consumes this metric, returning it with an updated series based on the given `name`.
102    #[inline]
103    #[must_use]
104    pub fn with_name(mut self, name: impl Into<String>) -> Self {
105        self.series.name.name = name.into();
106        self
107    }
108
109    /// Consumes this metric, returning it with an updated series based on the given `namespace`.
110    #[inline]
111    #[must_use]
112    pub fn with_namespace<T: Into<String>>(mut self, namespace: Option<T>) -> Self {
113        self.series.name.namespace = namespace.map(Into::into);
114        self
115    }
116
117    /// Consumes this metric, returning it with an updated timestamp.
118    #[inline]
119    #[must_use]
120    pub fn with_timestamp(mut self, timestamp: Option<DateTime<Utc>>) -> Self {
121        self.data.time.timestamp = timestamp;
122        self
123    }
124
125    /// Consumes this metric, returning it with an updated interval.
126    #[inline]
127    #[must_use]
128    pub fn with_interval_ms(mut self, interval_ms: Option<NonZeroU32>) -> Self {
129        self.data.time.interval_ms = interval_ms;
130        self
131    }
132
133    pub fn add_finalizer(&mut self, finalizer: EventFinalizer) {
134        self.metadata.add_finalizer(finalizer);
135    }
136
137    /// Consumes this metric, returning it with an updated set of event finalizers attached to `batch`.
138    #[must_use]
139    pub fn with_batch_notifier(mut self, batch: &BatchNotifier) -> Self {
140        self.metadata = self.metadata.with_batch_notifier(batch);
141        self
142    }
143
144    /// Consumes this metric, returning it with an optionally updated set of event finalizers attached to `batch`.
145    #[must_use]
146    pub fn with_batch_notifier_option(mut self, batch: &Option<BatchNotifier>) -> Self {
147        self.metadata = self.metadata.with_batch_notifier_option(batch);
148        self
149    }
150
151    /// Consumes this metric, returning it with an updated series based on the given `tags`.
152    #[inline]
153    #[must_use]
154    pub fn with_tags(mut self, tags: Option<MetricTags>) -> Self {
155        self.series.tags = tags;
156        self
157    }
158
159    /// Consumes this metric, returning it with an updated value.
160    #[inline]
161    #[must_use]
162    pub fn with_value(mut self, value: MetricValue) -> Self {
163        self.data.value = value;
164        self
165    }
166
167    /// Gets a reference to the series of this metric.
168    ///
169    /// The "series" is the name of the metric itself, including any tags. In other words, it is the unique identifier
170    /// for a metric, although metrics of different values (counter vs gauge) may be able to co-exist in outside metrics
171    /// implementations with identical series.
172    pub fn series(&self) -> &MetricSeries {
173        &self.series
174    }
175
176    /// Gets a reference to the data of this metric.
177    pub fn data(&self) -> &MetricData {
178        &self.data
179    }
180
181    /// Gets a mutable reference to the data of this metric.
182    pub fn data_mut(&mut self) -> &mut MetricData {
183        &mut self.data
184    }
185
186    /// Gets a reference to the metadata of this metric.
187    pub fn metadata(&self) -> &EventMetadata {
188        &self.metadata
189    }
190
191    /// Gets a mutable reference to the metadata of this metric.
192    pub fn metadata_mut(&mut self) -> &mut EventMetadata {
193        &mut self.metadata
194    }
195
196    /// Gets a reference to the name of this metric.
197    ///
198    /// The name of the metric does not include the namespace or tags.
199    #[inline]
200    pub fn name(&self) -> &str {
201        &self.series.name.name
202    }
203
204    /// Gets a reference to the namespace of this metric, if it exists.
205    #[inline]
206    pub fn namespace(&self) -> Option<&str> {
207        self.series.name.namespace.as_deref()
208    }
209
210    /// Takes the namespace out of this metric, if it exists, leaving it empty.
211    #[inline]
212    pub fn take_namespace(&mut self) -> Option<String> {
213        self.series.name.namespace.take()
214    }
215
216    /// Gets a reference to the tags of this metric, if they exist.
217    #[inline]
218    pub fn tags(&self) -> Option<&MetricTags> {
219        self.series.tags.as_ref()
220    }
221
222    /// Gets a mutable reference to the tags of this metric, if they exist.
223    #[inline]
224    pub fn tags_mut(&mut self) -> Option<&mut MetricTags> {
225        self.series.tags.as_mut()
226    }
227
228    /// Gets a reference to the timestamp of this metric, if it exists.
229    #[inline]
230    pub fn timestamp(&self) -> Option<DateTime<Utc>> {
231        self.data.time.timestamp
232    }
233
234    /// Gets a reference to the interval (in milliseconds) covered by this metric, if it exists.
235    #[inline]
236    pub fn interval_ms(&self) -> Option<NonZeroU32> {
237        self.data.time.interval_ms
238    }
239
240    /// Gets a reference to the value of this metric.
241    #[inline]
242    pub fn value(&self) -> &MetricValue {
243        &self.data.value
244    }
245
246    /// Gets a mutable reference to the value of this metric.
247    #[inline]
248    pub fn value_mut(&mut self) -> &mut MetricValue {
249        &mut self.data.value
250    }
251
252    /// Gets the kind of this metric.
253    #[inline]
254    pub fn kind(&self) -> MetricKind {
255        self.data.kind
256    }
257
258    /// Gets the time information of this metric.
259    #[inline]
260    pub fn time(&self) -> MetricTime {
261        self.data.time
262    }
263
264    /// Decomposes a `Metric` into its individual parts.
265    #[inline]
266    pub fn into_parts(self) -> (MetricSeries, MetricData, EventMetadata) {
267        (self.series, self.data, self.metadata)
268    }
269
270    /// Creates a `Metric` directly from the raw components of another metric.
271    #[inline]
272    pub fn from_parts(series: MetricSeries, data: MetricData, metadata: EventMetadata) -> Self {
273        Self {
274            series,
275            data,
276            metadata,
277        }
278    }
279
280    /// Consumes this metric, returning it as an absolute metric.
281    ///
282    /// If the metric was already absolute, nothing is changed.
283    #[must_use]
284    pub fn into_absolute(self) -> Self {
285        Self {
286            series: self.series,
287            data: self.data.into_absolute(),
288            metadata: self.metadata,
289        }
290    }
291
292    /// Consumes this metric, returning it as an incremental metric.
293    ///
294    /// If the metric was already incremental, nothing is changed.
295    #[must_use]
296    pub fn into_incremental(self) -> Self {
297        Self {
298            series: self.series,
299            data: self.data.into_incremental(),
300            metadata: self.metadata,
301        }
302    }
303
304    /// Creates a new metric from components specific to a metric emitted by `metrics`.
305    #[allow(clippy::cast_precision_loss)]
306    pub(crate) fn from_metric_kv(
307        key: &metrics::Key,
308        value: MetricValue,
309        timestamp: DateTime<Utc>,
310    ) -> Self {
311        let labels = key
312            .labels()
313            .map(|label| (String::from(label.key()), String::from(label.value())))
314            .collect::<MetricTags>();
315
316        Self::new(key.name().to_string(), MetricKind::Absolute, value)
317            .with_namespace(Some("vector"))
318            .with_timestamp(Some(timestamp))
319            .with_tags((!labels.is_empty()).then_some(labels))
320    }
321
322    /// Removes a tag from this metric, returning the value of the tag if the tag was previously in the metric.
323    pub fn remove_tag(&mut self, key: &str) -> Option<String> {
324        self.series.remove_tag(key)
325    }
326
327    /// Removes a tag from this metric, returning its full value set.
328    pub fn remove_tag_set(&mut self, key: &str) -> Option<TagValueSet> {
329        match &mut self.series.tags {
330            None => None,
331            Some(tags) => {
332                let result = tags.remove_set(key);
333                if tags.is_empty() {
334                    self.series.tags = None;
335                }
336                result
337            }
338        }
339    }
340
341    /// Removes all the tags.
342    pub fn remove_tags(&mut self) {
343        self.series.remove_tags();
344    }
345
346    /// Returns `true` if `name` tag is present, and matches the provided `value`
347    pub fn tag_matches(&self, name: &str, value: &str) -> bool {
348        self.tags()
349            .as_ref()
350            .is_some_and(|t| t.get(name).as_ref().is_some_and(|v| *v == value))
351    }
352
353    /// Returns the string value of a tag, if it exists
354    pub fn tag_value(&self, name: &str) -> Option<String> {
355        self.tags().and_then(|t| t.get(name)).map(ToOwned::to_owned)
356    }
357
358    /// Inserts a tag into this metric.
359    ///
360    /// If the metric did not have this tag, `None` will be returned. Otherwise, `Some(String)` will be returned,
361    /// containing the previous value of the tag.
362    ///
363    /// *Note:* This will create the tags map if it is not present.
364    pub fn replace_tag(&mut self, name: String, value: String) -> Option<String> {
365        self.series.replace_tag(name, value)
366    }
367
368    pub fn set_multi_value_tag(
369        &mut self,
370        name: String,
371        values: impl IntoIterator<Item = TagValue>,
372    ) {
373        self.series.set_multi_value_tag(name, values);
374    }
375
376    /// Zeroes out the data in this metric.
377    pub fn zero(&mut self) {
378        self.data.zero();
379    }
380
381    /// Adds the data from the `other` metric to this one.
382    ///
383    /// The other metric must be incremental and contain the same value type as this one.
384    #[must_use]
385    pub fn add(&mut self, other: impl AsRef<MetricData>) -> bool {
386        self.data.add(other.as_ref())
387    }
388
389    /// Updates this metric by adding the data from `other`.
390    #[must_use]
391    pub fn update(&mut self, other: impl AsRef<MetricData>) -> bool {
392        self.data.update(other.as_ref())
393    }
394
395    /// Subtracts the data from the `other` metric from this one.
396    ///
397    /// The other metric must contain the same value type as this one.
398    #[must_use]
399    pub fn subtract(&mut self, other: impl AsRef<MetricData>) -> bool {
400        self.data.subtract(other.as_ref())
401    }
402
403    /// Reduces all the tag values to their single value, discarding any for which that value would
404    /// be null. If the result is empty, the tag set is dropped.
405    pub fn reduce_tags_to_single(&mut self) {
406        if let Some(tags) = &mut self.series.tags {
407            tags.reduce_to_single();
408            if tags.is_empty() {
409                self.series.tags = None;
410            }
411        }
412    }
413}
414
415impl AsRef<MetricData> for Metric {
416    fn as_ref(&self) -> &MetricData {
417        &self.data
418    }
419}
420
421impl AsRef<MetricValue> for Metric {
422    fn as_ref(&self) -> &MetricValue {
423        &self.data.value
424    }
425}
426
427impl Display for Metric {
428    /// Display a metric using something like Prometheus' text format:
429    ///
430    /// ```text
431    /// TIMESTAMP NAMESPACE_NAME{TAGS} KIND DATA
432    /// ```
433    ///
434    /// TIMESTAMP is in ISO 8601 format with UTC time zone.
435    ///
436    /// KIND is either `=` for absolute metrics, or `+` for incremental
437    /// metrics.
438    ///
439    /// DATA is dependent on the type of metric, and is a simplified
440    /// representation of the data contents. In particular,
441    /// distributions, histograms, and summaries are represented as a
442    /// list of `X@Y` words, where `X` is the rate, count, or quantile,
443    /// and `Y` is the value or bucket.
444    ///
445    /// example:
446    /// ```text
447    /// 2020-08-12T20:23:37.248661343Z vector_received_bytes_total{component_kind="sink",component_type="blackhole"} = 6391
448    /// ```
449    fn fmt(&self, fmt: &mut Formatter<'_>) -> Result<(), fmt::Error> {
450        if let Some(timestamp) = &self.data.time.timestamp {
451            write!(fmt, "{timestamp:?} ")?;
452        }
453        let kind = match self.data.kind {
454            MetricKind::Absolute => '=',
455            MetricKind::Incremental => '+',
456        };
457        self.series.fmt(fmt)?;
458        write!(fmt, " {kind} ")?;
459        self.data.value.fmt(fmt)
460    }
461}
462
463impl EventDataEq for Metric {
464    fn event_data_eq(&self, other: &Self) -> bool {
465        self.series == other.series
466            && self.data == other.data
467            && self.metadata.event_data_eq(&other.metadata)
468    }
469}
470
471impl ByteSizeOf for Metric {
472    fn allocated_bytes(&self) -> usize {
473        self.series.allocated_bytes()
474            + self.data.allocated_bytes()
475            + self.metadata.allocated_bytes()
476    }
477}
478
479impl EstimatedJsonEncodedSizeOf for Metric {
480    fn estimated_json_encoded_size_of(&self) -> JsonSize {
481        // TODO: For now we're using the in-memory representation of the metric, but we'll convert
482        // this to actually calculate the JSON encoded size in the near future.
483        self.size_of().into()
484    }
485}
486
487impl Finalizable for Metric {
488    fn take_finalizers(&mut self) -> EventFinalizers {
489        self.metadata.take_finalizers()
490    }
491}
492
493impl MergeFinalizable for Metric {
494    fn merge_finalizers(&mut self, finalizers: EventFinalizers) {
495        self.metadata.merge_finalizers(finalizers);
496    }
497}
498
499impl GetEventCountTags for Metric {
500    fn get_tags(&self) -> TaggedEventsSent {
501        let source = if telemetry().tags().emit_source {
502            self.metadata().source_id().cloned().into()
503        } else {
504            OptionalTag::Ignored
505        };
506
507        // Currently there is no way to specify a tag that means the service,
508        // so we will be hardcoding it to "service".
509        let service = if telemetry().tags().emit_service {
510            self.tags()
511                .and_then(|tags| tags.get("service").map(ToString::to_string))
512                .into()
513        } else {
514            OptionalTag::Ignored
515        };
516
517        TaggedEventsSent { source, service }
518    }
519}
520
521/// Metric kind.
522///
523/// Metrics can be either absolute or incremental. Absolute metrics represent a sort of "last write wins" scenario,
524/// where the latest absolute value seen is meant to be the actual metric value.  In contrast, and perhaps intuitively,
525/// incremental metrics are meant to be additive, such that we don't know what total value of the metric is, but we know
526/// that we'll be adding or subtracting the given value from it.
527///
528/// Generally speaking, most metrics storage systems deal with incremental updates. A notable exception is Prometheus,
529/// which deals with, and expects, absolute values from clients.
530#[configurable_component]
531#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, PartialOrd)]
532#[serde(rename_all = "snake_case")]
533pub enum MetricKind {
534    /// Incremental metric.
535    Incremental,
536
537    /// Absolute metric.
538    Absolute,
539}
540
541impl TryFrom<vrl::value::Value> for MetricKind {
542    type Error = String;
543
544    fn try_from(value: vrl::value::Value) -> Result<Self, Self::Error> {
545        let value = value.try_bytes().map_err(|e| e.to_string())?;
546        match std::str::from_utf8(&value).map_err(|e| e.to_string())? {
547            "incremental" => Ok(Self::Incremental),
548            "absolute" => Ok(Self::Absolute),
549            value => Err(format!(
550                "invalid metric kind {value}, metric kind must be `absolute` or `incremental`"
551            )),
552        }
553    }
554}
555
556impl From<MetricKind> for vrl::value::Value {
557    fn from(kind: MetricKind) -> Self {
558        match kind {
559            MetricKind::Incremental => "incremental".into(),
560            MetricKind::Absolute => "absolute".into(),
561        }
562    }
563}
564
565#[macro_export]
566macro_rules! samples {
567    ( $( $value:expr => $rate:expr ),* ) => {
568        vec![ $( $crate::event::metric::Sample { value: $value, rate: $rate }, )* ]
569    }
570}
571
572#[macro_export]
573macro_rules! buckets {
574    ( $( $limit:expr => $count:expr ),* ) => {
575        vec![ $( $crate::event::metric::Bucket { upper_limit: $limit, count: $count }, )* ]
576    }
577}
578
579#[macro_export]
580macro_rules! quantiles {
581    ( $( $q:expr => $value:expr ),* ) => {
582        vec![ $( $crate::event::metric::Quantile { quantile: $q, value: $value }, )* ]
583    }
584}
585
586#[inline]
587pub(crate) fn zip_samples(
588    values: impl IntoIterator<Item = f64>,
589    rates: impl IntoIterator<Item = u32>,
590) -> Vec<Sample> {
591    values
592        .into_iter()
593        .zip(rates)
594        .map(|(value, rate)| Sample { value, rate })
595        .collect()
596}
597
598#[inline]
599pub(crate) fn zip_buckets(
600    limits: impl IntoIterator<Item = f64>,
601    counts: impl IntoIterator<Item = u64>,
602) -> Vec<Bucket> {
603    limits
604        .into_iter()
605        .zip(counts)
606        .map(|(upper_limit, count)| Bucket { upper_limit, count })
607        .collect()
608}
609
610#[inline]
611pub(crate) fn zip_quantiles(
612    quantiles: impl IntoIterator<Item = f64>,
613    values: impl IntoIterator<Item = f64>,
614) -> Vec<Quantile> {
615    quantiles
616        .into_iter()
617        .zip(values)
618        .map(|(quantile, value)| Quantile { quantile, value })
619        .collect()
620}
621
622fn write_list<I, T, W>(
623    fmt: &mut Formatter<'_>,
624    sep: &str,
625    items: I,
626    writer: W,
627) -> Result<(), fmt::Error>
628where
629    I: IntoIterator<Item = T>,
630    W: Fn(&mut Formatter<'_>, T) -> Result<(), fmt::Error>,
631{
632    let mut this_sep = "";
633    for item in items {
634        write!(fmt, "{this_sep}")?;
635        writer(fmt, item)?;
636        this_sep = sep;
637    }
638    Ok(())
639}
640
641fn write_word(fmt: &mut Formatter<'_>, word: &str) -> Result<(), fmt::Error> {
642    if word.contains(|c: char| !c.is_ascii_alphanumeric() && c != '_') {
643        write!(fmt, "{word:?}")
644    } else {
645        write!(fmt, "{word}")
646    }
647}
648
649pub fn samples_to_buckets(samples: &[Sample], buckets: &[f64]) -> (Vec<Bucket>, u64, f64) {
650    let mut counts = vec![0; buckets.len()];
651    let mut sum = 0.0;
652    let mut count = 0;
653    for sample in samples {
654        let rate = u64::from(sample.rate);
655
656        if let Some((i, _)) = buckets
657            .iter()
658            .enumerate()
659            .find(|&(_, b)| *b >= sample.value)
660        {
661            counts[i] += rate;
662        }
663
664        sum += sample.value * f64::from(sample.rate);
665        count += rate;
666    }
667
668    let buckets = buckets
669        .iter()
670        .zip(counts.iter())
671        .map(|(b, c)| Bucket {
672            upper_limit: *b,
673            count: *c,
674        })
675        .collect();
676
677    (buckets, count, sum)
678}
679
680#[cfg(test)]
681mod test {
682    use std::collections::BTreeSet;
683
684    use chrono::{DateTime, Timelike, Utc, offset::TimeZone};
685    use similar_asserts::assert_eq;
686
687    use super::*;
688
689    fn ts() -> DateTime<Utc> {
690        Utc.with_ymd_and_hms(2018, 11, 14, 8, 9, 10)
691            .single()
692            .and_then(|t| t.with_nanosecond(11))
693            .expect("invalid timestamp")
694    }
695
696    fn tags() -> MetricTags {
697        metric_tags!(
698            "normal_tag" => "value",
699            "true_tag" => "true",
700            "empty_tag" => "",
701        )
702    }
703
704    #[test]
705    fn merge_counters() {
706        let mut counter = Metric::new(
707            "counter",
708            MetricKind::Incremental,
709            MetricValue::Counter { value: 1.0 },
710        );
711
712        let delta = Metric::new(
713            "counter",
714            MetricKind::Incremental,
715            MetricValue::Counter { value: 2.0 },
716        )
717        .with_namespace(Some("vector"))
718        .with_tags(Some(tags()))
719        .with_timestamp(Some(ts()));
720
721        let expected = counter
722            .clone()
723            .with_value(MetricValue::Counter { value: 3.0 })
724            .with_timestamp(Some(ts()));
725
726        assert!(counter.data.add(&delta.data));
727        assert_eq!(counter, expected);
728    }
729
730    #[test]
731    fn merge_gauges() {
732        let mut gauge = Metric::new(
733            "gauge",
734            MetricKind::Incremental,
735            MetricValue::Gauge { value: 1.0 },
736        );
737
738        let delta = Metric::new(
739            "gauge",
740            MetricKind::Incremental,
741            MetricValue::Gauge { value: -2.0 },
742        )
743        .with_namespace(Some("vector"))
744        .with_tags(Some(tags()))
745        .with_timestamp(Some(ts()));
746
747        let expected = gauge
748            .clone()
749            .with_value(MetricValue::Gauge { value: -1.0 })
750            .with_timestamp(Some(ts()));
751
752        assert!(gauge.data.add(&delta.data));
753        assert_eq!(gauge, expected);
754    }
755
756    #[test]
757    fn merge_sets() {
758        let mut set = Metric::new(
759            "set",
760            MetricKind::Incremental,
761            MetricValue::Set {
762                values: vec!["old".into()].into_iter().collect(),
763            },
764        );
765
766        let delta = Metric::new(
767            "set",
768            MetricKind::Incremental,
769            MetricValue::Set {
770                values: vec!["new".into()].into_iter().collect(),
771            },
772        )
773        .with_namespace(Some("vector"))
774        .with_tags(Some(tags()))
775        .with_timestamp(Some(ts()));
776
777        let expected = set
778            .clone()
779            .with_value(MetricValue::Set {
780                values: vec!["old".into(), "new".into()].into_iter().collect(),
781            })
782            .with_timestamp(Some(ts()));
783
784        assert!(set.data.add(&delta.data));
785        assert_eq!(set, expected);
786    }
787
788    #[test]
789    fn merge_histograms() {
790        let mut dist = Metric::new(
791            "hist",
792            MetricKind::Incremental,
793            MetricValue::Distribution {
794                samples: samples![1.0 => 10],
795                statistic: StatisticKind::Histogram,
796            },
797        );
798
799        let delta = Metric::new(
800            "hist",
801            MetricKind::Incremental,
802            MetricValue::Distribution {
803                samples: samples![1.0 => 20],
804                statistic: StatisticKind::Histogram,
805            },
806        )
807        .with_namespace(Some("vector"))
808        .with_tags(Some(tags()))
809        .with_timestamp(Some(ts()));
810
811        let expected = dist
812            .clone()
813            .with_value(MetricValue::Distribution {
814                samples: samples![1.0 => 10, 1.0 => 20],
815                statistic: StatisticKind::Histogram,
816            })
817            .with_timestamp(Some(ts()));
818
819        assert!(dist.data.add(&delta.data));
820        assert_eq!(dist, expected);
821    }
822
823    #[test]
824    fn subtract_counters() {
825        // Make sure a newer/higher value counter can subtract an older/lesser value counter:
826        let old_counter = Metric::new(
827            "counter",
828            MetricKind::Absolute,
829            MetricValue::Counter { value: 4.0 },
830        );
831
832        let mut new_counter = Metric::new(
833            "counter",
834            MetricKind::Absolute,
835            MetricValue::Counter { value: 6.0 },
836        );
837
838        assert!(new_counter.subtract(&old_counter));
839        assert_eq!(new_counter.value(), &MetricValue::Counter { value: 2.0 });
840
841        // But not the other way around:
842        let old_counter = Metric::new(
843            "counter",
844            MetricKind::Absolute,
845            MetricValue::Counter { value: 6.0 },
846        );
847
848        let mut new_reset_counter = Metric::new(
849            "counter",
850            MetricKind::Absolute,
851            MetricValue::Counter { value: 1.0 },
852        );
853
854        assert!(!new_reset_counter.subtract(&old_counter));
855    }
856
857    #[test]
858    fn subtract_aggregated_histograms() {
859        // Make sure a newer/higher count aggregated histogram can subtract an older/lower count
860        // aggregated histogram:
861        let old_histogram = Metric::new(
862            "histogram",
863            MetricKind::Absolute,
864            MetricValue::AggregatedHistogram {
865                count: 1,
866                sum: 1.0,
867                buckets: buckets!(2.0 => 1),
868            },
869        );
870
871        let mut new_histogram = Metric::new(
872            "histogram",
873            MetricKind::Absolute,
874            MetricValue::AggregatedHistogram {
875                count: 3,
876                sum: 3.0,
877                buckets: buckets!(2.0 => 3),
878            },
879        );
880
881        assert!(new_histogram.subtract(&old_histogram));
882        assert_eq!(
883            new_histogram.value(),
884            &MetricValue::AggregatedHistogram {
885                count: 2,
886                sum: 2.0,
887                buckets: buckets!(2.0 => 2),
888            }
889        );
890
891        // But not the other way around:
892        let old_histogram = Metric::new(
893            "histogram",
894            MetricKind::Absolute,
895            MetricValue::AggregatedHistogram {
896                count: 3,
897                sum: 3.0,
898                buckets: buckets!(2.0 => 3),
899            },
900        );
901
902        let mut new_reset_histogram = Metric::new(
903            "histogram",
904            MetricKind::Absolute,
905            MetricValue::AggregatedHistogram {
906                count: 1,
907                sum: 1.0,
908                buckets: buckets!(2.0 => 1),
909            },
910        );
911
912        assert!(!new_reset_histogram.subtract(&old_histogram));
913    }
914
915    #[test]
916    fn subtract_aggregated_histograms_bucket_redistribution() {
917        // Test for issue #24415: when total count is higher but individual bucket counts is sometimes lower
918        let old_histogram = Metric::new(
919            "histogram",
920            MetricKind::Absolute,
921            MetricValue::AggregatedHistogram {
922                count: 15,
923                sum: 15.0,
924                buckets: buckets!(1.0 => 10, 2.0 => 5),
925            },
926        );
927
928        let mut new_histogram_with_redistribution = Metric::new(
929            "histogram",
930            MetricKind::Absolute,
931            MetricValue::AggregatedHistogram {
932                count: 20,
933                sum: 20.0,
934                // Total count is higher (20 > 15), but bucket1 count is lower (8 < 10)
935                buckets: buckets!(1.0 => 8, 2.0 => 12),
936            },
937        );
938
939        assert!(!new_histogram_with_redistribution.subtract(&old_histogram));
940    }
941
942    #[test]
943    // `too_many_lines` is mostly just useful for production code but we're not
944    // able to flag the lint on only for non-test.
945    #[allow(clippy::too_many_lines)]
946    fn display() {
947        assert_eq!(
948            format!(
949                "{}",
950                Metric::new(
951                    "one",
952                    MetricKind::Absolute,
953                    MetricValue::Counter { value: 1.23 },
954                )
955                .with_tags(Some(tags()))
956            ),
957            r#"one{empty_tag="",normal_tag="value",true_tag="true"} = 1.23"#
958        );
959
960        assert_eq!(
961            format!(
962                "{}",
963                Metric::new(
964                    "two word",
965                    MetricKind::Incremental,
966                    MetricValue::Gauge { value: 2.0 }
967                )
968                .with_timestamp(Some(ts()))
969            ),
970            r#"2018-11-14T08:09:10.000000011Z "two word"{} + 2"#
971        );
972
973        assert_eq!(
974            format!(
975                "{}",
976                Metric::new(
977                    "namespace",
978                    MetricKind::Absolute,
979                    MetricValue::Counter { value: 1.23 },
980                )
981                .with_namespace(Some("vector"))
982            ),
983            r"vector_namespace{} = 1.23"
984        );
985
986        assert_eq!(
987            format!(
988                "{}",
989                Metric::new(
990                    "namespace",
991                    MetricKind::Absolute,
992                    MetricValue::Counter { value: 1.23 },
993                )
994                .with_namespace(Some("vector host"))
995            ),
996            r#""vector host"_namespace{} = 1.23"#
997        );
998
999        let mut values = BTreeSet::<String>::new();
1000        values.insert("v1".into());
1001        values.insert("v2_two".into());
1002        values.insert("thrəë".into());
1003        values.insert("four=4".into());
1004        assert_eq!(
1005            format!(
1006                "{}",
1007                Metric::new("three", MetricKind::Absolute, MetricValue::Set { values })
1008            ),
1009            r#"three{} = "four=4" "thrəë" v1 v2_two"#
1010        );
1011
1012        assert_eq!(
1013            format!(
1014                "{}",
1015                Metric::new(
1016                    "four",
1017                    MetricKind::Absolute,
1018                    MetricValue::Distribution {
1019                        samples: samples![1.0 => 3, 2.0 => 4],
1020                        statistic: StatisticKind::Histogram,
1021                    }
1022                )
1023            ),
1024            r"four{} = histogram 3@1 4@2"
1025        );
1026
1027        assert_eq!(
1028            format!(
1029                "{}",
1030                Metric::new(
1031                    "five",
1032                    MetricKind::Absolute,
1033                    MetricValue::AggregatedHistogram {
1034                        buckets: buckets![51.0 => 53, 52.0 => 54],
1035                        count: 107,
1036                        sum: 103.0,
1037                    }
1038                )
1039            ),
1040            r"five{} = count=107 sum=103 53@51 54@52"
1041        );
1042
1043        assert_eq!(
1044            format!(
1045                "{}",
1046                Metric::new(
1047                    "six",
1048                    MetricKind::Absolute,
1049                    MetricValue::AggregatedSummary {
1050                        quantiles: quantiles![1.0 => 63.0, 2.0 => 64.0],
1051                        count: 2,
1052                        sum: 127.0,
1053                    }
1054                )
1055            ),
1056            r"six{} = count=2 sum=127 1@63 2@64"
1057        );
1058    }
1059
1060    #[test]
1061    fn quantile_to_percentile_string() {
1062        let quantiles = [
1063            (-1.0, "0"),
1064            (0.0, "0"),
1065            (0.25, "25"),
1066            (0.50, "50"),
1067            (0.999, "999"),
1068            (0.9999, "9999"),
1069            (0.99999, "9999"),
1070            (1.0, "100"),
1071            (3.0, "100"),
1072        ];
1073
1074        for (quantile, expected) in quantiles {
1075            let quantile = Quantile {
1076                quantile,
1077                value: 1.0,
1078            };
1079            let result = quantile.to_percentile_string();
1080            assert_eq!(result, expected);
1081        }
1082    }
1083
1084    #[test]
1085    fn quantile_to_string() {
1086        let quantiles = [
1087            (-1.0, "0"),
1088            (0.0, "0"),
1089            (0.25, "0.25"),
1090            (0.50, "0.5"),
1091            (0.999, "0.999"),
1092            (0.9999, "0.9999"),
1093            (0.99999, "0.9999"),
1094            (1.0, "1"),
1095            (3.0, "1"),
1096        ];
1097
1098        for (quantile, expected) in quantiles {
1099            let quantile = Quantile {
1100                quantile,
1101                value: 1.0,
1102            };
1103            let result = quantile.to_quantile_string();
1104            assert_eq!(result, expected);
1105        }
1106    }
1107
1108    #[test]
1109    fn value_conversions() {
1110        let counter_value = MetricValue::Counter { value: 3.13 };
1111        assert_eq!(counter_value.distribution_to_agg_histogram(&[1.0]), None);
1112
1113        let counter_value = MetricValue::Counter { value: 3.13 };
1114        assert_eq!(counter_value.distribution_to_sketch(), None);
1115
1116        let distrib_value = MetricValue::Distribution {
1117            samples: samples!(1.0 => 10, 2.0 => 5, 5.0 => 2),
1118            statistic: StatisticKind::Summary,
1119        };
1120        let converted = distrib_value.distribution_to_agg_histogram(&[1.0, 5.0, 10.0]);
1121        assert_eq!(
1122            converted,
1123            Some(MetricValue::AggregatedHistogram {
1124                buckets: vec![
1125                    Bucket {
1126                        upper_limit: 1.0,
1127                        count: 10,
1128                    },
1129                    Bucket {
1130                        upper_limit: 5.0,
1131                        count: 7,
1132                    },
1133                    Bucket {
1134                        upper_limit: 10.0,
1135                        count: 0,
1136                    },
1137                ],
1138                sum: 30.0,
1139                count: 17,
1140            })
1141        );
1142
1143        let distrib_value = MetricValue::Distribution {
1144            samples: samples!(1.0 => 1),
1145            statistic: StatisticKind::Summary,
1146        };
1147        let converted = distrib_value.distribution_to_sketch();
1148        assert!(matches!(converted, Some(MetricValue::Sketch { .. })));
1149    }
1150
1151    #[test]
1152    fn merge_non_contiguous_interval() {
1153        let mut gauge = Metric::new(
1154            "gauge",
1155            MetricKind::Incremental,
1156            MetricValue::Gauge { value: 12.0 },
1157        )
1158        .with_timestamp(Some(ts()))
1159        .with_interval_ms(std::num::NonZeroU32::new(10));
1160
1161        let delta = Metric::new(
1162            "gauge",
1163            MetricKind::Incremental,
1164            MetricValue::Gauge { value: -5.0 },
1165        )
1166        .with_timestamp(Some(ts() + chrono::Duration::milliseconds(20)))
1167        .with_interval_ms(std::num::NonZeroU32::new(15));
1168
1169        let expected = gauge
1170            .clone()
1171            .with_value(MetricValue::Gauge { value: 7.0 })
1172            .with_timestamp(Some(ts()))
1173            .with_interval_ms(std::num::NonZeroU32::new(35));
1174
1175        assert!(gauge.data.add(&delta.data));
1176        assert_eq!(gauge, expected);
1177    }
1178
1179    #[test]
1180    fn merge_contiguous_interval() {
1181        let mut gauge = Metric::new(
1182            "gauge",
1183            MetricKind::Incremental,
1184            MetricValue::Gauge { value: 12.0 },
1185        )
1186        .with_timestamp(Some(ts()))
1187        .with_interval_ms(std::num::NonZeroU32::new(10));
1188
1189        let delta = Metric::new(
1190            "gauge",
1191            MetricKind::Incremental,
1192            MetricValue::Gauge { value: -5.0 },
1193        )
1194        .with_timestamp(Some(ts() + chrono::Duration::milliseconds(5)))
1195        .with_interval_ms(std::num::NonZeroU32::new(15));
1196
1197        let expected = gauge
1198            .clone()
1199            .with_value(MetricValue::Gauge { value: 7.0 })
1200            .with_timestamp(Some(ts()))
1201            .with_interval_ms(std::num::NonZeroU32::new(20));
1202
1203        assert!(gauge.data.add(&delta.data));
1204        assert_eq!(gauge, expected);
1205    }
1206}
vector_core/event/metric/mod.rs

vector_core/event/metric/
mod.rs
