relay_metrics/
view.rs

1use relay_common::time::UnixTimestamp;
2use serde::Serialize;
3use serde::ser::{SerializeMap, SerializeSeq};
4
5use crate::{
6    BucketMetadata, CounterType, DistributionType, GaugeValue, MetricName, SetType, SetValue,
7};
8use relay_base_schema::metrics::MetricType;
9use std::collections::BTreeMap;
10use std::fmt;
11use std::ops::Range;
12
13use crate::BucketValue;
14use crate::bucket::Bucket;
15
16/// The fraction of size passed to [`BucketsView::by_size()`] at which buckets will be split. A value of
17/// `2` means that all buckets smaller than half of `metrics_max_batch_size` will be moved in their entirety,
18/// and buckets larger will be split up.
19const BUCKET_SPLIT_FACTOR: usize = 32;
20
21/// The base size of a serialized bucket in bytes.
22///
23/// This is the size of a bucket's fixed fields in JSON format, excluding the value and tags.
24const BUCKET_SIZE: usize = 50;
25
26/// The average size of values when serialized.
27const AVG_VALUE_SIZE: usize = 8;
28
29/// An internal type representing an index into a slice of buckets.
30///
31/// Note: the meaning of fields depends on the context of the index.
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33struct Index {
34    /// Index into the slice.
35    slice: usize,
36    /// Index into a bucket.
37    bucket: usize,
38}
39
40/// A view into a slice of metric buckets.
41///
42/// The view can be used to iterate over a large slice
43/// of metric data slicing even into the buckets themselves.
44///
45/// ```txt
46///                    Full View
47///  /---------------------------------------------\
48/// [[C:1], [C:12], [D:0, 1, 2, 3, 5, 5], [S:42, 75]]
49///  \--------------------------/  \---------------/
50///            View 1                    View 2
51/// ```
52///
53/// Iterating over a [`BucketsView`] yields [`BucketView`] items,
54/// only the first and last elements may be partial buckets.
55///
56/// In the above example `View 1` has a partial bucket at the end and
57/// `View 2` has a partial bucket in the beginning.
58///
59/// Using the above example, iterating over `View 1` yields the buckets:
60/// `[C:1], [C:12], [D:0, 1, 2, 3]`.
61#[derive(Clone, Copy)]
62pub struct BucketsView<T> {
63    /// The contained buckets.
64    inner: T,
65    /// Start index.
66    ///
67    /// - Slice index indicates bucket.
68    /// - Bucket index indicates offset in the selected bucket.
69    start: Index,
70    /// End index.
71    ///
72    /// - Slice index indicates exclusive end.
73    /// - Bucket index, indicates offset into the *next* bucket past the end.
74    end: Index,
75}
76
77impl<T> BucketsView<T>
78where
79    T: AsRef<[Bucket]>,
80{
81    /// Creates a new buckets view containing all data from the slice.
82    pub fn new(buckets: T) -> Self {
83        let len = buckets.as_ref().len();
84        Self {
85            inner: buckets,
86            start: Index {
87                slice: 0,
88                bucket: 0,
89            },
90            end: Index {
91                slice: len,
92                bucket: 0,
93            },
94        }
95    }
96
97    /// Returns the amount of partial or full buckets in the view.
98    pub fn len(&self) -> usize {
99        let mut len = self.end.slice - self.start.slice;
100        if self.end.bucket != 0 {
101            len += 1;
102        }
103        len
104    }
105
106    /// Returns whether the view contains any buckets.
107    pub fn is_empty(&self) -> bool {
108        self.len() == 0
109    }
110
111    /// Returns the same bucket view as a bucket view over a slice.
112    pub fn as_slice(&self) -> BucketsView<&[Bucket]> {
113        BucketsView {
114            inner: self.inner.as_ref(),
115            start: self.start,
116            end: self.end,
117        }
118    }
119
120    /// Iterator over all buckets in the view.
121    pub fn iter(&self) -> BucketsViewIter<'_> {
122        BucketsViewIter::new(self.inner.as_ref(), self.start, self.end)
123    }
124
125    /// Iterator which slices the source view into segments with an approximate size of `size_in_bytes`.
126    pub fn by_size(self, size_in_bytes: usize) -> BucketsViewBySizeIter<T> {
127        BucketsViewBySizeIter::new(self.inner, self.start, self.end, size_in_bytes)
128    }
129}
130
131impl<'a> From<&'a [Bucket]> for BucketsView<&'a [Bucket]> {
132    fn from(value: &'a [Bucket]) -> Self {
133        Self::new(value)
134    }
135}
136
137impl<'a> From<&'a Vec<Bucket>> for BucketsView<&'a [Bucket]> {
138    fn from(value: &'a Vec<Bucket>) -> Self {
139        Self::new(value.as_slice())
140    }
141}
142
143impl<T> fmt::Debug for BucketsView<T>
144where
145    T: AsRef<[Bucket]>,
146{
147    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
148        let contents = self.iter().collect::<Vec<_>>();
149        f.debug_tuple("BucketsView").field(&contents).finish()
150    }
151}
152
153impl<'a> IntoIterator for BucketsView<&'a [Bucket]> {
154    type Item = BucketView<'a>;
155    type IntoIter = BucketsViewIter<'a>;
156
157    fn into_iter(self) -> Self::IntoIter {
158        BucketsViewIter::new(self.inner, self.start, self.end)
159    }
160}
161
162impl<'a, T> IntoIterator for &'a BucketsView<T>
163where
164    T: AsRef<[Bucket]>,
165{
166    type Item = BucketView<'a>;
167    type IntoIter = BucketsViewIter<'a>;
168
169    fn into_iter(self) -> Self::IntoIter {
170        BucketsViewIter::new(self.inner.as_ref(), self.start, self.end)
171    }
172}
173
174/// Iterator yielding all items contained in a [`BucketsView`].
175///
176/// First and/or last item may be partial buckets.
177pub struct BucketsViewIter<'a> {
178    /// Source slice of buckets.
179    inner: &'a [Bucket],
180    /// Current index.
181    current: Index,
182    /// End index.
183    end: Index,
184}
185
186impl<'a> BucketsViewIter<'a> {
187    /// Creates a new iterator.
188    ///
189    /// Start and end must be valid indices or iterator may end early.
190    fn new(inner: &'a [Bucket], start: Index, end: Index) -> Self {
191        Self {
192            inner,
193            end,
194            current: start,
195        }
196    }
197}
198
199impl<'a> Iterator for BucketsViewIter<'a> {
200    type Item = BucketView<'a>;
201
202    fn next(&mut self) -> Option<Self::Item> {
203        // We reached the exact end, there is no sub-bucket index.
204        if self.current.slice == self.end.slice && self.end.bucket == 0 {
205            return None;
206        }
207        // We are way past, including sub-bucket offset.
208        if self.current.slice > self.end.slice {
209            return None;
210        }
211
212        // This doesn't overflow because the last bucket in the inner slice will always have a 0 bucket index.
213        debug_assert!(
214            self.current.slice < self.inner.len(),
215            "invariant violated, iterator pointing past the slice"
216        );
217        let next = self.inner.get(self.current.slice)?;
218
219        // Choose the bucket end, this will always be the full bucket except if it is the last.
220        let end = match self.current.slice == self.end.slice {
221            false => next.value.len(),
222            true => self.end.bucket,
223        };
224
225        let next = BucketView::new(next).select(self.current.bucket..end);
226        let Some(next) = next else {
227            debug_assert!(false, "invariant violated, invalid bucket split");
228            relay_log::error!(
229                "Internal invariant violated, invalid bucket split, dropping all remaining buckets."
230            );
231            return None;
232        };
233
234        // Even if the current Bucket was partial, the next one will be full,
235        // except if it is the last one.
236        self.current = Index {
237            slice: self.current.slice + 1,
238            bucket: 0,
239        };
240
241        Some(next)
242    }
243}
244
245/// Iterator slicing a [`BucketsView`] into smaller views constrained by a given size in bytes.
246///
247// See [`estimate_size`] for how the size of a bucket is calculated.
248pub struct BucketsViewBySizeIter<T> {
249    /// Source slice.
250    inner: T,
251    /// Current position in the slice.
252    current: Index,
253    /// Terminal position.
254    end: Index,
255    /// Maximum size in bytes of each slice.
256    max_size_bytes: usize,
257}
258
259impl<T> BucketsViewBySizeIter<T> {
260    /// Creates a new iterator.
261    ///
262    /// Start and end must be valid indices or iterator may end early.
263    fn new(inner: T, start: Index, end: Index, max_size_bytes: usize) -> Self {
264        Self {
265            inner,
266            end,
267            current: start,
268            max_size_bytes,
269        }
270    }
271}
272
273impl<T> Iterator for BucketsViewBySizeIter<T>
274where
275    T: AsRef<[Bucket]>,
276    T: Clone,
277{
278    type Item = BucketsView<T>;
279
280    fn next(&mut self) -> Option<Self::Item> {
281        let start = self.current;
282
283        let mut remaining_bytes = self.max_size_bytes;
284        loop {
285            // Make sure, we don't shoot past the end ...
286            if (self.current.slice > self.end.slice)
287                || (self.current.slice == self.end.slice && self.end.bucket == 0)
288            {
289                break;
290            }
291
292            let inner = self.inner.as_ref();
293            // Select next potential bucket,
294            // this should never overflow because `end` will never go past the slice and
295            // we just validated that current is constrained by end.
296            debug_assert!(
297                self.current.slice < inner.len(),
298                "invariant violated, iterator pointing past the slice"
299            );
300            let bucket = inner.get(self.current.slice)?;
301
302            // Selection should never fail, because either we select the entire range,
303            // or we previously already split the bucket, which means this range is good.
304            let bucket = BucketView::new(bucket).select(self.current.bucket..bucket.value.len());
305            let Some(bucket) = bucket else {
306                debug_assert!(false, "internal invariant violated, invalid bucket split");
307                relay_log::error!(
308                    "Internal invariant violated, invalid bucket split, dropping all remaining buckets."
309                );
310                return None;
311            };
312
313            match split(
314                &bucket,
315                remaining_bytes,
316                self.max_size_bytes / BUCKET_SPLIT_FACTOR,
317            ) {
318                SplitDecision::BucketFits(size) => {
319                    remaining_bytes -= size;
320                    self.current = Index {
321                        slice: self.current.slice + 1,
322                        bucket: 0,
323                    };
324                    continue;
325                }
326                SplitDecision::MoveToNextBatch => break,
327                SplitDecision::Split(n) => {
328                    self.current = Index {
329                        slice: self.current.slice,
330                        bucket: self.current.bucket + n,
331                    };
332                    break;
333                }
334            }
335        }
336
337        if start == self.current {
338            // Either no progress could be made (not enough space to fit a bucket),
339            // or we're done.
340            return None;
341        }
342
343        // Current is the current for the next batch now,
344        // which means, current is the end for this batch.
345        Some(BucketsView {
346            inner: self.inner.clone(),
347            start,
348            end: self.current,
349        })
350    }
351}
352
353impl<T> Serialize for BucketsView<T>
354where
355    T: AsRef<[Bucket]>,
356{
357    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
358    where
359        S: serde::Serializer,
360    {
361        let mut state = serializer.serialize_seq(Some(self.len()))?;
362
363        for bucket in self.iter() {
364            state.serialize_element(&bucket)?;
365        }
366
367        state.end()
368    }
369}
370
371/// A view into a metrics bucket. Sometimes also called a partial bucket.
372/// A view contains a subset of datapoints of the original bucket.
373///
374/// ```txt
375///             Full Bucket
376/// /-------------------------------\
377/// [0, 1, 2, 3, 5, 5, 5, 10, 11, 11]
378/// \----------------/\-------------/
379///       View 1          View 2
380/// ```
381///
382/// A view can be split again into multiple smaller views.
383#[derive(Clone)]
384pub struct BucketView<'a> {
385    /// The source bucket.
386    inner: &'a Bucket,
387    /// Non-empty and valid range into the bucket.
388    /// The full range is constrained by `0..bucket.value.len()`
389    range: Range<usize>,
390}
391
392impl<'a> BucketView<'a> {
393    /// Creates a new bucket view of a bucket.
394    ///
395    /// The resulting view contains the entire bucket.
396    pub fn new(bucket: &'a Bucket) -> Self {
397        Self {
398            inner: bucket,
399            range: 0..bucket.value.len(),
400        }
401    }
402
403    /// Timestamp of the bucket.
404    ///
405    /// See also: [`Bucket::timestamp`]
406    pub fn timestamp(&self) -> UnixTimestamp {
407        self.inner.timestamp
408    }
409
410    /// Width of the bucket.
411    ///
412    /// See also: [`Bucket::width`]
413    pub fn width(&self) -> u64 {
414        self.inner.width
415    }
416
417    /// Name of the bucket.
418    ///
419    /// See also: [`Bucket::name`]
420    pub fn name(&self) -> &'a MetricName {
421        &self.inner.name
422    }
423
424    /// Value of the bucket view.
425    pub fn value(&self) -> BucketViewValue<'a> {
426        match &self.inner.value {
427            BucketValue::Counter(c) => BucketViewValue::Counter(*c),
428            BucketValue::Distribution(d) => BucketViewValue::Distribution(&d[self.range.clone()]),
429            BucketValue::Set(s) => BucketViewValue::Set(SetView::new(s, self.range.clone())),
430            BucketValue::Gauge(g) => BucketViewValue::Gauge(*g),
431        }
432    }
433
434    /// Type of the value of the bucket view.
435    pub fn ty(&self) -> MetricType {
436        match &self.inner.value {
437            BucketValue::Counter(_) => MetricType::Counter,
438            BucketValue::Distribution(_) => MetricType::Distribution,
439            BucketValue::Set(_) => MetricType::Set,
440            BucketValue::Gauge(_) => MetricType::Gauge,
441        }
442    }
443
444    /// Name of the bucket.
445    ///
446    /// See also: [`Bucket::tags`]
447    pub fn tags(&self) -> &'a BTreeMap<String, String> {
448        &self.inner.tags
449    }
450
451    /// Returns the value of the specified tag if it exists.
452    ///
453    /// See also: [`Bucket::tag()`]
454    pub fn tag(&self, name: &str) -> Option<&'a str> {
455        self.inner.tag(name)
456    }
457
458    /// Returns the metadata for this bucket.
459    ///
460    /// The aggregation process of metadata is inheritly lossy, which means
461    /// some metadata, for example the amount of merges, can not be accurately split
462    /// or divided over multiple bucket views.
463    ///
464    /// To compensate for this only a bucket view which contains the start of a bucket
465    /// will yield this metadata, all other views created from the bucket return an
466    /// identity value. Merging all metadata from non-overlapping bucket views must
467    /// yield the same values as stored on the original bucket.
468    ///
469    /// This causes some problems when operations on partial buckets are fallible,
470    /// for example transmitting two bucket views in separate http requests.
471    /// To deal with this Relay needs to prevent the splitting of buckets in the first place,
472    /// by never not creating too large buckets via aggregation in the first place.
473    ///
474    /// See also: [`Bucket::metadata`].
475    pub fn metadata(&self) -> BucketMetadata {
476        let merges = if self.range.start == 0 {
477            self.inner.metadata.merges
478        } else {
479            0
480        };
481
482        BucketMetadata {
483            merges,
484            ..self.inner.metadata
485        }
486    }
487
488    /// Number of raw datapoints in this view.
489    ///
490    /// See also: [`BucketValue::len()`]
491    pub fn len(&self) -> usize {
492        self.range.len()
493    }
494
495    /// Returns `true` if this bucket view contains no values.
496    pub fn is_empty(&self) -> bool {
497        self.len() == 0
498    }
499
500    /// Selects a sub-view of the current view.
501    ///
502    /// Returns `None` when:
503    /// - the passed range is not contained in the current view.
504    /// - trying to split a counter or gauge bucket.
505    pub fn select(mut self, range: Range<usize>) -> Option<Self> {
506        if range.start < self.range.start || range.end > self.range.end {
507            return None;
508        }
509
510        // Make sure the bucket can be split, or the entire bucket range is passed.
511        if !self.can_split() && range != (0..self.inner.value.len()) {
512            return None;
513        }
514
515        self.range = range;
516        Some(self)
517    }
518
519    /// Estimates the number of bytes needed to serialize the bucket without value.
520    ///
521    /// Note that this does not match the exact size of the serialized payload. Instead, the size is
522    /// approximated through tags and a static overhead.
523    fn estimated_base_size(&self) -> usize {
524        BUCKET_SIZE + self.name().len() + crate::utils::tags_cost(self.tags())
525    }
526
527    /// Estimates the number of bytes needed to serialize the bucket.
528    ///
529    /// Note that this does not match the exact size of the serialized payload. Instead, the size is
530    /// approximated through the number of contained values, assuming an average size of serialized
531    /// values.
532    pub fn estimated_size(&self) -> usize {
533        self.estimated_base_size() + self.len() * AVG_VALUE_SIZE
534    }
535
536    /// Calculates a split for this bucket if its estimated serialization size exceeds a threshold.
537    ///
538    /// There are three possible return values:
539    ///  - `(Some, None)` if the bucket fits entirely into the size budget. There is no split.
540    ///  - `(None, Some)` if the size budget cannot even hold the bucket name and tags. There is no
541    ///    split, the entire bucket is moved.
542    ///  - `(Some, Some)` if the bucket fits partially. Remaining values are moved into a new bucket
543    ///    with all other information cloned.
544    ///
545    /// This is an approximate function. The bucket is not actually serialized, but rather its
546    /// footprint is estimated through the number of data points contained. See
547    /// [`estimated_size`](Self::estimated_size) for more information.
548    pub fn split(self, size: usize, max_size: Option<usize>) -> (Option<Self>, Option<Self>) {
549        match split(&self, size, max_size.unwrap_or(0) / BUCKET_SPLIT_FACTOR) {
550            SplitDecision::BucketFits(_) => (Some(self), None),
551            SplitDecision::MoveToNextBatch => (None, Some(self)),
552            SplitDecision::Split(n) => {
553                let Range { start, end } = self.range;
554                let at = start + n;
555                (self.clone().select(start..at), self.select(at..end))
556            }
557        }
558    }
559
560    /// Whether the bucket can be split into multiple.
561    ///
562    /// Only set and distribution buckets can be split.
563    fn can_split(&self) -> bool {
564        matches!(
565            self.inner.value,
566            BucketValue::Distribution(_) | BucketValue::Set(_)
567        )
568    }
569
570    /// Returns `true` when this view contains the entire bucket.
571    fn is_full_bucket(&self) -> bool {
572        self.range.start == 0 && self.range.end == self.inner.value.len()
573    }
574}
575
576impl fmt::Debug for BucketView<'_> {
577    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
578        f.debug_struct("BucketView")
579            .field("timestamp", &self.inner.timestamp)
580            .field("width", &self.inner.width)
581            .field("name", &self.inner.name)
582            .field("value", &self.value())
583            .field("tags", &self.inner.tags)
584            .finish()
585    }
586}
587
588impl<'a> From<&'a Bucket> for BucketView<'a> {
589    fn from(value: &'a Bucket) -> Self {
590        BucketView::new(value)
591    }
592}
593
594impl Serialize for BucketView<'_> {
595    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
596    where
597        S: serde::Serializer,
598    {
599        let Bucket {
600            timestamp,
601            width,
602            name,
603            value: _,
604            tags,
605            metadata,
606        } = self.inner;
607
608        let len = match tags.is_empty() {
609            true => 4,
610            false => 5,
611        };
612
613        let mut state = serializer.serialize_map(Some(len))?;
614
615        state.serialize_entry("timestamp", timestamp)?;
616        state.serialize_entry("width", width)?;
617        state.serialize_entry("name", name)?;
618
619        if self.is_full_bucket() {
620            self.inner
621                .value
622                .serialize(serde::__private228::ser::FlatMapSerializer(&mut state))?;
623        } else {
624            self.value()
625                .serialize(serde::__private228::ser::FlatMapSerializer(&mut state))?;
626        }
627
628        if !tags.is_empty() {
629            state.serialize_entry("tags", tags)?;
630        }
631        if !metadata.is_default() {
632            state.serialize_entry("metadata", metadata)?;
633        }
634
635        state.end()
636    }
637}
638
639/// A view into the datapoints of a [`BucketValue`].
640#[derive(Debug, Clone, PartialEq, Serialize)]
641#[serde(tag = "type", content = "value")]
642pub enum BucketViewValue<'a> {
643    /// A counter metric.
644    ///
645    /// See: [`BucketValue::Counter`].
646    #[serde(rename = "c")]
647    Counter(CounterType),
648    /// A distribution metric.
649    ///
650    /// See: [`BucketValue::Distribution`].
651    #[serde(rename = "d")]
652    Distribution(&'a [DistributionType]),
653    /// A set metric.
654    ///
655    /// See: [`BucketValue::Set`].
656    #[serde(rename = "s")]
657    Set(SetView<'a>),
658    /// A gauage metric.
659    ///
660    /// See: [`BucketValue::Gauge`].
661    #[serde(rename = "g")]
662    Gauge(GaugeValue),
663}
664
665impl<'a> From<&'a BucketValue> for BucketViewValue<'a> {
666    fn from(value: &'a BucketValue) -> Self {
667        match value {
668            BucketValue::Counter(c) => BucketViewValue::Counter(*c),
669            BucketValue::Distribution(d) => BucketViewValue::Distribution(d),
670            BucketValue::Set(s) => BucketViewValue::Set(SetView::new(s, 0..s.len())),
671            BucketValue::Gauge(g) => BucketViewValue::Gauge(*g),
672        }
673    }
674}
675
676/// A view into the datapoints of a set metric.
677#[derive(Clone)]
678pub struct SetView<'a> {
679    source: &'a SetValue,
680    range: Range<usize>,
681}
682
683impl<'a> SetView<'a> {
684    fn new(source: &'a SetValue, range: Range<usize>) -> Self {
685        Self { source, range }
686    }
687
688    /// Amount of datapoints contained within the set view.
689    pub fn len(&self) -> usize {
690        self.range.len()
691    }
692
693    /// Returns `true` if this set contains no values.
694    pub fn is_empty(&self) -> bool {
695        self.len() == 0
696    }
697
698    /// Iterator over all datapoints contained in this set metric.
699    pub fn iter(&self) -> impl Iterator<Item = &SetType> {
700        self.source
701            .iter()
702            .skip(self.range.start)
703            .take(self.range.len())
704    }
705}
706
707impl PartialEq for SetView<'_> {
708    fn eq(&self, other: &Self) -> bool {
709        self.iter().eq(other.iter())
710    }
711}
712
713impl fmt::Debug for SetView<'_> {
714    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
715        f.debug_tuple("SetView")
716            .field(&self.iter().collect::<Vec<_>>())
717            .finish()
718    }
719}
720
721impl Serialize for SetView<'_> {
722    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
723    where
724        S: serde::Serializer,
725    {
726        let mut state = serializer.serialize_seq(Some(self.len()))?;
727
728        for item in self.iter() {
729            state.serialize_element(item)?;
730        }
731
732        state.end()
733    }
734}
735
736/// Result of [`split`].
737enum SplitDecision {
738    /// Bucket fits within the current budget.
739    ///
740    /// Contains the size of the bucket to subtract from the budget.
741    BucketFits(usize),
742    /// Bucket does not fit within the current budget and cannot be split.
743    MoveToNextBatch,
744    /// The bucket should be split at the specified position.
745    Split(usize),
746}
747
748/// Calculates a split for this bucket if its estimated serialization size exceeds a threshold.
749///
750/// There are three possible return values:
751///  - `BucketFits(size)` if the bucket fits entirely into the budget and consumes `size` bytes.
752///  - `MoveToNextBatch` if the size budget cannot even hold the bucket name and tags. There is no
753///    split, the entire bucket is moved.
754///  - `Split(n)` if the bucket fits partially, the bucket should be split after `n` elements.
755///
756/// This is an approximate function. The bucket is not actually serialized, but rather its
757/// footprint is estimated through the number of data points contained. See
758/// `estimate_size` for more information.
759fn split(bucket: &BucketView<'_>, max_size: usize, min_split_size: usize) -> SplitDecision {
760    // If there's enough space for the entire bucket, do not perform a split.
761    let bucket_size = bucket.estimated_size();
762    if max_size >= bucket_size {
763        return SplitDecision::BucketFits(bucket_size);
764    }
765
766    if !bucket.can_split() {
767        return SplitDecision::MoveToNextBatch;
768    }
769
770    // If the bucket key can't even fit into the remaining length, move the entire bucket into
771    // the right-hand side.
772    let own_size = bucket.estimated_base_size();
773    if max_size < (own_size + AVG_VALUE_SIZE) {
774        // split_at must not be zero
775        return SplitDecision::MoveToNextBatch;
776    }
777
778    if bucket_size < min_split_size {
779        return SplitDecision::MoveToNextBatch;
780    }
781
782    // Perform a split with the remaining space after adding the key. We assume an average
783    // length of 8 bytes per value and compute the number of items fitting into the left side.
784    let split_at = (max_size - own_size) / AVG_VALUE_SIZE;
785    SplitDecision::Split(split_at)
786}
787
788#[cfg(test)]
789mod tests {
790    use std::sync::Arc;
791
792    use insta::assert_json_snapshot;
793
794    use crate::dist;
795
796    use super::*;
797
798    fn bucket(name: &str, value: BucketValue) -> Bucket {
799        Bucket {
800            timestamp: UnixTimestamp::from_secs(5000),
801            width: 0,
802            name: name.into(),
803            value,
804            tags: Default::default(),
805            metadata: Default::default(),
806        }
807    }
808
809    fn gauge() -> BucketValue {
810        BucketValue::Gauge(GaugeValue {
811            last: 25.into(),
812            min: 17.into(),
813            max: 42.into(),
814            sum: 220.into(),
815            count: 85,
816        })
817    }
818
819    fn buckets() -> Vec<Bucket> {
820        vec![
821            bucket("c:spans/b0@none", BucketValue::Counter(1.into())),
822            bucket("c:spans/b1@none", BucketValue::Counter(12.into())),
823            bucket(
824                "d:spans/b2@none",
825                BucketValue::Distribution(dist![1, 2, 3, 5, 5]),
826            ),
827            bucket("s:spans/b3@none", BucketValue::Set([42, 75].into())),
828        ]
829    }
830
831    #[test]
832    fn test_bucket_view_select_counter() {
833        let bucket = bucket("c:spans/b0@none", BucketValue::Counter(1.into()));
834
835        let view = BucketView::new(&bucket).select(0..1).unwrap();
836        assert_eq!(view.len(), 1);
837        assert_eq!(
838            serde_json::to_string(&view).unwrap(),
839            serde_json::to_string(&bucket).unwrap()
840        );
841    }
842
843    #[test]
844    fn test_bucket_view_select_invalid_counter() {
845        let bucket = bucket("c:spans/b0@none", BucketValue::Counter(1.into()));
846
847        assert!(BucketView::new(&bucket).select(0..0).is_none());
848        assert!(BucketView::new(&bucket).select(0..2).is_none());
849        assert!(BucketView::new(&bucket).select(1..1).is_none());
850    }
851
852    #[test]
853    fn test_bucket_view_counter_metadata() {
854        let bucket = bucket("c:spans/b0@none", BucketValue::Counter(1.into()));
855        assert_eq!(bucket.metadata, BucketView::new(&bucket).metadata());
856    }
857
858    #[test]
859    fn test_bucket_view_select_distribution() {
860        let bucket = bucket(
861            "d:spans/b2@none",
862            BucketValue::Distribution(dist![1, 2, 3, 5, 5]),
863        );
864
865        let view = BucketView::new(&bucket).select(0..3).unwrap();
866        assert_eq!(view.len(), 3);
867        assert_eq!(
868            view.value(),
869            BucketViewValue::Distribution(&[1.into(), 2.into(), 3.into()])
870        );
871        let view = BucketView::new(&bucket).select(1..3).unwrap();
872        assert_eq!(view.len(), 2);
873        assert_eq!(
874            view.value(),
875            BucketViewValue::Distribution(&[2.into(), 3.into()])
876        );
877        let view = BucketView::new(&bucket).select(1..5).unwrap();
878        assert_eq!(view.len(), 4);
879        assert_eq!(
880            view.value(),
881            BucketViewValue::Distribution(&[2.into(), 3.into(), 5.into(), 5.into()])
882        );
883    }
884
885    #[test]
886    fn test_bucket_view_select_invalid_distribution() {
887        let bucket = bucket(
888            "d:spans/b2@none",
889            BucketValue::Distribution(dist![1, 2, 3, 5, 5]),
890        );
891
892        assert!(BucketView::new(&bucket).select(0..6).is_none());
893        assert!(BucketView::new(&bucket).select(5..6).is_none());
894        assert!(BucketView::new(&bucket).select(77..99).is_none());
895    }
896
897    #[test]
898    fn test_bucket_view_distribution_metadata() {
899        let bucket = bucket(
900            "d:spans/b2@none",
901            BucketValue::Distribution(dist![1, 2, 3, 5, 5]),
902        );
903        assert_eq!(bucket.metadata, BucketView::new(&bucket).metadata());
904
905        assert_eq!(
906            BucketView::new(&bucket).select(0..3).unwrap().metadata(),
907            bucket.metadata
908        );
909
910        let m = BucketView::new(&bucket).select(1..3).unwrap().metadata();
911        assert_eq!(
912            m,
913            BucketMetadata {
914                merges: 0,
915                ..bucket.metadata
916            }
917        );
918    }
919
920    #[test]
921    fn test_bucket_view_select_set() {
922        let bucket = bucket("s:spans/b3@none", BucketValue::Set([42, 75].into()));
923        let s = [42, 75].into();
924
925        let view = BucketView::new(&bucket).select(0..2).unwrap();
926        assert_eq!(view.len(), 2);
927        assert_eq!(view.value(), BucketViewValue::Set(SetView::new(&s, 0..2)));
928        let view = BucketView::new(&bucket).select(1..2).unwrap();
929        assert_eq!(view.len(), 1);
930        assert_eq!(view.value(), BucketViewValue::Set(SetView::new(&s, 1..2)));
931        let view = BucketView::new(&bucket).select(0..1).unwrap();
932        assert_eq!(view.len(), 1);
933        assert_eq!(view.value(), BucketViewValue::Set(SetView::new(&s, 0..1)));
934    }
935
936    #[test]
937    fn test_bucket_view_select_invalid_set() {
938        let bucket = bucket("s:spans/b3@none", BucketValue::Set([42, 75].into()));
939
940        assert!(BucketView::new(&bucket).select(0..3).is_none());
941        assert!(BucketView::new(&bucket).select(2..5).is_none());
942        assert!(BucketView::new(&bucket).select(77..99).is_none());
943    }
944
945    #[test]
946    fn test_bucket_view_set_metadata() {
947        let bucket = bucket("s:spans/b2@none", BucketValue::Set([1, 2, 3, 5, 5].into()));
948        assert_eq!(bucket.metadata, BucketView::new(&bucket).metadata());
949
950        assert_eq!(
951            BucketView::new(&bucket).select(0..3).unwrap().metadata(),
952            bucket.metadata
953        );
954
955        let m = BucketView::new(&bucket).select(1..3).unwrap().metadata();
956        assert_eq!(
957            m,
958            BucketMetadata {
959                merges: 0,
960                ..bucket.metadata
961            }
962        );
963    }
964
965    #[test]
966    fn test_bucket_view_select_gauge() {
967        let bucket = bucket("g:spans/b4@none", gauge());
968
969        let view = BucketView::new(&bucket).select(0..5).unwrap();
970        assert_eq!(view.len(), 5);
971        assert_eq!(
972            view.value(),
973            BucketViewValue::Gauge(GaugeValue {
974                last: 25.into(),
975                min: 17.into(),
976                max: 42.into(),
977                sum: 220.into(),
978                count: 85
979            })
980        );
981    }
982
983    #[test]
984    fn test_bucket_view_select_invalid_gauge() {
985        let bucket = bucket("g:spans/b4@none", gauge());
986
987        assert!(BucketView::new(&bucket).select(0..1).is_none());
988        assert!(BucketView::new(&bucket).select(0..4).is_none());
989        assert!(BucketView::new(&bucket).select(5..5).is_none());
990        assert!(BucketView::new(&bucket).select(5..6).is_none());
991    }
992
993    #[test]
994    fn test_bucket_view_gauge_metadata() {
995        let bucket = bucket("g:spans/b4@none", gauge());
996        assert_eq!(BucketView::new(&bucket).metadata(), bucket.metadata);
997    }
998
999    #[test]
1000    fn test_buckets_view_empty() {
1001        let view = BucketsView::new(Vec::new());
1002        assert_eq!(view.len(), 0);
1003        assert!(view.is_empty());
1004        let partials = view.iter().collect::<Vec<_>>();
1005        assert!(partials.is_empty());
1006    }
1007
1008    #[test]
1009    fn test_buckets_view_iter_full() {
1010        let buckets = buckets();
1011
1012        let view = BucketsView::from(&buckets);
1013        assert_eq!(view.len(), 4);
1014        assert!(!view.is_empty());
1015        let partials = view.iter().collect::<Vec<_>>();
1016        assert_eq!(partials.len(), 4);
1017        assert_eq!(partials[0].name(), "c:spans/b0@none");
1018        assert_eq!(partials[0].len(), 1);
1019        assert_eq!(partials[1].name(), "c:spans/b1@none");
1020        assert_eq!(partials[1].len(), 1);
1021        assert_eq!(partials[2].name(), "d:spans/b2@none");
1022        assert_eq!(partials[2].len(), 5);
1023        assert_eq!(partials[3].name(), "s:spans/b3@none");
1024        assert_eq!(partials[3].len(), 2);
1025    }
1026
1027    #[test]
1028    fn test_buckets_view_iter_partial_end() {
1029        let buckets = buckets();
1030
1031        let mut view = BucketsView::new(&buckets);
1032        view.end.slice = 2;
1033        view.end.bucket = 3;
1034        assert_eq!(view.len(), 3);
1035        assert!(!view.is_empty());
1036
1037        let partials = view.iter().collect::<Vec<_>>();
1038        assert_eq!(partials.len(), 3);
1039        assert_eq!(partials[0].name(), "c:spans/b0@none");
1040        assert_eq!(partials[0].len(), 1);
1041        assert_eq!(partials[1].name(), "c:spans/b1@none");
1042        assert_eq!(partials[1].len(), 1);
1043        assert_eq!(partials[2].name(), "d:spans/b2@none");
1044        assert_eq!(partials[2].len(), 3);
1045    }
1046
1047    #[test]
1048    fn test_buckets_view_iter_partial_start() {
1049        let buckets = buckets();
1050
1051        let mut view = BucketsView::new(buckets);
1052        view.start.slice = 2;
1053        view.start.bucket = 3;
1054        assert_eq!(view.len(), 2);
1055        assert!(!view.is_empty());
1056
1057        let partials = view.iter().collect::<Vec<_>>();
1058        assert_eq!(partials.len(), 2);
1059        assert_eq!(partials[0].name(), "d:spans/b2@none");
1060        assert_eq!(partials[0].len(), 2);
1061        assert_eq!(partials[1].name(), "s:spans/b3@none");
1062        assert_eq!(partials[1].len(), 2);
1063    }
1064
1065    #[test]
1066    fn test_buckets_view_iter_partial_start_and_end() {
1067        let buckets = buckets();
1068
1069        let mut view = BucketsView::from(&buckets);
1070        view.start.slice = 2;
1071        view.start.bucket = 1;
1072        view.end.slice = 3;
1073        view.end.bucket = 1;
1074        assert_eq!(view.len(), 2);
1075        assert!(!view.is_empty());
1076
1077        let partials = view.iter().collect::<Vec<_>>();
1078        assert_eq!(partials.len(), 2);
1079        assert_eq!(partials[0].name(), "d:spans/b2@none");
1080        assert_eq!(partials[0].len(), 4);
1081        assert_eq!(partials[1].name(), "s:spans/b3@none");
1082        assert_eq!(partials[1].len(), 1);
1083    }
1084
1085    #[test]
1086    fn test_buckets_view_by_size_small() {
1087        let buckets = buckets();
1088
1089        let view = BucketsView::from(&buckets);
1090        let partials = view
1091            .by_size(100)
1092            .map(|bv| {
1093                let len: usize = bv.iter().map(|b| b.len()).sum();
1094                let size: usize = bv.iter().map(|b| b.estimated_size()).sum();
1095                (len, size)
1096            })
1097            .collect::<Vec<_>>();
1098
1099        assert_eq!(partials, vec![(1, 73), (1, 73), (4, 97), (1, 73), (2, 81),]);
1100    }
1101
1102    #[test]
1103    fn test_buckets_view_by_size_small_as_arc() {
1104        let buckets: Arc<_> = buckets().into_boxed_slice().into();
1105
1106        let view = BucketsView::new(buckets);
1107        let partials = view
1108            .by_size(100)
1109            .map(|bv| {
1110                let len: usize = bv.iter().map(|b| b.len()).sum();
1111                let size: usize = bv.iter().map(|b| b.estimated_size()).sum();
1112                (len, size)
1113            })
1114            .collect::<Vec<_>>();
1115
1116        assert_eq!(partials, vec![(1, 73), (1, 73), (4, 97), (1, 73), (2, 81),]);
1117    }
1118
1119    #[test]
1120    fn test_buckets_view_by_size_one_split() {
1121        let buckets = buckets();
1122
1123        let view = BucketsView::from(&buckets);
1124        let partials = view
1125            .by_size(250)
1126            .map(|bv| {
1127                let len: usize = bv.iter().map(|b| b.len()).sum();
1128                let size: usize = bv.iter().map(|b| b.estimated_size()).sum();
1129                (len, size)
1130            })
1131            .collect::<Vec<_>>();
1132
1133        assert_eq!(partials, vec![(6, 243), (3, 154)]);
1134    }
1135
1136    #[test]
1137    fn test_buckets_view_by_size_no_split() {
1138        let buckets = buckets();
1139
1140        let view = BucketsView::from(&buckets);
1141        let partials = view
1142            .by_size(500)
1143            .map(|bv| {
1144                let len: usize = bv.iter().map(|b| b.len()).sum();
1145                let size: usize = bv.iter().map(|b| b.estimated_size()).sum();
1146                (len, size)
1147            })
1148            .collect::<Vec<_>>();
1149
1150        assert_eq!(partials, vec![(9, 332)]);
1151    }
1152
1153    #[test]
1154    fn test_buckets_view_by_size_no_too_small_no_bucket_fits() {
1155        let buckets = buckets();
1156
1157        let view = BucketsView::from(&buckets);
1158        let partials = view
1159            .by_size(50) // Too small, a bucket requires at least 74 bytes
1160            .count();
1161
1162        assert_eq!(partials, 0);
1163    }
1164
1165    #[test]
1166    fn test_buckets_view_by_size_do_not_split_gauge() {
1167        let buckets: Vec<_> = vec![bucket("g:transactions/foo@none", gauge())];
1168
1169        let view = BucketsView::from(&buckets);
1170        // 100 is too small to fit the gauge, but it is big enough to fit half a gauage,
1171        // make sure the gauge does not actually get split.
1172        let partials = view.by_size(100).count();
1173
1174        assert_eq!(partials, 0);
1175    }
1176
1177    #[test]
1178    fn test_buckets_view_serialize_full() {
1179        let mut buckets = buckets();
1180        for bucket in &mut buckets[1..3] {
1181            bucket.tags = [("foo".into(), "".into()), ("bar".into(), "baz".into())].into();
1182        }
1183        buckets.push(bucket("g:transactions/foo@none", gauge()));
1184
1185        assert_eq!(
1186            serde_json::to_string(&BucketsView::from(&buckets)).unwrap(),
1187            serde_json::to_string(&buckets).unwrap()
1188        );
1189    }
1190
1191    #[test]
1192    fn test_buckets_view_serialize_partial() {
1193        let mut buckets = buckets();
1194        buckets.remove(0);
1195        for bucket in &mut buckets[..2] {
1196            bucket.tags = [("foo".into(), "".into()), ("bar".into(), "baz".into())].into();
1197        }
1198        buckets.push(bucket("g:spans/b4@none", gauge()));
1199        let buckets: Arc<[_]> = buckets.into_boxed_slice().into();
1200
1201        let view = BucketsView::new(buckets);
1202        // This creates 4 separate views, spanning 1-2, 2-3, 3, 4.
1203        // 4 is too big to fit into a view together with the remainder of 3.
1204        let partials = view.by_size(178).collect::<Vec<_>>();
1205
1206        assert_json_snapshot!(partials);
1207    }
1208
1209    #[test]
1210    fn test_split_repeatedly() {
1211        let bucket = bucket(
1212            "d:spans/b2@none",
1213            BucketValue::Distribution(dist![1, 2, 3, 5, 5]),
1214        );
1215        let view = BucketView::new(&bucket);
1216
1217        // construct this so that we can take 2 values per split and result in 3 parts.
1218        let split_size = view.estimated_base_size() + 2 * AVG_VALUE_SIZE;
1219
1220        let (first, rest) = view.split(split_size, None);
1221        let (second, rest) = rest.unwrap().split(split_size, None);
1222        let (third, rest) = rest.unwrap().split(split_size, None);
1223
1224        assert_eq!(first.unwrap().range, 0..2);
1225        assert_eq!(second.unwrap().range, 2..4);
1226        assert_eq!(third.unwrap().range, 4..5);
1227        assert!(rest.is_none());
1228    }
1229}
relay_metrics/view.rs

relay_metrics/
view.rs
