micromegas_analytics/
thread_block_processor.rs

1use crate::metadata::StreamMetadata;
2use crate::payload::{fetch_block_payload, parse_block};
3use crate::scope::BorrowedScopeDesc;
4use anyhow::{Context, Result};
5use micromegas_telemetry::blob_storage::BlobStorage;
6use micromegas_tracing::prelude::*;
7use micromegas_transit::value::{Object, Value};
8use std::sync::Arc;
9
10/// A trait for processing thread event blocks.
11pub trait ThreadBlockProcessor {
12    // return true to continue
13    fn on_begin_thread_scope(
14        &mut self,
15        block_id: &str,
16        event_id: i64,
17        scope: BorrowedScopeDesc<'_>,
18        ts: i64,
19    ) -> Result<bool>;
20    fn on_end_thread_scope(
21        &mut self,
22        block_id: &str,
23        event_id: i64,
24        scope: BorrowedScopeDesc<'_>,
25        ts: i64,
26    ) -> Result<bool>;
27}
28
29fn on_thread_event<'a, F>(obj: &Object<'a>, mut fun: F) -> Result<bool>
30where
31    F: FnMut(&'a Object<'a>, i64) -> Result<bool>,
32{
33    let tick = obj.get::<i64>("time")?;
34    let scope = obj.get::<&Object>("thread_span_desc")?;
35    fun(scope, tick)
36}
37
38fn on_thread_named_event<'a, F>(obj: &Object<'a>, mut fun: F) -> Result<bool>
39where
40    F: FnMut(&'a Object<'a>, &'a str, i64) -> Result<bool>,
41{
42    let tick = obj.get::<i64>("time")?;
43    let scope = obj.get::<&Object>("thread_span_location")?;
44    let name = obj.get::<&str>("name")?;
45    fun(scope, name, tick)
46}
47
48/// Parses a thread event block from a payload.
49#[span_fn]
50pub fn parse_thread_block_payload<Proc: ThreadBlockProcessor>(
51    block_id: &str,
52    object_offset: i64,
53    payload: &micromegas_telemetry::block_wire_format::BlockPayload,
54    stream: &StreamMetadata,
55    processor: &mut Proc,
56) -> Result<bool> {
57    let mut event_id = object_offset;
58    parse_block(stream, payload, |val| {
59        let res = if let Value::Object(obj) = val {
60            match obj.type_name {
61                "BeginThreadSpanEvent" => on_thread_event(obj, |scope, ts| {
62                    let name = scope.get::<&str>("name")?;
63                    let filename = scope.get::<&str>("file")?;
64                    let target = scope.get::<&str>("target")?;
65                    let line = scope.get::<u32>("line")?;
66                    let scope_desc = BorrowedScopeDesc::new(name, filename, target, line);
67                    processor.on_begin_thread_scope(block_id, event_id, scope_desc, ts)
68                })
69                .with_context(|| "reading BeginThreadSpanEvent"),
70                "EndThreadSpanEvent" => on_thread_event(obj, |scope, ts| {
71                    let name = scope.get::<&str>("name")?;
72                    let filename = scope.get::<&str>("file")?;
73                    let target = scope.get::<&str>("target")?;
74                    let line = scope.get::<u32>("line")?;
75                    let scope_desc = BorrowedScopeDesc::new(name, filename, target, line);
76                    processor.on_end_thread_scope(block_id, event_id, scope_desc, ts)
77                })
78                .with_context(|| "reading EndThreadSpanEvent"),
79                "BeginThreadNamedSpanEvent" | "BeginThreadDynNamedSpanEvent" => {
80                    on_thread_named_event(obj, |scope, name, ts| {
81                        let filename = scope.get::<&str>("file")?;
82                        let target = scope.get::<&str>("target")?;
83                        let line = scope.get::<u32>("line")?;
84                        let scope_desc = BorrowedScopeDesc::new(name, filename, target, line);
85                        processor.on_begin_thread_scope(block_id, event_id, scope_desc, ts)
86                    })
87                    .with_context(|| "reading BeginThreadNamedSpanEvent")
88                }
89                "EndThreadNamedSpanEvent" | "EndThreadDynNamedSpanEvent" => {
90                    on_thread_named_event(obj, |scope, name, ts| {
91                        let filename = scope.get::<&str>("file")?;
92                        let target = scope.get::<&str>("target")?;
93                        let line = scope.get::<u32>("line")?;
94                        let scope_desc = BorrowedScopeDesc::new(name, filename, target, line);
95                        processor.on_end_thread_scope(block_id, event_id, scope_desc, ts)
96                    })
97                    .with_context(|| "reading EndThreadNamedSpanEvent")
98                }
99                "BeginAsyncSpanEvent"
100                | "EndAsyncSpanEvent"
101                | "BeginAsyncNamedSpanEvent"
102                | "EndAsyncNamedSpanEvent"
103                | "BeginAsyncDynNamedSpanEvent"
104                | "EndAsyncDynNamedSpanEvent" => {
105                    // Ignore async span events as they are not relevant for thread spans view
106                    Ok(true)
107                }
108                event_type => {
109                    warn!("unknown event type {}", event_type);
110                    Ok(true)
111                }
112            }
113        } else {
114            Ok(true) // continue
115        };
116        event_id += 1;
117        res
118    })
119}
120
121/// Parses a thread event block.
122#[span_fn]
123pub async fn parse_thread_block<Proc: ThreadBlockProcessor>(
124    blob_storage: Arc<BlobStorage>,
125    stream: &StreamMetadata,
126    block_id: sqlx::types::Uuid,
127    object_offset: i64,
128    processor: &mut Proc,
129) -> Result<bool> {
130    let payload =
131        fetch_block_payload(blob_storage, stream.process_id, stream.stream_id, block_id).await?;
132    let block_id_str = block_id
133        .hyphenated()
134        .encode_lower(&mut sqlx::types::uuid::Uuid::encode_buffer())
135        .to_owned();
136    info!(
137        "parse_thread_block stream_id={} block_id={block_id_str}",
138        stream.stream_id
139    );
140    parse_thread_block_payload(&block_id_str, object_offset, &payload, stream, processor)
141}
micromegas_analytics/thread_block_processor.rs

micromegas_analytics/
thread_block_processor.rs
