micromegas_analytics/
measure.rs

1use crate::{
2    metadata::{ProcessMetadata, StreamMetadata},
3    payload::{fetch_block_payload, parse_block},
4    properties::property_set::PropertySet,
5    time::ConvertTicks,
6};
7use anyhow::{Context, Result};
8use micromegas_telemetry::{blob_storage::BlobStorage, types::block::BlockMetadata};
9use micromegas_tracing::prelude::*;
10use micromegas_transit::value::{Object, Value};
11use std::sync::Arc;
12
13/// Unit substituted for integer "ticks" metrics once converted to seconds.
14const SECONDS_METRIC_UNIT: &str = "seconds";
15
16/// Represents a single metric measurement.
17///
18/// String fields borrow the per-block parse arena (see [`crate::log_entry::LogEntry`]).
19pub struct Measure<'a> {
20    pub process: Arc<ProcessMetadata>,
21    pub stream_id: Arc<String>,
22    pub block_id: Arc<String>,
23    pub insert_time: i64, // nanoseconds
24    pub time: i64,        // nanoseconds
25    pub target: &'a str,
26    pub name: &'a str,
27    pub unit: &'a str,
28    pub value: f64,
29    pub properties: PropertySet<'a>,
30}
31
32/// Creates a `Measure` from a `Value`.
33pub fn measure_from_value<'a>(
34    process: Arc<ProcessMetadata>,
35    stream_id: Arc<String>,
36    block_id: Arc<String>,
37    block_insert_time_ns: i64,
38    convert_ticks: &ConvertTicks,
39    val: Value<'a>,
40) -> Result<Option<Measure<'a>>> {
41    if let Value::Object(obj) = val {
42        match obj.type_name {
43            "FloatMetricEvent" => {
44                let ticks = obj
45                    .get::<i64>("time")
46                    .with_context(|| "reading time from FloatMetricEvent")?;
47                let value = obj
48                    .get::<f64>("value")
49                    .with_context(|| "reading value from FloatMetricEvent")?;
50                let desc = obj
51                    .get::<&Object>("desc")
52                    .with_context(|| "reading desc from FloatMetricEvent")?;
53                let target = desc
54                    .get::<&str>("target")
55                    .with_context(|| "reading target from FloatMetricEvent")?;
56                let name = desc
57                    .get::<&str>("name")
58                    .with_context(|| "reading name from FloatMetricEvent")?;
59                let unit = desc
60                    .get::<&str>("unit")
61                    .with_context(|| "reading unit from FloatMetricEvent")?;
62                Ok(Some(Measure {
63                    process,
64                    stream_id,
65                    block_id,
66                    insert_time: block_insert_time_ns,
67                    time: convert_ticks.ticks_to_nanoseconds(ticks),
68                    target,
69                    name,
70                    unit,
71                    value,
72                    properties: PropertySet::empty(),
73                }))
74            }
75            "IntegerMetricEvent" => {
76                let ticks = obj
77                    .get::<i64>("time")
78                    .with_context(|| "reading time from IntegerMetricEvent")?;
79                let time = convert_ticks.ticks_to_nanoseconds(ticks);
80                let value = obj
81                    .get::<u64>("value")
82                    .with_context(|| "reading value from IntegerMetricEvent")?;
83                let desc = obj
84                    .get::<&Object>("desc")
85                    .with_context(|| "reading desc from IntegerMetricEvent")?;
86                let target = desc
87                    .get::<&str>("target")
88                    .with_context(|| "reading target from IntegerMetricEvent")?;
89                let name = desc
90                    .get::<&str>("name")
91                    .with_context(|| "reading name from IntegerMetricEvent")?;
92                let unit = desc
93                    .get::<&str>("unit")
94                    .with_context(|| "reading unit from IntegerMetricEvent")?;
95                if unit == "ticks" {
96                    Ok(Some(Measure {
97                        process,
98                        stream_id,
99                        block_id,
100                        insert_time: block_insert_time_ns,
101                        time,
102                        target,
103                        name,
104                        unit: SECONDS_METRIC_UNIT,
105                        value: convert_ticks.delta_ticks_to_ms(value as i64) / 1000.0,
106                        properties: PropertySet::empty(),
107                    }))
108                } else {
109                    Ok(Some(Measure {
110                        process,
111                        stream_id,
112                        block_id,
113                        insert_time: block_insert_time_ns,
114                        time,
115                        target,
116                        name,
117                        unit,
118                        value: value as f64,
119                        properties: PropertySet::empty(),
120                    }))
121                }
122            }
123            "TaggedIntegerMetricEvent" => {
124                let ticks = obj
125                    .get::<i64>("time")
126                    .with_context(|| "reading time from TaggedIntegerMetricEvent")?;
127                let time = convert_ticks.ticks_to_nanoseconds(ticks);
128                let value = obj
129                    .get::<u64>("value")
130                    .with_context(|| "reading value from TaggedIntegerMetricEvent")?;
131                let desc = obj
132                    .get::<&Object>("desc")
133                    .with_context(|| "reading desc from IntegerMetricEvent")?;
134                let mut target = desc
135                    .get::<&str>("target")
136                    .with_context(|| "reading target from IntegerMetricEvent")?;
137                let mut name = desc
138                    .get::<&str>("name")
139                    .with_context(|| "reading name from IntegerMetricEvent")?;
140                let mut unit = desc
141                    .get::<&str>("unit")
142                    .with_context(|| "reading unit from IntegerMetricEvent")?;
143                let properties = obj
144                    .get::<&Object>("properties")
145                    .with_context(|| "reading properties from TaggedIntegerMetricEvent")?;
146                for &(prop_name, prop_value) in properties.members {
147                    match (prop_name, prop_value) {
148                        ("target", Value::String(value_str)) => {
149                            target = value_str;
150                        }
151                        ("name", Value::String(value_str)) => {
152                            name = value_str;
153                        }
154                        ("unit", Value::String(value_str)) => {
155                            unit = value_str;
156                        }
157                        (_, _) => {}
158                    }
159                }
160
161                if unit == "ticks" {
162                    Ok(Some(Measure {
163                        process,
164                        stream_id,
165                        block_id,
166                        insert_time: block_insert_time_ns,
167                        time,
168                        target,
169                        name,
170                        unit: SECONDS_METRIC_UNIT,
171                        value: convert_ticks.delta_ticks_to_ms(value as i64) / 1000.0,
172                        properties: properties.into(),
173                    }))
174                } else {
175                    Ok(Some(Measure {
176                        process,
177                        stream_id,
178                        block_id,
179                        insert_time: block_insert_time_ns,
180                        time,
181                        target,
182                        name,
183                        unit,
184                        value: value as f64,
185                        properties: properties.into(),
186                    }))
187                }
188            }
189            "TaggedFloatMetricEvent" => {
190                let ticks = obj
191                    .get::<i64>("time")
192                    .with_context(|| "reading time from TaggedFloatMetricEvent")?;
193                let time = convert_ticks.ticks_to_nanoseconds(ticks);
194                let value = obj
195                    .get::<f64>("value")
196                    .with_context(|| "reading value from TaggedFloatMetricEvent")?;
197                let desc = obj
198                    .get::<&Object>("desc")
199                    .with_context(|| "reading desc from TaggedFloatMetricEvent")?;
200                let mut target = desc
201                    .get::<&str>("target")
202                    .with_context(|| "reading target from TaggedFloatMetricEvent")?;
203                let mut name = desc
204                    .get::<&str>("name")
205                    .with_context(|| "reading name from TaggedFloatMetricEvent")?;
206                let mut unit = desc
207                    .get::<&str>("unit")
208                    .with_context(|| "reading unit from TaggedFloatMetricEvent")?;
209                let properties = obj
210                    .get::<&Object>("properties")
211                    .with_context(|| "reading properties from TaggedFloatMetricEvent")?;
212                for &(prop_name, prop_value) in properties.members {
213                    match (prop_name, prop_value) {
214                        ("target", Value::String(value_str)) => {
215                            target = value_str;
216                        }
217                        ("name", Value::String(value_str)) => {
218                            name = value_str;
219                        }
220                        ("unit", Value::String(value_str)) => {
221                            unit = value_str;
222                        }
223                        (_, _) => {}
224                    }
225                }
226                Ok(Some(Measure {
227                    process,
228                    stream_id,
229                    block_id,
230                    insert_time: block_insert_time_ns,
231                    time,
232                    target,
233                    name,
234                    unit,
235                    value,
236                    properties: properties.into(),
237                }))
238            }
239
240            _ => {
241                warn!("unknown metric event {:?}", obj);
242                Ok(None)
243            }
244        }
245    } else {
246        Ok(None)
247    }
248}
249
250/// Iterates over each metric measurement in a block.
251#[span_fn]
252pub async fn for_each_measure_in_block<Predicate>(
253    blob_storage: Arc<BlobStorage>,
254    convert_ticks: &ConvertTicks,
255    process: Arc<ProcessMetadata>,
256    stream: &StreamMetadata,
257    block: &BlockMetadata,
258    mut fun: Predicate,
259) -> Result<bool>
260where
261    Predicate: for<'a> FnMut(Measure<'a>) -> Result<bool>,
262{
263    let payload = fetch_block_payload(
264        blob_storage,
265        stream.process_id,
266        stream.stream_id,
267        block.block_id,
268    )
269    .await?;
270    let stream_id = Arc::new(stream.stream_id.to_string());
271    let block_id = Arc::new(block.block_id.to_string());
272    let block_insert_time_ns = block.insert_time.timestamp_nanos_opt().unwrap_or_default();
273    let continue_iterating = parse_block(stream, &payload, |val| {
274        if let Some(measure) = measure_from_value(
275            process.clone(),
276            stream_id.clone(),
277            block_id.clone(),
278            block_insert_time_ns,
279            convert_ticks,
280            val,
281        )
282        .with_context(|| "measure_from_value")?
283            && !fun(measure)?
284        {
285            return Ok(false); //do not continue
286        }
287        Ok(true) //continue
288    })
289    .with_context(|| format!("parse_block {}", block.block_id))?;
290    Ok(continue_iterating)
291}
micromegas_analytics/measure.rs

micromegas_analytics/
measure.rs
