micromegas_analytics/lakehouse/otel/
spans_view.rs

1//! Per-process JIT view of OTel spans.
2//!
3//! `otel_spans` mirrors the `AsyncEventsView` pattern (no batch path; JIT-only;
4//! materialized per-process) but the time-conversion plumbing is simpler: OTLP
5//! timestamps are absolute nanoseconds (`tsc_frequency = 1_000_000_000`), so we
6//! use the plain `find_process` flow instead of the latest-timing variant.
7//!
8//! Cross-process trace traversal (`WHERE trace_id = X` across services) is a
9//! documented v1 limitation — users supply the `process_id` to
10//! `view_instance('otel_spans', '<uuid>')` or UNION across multiple instances.
11
12use super::spans_block_processor::OtelSpansBlockProcessor;
13use super::spans_table::otel_spans_table_schema;
14use crate::lakehouse::{
15    batch_update::PartitionCreationStrategy,
16    block_partition_spec::{BlockProcessor, BlockProcessorMap},
17    blocks_view::BlocksView,
18    dataframe_time_bounds::{DataFrameTimeBounds, NamedColumnsTimeBounds},
19    jit_partitions::{
20        BlockOrder, JitPartitionConfig, find_up_to_date_partitions,
21        generate_process_jit_partitions, write_partition_from_blocks,
22    },
23    lakehouse_context::LakehouseContext,
24    partition_cache::PartitionCache,
25    view::{PartitionSpec, View, ViewMetadata},
26    view_factory::ViewMaker,
27};
28use crate::{
29    metadata::find_process,
30    time::{TimeRange, datetime_to_scalar},
31};
32use anyhow::{Context, Result};
33use async_trait::async_trait;
34use chrono::{DateTime, TimeDelta, Utc};
35use datafusion::{
36    arrow::datatypes::Schema,
37    logical_expr::{BinaryExpr, Expr, Operator, col},
38};
39use micromegas_ingestion::web_ingestion_service::FORMAT_OTLP_TRACES;
40use micromegas_tracing::prelude::*;
41use std::collections::HashMap;
42use std::sync::Arc;
43use uuid::Uuid;
44
45const VIEW_SET_NAME: &str = "otel_spans";
46const SCHEMA_VERSION: u8 = 1;
47lazy_static::lazy_static! {
48    static ref TIME_COLUMN: Arc<String> = Arc::new(String::from("start_time"));
49    static ref END_TIME_COLUMN: Arc<String> = Arc::new(String::from("end_time"));
50}
51
52/// `ViewMaker` for `OtelSpansView`. Per-process only — `"global"` is rejected.
53#[derive(Debug)]
54pub struct OtelSpansViewMaker {}
55
56impl ViewMaker for OtelSpansViewMaker {
57    fn make_view(&self, view_instance_id: &str) -> Result<Arc<dyn View>> {
58        Ok(Arc::new(OtelSpansView::new(view_instance_id)?))
59    }
60
61    fn get_schema_hash(&self) -> Vec<u8> {
62        vec![SCHEMA_VERSION]
63    }
64
65    fn get_schema(&self) -> Arc<Schema> {
66        Arc::new(otel_spans_table_schema())
67    }
68}
69
70#[derive(Debug)]
71pub struct OtelSpansView {
72    view_set_name: Arc<String>,
73    view_instance_id: Arc<String>,
74    process_id: sqlx::types::Uuid,
75}
76
77impl OtelSpansView {
78    pub fn new(view_instance_id: &str) -> Result<Self> {
79        if view_instance_id == "global" {
80            anyhow::bail!("OtelSpansView does not support global view access");
81        }
82        let process_id = Uuid::parse_str(view_instance_id).with_context(|| "Uuid::parse_str")?;
83        Ok(Self {
84            view_set_name: Arc::new(String::from(VIEW_SET_NAME)),
85            view_instance_id: Arc::new(view_instance_id.into()),
86            process_id,
87        })
88    }
89}
90
91#[async_trait]
92impl View for OtelSpansView {
93    fn get_view_set_name(&self) -> Arc<String> {
94        self.view_set_name.clone()
95    }
96
97    fn get_view_instance_id(&self) -> Arc<String> {
98        self.view_instance_id.clone()
99    }
100
101    async fn make_batch_partition_spec(
102        &self,
103        _lakehouse: Arc<LakehouseContext>,
104        _existing_partitions: Arc<PartitionCache>,
105        _insert_range: TimeRange,
106    ) -> Result<Arc<dyn PartitionSpec>> {
107        anyhow::bail!("OtelSpansView does not support batch partition specs");
108    }
109
110    fn get_file_schema_hash(&self) -> Vec<u8> {
111        vec![SCHEMA_VERSION]
112    }
113
114    fn get_file_schema(&self) -> Arc<Schema> {
115        Arc::new(otel_spans_table_schema())
116    }
117
118    #[span_fn]
119    async fn jit_update(
120        &self,
121        lakehouse: Arc<LakehouseContext>,
122        query_range: Option<TimeRange>,
123    ) -> Result<()> {
124        let process = Arc::new(
125            find_process(
126                &lakehouse.lake().db_pool,
127                &self.process_id,
128                &lakehouse.default_audience(),
129            )
130            .await
131            .with_context(|| "find_process")?,
132        );
133        let query_range =
134            query_range.unwrap_or_else(|| TimeRange::new(process.start_time, chrono::Utc::now()));
135
136        let blocks_view = BlocksView::new(lakehouse.default_audience())?;
137        let all_partitions = generate_process_jit_partitions(
138            &JitPartitionConfig::default(),
139            lakehouse.clone(),
140            &blocks_view,
141            &query_range,
142            process.clone(),
143            "trace",
144        )
145        .await
146        .with_context(|| "generate_process_jit_partitions")?;
147        let view_meta = ViewMetadata {
148            view_set_name: self.get_view_set_name(),
149            view_instance_id: self.get_view_instance_id(),
150            file_schema_hash: self.get_file_schema_hash(),
151        };
152
153        let mut block_processors: BlockProcessorMap = HashMap::new();
154        block_processors.insert(
155            FORMAT_OTLP_TRACES,
156            Arc::new(OtelSpansBlockProcessor {}) as Arc<dyn BlockProcessor>,
157        );
158        let block_processors = Arc::new(block_processors);
159
160        let up_to_date = find_up_to_date_partitions(
161            &lakehouse.lake().db_pool,
162            view_meta.clone(),
163            BlockOrder::InsertTime,
164            &all_partitions,
165        )
166        .await
167        .with_context(|| "find_up_to_date_partitions")?;
168        for (part, up_to_date) in all_partitions.into_iter().zip(up_to_date) {
169            if !up_to_date {
170                write_partition_from_blocks(
171                    lakehouse.lake().clone(),
172                    view_meta.clone(),
173                    self.get_file_schema(),
174                    part,
175                    block_processors.clone(),
176                )
177                .await
178                .with_context(|| "write_partition_from_blocks")?;
179            }
180        }
181        Ok(())
182    }
183
184    fn make_time_filter(&self, begin: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Expr>> {
185        // Dual-bound (start_time <= end AND end_time >= begin) so spans whose
186        // duration crosses the query window — i.e. start before `begin` or end
187        // after `end` — are still included. A `start_time BETWEEN begin AND end`
188        // filter would silently drop long-running spans that overlap the window.
189        Ok(vec![
190            Expr::BinaryExpr(BinaryExpr::new(
191                col("start_time").into(),
192                Operator::LtEq,
193                Expr::Literal(datetime_to_scalar(end), None).into(),
194            )),
195            Expr::BinaryExpr(BinaryExpr::new(
196                col("end_time").into(),
197                Operator::GtEq,
198                Expr::Literal(datetime_to_scalar(begin), None).into(),
199            )),
200        ])
201    }
202
203    fn get_time_bounds(&self) -> Arc<dyn DataFrameTimeBounds> {
204        Arc::new(NamedColumnsTimeBounds::new(
205            TIME_COLUMN.clone(),
206            END_TIME_COLUMN.clone(),
207        ))
208    }
209
210    fn get_update_group(&self) -> Option<i32> {
211        None // process-specific JIT view; no scheduled updates
212    }
213
214    fn get_max_partition_time_delta(&self, _strategy: &PartitionCreationStrategy) -> TimeDelta {
215        TimeDelta::hours(1)
216    }
217}
micromegas_analytics/lakehouse/otel/spans_view.rs

micromegas_analytics/lakehouse/otel/
spans_view.rs
