micromegas_analytics/
net_block_processing.rs

1use crate::metadata::StreamMetadata;
2use crate::payload::{fetch_block_payload, parse_block};
3use anyhow::{Context, Result};
4use micromegas_telemetry::blob_storage::BlobStorage;
5use micromegas_telemetry::block_wire_format::BlockPayload;
6use micromegas_tracing::prelude::*;
7use micromegas_transit::value::{Object, Value};
8use std::sync::Arc;
9
10/// A trait for processing network tracing event blocks.
11///
12/// Implementors receive one callback per decoded net event. Returning `Ok(true)`
13/// continues iteration; returning `Ok(false)` stops parsing the current block.
14///
15/// The string arguments borrow the per-block parse arena; an implementor that
16/// retains them across blocks (e.g. an open-span stack) must copy them out.
17pub trait NetBlockProcessor {
18    fn on_connection_begin(
19        &mut self,
20        event_id: i64,
21        time: i64,
22        connection_name: &str,
23        is_outgoing: bool,
24    ) -> Result<bool>;
25
26    fn on_connection_end(&mut self, event_id: i64, time: i64, bit_size: i64) -> Result<bool>;
27
28    fn on_object_begin(&mut self, event_id: i64, time: i64, object_name: &str) -> Result<bool>;
29
30    fn on_object_end(&mut self, event_id: i64, time: i64, bit_size: i64) -> Result<bool>;
31
32    fn on_property(
33        &mut self,
34        event_id: i64,
35        time: i64,
36        property_name: &str,
37        bit_size: i64,
38    ) -> Result<bool>;
39
40    fn on_rpc_begin(&mut self, event_id: i64, time: i64, function_name: &str) -> Result<bool>;
41
42    fn on_rpc_end(&mut self, event_id: i64, time: i64, bit_size: i64) -> Result<bool>;
43}
44
45fn read_time(obj: &Object<'_>) -> Result<i64> {
46    obj.get::<i64>("time")
47}
48
49fn read_bit_size(obj: &Object<'_>) -> Result<i64> {
50    Ok(obj.get::<u32>("bit_size")? as i64)
51}
52
53/// Parses a net event block payload and calls the appropriate processor callback for each event.
54#[span_fn]
55pub fn parse_net_block_payload<Proc: NetBlockProcessor>(
56    object_offset: i64,
57    payload: &BlockPayload,
58    stream: &StreamMetadata,
59    processor: &mut Proc,
60) -> Result<bool> {
61    let mut event_id = object_offset;
62    parse_block(stream, payload, |val| {
63        let res = if let Value::Object(obj) = val {
64            match obj.type_name {
65                "NetConnectionBeginEvent" => {
66                    let time = read_time(obj).with_context(|| "NetConnectionBeginEvent.time")?;
67                    let connection_name = obj
68                        .get::<&str>("connection_name")
69                        .with_context(|| "NetConnectionBeginEvent.connection_name")?;
70                    let is_outgoing = obj
71                        .get::<u8>("is_outgoing")
72                        .with_context(|| "NetConnectionBeginEvent.is_outgoing")?
73                        != 0;
74                    processor.on_connection_begin(event_id, time, connection_name, is_outgoing)
75                }
76                "NetConnectionEndEvent" => {
77                    let time = read_time(obj).with_context(|| "NetConnectionEndEvent.time")?;
78                    let bit_size =
79                        read_bit_size(obj).with_context(|| "NetConnectionEndEvent.bit_size")?;
80                    processor.on_connection_end(event_id, time, bit_size)
81                }
82                "NetObjectBeginEvent" => {
83                    let time = read_time(obj).with_context(|| "NetObjectBeginEvent.time")?;
84                    let object_name = obj
85                        .get::<&str>("object_name")
86                        .with_context(|| "NetObjectBeginEvent.object_name")?;
87                    processor.on_object_begin(event_id, time, object_name)
88                }
89                "NetObjectEndEvent" => {
90                    let time = read_time(obj).with_context(|| "NetObjectEndEvent.time")?;
91                    let bit_size =
92                        read_bit_size(obj).with_context(|| "NetObjectEndEvent.bit_size")?;
93                    processor.on_object_end(event_id, time, bit_size)
94                }
95                "NetPropertyEvent" => {
96                    let time = read_time(obj).with_context(|| "NetPropertyEvent.time")?;
97                    let property_name = obj
98                        .get::<&str>("property_name")
99                        .with_context(|| "NetPropertyEvent.property_name")?;
100                    let bit_size =
101                        read_bit_size(obj).with_context(|| "NetPropertyEvent.bit_size")?;
102                    processor.on_property(event_id, time, property_name, bit_size)
103                }
104                "NetRPCBeginEvent" => {
105                    let time = read_time(obj).with_context(|| "NetRPCBeginEvent.time")?;
106                    let function_name = obj
107                        .get::<&str>("function_name")
108                        .with_context(|| "NetRPCBeginEvent.function_name")?;
109                    processor.on_rpc_begin(event_id, time, function_name)
110                }
111                "NetRPCEndEvent" => {
112                    let time = read_time(obj).with_context(|| "NetRPCEndEvent.time")?;
113                    let bit_size = read_bit_size(obj).with_context(|| "NetRPCEndEvent.bit_size")?;
114                    processor.on_rpc_end(event_id, time, bit_size)
115                }
116                event_type => {
117                    warn!("unknown event type in net block: {}", event_type);
118                    Ok(true)
119                }
120            }
121        } else {
122            Ok(true)
123        };
124        event_id += 1;
125        res
126    })
127}
128
129/// Fetches and parses a net event block.
130#[span_fn]
131pub async fn parse_net_block<Proc: NetBlockProcessor>(
132    blob_storage: Arc<BlobStorage>,
133    stream: &StreamMetadata,
134    block_id: sqlx::types::Uuid,
135    object_offset: i64,
136    processor: &mut Proc,
137) -> Result<bool> {
138    let payload =
139        fetch_block_payload(blob_storage, stream.process_id, stream.stream_id, block_id).await?;
140    parse_net_block_payload(object_offset, &payload, stream, processor)
141}
micromegas_analytics/net_block_processing.rs

micromegas_analytics/
net_block_processing.rs
