metrics_procession/
procession.rs

1use time::Duration;
2
3use metrics::{Key, Label};
4use serde::{Deserialize, Serialize};
5use time::OffsetDateTime;
6
7use crate::{
8    chunk::Chunk,
9    event::{Entry, Event},
10    iter::{Metric, MetricRef, MetricsIterator, MetricsRefIterator},
11    label_set::LabelSet,
12};
13
14/// This represents a time series of metrics collected over some length of time
15#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, Default)]
16pub struct Procession {
17    /// The series of chunks representing ~65 seconds of time in each chunk
18    pub chunks: Vec<Chunk>,
19    /// The set of all unique keys and labels currently in the set
20    pub labels: LabelSet,
21}
22
23impl Procession {
24    /// A naive attempt to calculate the memory size of the current state
25    pub fn memory_size(&self) -> usize {
26        use std::{collections::HashSet, mem::size_of};
27        let mut shared_string_set = HashSet::new();
28        let labels_size = self
29            .labels
30            .0
31            .keys()
32            .map(|k| {
33                let k_size = if shared_string_set.insert(k.name()) {
34                    k.name().len()
35                } else {
36                    0
37                } + size_of::<Key>();
38                let l_size = k.labels().fold(0, |acc, l| {
39                    let l_size = if shared_string_set.insert(l.key()) {
40                        l.key().len()
41                    } else {
42                        0
43                    } + size_of::<Label>();
44                    let v_size = if shared_string_set.insert(l.value()) {
45                        l.value().len()
46                    } else {
47                        0
48                    } + size_of::<Label>();
49                    acc + l_size + v_size
50                });
51                k_size + l_size + size_of::<u16>()
52            })
53            .sum::<usize>();
54        let chunk_size = self.chunks.iter().map(|c| c.memory_size()).sum::<usize>();
55        labels_size + chunk_size + size_of::<Self>()
56    }
57
58    /// Insert a new entry into the last (or newly last) [`Chunk`]
59    pub fn insert_entry(&mut self, entry: Entry, label: u16) {
60        let now = OffsetDateTime::now_utc();
61        let (last, ms) = self.last_chunk_and_ms(now);
62        last.push(Event { entry, ms, label });
63    }
64
65    /// Find the last chunk in this [Procession] along with the number of milliseconds
66    /// since the reference time on that chunk. If either there are no chunks already
67    /// available _or_ the number of milliseconds since the last chunk's reference time
68    /// would exceed [u16::MAX] a new chunk is added and a mutable reference to that chunk
69    /// is returned with a ms value of 0
70    pub fn last_chunk_and_ms(&mut self, now: OffsetDateTime) -> (&mut Chunk, u16) {
71        if self.chunks.is_empty() {
72            self.chunks.push(Chunk::default());
73        }
74        let mut duration = self
75            .chunks
76            .last()
77            .map(|c| (now - c.reference_time))
78            .unwrap_or_default();
79        if duration > Duration::milliseconds(i64::from(u16::MAX)) {
80            self.chunks.push(Chunk::new(now));
81            duration = Duration::ZERO;
82        }
83        let ms = u16::try_from(duration.whole_milliseconds()).unwrap_or(u16::MAX);
84        (self.chunks.last_mut().unwrap(), ms)
85    }
86
87    /// Ensure the provided key is in the [`labels`]
88    pub fn ensure_label(&mut self, k: &Key) -> u16 {
89        self.labels.ensure_key(k)
90    }
91
92    /// create an iterator for the raw metric events currently recorded that will be tied to the
93    /// lifetime of this instance of the [`Procession`]
94    pub fn iter(&self) -> MetricsRefIterator {
95        MetricsRefIterator::from(self)
96    }
97
98    /// create an iterator for the raw metric events currently recorded providing owned
99    /// version of all events
100    pub fn iter_owned(&self) -> MetricsIterator {
101        self.iter().into()
102    }
103}
104
105impl FromIterator<Metric> for Procession {
106    fn from_iter<T: IntoIterator<Item = Metric>>(iter: T) -> Self {
107        let mut iter = iter.into_iter().peekable();
108        let mut ret = Self::default();
109        if let Some(first) = iter.peek() {
110            let start = first.when;
111            ret.chunks.push(Chunk::new(start));
112        }
113        for event in iter {
114            let labels = event
115                .labels
116                .into_iter()
117                .map(|(k, v)| Label::new(k, v))
118                .collect::<Vec<_>>();
119            let label = ret.ensure_label(&Key::from_parts(event.key, labels));
120            ret.insert_entry(event.event, label);
121        }
122        ret
123    }
124}
125
126impl<'a> FromIterator<MetricRef<'a>> for Procession {
127    fn from_iter<T: IntoIterator<Item = MetricRef<'a>>>(iter: T) -> Self {
128        let mut iter = iter.into_iter().peekable();
129        let mut ret = Self::default();
130        if let Some(first) = iter.peek() {
131            let start = first.when;
132            ret.chunks.push(Chunk::new(start));
133        }
134        for event in iter {
135            let label = ret.ensure_label(event.key);
136            ret.insert_entry(event.event, label);
137        }
138        ret
139    }
140}
metrics_procession/procession.rs

metrics_procession/
procession.rs
