micromegas_analytics/
log_entry.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/// A single log entry.
14///
15/// String fields borrow the per-block parse arena, so a `LogEntry` is valid only
16/// within the `parse_block` callback that produced it; it must be appended to
17/// Arrow (which copies the bytes) before the arena is dropped.
18#[derive(Debug)]
19pub struct LogEntry<'a> {
20    pub process: Arc<ProcessMetadata>,
21    pub stream_id: Arc<String>,
22    pub block_id: Arc<String>,
23    pub insert_time: i64,
24    pub time: i64,
25    pub level: i32,
26    pub target: &'a str,
27    pub msg: &'a str,
28    pub properties: PropertySet<'a>,
29}
30
31/// Creates a `LogEntry` from a `Value`.
32#[span_fn]
33pub fn log_entry_from_value<'a>(
34    convert_ticks: &ConvertTicks,
35    process: Arc<ProcessMetadata>,
36    stream_id: Arc<String>,
37    block_id: Arc<String>,
38    block_insert_time_ns: i64,
39    val: Value<'a>,
40) -> Result<Option<LogEntry<'a>>> {
41    if let Value::Object(obj) = val {
42        match obj.type_name {
43            "LogStaticStrEvent" => {
44                let ticks = obj
45                    .get::<i64>("time")
46                    .with_context(|| "reading time from LogStaticStrEvent")?;
47                let desc = obj
48                    .get::<&Object>("desc")
49                    .with_context(|| "reading desc from LogStaticStrEvent")?;
50                let level = desc
51                    .get::<u32>("level")
52                    .with_context(|| "reading level from LogStaticStrEvent")?;
53                let target = desc
54                    .get::<&str>("target")
55                    .with_context(|| "reading target from LogStaticStrEvent")?;
56                let msg = desc
57                    .get::<&str>("fmt_str")
58                    .with_context(|| "reading fmt_str from LogStaticStrEvent")?;
59                Ok(Some(LogEntry {
60                    process,
61                    stream_id,
62                    block_id,
63                    insert_time: block_insert_time_ns,
64                    time: convert_ticks.ticks_to_nanoseconds(ticks),
65                    level: level as i32,
66                    target,
67                    msg,
68                    properties: PropertySet::empty(),
69                }))
70            }
71            "LogStringEvent" | "LogStringEventV2" => {
72                let ticks = obj
73                    .get::<i64>("time")
74                    .with_context(|| "reading time from LogStringEvent")?;
75                let desc = obj
76                    .get::<&Object>("desc")
77                    .with_context(|| "reading desc from LogStringEvent")?;
78                let level = desc
79                    .get::<u32>("level")
80                    .with_context(|| "reading level from LogStringEvent")?;
81                let target = desc
82                    .get::<&str>("target")
83                    .with_context(|| "reading target from LogStringEvent")?;
84                let msg = obj
85                    .get::<&str>("msg")
86                    .with_context(|| "reading msg from LogStringEvent")?;
87                Ok(Some(LogEntry {
88                    process,
89                    stream_id,
90                    block_id,
91                    insert_time: block_insert_time_ns,
92                    time: convert_ticks.ticks_to_nanoseconds(ticks),
93                    level: level as i32,
94                    target,
95                    msg,
96                    properties: PropertySet::empty(),
97                }))
98            }
99            "LogStaticStrInteropEvent"
100            | "LogStaticStrInteropEventV2"
101            | "LogStringInteropEventV2"
102            | "LogStringInteropEventV3" => {
103                let ticks = obj
104                    .get::<i64>("time")
105                    .with_context(|| format!("reading time from {}", obj.type_name))?;
106                let level = obj
107                    .get::<u32>("level")
108                    .with_context(|| format!("reading level from {}", obj.type_name))?;
109                let target = obj
110                    .get::<&str>("target")
111                    .with_context(|| format!("reading target from {}", obj.type_name))?;
112                let msg = obj
113                    .get::<&str>("msg")
114                    .with_context(|| format!("reading msg from {}", obj.type_name))?;
115                Ok(Some(LogEntry {
116                    process,
117                    stream_id,
118                    block_id,
119                    insert_time: block_insert_time_ns,
120                    time: convert_ticks.ticks_to_nanoseconds(ticks),
121                    level: level as i32,
122                    target,
123                    msg,
124                    properties: PropertySet::empty(),
125                }))
126            }
127            "TaggedLogInteropEvent" => {
128                let ticks = obj
129                    .get::<i64>("time")
130                    .with_context(|| format!("reading time from {}", obj.type_name))?;
131                let level = obj
132                    .get::<u32>("level")
133                    .with_context(|| format!("reading level from {}", obj.type_name))?;
134                let target = obj
135                    .get::<&str>("target")
136                    .with_context(|| format!("reading target from {}", obj.type_name))?;
137                let msg = obj
138                    .get::<&str>("msg")
139                    .with_context(|| format!("reading msg from {}", obj.type_name))?;
140                let properties = obj
141                    .get::<&Object>("properties")
142                    .with_context(|| format!("reading properties from {}", obj.type_name))?;
143                let time = convert_ticks.ticks_to_nanoseconds(ticks);
144                Ok(Some(LogEntry {
145                    process,
146                    stream_id,
147                    block_id,
148                    insert_time: block_insert_time_ns,
149                    time,
150                    level: level as i32,
151                    target,
152                    msg,
153                    properties: properties.into(),
154                }))
155            }
156            "TaggedLogString" => {
157                let msg = obj
158                    .get::<&str>("msg")
159                    .with_context(|| format!("reading msg from {}", obj.type_name))?;
160                tagged_log_entry(
161                    convert_ticks,
162                    process,
163                    stream_id,
164                    block_id,
165                    block_insert_time_ns,
166                    obj,
167                    msg,
168                )
169            }
170            "TaggedLogStaticStrEvent" => {
171                let msg = obj
172                    .get::<&Object>("desc")
173                    .with_context(|| format!("reading desc from {}", obj.type_name))?
174                    .get::<&str>("fmt_str")
175                    .with_context(|| format!("reading fmt_str from {}", obj.type_name))?;
176                tagged_log_entry(
177                    convert_ticks,
178                    process,
179                    stream_id,
180                    block_id,
181                    block_insert_time_ns,
182                    obj,
183                    msg,
184                )
185            }
186
187            _ => {
188                warn!("unknown log event {:?}", obj);
189                Ok(None)
190            }
191        }
192    } else {
193        Ok(None)
194    }
195}
196
197/// Iterates over all log entries in a block.
198#[span_fn]
199pub async fn for_each_log_entry_in_block<Predicate>(
200    blob_storage: Arc<BlobStorage>,
201    convert_ticks: &ConvertTicks,
202    process: Arc<ProcessMetadata>,
203    stream: &StreamMetadata,
204    block: &BlockMetadata,
205    mut fun: Predicate,
206) -> Result<bool>
207where
208    Predicate: for<'a> FnMut(LogEntry<'a>) -> Result<bool>,
209{
210    let payload = fetch_block_payload(
211        blob_storage,
212        stream.process_id,
213        stream.stream_id,
214        block.block_id,
215    )
216    .await?;
217    let stream_id = Arc::new(stream.stream_id.to_string());
218    let block_id = Arc::new(block.block_id.to_string());
219    let block_insert_time_ns = block.insert_time.timestamp_nanos_opt().unwrap_or_default();
220    let continue_iterating = parse_block(stream, &payload, |val| {
221        if let Some(log_entry) = log_entry_from_value(
222            convert_ticks,
223            process.clone(),
224            stream_id.clone(),
225            block_id.clone(),
226            block_insert_time_ns,
227            val,
228        )
229        .with_context(|| "log_entry_from_value")?
230            && !fun(log_entry)?
231        {
232            return Ok(false); //do not continue
233        }
234        Ok(true) //continue
235    })
236    .with_context(|| format!("parse_block {}", block.block_id))?;
237    Ok(continue_iterating)
238}
239
240fn tagged_log_entry<'a>(
241    convert_ticks: &ConvertTicks,
242    process: Arc<ProcessMetadata>,
243    stream_id: Arc<String>,
244    block_id: Arc<String>,
245    block_insert_time_ns: i64,
246    obj: &Object<'a>,
247    msg: &'a str,
248) -> Result<Option<LogEntry<'a>>> {
249    let ticks = obj
250        .get::<i64>("time")
251        .with_context(|| format!("reading time from {}", obj.type_name))?;
252    let desc = obj
253        .get::<&Object>("desc")
254        .with_context(|| format!("reading desc from {}", obj.type_name))?;
255    let mut level = desc
256        .get::<u32>("level")
257        .with_context(|| format!("reading level from {}", obj.type_name))?;
258    let mut target = desc
259        .get::<&str>("target")
260        .with_context(|| format!("reading target from {}", obj.type_name))?;
261    let properties = obj
262        .get::<&Object>("properties")
263        .with_context(|| format!("reading properties from {}", obj.type_name))?;
264    for &(prop_name, prop_value) in properties.members {
265        match (prop_name, prop_value) {
266            ("target", Value::String(value_str)) => {
267                target = value_str;
268            }
269            ("level", Value::String(level_str)) => {
270                level = Level::parse(level_str).with_context(|| "parsing log level")? as u32;
271            }
272            (_, _) => {}
273        }
274    }
275    Ok(Some(LogEntry {
276        process,
277        stream_id,
278        block_id,
279        insert_time: block_insert_time_ns,
280        time: convert_ticks.ticks_to_nanoseconds(ticks),
281        level: level as i32,
282        target,
283        msg,
284        properties: properties.into(),
285    }))
286}
micromegas_analytics/log_entry.rs

micromegas_analytics/
log_entry.rs
