relay_metrics/bucket.rs
1use std::collections::{BTreeMap, BTreeSet};
2use std::iter::FusedIterator;
3use std::{fmt, mem};
4
5use relay_common::time::UnixTimestamp;
6use relay_protocol::FiniteF64;
7use serde::{Deserialize, Serialize};
8use smallvec::SmallVec;
9
10use crate::ParseMetricError;
11use crate::protocol::{
12 self, CounterType, DistributionType, GaugeType, MetricName, MetricResourceIdentifier,
13 MetricType, SetType, hash_set_value,
14};
15
16const VALUE_SEPARATOR: char = ':';
17
18/// Type of [`Bucket::tags`].
19pub type MetricTags = BTreeMap<String, String>;
20
21/// A snapshot of values within a [`Bucket`].
22#[derive(Clone, Copy, Debug, PartialEq, Deserialize, Serialize)]
23pub struct GaugeValue {
24 /// The last value reported in the bucket.
25 ///
26 /// This aggregation is not commutative.
27 pub last: GaugeType,
28 /// The minimum value reported in the bucket.
29 pub min: GaugeType,
30 /// The maximum value reported in the bucket.
31 pub max: GaugeType,
32 /// The sum of all values reported in the bucket.
33 pub sum: GaugeType,
34 /// The number of times this bucket was updated with a new value.
35 pub count: u64,
36}
37
38impl GaugeValue {
39 /// Creates a gauge snapshot from a single value.
40 pub fn single(value: GaugeType) -> Self {
41 Self {
42 last: value,
43 min: value,
44 max: value,
45 sum: value,
46 count: 1,
47 }
48 }
49
50 /// Inserts a new value into the gauge.
51 pub fn insert(&mut self, value: GaugeType) {
52 self.last = value;
53 self.min = self.min.min(value);
54 self.max = self.max.max(value);
55 self.sum = self.sum.saturating_add(value);
56 self.count += 1;
57 }
58
59 /// Merges two gauge snapshots.
60 pub fn merge(&mut self, other: Self) {
61 self.last = other.last;
62 self.min = self.min.min(other.min);
63 self.max = self.max.max(other.max);
64 self.sum = self.sum.saturating_add(other.sum);
65 self.count += other.count;
66 }
67
68 /// Returns the average of all values reported in this bucket.
69 pub fn avg(&self) -> Option<GaugeType> {
70 self.sum / FiniteF64::new(self.count as f64)?
71 }
72}
73
74/// A distribution of values within a [`Bucket`].
75///
76/// Distributions logically store a histogram of values. Based on individual reported values,
77/// distributions allow to query the maximum, minimum, or average of the reported values, as well as
78/// statistical quantiles.
79///
80/// # Example
81///
82/// ```
83/// use relay_metrics::dist;
84///
85/// let mut dist = dist![1, 1, 1, 2];
86/// dist.push(5.into());
87/// dist.extend(std::iter::repeat(3.into()).take(7));
88/// ```
89///
90/// Logically, this distribution is equivalent to this visualization:
91///
92/// ```plain
93/// value | count
94/// 1.0 | ***
95/// 2.0 | *
96/// 3.0 | *******
97/// 4.0 |
98/// 5.0 | *
99/// ```
100///
101/// # Serialization
102///
103/// Distributions serialize as lists of floating point values. The list contains one entry for each
104/// value in the distribution, including duplicates.
105pub type DistributionValue = SmallVec<[DistributionType; 3]>;
106
107#[doc(hidden)]
108pub use smallvec::smallvec as _smallvec;
109
110/// Creates a [`DistributionValue`] containing the given arguments.
111///
112/// `dist!` allows `DistributionValue` to be defined with the same syntax as array expressions.
113///
114/// # Example
115///
116/// ```
117/// let dist = relay_metrics::dist![1, 2];
118/// ```
119#[macro_export]
120macro_rules! dist {
121 ($($x:expr),*$(,)*) => {
122 $crate::_smallvec!($($crate::DistributionType::from($x)),*) as $crate::DistributionValue
123 };
124}
125
126/// A set of unique values.
127///
128/// Set values can be specified as strings in the submission protocol. They are always hashed
129/// into a 32-bit value and the original value is dropped. If the submission protocol contains a
130/// 32-bit integer, it will be used directly, instead.
131///
132/// See the [bucket docs](crate::Bucket) for more information on set hashing.
133pub type SetValue = BTreeSet<SetType>;
134
135/// The [aggregated value](Bucket::value) of a metric bucket.
136#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
137#[serde(tag = "type", content = "value")]
138pub enum BucketValue {
139 /// Counts instances of an event ([`MetricType::Counter`]).
140 ///
141 /// Counters can be incremented and decremented. The default operation is to increment a counter
142 /// by `1`, although increments by larger values are equally possible.
143 ///
144 /// # Statsd Format
145 ///
146 /// Counters are declared as `"c"`. Alternatively, `"m"` is allowed.
147 ///
148 /// There can be a variable number of floating point values. If more than one value is given,
149 /// the values are summed into a single counter value:
150 ///
151 /// ```text
152 /// endpoint.hits:4.5:21:17.0|c
153 /// ```
154 ///
155 /// # Serialization
156 ///
157 /// This variant serializes to a double precision float.
158 ///
159 /// # Aggregation
160 ///
161 /// Counters aggregate by folding individual values into a single sum value per bucket. The sum
162 /// is ingested and stored directly.
163 #[serde(rename = "c")]
164 Counter(CounterType),
165
166 /// Builds a statistical distribution over values reported ([`MetricType::Distribution`]).
167 ///
168 /// Based on individual reported values, distributions allow to query the maximum, minimum, or
169 /// average of the reported values, as well as statistical quantiles. With an increasing number
170 /// of values in the distribution, its accuracy becomes approximate.
171 ///
172 /// # Statsd Format
173 ///
174 /// Distributions are declared as `"d"`. Alternatively, `"d"` and `"ms"` are allowed.
175 ///
176 /// There can be a variable number of floating point values. These values are collected directly
177 /// in a list per bucket.
178 ///
179 /// ```text
180 /// endpoint.response_time@millisecond:36:49:57:68|d
181 /// ```
182 ///
183 /// # Serialization
184 ///
185 /// This variant serializes to a list of double precision floats, see [`DistributionValue`].
186 ///
187 /// # Aggregation
188 ///
189 /// During ingestion, all individual reported values are collected in a lossless format. In
190 /// storage, these values are compressed into data sketches that allow to query quantiles.
191 /// Separately, the count and sum of the reported values is stored, which makes distributions a
192 /// strict superset of counters.
193 #[serde(rename = "d")]
194 Distribution(DistributionValue),
195
196 /// Counts the number of unique reported values.
197 ///
198 /// Sets allow sending arbitrary discrete values, including strings, and store the deduplicated
199 /// count. With an increasing number of unique values in the set, its accuracy becomes
200 /// approximate. It is not possible to query individual values from a set.
201 ///
202 /// # Statsd Format
203 ///
204 /// Sets are declared as `"s"`. Values in the list should be deduplicated.
205 ///
206 ///
207 /// ```text
208 /// endpoint.users:3182887624:4267882815|s
209 /// endpoint.users:e2546e4c-ecd0-43ad-ae27-87960e57a658|s
210 /// ```
211 ///
212 /// # Serialization
213 ///
214 /// This variant serializes to a list of 32-bit integers.
215 ///
216 /// # Aggregation
217 ///
218 /// Set values are internally represented as 32-bit integer hashes of the original value. These
219 /// hashes can be ingested directly as seen in the first example above. If raw strings are sent,
220 /// they will be hashed on-the-fly.
221 ///
222 /// Internally, set metrics are stored in data sketches that expose an approximate cardinality.
223 #[serde(rename = "s")]
224 Set(SetValue),
225
226 /// Stores absolute snapshots of values.
227 ///
228 /// In addition to plain [counters](Self::Counter), gauges store a snapshot of the maximum,
229 /// minimum and sum of all values, as well as the last reported value. Note that the "last"
230 /// component of this aggregation is not commutative. Which value is preserved as last value is
231 /// implementation-defined.
232 ///
233 /// # Statsd Format
234 ///
235 /// Gauges are declared as `"g"`. There are two ways to ingest gauges:
236 /// 1. As a single value. In this case, the provided value is assumed as the last, minimum,
237 /// maximum, and the sum.
238 /// 2. As a sequence of five values in the order: `last`, `min`, `max`, `sum`, `count`.
239 ///
240 /// ```text
241 /// endpoint.parallel_requests:25|g
242 /// endpoint.parallel_requests:25:17:42:220:85|g
243 /// ```
244 ///
245 /// # Serialization
246 ///
247 /// This variant serializes to a structure with named fields, see [`GaugeValue`].
248 ///
249 /// # Aggregation
250 ///
251 /// Gauges aggregate by folding each of the components based on their semantics:
252 /// - `last` assumes the newly added value
253 /// - `min` retains the smaller value
254 /// - `max` retains the larger value
255 /// - `sum` adds the new value to the existing sum
256 /// - `count` adds the count of the newly added gauge (defaulting to `1`)
257 #[serde(rename = "g")]
258 Gauge(GaugeValue),
259}
260
261impl BucketValue {
262 /// Returns a bucket value representing a counter with the given value.
263 pub fn counter(value: CounterType) -> Self {
264 Self::Counter(value)
265 }
266
267 /// Returns a bucket value representing a distribution with a single given value.
268 pub fn distribution(value: DistributionType) -> Self {
269 Self::Distribution(dist![value])
270 }
271
272 /// Returns a bucket value representing a set with a single given hash value.
273 pub fn set(value: SetType) -> Self {
274 Self::Set(std::iter::once(value).collect())
275 }
276
277 /// Returns a bucket value representing a set with a single given string value.
278 pub fn set_from_str(string: &str) -> Self {
279 Self::set(hash_set_value(string))
280 }
281
282 /// Returns a bucket value representing a set with a single given value.
283 pub fn set_from_display(display: impl fmt::Display) -> Self {
284 Self::set(hash_set_value(&display.to_string()))
285 }
286
287 /// Returns a bucket value representing a gauge with a single given value.
288 pub fn gauge(value: GaugeType) -> Self {
289 Self::Gauge(GaugeValue::single(value))
290 }
291
292 /// Returns the type of this value.
293 pub fn ty(&self) -> MetricType {
294 match self {
295 Self::Counter(_) => MetricType::Counter,
296 Self::Distribution(_) => MetricType::Distribution,
297 Self::Set(_) => MetricType::Set,
298 Self::Gauge(_) => MetricType::Gauge,
299 }
300 }
301
302 /// Returns the number of raw data points in this value.
303 pub fn len(&self) -> usize {
304 match self {
305 BucketValue::Counter(_) => 1,
306 BucketValue::Distribution(distribution) => distribution.len(),
307 BucketValue::Set(set) => set.len(),
308 BucketValue::Gauge(_) => 5,
309 }
310 }
311
312 /// Returns `true` if this bucket contains no values.
313 pub fn is_empty(&self) -> bool {
314 self.len() == 0
315 }
316
317 /// Estimates the number of bytes needed to encode the bucket value.
318 ///
319 /// Note that this does not necessarily match the exact memory footprint of the value,
320 /// because data structures have a memory overhead.
321 pub fn cost(&self) -> usize {
322 // Beside the size of [`BucketValue`], we also need to account for the cost of values
323 // allocated dynamically.
324 let allocated_cost = match self {
325 Self::Counter(_) => 0,
326 Self::Set(s) => mem::size_of::<SetType>() * s.len(),
327 Self::Gauge(_) => 0,
328 Self::Distribution(d) => d.len() * mem::size_of::<DistributionType>(),
329 };
330
331 mem::size_of::<Self>() + allocated_cost
332 }
333
334 /// Merges the given `bucket_value` into `self`.
335 ///
336 /// Returns `Ok(())` if the two bucket values can be merged. This is the case when both bucket
337 /// values are of the same variant. Otherwise, this returns `Err(other)`.
338 pub fn merge(&mut self, other: Self) -> Result<(), Self> {
339 match (self, other) {
340 (Self::Counter(slf), Self::Counter(other)) => *slf = slf.saturating_add(other),
341 (Self::Distribution(slf), Self::Distribution(other)) => slf.extend_from_slice(&other),
342 (Self::Set(slf), Self::Set(other)) => slf.extend(other),
343 (Self::Gauge(slf), Self::Gauge(other)) => slf.merge(other),
344 (_, other) => return Err(other),
345 }
346
347 Ok(())
348 }
349}
350
351/// Parses a list of counter values separated by colons and sums them up.
352fn parse_counter(string: &str) -> Option<CounterType> {
353 let mut sum = CounterType::default();
354 for component in string.split(VALUE_SEPARATOR) {
355 sum = sum.saturating_add(component.parse().ok()?);
356 }
357 Some(sum)
358}
359
360/// Parses a distribution from a list of floating point values separated by colons.
361fn parse_distribution(string: &str) -> Option<DistributionValue> {
362 let mut dist = DistributionValue::default();
363 for component in string.split(VALUE_SEPARATOR) {
364 dist.push(component.parse().ok()?);
365 }
366 Some(dist)
367}
368
369/// Parses a set of hashed numeric values.
370fn parse_set(string: &str) -> Option<SetValue> {
371 let mut set = SetValue::default();
372 for component in string.split(VALUE_SEPARATOR) {
373 let hash = component
374 .parse()
375 .unwrap_or_else(|_| protocol::hash_set_value(component));
376 set.insert(hash);
377 }
378 Some(set)
379}
380
381/// Parses a gauge from a value.
382///
383/// The gauge can either be given as a single floating point value, or as a list of exactly five
384/// values in the order of [`GaugeValue`] fields.
385fn parse_gauge(string: &str) -> Option<GaugeValue> {
386 let mut components = string.split(VALUE_SEPARATOR);
387
388 let last = components.next()?.parse().ok()?;
389 Some(if let Some(min) = components.next() {
390 GaugeValue {
391 last,
392 min: min.parse().ok()?,
393 max: components.next()?.parse().ok()?,
394 sum: components.next()?.parse().ok()?,
395 count: components.next()?.parse().ok()?,
396 }
397 } else {
398 GaugeValue::single(last)
399 })
400}
401
402/// Parses tags in the format `tag1,tag2:value`.
403///
404/// Tag values are optional. For tags with missing values, an empty `""` value is assumed.
405fn parse_tags(string: &str) -> Option<MetricTags> {
406 let mut map = MetricTags::new();
407
408 for pair in string.split(',') {
409 let mut name_value = pair.splitn(2, ':');
410
411 let name = name_value.next()?;
412 if !protocol::is_valid_tag_key(name) {
413 continue;
414 }
415
416 if let Ok(value) = protocol::unescape_tag_value(name_value.next().unwrap_or_default()) {
417 map.insert(name.to_owned(), value);
418 }
419 }
420
421 Some(map)
422}
423
424/// Parses a unix UTC timestamp.
425fn parse_timestamp(string: &str) -> Option<UnixTimestamp> {
426 string.parse().ok().map(UnixTimestamp::from_secs)
427}
428
429/// An aggregation of metric values.
430///
431/// As opposed to single metric values, bucket aggregations can carry multiple values. See
432/// [`MetricType`] for a description on how values are aggregated in buckets. Values are aggregated
433/// by metric name, type, time window, and all tags. Particularly, this allows metrics to have the
434/// same name even if their types differ.
435///
436/// See the [crate documentation](crate) for general information on Metrics.
437///
438/// # Values
439///
440/// The contents of a bucket, especially their representation and serialization, depend on the
441/// metric type:
442///
443/// - [Counters](BucketValue::Counter) store a single value, serialized as floating point.
444/// - [Distributions](MetricType::Distribution) and [sets](MetricType::Set) store the full set of
445/// reported values.
446/// - [Gauges](BucketValue::Gauge) store a snapshot of reported values, see [`GaugeValue`].
447///
448/// # Submission Protocol
449///
450/// ```text
451/// <name>[@unit]:<value>[:<value>...]|<type>[|#<tag_key>:<tag_value>,<tag>][|T<timestamp>]
452/// ```
453///
454/// See the [field documentation](Bucket#fields) for more information on the components. An example
455/// submission looks like this:
456///
457/// ```text
458#[doc = include_str!("../tests/fixtures/buckets.statsd.txt")]
459/// ```
460///
461/// To parse a submission payload, use [`Bucket::parse_all`].
462///
463/// # JSON Representation
464///
465/// Alternatively to the submission protocol, metrics can be represented as structured data in JSON.
466/// The data type of the `value` field is determined by the metric type.
467///
468/// In addition to the submission protocol, buckets have a required [`width`](Self::width) field in
469/// their JSON representation.
470///
471/// ```json
472#[doc = include_str!("../tests/fixtures/buckets.json")]
473/// ```
474///
475/// To parse a JSON payload, use [`serde_json`].
476///
477/// # Hashing of Sets
478///
479/// Set values can be specified as strings in the submission protocol. They are always hashed
480/// into a 32-bit value and the original value is dropped. If the submission protocol contains a
481/// 32-bit integer, it will be used directly, instead.
482///
483/// **Example**:
484///
485/// ```text
486#[doc = include_str!("../tests/fixtures/set.statsd.txt")]
487/// ```
488///
489/// The above submission is represented as:
490///
491/// ```json
492#[doc = include_str!("../tests/fixtures/set.json")]
493/// ```
494#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
495pub struct Bucket {
496 /// The start time of the bucket's time window.
497 ///
498 /// If a timestamp is not supplied as part of the submission payload, the default timestamp
499 /// supplied to [`Bucket::parse`] or [`Bucket::parse_all`] is associated with the metric. It is
500 /// then aligned with the aggregation window.
501 ///
502 /// # Statsd Format
503 ///
504 /// In statsd, timestamps are part of the `|`-separated list following values. Timestamps start
505 /// with the literal character `'T'` followed by the UNIX timestamp.
506 ///
507 /// The timestamp must be a positive integer in decimal notation representing the value of the
508 /// UNIX timestamp.
509 ///
510 /// # Example
511 ///
512 /// ```text
513 /// endpoint.hits:1|c|T1615889440
514 /// ```
515 pub timestamp: UnixTimestamp,
516
517 /// The length of the time window in seconds.
518 ///
519 /// To initialize a new bucket, choose `0` as width. Once the bucket is tracked by Relay's
520 /// aggregator, the width is aligned with configuration for the namespace and the timestamp is
521 /// adjusted accordingly.
522 ///
523 /// # Statsd Format
524 ///
525 /// Specifying the bucket width in statsd is not supported.
526 pub width: u64,
527
528 /// The name of the metric in MRI (metric resource identifier) format.
529 ///
530 /// MRIs have the format `<type>:<ns>/<name>@<unit>`. See [`MetricResourceIdentifier`] for
531 /// information on fields and representations.
532 ///
533 /// # Statsd Format
534 ///
535 /// MRIs are sent in a more relaxed format: `<namespace>/<name>[@unit]`. The value type is not
536 /// part of the metric name.
537 ///
538 /// Namespaces and units must consist of ASCII characters and match the regular expression
539 /// `/\w+/`. The name component of MRIs consist of unicode characters and must match the
540 /// regular expression `/\w[\w\-.]*/`. Note that the name must begin with a letter.
541 ///
542 /// Per convention, dots separate metric names into components, where the leading components are
543 /// considered namespaces and the final component is the name of the metric within its
544 /// namespace.
545 ///
546 /// # Examples
547 ///
548 /// ```text
549 /// transactions/endpoint.hits:1|c
550 /// transactions/endpoint.duration@millisecond:21.5|d
551 /// ```
552 pub name: MetricName,
553
554 /// The type and aggregated values of this bucket.
555 ///
556 /// Buckets support multiple values that are aggregated and can be accessed using a range of
557 /// aggregation functions depending on the value type. While always a variable number of values
558 /// can be sent in, some aggregations reduce the raw values to a fixed set of aggregates.
559 ///
560 /// See [`BucketValue`] for more examples and semantics.
561 ///
562 /// # Statsd Payload
563 ///
564 /// The bucket value and its type are specified in separate fields following the metric name in
565 /// the format: `<name>:<value>|<type>`. Values must be base-10 floating point numbers with
566 /// optional decimal places.
567 ///
568 /// It is possible to pack multiple values into a single datagram, but note that the type and
569 /// the value representation must match for this. Refer to the [`BucketValue`] docs for more
570 /// examples.
571 ///
572 /// # Example
573 ///
574 /// ```text
575 /// endpoint.hits:21|c
576 /// endpoint.hits:4.5|c
577 /// ```
578 #[serde(flatten)]
579 pub value: BucketValue,
580
581 /// A list of tags adding dimensions to the metric for filtering and aggregation.
582 ///
583 /// Tags allow to compute separate aggregates to filter or group metric values by any number of
584 /// dimensions. Tags consist of a unique tag key and one associated value. For tags with missing
585 /// values, an empty `""` value is assumed at query time.
586 ///
587 /// # Statsd Format
588 ///
589 /// Tags are preceded with a hash `#` and specified in a comma (`,`) separated list. Each tag
590 /// can either be a tag name, or a `name:value` combination. Tags are optional and can be
591 /// omitted.
592 ///
593 /// Tag keys are restricted to ASCII characters and must match the regular expression
594 /// `/[\w\-.\/]+/`.
595 ///
596 /// Tag values can contain unicode characters with the following escaping rules:
597 /// - Tab is escaped as `\t`.
598 /// - Carriage return is escaped as `\r`.
599 /// - Line feed is escaped as `\n`.
600 /// - Backslash is escaped as `\\`.
601 /// - Commas and pipes are given unicode escapes in the form `\u{2c}` and `\u{7c}`,
602 /// respectively.
603 ///
604 /// # Example
605 ///
606 /// ```text
607 /// endpoint.hits:1|c|#route:user_index,environment:production,release:1.4.0
608 /// ```
609 #[serde(default, skip_serializing_if = "MetricTags::is_empty")]
610 pub tags: MetricTags,
611
612 /// Relay internal metadata for a metric bucket.
613 ///
614 /// The metadata contains meta information about the metric bucket itself,
615 /// for example how many this bucket has been aggregated in total.
616 #[serde(default, skip_serializing_if = "BucketMetadata::is_default")]
617 pub metadata: BucketMetadata,
618}
619
620impl Bucket {
621 /// Parses a statsd-compatible payload.
622 ///
623 /// ```text
624 /// [<ns>/]<name>[@<unit>]:<value>|<type>[|#<tags>]`
625 /// ```
626 fn parse_str(string: &str, timestamp: UnixTimestamp) -> Option<Self> {
627 let mut components = string.split('|');
628
629 let (mri_str, values_str) = components.next()?.split_once(':')?;
630 let ty = components.next().and_then(|s| s.parse().ok())?;
631
632 let mri = MetricResourceIdentifier::parse_with_type(mri_str, ty).ok()?;
633 let value = match ty {
634 MetricType::Counter => BucketValue::Counter(parse_counter(values_str)?),
635 MetricType::Distribution => BucketValue::Distribution(parse_distribution(values_str)?),
636 MetricType::Set => BucketValue::Set(parse_set(values_str)?),
637 MetricType::Gauge => BucketValue::Gauge(parse_gauge(values_str)?),
638 };
639
640 let mut bucket = Bucket {
641 timestamp,
642 width: 0,
643 name: mri.to_string().into(),
644 value,
645 tags: Default::default(),
646 metadata: Default::default(),
647 };
648
649 for component in components {
650 match component.chars().next() {
651 Some('#') => {
652 bucket.tags = parse_tags(component.get(1..)?)?;
653 }
654 Some('T') => {
655 bucket.timestamp = parse_timestamp(component.get(1..)?)?;
656 }
657 _ => (),
658 }
659 }
660
661 Some(bucket)
662 }
663
664 /// Parses a single metric aggregate from the raw protocol.
665 ///
666 /// See the [`Bucket`] for more information on the protocol.
667 ///
668 /// # Example
669 ///
670 /// ```
671 /// use relay_metrics::{Bucket, UnixTimestamp};
672 ///
673 /// let bucket = Bucket::parse(b"transactions/response_time@millisecond:57|d", UnixTimestamp::now())
674 /// .expect("metric should parse");
675 /// ```
676 pub fn parse(slice: &[u8], timestamp: UnixTimestamp) -> Result<Self, ParseMetricError> {
677 let string = std::str::from_utf8(slice).map_err(|_| ParseMetricError)?;
678 Self::parse_str(string, timestamp).ok_or(ParseMetricError)
679 }
680
681 /// Parses a set of metric aggregates from the raw protocol.
682 ///
683 /// Returns a metric result for each line in `slice`, ignoring empty lines. Both UNIX newlines
684 /// (`\n`) and Windows newlines (`\r\n`) are supported.
685 ///
686 /// It is possible to continue consuming the iterator after `Err` is yielded.
687 ///
688 /// See [`Bucket`] for more information on the protocol.
689 ///
690 /// # Example
691 ///
692 /// ```
693 /// use relay_metrics::{Bucket, UnixTimestamp};
694 ///
695 /// let data = br#"
696 /// transactions/endpoint.response_time@millisecond:57|d
697 /// transactions/endpoint.hits:1|c
698 /// "#;
699 ///
700 /// for metric_result in Bucket::parse_all(data, UnixTimestamp::now()) {
701 /// let bucket = metric_result.expect("metric should parse");
702 /// println!("Metric {}: {:?}", bucket.name, bucket.value);
703 /// }
704 /// ```
705 pub fn parse_all(slice: &[u8], timestamp: UnixTimestamp) -> ParseBuckets<'_> {
706 ParseBuckets { slice, timestamp }
707 }
708
709 /// Returns the value of the specified tag if it exists.
710 pub fn tag(&self, name: &str) -> Option<&str> {
711 self.tags.get(name).map(|s| s.as_str())
712 }
713
714 /// Removes the value of the specified tag.
715 ///
716 /// If the tag exists, the removed value is returned.
717 pub fn remove_tag(&mut self, name: &str) -> Option<String> {
718 self.tags.remove(name)
719 }
720}
721
722/// Relay internal metadata for a metric bucket.
723#[derive(Clone, Copy, Debug, PartialEq, Deserialize, Serialize)]
724pub struct BucketMetadata {
725 /// How many times the bucket was merged.
726 ///
727 /// Creating a new bucket is the first merge.
728 /// Merging two buckets sums the amount of merges.
729 ///
730 /// For example: Merging two un-merged buckets will yield a total
731 /// of `2` merges.
732 ///
733 /// Due to how Relay aggregates metrics and later splits them into multiple
734 /// buckets again, the amount of merges can be zero.
735 /// When splitting a bucket the total volume of the bucket may only be attributed
736 /// to one part or distributed across the resulting buckets, in either case
737 /// values of `0` are possible.
738 pub merges: u32,
739
740 /// Received timestamp of the first metric in this bucket.
741 ///
742 /// This field should be set to the time in which the first metric of a specific bucket was
743 /// received in the outermost internal Relay.
744 pub received_at: Option<UnixTimestamp>,
745
746 /// Is `true` if this metric was extracted from a sampled/indexed envelope item.
747 ///
748 /// The final dynamic sampling decision is always made in processing Relays.
749 /// If a metric was extracted from an item which is sampled (i.e. retained by dynamic sampling), this flag is `true`.
750 ///
751 /// Since these metrics from samples carry additional information, e.g. they don't
752 /// require rate limiting since the sample they've been extracted from was already
753 /// rate limited, this flag must be included in the aggregation key when aggregation buckets.
754 #[serde(skip)]
755 pub extracted_from_indexed: bool,
756}
757
758impl BucketMetadata {
759 /// Creates a fresh metadata instance.
760 ///
761 /// The new metadata is initialized with `1` merge and a given `received_at` timestamp.
762 pub fn new(received_at: UnixTimestamp) -> Self {
763 Self {
764 merges: 1,
765 received_at: Some(received_at),
766 extracted_from_indexed: false,
767 }
768 }
769
770 /// Whether the metadata does not contain more information than the default.
771 pub fn is_default(&self) -> bool {
772 &Self::default() == self
773 }
774
775 /// Merges another metadata object into the current one.
776 pub fn merge(&mut self, other: Self) {
777 self.merges = self.merges.saturating_add(other.merges);
778 self.received_at = match (self.received_at, other.received_at) {
779 (Some(received_at), None) => Some(received_at),
780 (None, Some(received_at)) => Some(received_at),
781 (left, right) => left.min(right),
782 };
783 }
784}
785
786impl Default for BucketMetadata {
787 fn default() -> Self {
788 Self {
789 merges: 1,
790 received_at: None,
791 extracted_from_indexed: false,
792 }
793 }
794}
795
796/// Iterator over parsed metrics returned from [`Bucket::parse_all`].
797#[derive(Clone, Debug)]
798pub struct ParseBuckets<'a> {
799 slice: &'a [u8],
800 timestamp: UnixTimestamp,
801}
802
803impl Default for ParseBuckets<'_> {
804 fn default() -> Self {
805 Self {
806 slice: &[],
807 // The timestamp will never be returned.
808 timestamp: UnixTimestamp::from_secs(4711),
809 }
810 }
811}
812
813impl Iterator for ParseBuckets<'_> {
814 type Item = Result<Bucket, ParseMetricError>;
815
816 fn next(&mut self) -> Option<Self::Item> {
817 loop {
818 if self.slice.is_empty() {
819 return None;
820 }
821
822 let mut split = self.slice.splitn(2, |&b| b == b'\n');
823 let current = split.next()?;
824 self.slice = split.next().unwrap_or_default();
825
826 let string = match std::str::from_utf8(current) {
827 Ok(string) => string.strip_suffix('\r').unwrap_or(string),
828 Err(_) => return Some(Err(ParseMetricError)),
829 };
830
831 if !string.is_empty() {
832 return Some(Bucket::parse_str(string, self.timestamp).ok_or(ParseMetricError));
833 }
834 }
835 }
836}
837
838impl FusedIterator for ParseBuckets<'_> {}
839
840#[cfg(test)]
841mod tests {
842 use similar_asserts::assert_eq;
843
844 use crate::protocol::{DurationUnit, MetricUnit};
845
846 use super::*;
847
848 #[test]
849 fn test_distribution_value_size() {
850 // DistributionValue uses a SmallVec internally to prevent an additional allocation and
851 // indirection in cases where it needs to store only a small number of items. This is
852 // enabled by a comparably large `GaugeValue`, which stores five atoms. Ensure that the
853 // `DistributionValue`'s size does not exceed that of `GaugeValue`.
854 assert!(
855 std::mem::size_of::<DistributionValue>() <= std::mem::size_of::<GaugeValue>(),
856 "distribution value should not exceed gauge {}",
857 std::mem::size_of::<DistributionValue>()
858 );
859 }
860
861 #[test]
862 fn test_bucket_value_merge_counter() {
863 let mut value = BucketValue::Counter(42.into());
864 value.merge(BucketValue::Counter(43.into())).unwrap();
865 assert_eq!(value, BucketValue::Counter(85.into()));
866 }
867
868 #[test]
869 fn test_bucket_value_merge_distribution() {
870 let mut value = BucketValue::Distribution(dist![1, 2, 3]);
871 value.merge(BucketValue::Distribution(dist![2, 4])).unwrap();
872 assert_eq!(value, BucketValue::Distribution(dist![1, 2, 3, 2, 4]));
873 }
874
875 #[test]
876 fn test_bucket_value_merge_set() {
877 let mut value = BucketValue::Set(vec![1, 2].into_iter().collect());
878 value.merge(BucketValue::Set([2, 3].into())).unwrap();
879 assert_eq!(value, BucketValue::Set(vec![1, 2, 3].into_iter().collect()));
880 }
881
882 #[test]
883 fn test_bucket_value_merge_gauge() {
884 let mut value = BucketValue::Gauge(GaugeValue::single(42.into()));
885 value.merge(BucketValue::gauge(43.into())).unwrap();
886
887 assert_eq!(
888 value,
889 BucketValue::Gauge(GaugeValue {
890 last: 43.into(),
891 min: 42.into(),
892 max: 43.into(),
893 sum: 85.into(),
894 count: 2,
895 })
896 );
897 }
898
899 #[test]
900 fn test_parse_garbage() {
901 let s = "x23-408j17z4232@#34d\nc3456y7^😎";
902 let timestamp = UnixTimestamp::from_secs(4711);
903 let result = Bucket::parse(s.as_bytes(), timestamp);
904 assert!(result.is_err());
905 }
906
907 #[test]
908 fn test_parse_counter() {
909 let s = "spans/foo:42|c";
910 let timestamp = UnixTimestamp::from_secs(4711);
911 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
912 insta::assert_debug_snapshot!(metric, @r#"
913 Bucket {
914 timestamp: UnixTimestamp(4711),
915 width: 0,
916 name: MetricName(
917 "c:spans/foo@none",
918 ),
919 value: Counter(
920 42.0,
921 ),
922 tags: {},
923 metadata: BucketMetadata {
924 merges: 1,
925 received_at: None,
926 extracted_from_indexed: false,
927 },
928 }
929 "#);
930 }
931
932 #[test]
933 fn test_parse_counter_packed() {
934 let s = "spans/foo:42:17:21|c";
935 let timestamp = UnixTimestamp::from_secs(4711);
936 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
937 assert_eq!(metric.value, BucketValue::Counter(80.into()));
938 }
939
940 #[test]
941 fn test_parse_distribution() {
942 let s = "spans/foo:17.5|d";
943 let timestamp = UnixTimestamp::from_secs(4711);
944 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
945 insta::assert_debug_snapshot!(metric, @r#"
946 Bucket {
947 timestamp: UnixTimestamp(4711),
948 width: 0,
949 name: MetricName(
950 "d:spans/foo@none",
951 ),
952 value: Distribution(
953 [
954 17.5,
955 ],
956 ),
957 tags: {},
958 metadata: BucketMetadata {
959 merges: 1,
960 received_at: None,
961 extracted_from_indexed: false,
962 },
963 }
964 "#);
965 }
966
967 #[test]
968 fn test_parse_distribution_packed() {
969 let s = "transactions/foo:17.5:21.9:42.7|d";
970 let timestamp = UnixTimestamp::from_secs(4711);
971 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
972 assert_eq!(
973 metric.value,
974 BucketValue::Distribution(dist![
975 FiniteF64::new(17.5).unwrap(),
976 FiniteF64::new(21.9).unwrap(),
977 FiniteF64::new(42.7).unwrap()
978 ])
979 );
980 }
981
982 #[test]
983 fn test_parse_histogram() {
984 let s = "transactions/foo:17.5|h"; // common alias for distribution
985 let timestamp = UnixTimestamp::from_secs(4711);
986 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
987 assert_eq!(
988 metric.value,
989 BucketValue::Distribution(dist![FiniteF64::new(17.5).unwrap()])
990 );
991 }
992
993 #[test]
994 fn test_parse_set() {
995 let s = "spans/foo:4267882815|s";
996 let timestamp = UnixTimestamp::from_secs(4711);
997 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
998 insta::assert_debug_snapshot!(metric, @r#"
999 Bucket {
1000 timestamp: UnixTimestamp(4711),
1001 width: 0,
1002 name: MetricName(
1003 "s:spans/foo@none",
1004 ),
1005 value: Set(
1006 {
1007 4267882815,
1008 },
1009 ),
1010 tags: {},
1011 metadata: BucketMetadata {
1012 merges: 1,
1013 received_at: None,
1014 extracted_from_indexed: false,
1015 },
1016 }
1017 "#);
1018 }
1019
1020 #[test]
1021 fn test_parse_set_hashed() {
1022 let s = "transactions/foo:e2546e4c-ecd0-43ad-ae27-87960e57a658|s";
1023 let timestamp = UnixTimestamp::from_secs(4711);
1024 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1025 assert_eq!(metric.value, BucketValue::Set([4267882815].into()));
1026 }
1027
1028 #[test]
1029 fn test_parse_set_hashed_packed() {
1030 let s = "transactions/foo:e2546e4c-ecd0-43ad-ae27-87960e57a658:00449b66-d91f-4fb8-b324-4c8bdf2499f6|s";
1031 let timestamp = UnixTimestamp::from_secs(4711);
1032 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1033 assert_eq!(
1034 metric.value,
1035 BucketValue::Set([181348692, 4267882815].into())
1036 );
1037 }
1038
1039 #[test]
1040 fn test_parse_set_packed() {
1041 let s = "transactions/foo:3182887624:4267882815|s";
1042 let timestamp = UnixTimestamp::from_secs(4711);
1043 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1044 assert_eq!(
1045 metric.value,
1046 BucketValue::Set([3182887624, 4267882815].into())
1047 )
1048 }
1049
1050 #[test]
1051 fn test_parse_gauge() {
1052 let s = "spans/foo:42|g";
1053 let timestamp = UnixTimestamp::from_secs(4711);
1054 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1055 insta::assert_debug_snapshot!(metric, @r#"
1056 Bucket {
1057 timestamp: UnixTimestamp(4711),
1058 width: 0,
1059 name: MetricName(
1060 "g:spans/foo@none",
1061 ),
1062 value: Gauge(
1063 GaugeValue {
1064 last: 42.0,
1065 min: 42.0,
1066 max: 42.0,
1067 sum: 42.0,
1068 count: 1,
1069 },
1070 ),
1071 tags: {},
1072 metadata: BucketMetadata {
1073 merges: 1,
1074 received_at: None,
1075 extracted_from_indexed: false,
1076 },
1077 }
1078 "#);
1079 }
1080
1081 #[test]
1082 fn test_parse_gauge_packed() {
1083 let s = "spans/foo:25:17:42:220:85|g";
1084 let timestamp = UnixTimestamp::from_secs(4711);
1085 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1086 insta::assert_debug_snapshot!(metric, @r#"
1087 Bucket {
1088 timestamp: UnixTimestamp(4711),
1089 width: 0,
1090 name: MetricName(
1091 "g:spans/foo@none",
1092 ),
1093 value: Gauge(
1094 GaugeValue {
1095 last: 25.0,
1096 min: 17.0,
1097 max: 42.0,
1098 sum: 220.0,
1099 count: 85,
1100 },
1101 ),
1102 tags: {},
1103 metadata: BucketMetadata {
1104 merges: 1,
1105 received_at: None,
1106 extracted_from_indexed: false,
1107 },
1108 }
1109 "#);
1110 }
1111
1112 #[test]
1113 fn test_parse_missing_namespace() {
1114 let s = "foo:42|c";
1115 let timestamp = UnixTimestamp::from_secs(4711);
1116 let metric = Bucket::parse(s.as_bytes(), timestamp);
1117 assert!(metric.is_err());
1118 }
1119
1120 #[test]
1121 fn test_parse_unit() {
1122 let s = "transactions/foo@second:17.5|d";
1123 let timestamp = UnixTimestamp::from_secs(4711);
1124 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1125 let mri = MetricResourceIdentifier::parse(&metric.name).unwrap();
1126 assert_eq!(mri.unit, MetricUnit::Duration(DurationUnit::Second));
1127 }
1128
1129 #[test]
1130 fn test_parse_unit_regression() {
1131 let s = "transactions/foo@s:17.5|d";
1132 let timestamp = UnixTimestamp::from_secs(4711);
1133 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1134 let mri = MetricResourceIdentifier::parse(&metric.name).unwrap();
1135 assert_eq!(mri.unit, MetricUnit::Duration(DurationUnit::Second));
1136 }
1137
1138 #[test]
1139 fn test_parse_tags() {
1140 let s = "transactions/foo:17.5|d|#foo,bar:baz";
1141 let timestamp = UnixTimestamp::from_secs(4711);
1142 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1143 insta::assert_debug_snapshot!(metric.tags, @r###"
1144 {
1145 "bar": "baz",
1146 "foo": "",
1147 }
1148 "###);
1149 }
1150
1151 #[test]
1152 fn test_parse_tags_escaped() {
1153 let s = "transactions/foo:17.5|d|#foo:😅\\u{2c}🚀";
1154 let timestamp = UnixTimestamp::from_secs(4711);
1155 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1156 insta::assert_debug_snapshot!(metric.tags, @r###"
1157 {
1158 "foo": "😅,🚀",
1159 }
1160 "###);
1161 }
1162
1163 #[test]
1164 fn test_parse_timestamp() {
1165 let s = "transactions/foo:17.5|d|T1615889449";
1166 let timestamp = UnixTimestamp::from_secs(4711);
1167 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1168 assert_eq!(metric.timestamp, UnixTimestamp::from_secs(1615889449));
1169 }
1170
1171 #[test]
1172 fn test_parse_sample_rate() {
1173 // Sample rate should be ignored
1174 let s = "transactions/foo:17.5|d|@0.1";
1175 let timestamp = UnixTimestamp::from_secs(4711);
1176 Bucket::parse(s.as_bytes(), timestamp).unwrap();
1177 }
1178
1179 #[test]
1180 fn test_parse_invalid_name() {
1181 let s = "transactions/foo#bar:42|c";
1182 let timestamp = UnixTimestamp::from_secs(4711);
1183 let metric = Bucket::parse(s.as_bytes(), timestamp).unwrap();
1184 assert_eq!(metric.name.as_ref(), "c:transactions/foo_bar@none");
1185 }
1186
1187 #[test]
1188 fn test_parse_empty_name() {
1189 let s = ":42|c";
1190 let timestamp = UnixTimestamp::from_secs(4711);
1191 let metric = Bucket::parse(s.as_bytes(), timestamp);
1192 assert!(metric.is_err());
1193 }
1194
1195 #[test]
1196 fn test_parse_invalid_name_with_leading_digit() {
1197 let s = "64bit:42|c";
1198 let timestamp = UnixTimestamp::from_secs(4711);
1199 let metric = Bucket::parse(s.as_bytes(), timestamp);
1200 assert!(metric.is_err());
1201 }
1202
1203 #[test]
1204 fn test_parse_all() {
1205 let s = "transactions/foo:42|c\nspans/bar:17|c";
1206 let timestamp = UnixTimestamp::from_secs(4711);
1207
1208 let metrics: Vec<Bucket> = Bucket::parse_all(s.as_bytes(), timestamp)
1209 .collect::<Result<_, _>>()
1210 .unwrap();
1211
1212 assert_eq!(metrics.len(), 2);
1213 }
1214
1215 #[test]
1216 fn test_parse_all_crlf() {
1217 let s = "transactions/foo:42|c\r\nspans/bar:17|c";
1218 let timestamp = UnixTimestamp::from_secs(4711);
1219
1220 let metrics: Vec<Bucket> = Bucket::parse_all(s.as_bytes(), timestamp)
1221 .collect::<Result<_, _>>()
1222 .unwrap();
1223
1224 assert_eq!(metrics.len(), 2);
1225 }
1226
1227 #[test]
1228 fn test_parse_all_empty_lines() {
1229 let s = "transactions/foo:42|c\n\n\nspans/bar:17|c";
1230 let timestamp = UnixTimestamp::from_secs(4711);
1231
1232 let metric_count = Bucket::parse_all(s.as_bytes(), timestamp).count();
1233 assert_eq!(metric_count, 2);
1234 }
1235
1236 #[test]
1237 fn test_parse_all_trailing() {
1238 let s = "transactions/foo:42|c\nspans/bar:17|c\n";
1239 let timestamp = UnixTimestamp::from_secs(4711);
1240
1241 let metric_count = Bucket::parse_all(s.as_bytes(), timestamp).count();
1242 assert_eq!(metric_count, 2);
1243 }
1244
1245 #[test]
1246 fn test_metrics_docs() {
1247 let text = include_str!("../tests/fixtures/buckets.statsd.txt").trim_end();
1248 let json = include_str!("../tests/fixtures/buckets.json").trim_end();
1249
1250 let timestamp = UnixTimestamp::from_secs(0);
1251 let statsd_metrics = Bucket::parse_all(text.as_bytes(), timestamp)
1252 .collect::<Result<Vec<_>, _>>()
1253 .unwrap();
1254
1255 let json_metrics: Vec<Bucket> = serde_json::from_str(json).unwrap();
1256
1257 assert_eq!(statsd_metrics, json_metrics);
1258 }
1259
1260 #[test]
1261 fn test_set_docs() {
1262 let text = include_str!("../tests/fixtures/set.statsd.txt").trim_end();
1263 let json = include_str!("../tests/fixtures/set.json").trim_end();
1264
1265 let timestamp = UnixTimestamp::from_secs(1615889449);
1266 let statsd_metric = Bucket::parse(text.as_bytes(), timestamp).unwrap();
1267 let json_metric: Bucket = serde_json::from_str(json).unwrap();
1268
1269 assert_eq!(statsd_metric, json_metric);
1270 }
1271
1272 #[test]
1273 fn test_parse_buckets() {
1274 let json = r#"[
1275 {
1276 "name": "endpoint.response_time",
1277 "unit": "millisecond",
1278 "value": [36, 49, 57, 68],
1279 "type": "d",
1280 "timestamp": 1615889440,
1281 "width": 10,
1282 "tags": {
1283 "route": "user_index"
1284 },
1285 "metadata": {
1286 "merges": 1,
1287 "received_at": 1615889440
1288 }
1289 }
1290 ]"#;
1291
1292 let buckets = serde_json::from_str::<Vec<Bucket>>(json).unwrap();
1293
1294 insta::assert_debug_snapshot!(buckets, @r###"
1295 [
1296 Bucket {
1297 timestamp: UnixTimestamp(1615889440),
1298 width: 10,
1299 name: MetricName(
1300 "endpoint.response_time",
1301 ),
1302 value: Distribution(
1303 [
1304 36.0,
1305 49.0,
1306 57.0,
1307 68.0,
1308 ],
1309 ),
1310 tags: {
1311 "route": "user_index",
1312 },
1313 metadata: BucketMetadata {
1314 merges: 1,
1315 received_at: Some(
1316 UnixTimestamp(1615889440),
1317 ),
1318 extracted_from_indexed: false,
1319 },
1320 },
1321 ]
1322 "###);
1323 }
1324
1325 #[test]
1326 fn test_parse_bucket_defaults() {
1327 let json = r#"[
1328 {
1329 "name": "endpoint.hits",
1330 "value": 4,
1331 "type": "c",
1332 "timestamp": 1615889440,
1333 "width": 10,
1334 "metadata": {
1335 "merges": 1,
1336 "received_at": 1615889440
1337 }
1338 }
1339 ]"#;
1340
1341 let buckets = serde_json::from_str::<Vec<Bucket>>(json).unwrap();
1342
1343 insta::assert_debug_snapshot!(buckets, @r###"
1344 [
1345 Bucket {
1346 timestamp: UnixTimestamp(1615889440),
1347 width: 10,
1348 name: MetricName(
1349 "endpoint.hits",
1350 ),
1351 value: Counter(
1352 4.0,
1353 ),
1354 tags: {},
1355 metadata: BucketMetadata {
1356 merges: 1,
1357 received_at: Some(
1358 UnixTimestamp(1615889440),
1359 ),
1360 extracted_from_indexed: false,
1361 },
1362 },
1363 ]
1364 "###);
1365 }
1366
1367 #[test]
1368 fn test_buckets_roundtrip() {
1369 let json = r#"[
1370 {
1371 "timestamp": 1615889440,
1372 "width": 10,
1373 "name": "endpoint.response_time",
1374 "type": "d",
1375 "value": [
1376 36.0,
1377 49.0,
1378 57.0,
1379 68.0
1380 ],
1381 "tags": {
1382 "route": "user_index"
1383 }
1384 },
1385 {
1386 "timestamp": 1615889440,
1387 "width": 10,
1388 "name": "endpoint.hits",
1389 "type": "c",
1390 "value": 4.0,
1391 "tags": {
1392 "route": "user_index"
1393 }
1394 },
1395 {
1396 "timestamp": 1615889440,
1397 "width": 10,
1398 "name": "endpoint.parallel_requests",
1399 "type": "g",
1400 "value": {
1401 "last": 25.0,
1402 "min": 17.0,
1403 "max": 42.0,
1404 "sum": 2210.0,
1405 "count": 85
1406 }
1407 },
1408 {
1409 "timestamp": 1615889440,
1410 "width": 10,
1411 "name": "endpoint.users",
1412 "type": "s",
1413 "value": [
1414 3182887624,
1415 4267882815
1416 ],
1417 "tags": {
1418 "route": "user_index"
1419 }
1420 }
1421]"#;
1422
1423 let buckets = serde_json::from_str::<Vec<Bucket>>(json).unwrap();
1424 let serialized = serde_json::to_string_pretty(&buckets).unwrap();
1425 assert_eq!(json, serialized);
1426 }
1427
1428 #[test]
1429 fn test_bucket_docs_roundtrip() {
1430 let json = include_str!("../tests/fixtures/buckets.json")
1431 .trim_end()
1432 .replace("\r\n", "\n");
1433 let buckets = serde_json::from_str::<Vec<Bucket>>(&json).unwrap();
1434
1435 let serialized = serde_json::to_string_pretty(&buckets).unwrap();
1436 assert_eq!(json, serialized);
1437 }
1438
1439 #[test]
1440 fn test_bucket_metadata_merge() {
1441 let mut metadata = BucketMetadata::default();
1442
1443 let other_metadata = BucketMetadata::default();
1444 metadata.merge(other_metadata);
1445 assert_eq!(
1446 metadata,
1447 BucketMetadata {
1448 merges: 2,
1449 received_at: None,
1450 extracted_from_indexed: false,
1451 }
1452 );
1453
1454 let other_metadata = BucketMetadata::new(UnixTimestamp::from_secs(10));
1455 metadata.merge(other_metadata);
1456 assert_eq!(
1457 metadata,
1458 BucketMetadata {
1459 merges: 3,
1460 received_at: Some(UnixTimestamp::from_secs(10)),
1461 extracted_from_indexed: false,
1462 }
1463 );
1464
1465 let other_metadata = BucketMetadata::new(UnixTimestamp::from_secs(20));
1466 metadata.merge(other_metadata);
1467 assert_eq!(
1468 metadata,
1469 BucketMetadata {
1470 merges: 4,
1471 received_at: Some(UnixTimestamp::from_secs(10)),
1472 extracted_from_indexed: false,
1473 }
1474 );
1475 }
1476}