mprobe_diagnostics/
metrics.rs

1//! Defines an API for parsing diagnostic metrics.
2//!
3//! One usually gets [metric chunks] by iterating over the [diagnostic data].
4//!
5//! [metric chunks]: crate::metrics::MetricsChunk
6//! [diagnostic data]: crate::DiagnosticData
7
8mod raw;
9
10use std::fmt::Display;
11use std::io::Cursor;
12use std::io::Read;
13use std::sync::Arc;
14
15use bson::Document;
16use chrono::DateTime;
17use chrono::TimeZone;
18use chrono::Utc;
19
20use crate::bytes;
21use crate::compression;
22use crate::error::MetricParseError;
23use crate::metadata::Metadata;
24use crate::metrics::raw::MetricParser;
25use crate::metrics::raw::RawMetric;
26
27/// `MetricsChunk` contains a chunk of metrics in a specified time window,
28/// parsed from the diagnostic data.
29#[derive(Debug, Clone)]
30pub struct MetricsChunk {
31    /// Metadata associated with all the metrics in this chunk.
32    pub metadata: Metadata,
33
34    /// A list of diagnostic metrics.
35    pub metrics: Vec<Metric>,
36
37    /// Specifies the timestamp when the recording of these metrics started.
38    pub start: DateTime<Utc>,
39
40    /// Specifies the timestamp when the recording of these metrics ended.
41    pub end: DateTime<Utc>,
42}
43
44/// `Metric` represents a single diagnostic metric in a specified time window.
45#[derive(Debug, Clone)]
46pub struct Metric {
47    /// Name of the diagnostic metric.
48    pub name: Arc<str>,
49
50    /// A list of categories that this metric belongs to.
51    pub groups: Vec<String>,
52
53    /// A list of metric measurements.
54    pub measurements: Vec<Measurement>,
55
56    /// Specifies the timestamp when the recording of this metric started.
57    pub start: DateTime<Utc>,
58
59    /// Specifies the timestamp when the recording of this metric ended.
60    pub end: DateTime<Utc>,
61}
62
63/// `Measurement` represents a measurement of a metric at a single point in time.
64#[derive(Debug, PartialEq, PartialOrd, Clone, Copy)]
65pub struct Measurement {
66    /// Timestamp of the measurement.
67    pub timestamp: DateTime<Utc>,
68
69    /// Metric value.
70    pub value: MetricValue,
71}
72
73/// `MetricValue` defines the type of the metric value
74#[derive(Debug, PartialEq, PartialOrd, Clone, Copy)]
75pub enum MetricValue {
76    /// Unsigned 32-bit integer.
77    UInt32(u32),
78
79    /// Signed 32-bit integer.
80    Int32(i32),
81
82    /// Signed 64-bit integer.
83    Int64(i64),
84
85    /// Floating-point 64-bit number.
86    Float64(f64),
87
88    /// Boolean value.
89    Boolean(bool),
90
91    /// Date time.
92    DateTime(DateTime<Utc>),
93}
94
95impl From<MetricValue> for f64 {
96    fn from(value: MetricValue) -> f64 {
97        match value {
98            MetricValue::UInt32(v) => v as f64,
99            MetricValue::Int32(v) => v as f64,
100            MetricValue::Int64(v) => v as f64,
101            MetricValue::Float64(v) => v,
102            MetricValue::Boolean(b) => b as u64 as f64,
103            MetricValue::DateTime(dt) => dt.timestamp_millis() as f64,
104        }
105    }
106}
107
108impl Display for MetricValue {
109    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
110        match self {
111            MetricValue::UInt32(n) => Display::fmt(n, f),
112            MetricValue::Int32(n) => Display::fmt(n, f),
113            MetricValue::Int64(n) => Display::fmt(n, f),
114            MetricValue::Float64(n) => Display::fmt(n, f),
115            MetricValue::Boolean(b) => Display::fmt(b, f),
116            MetricValue::DateTime(dt) => Display::fmt(dt, f),
117        }
118    }
119}
120
121impl MetricsChunk {
122    const METRIC_NAME_DELIMITER: &str = " ";
123    const START_TIMESTAMP_METRIC_NAME: &str = "start";
124    const END_TIMESTAMP_METRIC_NAME: &str = "end";
125
126    pub(crate) fn from_reader<R: Read + ?Sized>(
127        reader: &mut R,
128    ) -> Result<MetricsChunk, MetricParseError> {
129        let data = compression::decompress(reader)?;
130        let mut cursor = Cursor::new(data.as_slice());
131
132        let reference_doc = Document::from_reader(&mut cursor)?;
133        let metrics_count: usize = bytes::read_le_u32(&mut cursor)?.try_into()?;
134        let samples_count: usize = bytes::read_le_u32(&mut cursor)?.try_into()?;
135        let metrics =
136            MetricParser::parse(&reference_doc, &mut cursor, metrics_count, samples_count)?;
137
138        MetricsChunk::from_raw(metrics, &reference_doc)
139    }
140
141    fn from_raw(
142        metrics: Vec<RawMetric>,
143        reference_doc: &Document,
144    ) -> Result<MetricsChunk, MetricParseError> {
145        let mut metrics_chunk: Vec<Metric> = Vec::with_capacity(metrics.len());
146        let mut chunk_timestamps: Vec<DateTime<Utc>> = Vec::new();
147        let mut timestamps: Vec<DateTime<Utc>> = Vec::new();
148
149        for metric in metrics.into_iter() {
150            if let Some(name) = metric.groups.last()
151                && name == Self::START_TIMESTAMP_METRIC_NAME
152            {
153                let ts = to_timestamps(metric.values).collect();
154
155                if metric.groups.len() == 1 {
156                    chunk_timestamps = ts;
157                } else {
158                    timestamps = ts;
159                };
160
161                continue;
162            }
163
164            if let Some(name) = metric.groups.last()
165                && name == Self::END_TIMESTAMP_METRIC_NAME
166            {
167                continue;
168            }
169
170            let name: Arc<str> = Arc::from(metric.groups.join(Self::METRIC_NAME_DELIMITER));
171            let measurements = timestamps
172                .iter()
173                .zip(metric.values)
174                .map(|(start, value)| Measurement {
175                    timestamp: start.to_owned(),
176                    value: metric.vtype.convert(value),
177                })
178                .collect::<Vec<Measurement>>();
179
180            let ts_err = || MetricParseError::MetricTimestampNotFound {
181                name: Arc::clone(&name),
182            };
183            let start = timestamps.first().ok_or_else(ts_err)?.to_owned();
184            let end = timestamps.last().ok_or_else(ts_err)?.to_owned();
185
186            metrics_chunk.push(Metric {
187                name,
188                groups: metric.groups,
189                start,
190                end,
191                measurements,
192            })
193        }
194
195        let ts_err = || MetricParseError::MetricTimestampNotFound {
196            name: Arc::from(Self::START_TIMESTAMP_METRIC_NAME),
197        };
198        let start_chunk = chunk_timestamps.first().ok_or_else(ts_err)?.to_owned();
199        let end_chunk = chunk_timestamps.last().ok_or_else(ts_err)?.to_owned();
200
201        let metadata = Metadata::from_reference_document(reference_doc)?;
202
203        Ok(MetricsChunk {
204            start: start_chunk,
205            end: end_chunk,
206            metrics: metrics_chunk,
207            metadata,
208        })
209    }
210}
211
212fn to_timestamps(values: Vec<u64>) -> impl Iterator<Item = DateTime<Utc>> {
213    values.into_iter().map(|v| {
214        Utc.timestamp_millis_opt(v as i64)
215            .single()
216            .expect("timestamp to be converted to UTC")
217    })
218}
mprobe_diagnostics/metrics.rs

mprobe_diagnostics/
metrics.rs
