mira_core/
traces.rs

1//! OTLP traces -> Arrow, following the same shape as [`crate::logs`].
2//!
3//! Nine tables, because a span is not flat. `spans` is the root; `span_events`
4//! and `span_links` are child tables carrying their own dense block-local `id`
5//! so that `span_event_attrs` and `span_link_attrs` can key on them the same way
6//! `span_attrs` keys on a span. The alternative — `List<Struct>` columns for
7//! events and links — would need a second, different mechanism to attach
8//! attributes to list *elements*, and the whole point of the EAV table is that
9//! there is only one.
10//!
11//! The id story, which is the part that is easy to get wrong: `id` and
12//! `parent_id` are rebased to be dense within this block, because they name rows
13//! in these files. `trace_id`, `span_id` and a link's target ids are **not**
14//! rebased, because they name things outside it — usually on another node
15//! entirely. That distinction is what makes a join inside a block
16//! unconditionally correct without a partition discriminant.
17
18use arrow_array::builder::{
19    FixedSizeBinaryBuilder, StringBuilder, TimestampNanosecondBuilder, UInt8Builder, UInt16Builder,
20    UInt32Builder, UInt64Builder,
21};
22use arrow_array::{ArrayRef, RecordBatch};
23use std::sync::Arc;
24
25use mira_proto::collector::trace::v1::ExportTraceServiceRequest;
26use mira_proto::trace::v1::Span;
27
28use crate::attrs::{AttrsBuilder, DictColumn, ResourceScope, resource_kv, scope_kv};
29use crate::error::Result;
30use crate::logs::{append_fixed, nanos};
31use crate::schema::{SPAN_EVENTS, SPAN_LINKS, SPANS};
32use crate::signal::{Sealed, Sidecars, SignalBuilder};
33
34pub struct TracesBuilder {
35    id: UInt32Builder,
36    trace_id: FixedSizeBinaryBuilder,
37    span_id: FixedSizeBinaryBuilder,
38    parent_span_id: FixedSizeBinaryBuilder,
39    trace_state: StringBuilder,
40    flags: UInt32Builder,
41    name: DictColumn,
42    kind: UInt8Builder,
43    start: TimestampNanosecondBuilder,
44    duration: UInt64Builder,
45    status_code: UInt8Builder,
46    status_message: StringBuilder,
47    dropped_attrs: UInt32Builder,
48    dropped_events: UInt32Builder,
49    dropped_links: UInt32Builder,
50    resource_id: UInt16Builder,
51    scope_id: UInt16Builder,
52    span_attrs: AttrsBuilder,
53
54    ev_id: UInt32Builder,
55    ev_parent: UInt32Builder,
56    ev_time: TimestampNanosecondBuilder,
57    ev_name: DictColumn,
58    ev_dropped: UInt32Builder,
59    event_attrs: AttrsBuilder,
60    next_event_id: u32,
61
62    ln_id: UInt32Builder,
63    ln_parent: UInt32Builder,
64    ln_trace_id: FixedSizeBinaryBuilder,
65    ln_span_id: FixedSizeBinaryBuilder,
66    ln_trace_state: StringBuilder,
67    ln_flags: UInt32Builder,
68    ln_dropped: UInt32Builder,
69    link_attrs: AttrsBuilder,
70    next_link_id: u32,
71
72    rs: ResourceScope,
73    next_id: u32,
74    min_ts: i64,
75    max_ts: i64,
76}
77
78impl Default for TracesBuilder {
79    fn default() -> Self {
80        Self::new()
81    }
82}
83
84impl TracesBuilder {
85    pub fn new() -> Self {
86        Self {
87            id: UInt32Builder::new(),
88            trace_id: FixedSizeBinaryBuilder::new(16),
89            span_id: FixedSizeBinaryBuilder::new(8),
90            parent_span_id: FixedSizeBinaryBuilder::new(8),
91            trace_state: StringBuilder::new(),
92            flags: UInt32Builder::new(),
93            name: DictColumn::new("spans.name"),
94            kind: UInt8Builder::new(),
95            start: TimestampNanosecondBuilder::new(),
96            duration: UInt64Builder::new(),
97            status_code: UInt8Builder::new(),
98            status_message: StringBuilder::new(),
99            dropped_attrs: UInt32Builder::new(),
100            dropped_events: UInt32Builder::new(),
101            dropped_links: UInt32Builder::new(),
102            resource_id: UInt16Builder::new(),
103            scope_id: UInt16Builder::new(),
104            span_attrs: AttrsBuilder::new("span_attrs.key"),
105
106            ev_id: UInt32Builder::new(),
107            ev_parent: UInt32Builder::new(),
108            ev_time: TimestampNanosecondBuilder::new(),
109            ev_name: DictColumn::new("span_events.name"),
110            ev_dropped: UInt32Builder::new(),
111            event_attrs: AttrsBuilder::new("span_event_attrs.key"),
112            next_event_id: 0,
113
114            ln_id: UInt32Builder::new(),
115            ln_parent: UInt32Builder::new(),
116            ln_trace_id: FixedSizeBinaryBuilder::new(16),
117            ln_span_id: FixedSizeBinaryBuilder::new(8),
118            ln_trace_state: StringBuilder::new(),
119            ln_flags: UInt32Builder::new(),
120            ln_dropped: UInt32Builder::new(),
121            link_attrs: AttrsBuilder::new("span_link_attrs.key"),
122            next_link_id: 0,
123
124            rs: ResourceScope::new(),
125            next_id: 0,
126            min_ts: i64::MAX,
127            max_ts: i64::MIN,
128        }
129    }
130
131    pub fn num_rows(&self) -> usize {
132        self.next_id as usize
133    }
134
135    pub fn is_empty(&self) -> bool {
136        self.next_id == 0
137    }
138
139    /// Same accounting as [`crate::logs::LogsBuilder::approx_bytes`]: fixed-width
140    /// columns estimated from row counts, variable-width heaps measured. A span
141    /// is wider than a log record, and events and links are rows of their own.
142    pub fn approx_bytes(&self) -> usize {
143        self.next_id as usize * 96
144            + self.next_event_id as usize * 24
145            + self.next_link_id as usize * 48
146            + (self.span_attrs.len()
147                + self.event_attrs.len()
148                + self.link_attrs.len()
149                + self.rs.len())
150                * 48
151            + self.trace_state.values_slice().len()
152            + self.status_message.values_slice().len()
153            + self.ln_trace_state.values_slice().len()
154            + self.span_attrs.heap_bytes()
155            + self.event_attrs.heap_bytes()
156            + self.link_attrs.heap_bytes()
157            + self.rs.heap_bytes()
158    }
159
160    pub fn has_headroom_for(&self, req: &ExportTraceServiceRequest) -> bool {
161        let (mut resources, mut scopes) = (0usize, 0usize);
162        let (mut res_kv, mut sc_kv) = (0usize, 0usize);
163        // Links are not counted: they have no dictionary column of their own, so
164        // the only ceiling they can hit is `link_attrs`.
165        let (mut spans, mut events) = (0usize, 0usize);
166        let (mut span_kv, mut ev_kv, mut ln_kv) = (0usize, 0usize, 0usize);
167        for rs in &req.resource_spans {
168            resources += 1;
169            res_kv += resource_kv(rs.resource.as_ref());
170            for ss in &rs.scope_spans {
171                scopes += 1;
172                sc_kv += scope_kv(ss.scope.as_ref());
173                spans += ss.spans.len();
174                for s in &ss.spans {
175                    span_kv += s.attributes.len();
176                    events += s.events.len();
177                    ev_kv += s.events.iter().map(|e| e.attributes.len()).sum::<usize>();
178                    ln_kv += s.links.iter().map(|l| l.attributes.len()).sum::<usize>();
179                }
180            }
181        }
182        self.rs.has_headroom(resources, scopes, res_kv, sc_kv)
183            && self.span_attrs.has_headroom(span_kv)
184            && self.event_attrs.has_headroom(ev_kv)
185            && self.link_attrs.has_headroom(ln_kv)
186            && self.name.has_headroom(spans)
187            && self.ev_name.has_headroom(events)
188    }
189
190    pub fn append_request(&mut self, req: &ExportTraceServiceRequest) -> Result<usize> {
191        let mut added = 0;
192        for rs in &req.resource_spans {
193            let rid = self.rs.resource(rs.resource.as_ref())?;
194            for ss in &rs.scope_spans {
195                let sid = self.rs.scope(ss.scope.as_ref())?;
196                for span in &ss.spans {
197                    self.append_span(span, rid, sid)?;
198                    added += 1;
199                }
200            }
201        }
202        Ok(added)
203    }
204
205    fn append_span(&mut self, s: &Span, rid: u16, sid: u16) -> Result<()> {
206        // Fallible first, before `next_id` moves: an overflow here must leave the
207        // block sealable and the ids dense. See `AttrsBuilder::append`.
208        self.name.append(&s.name)?;
209
210        let id = self.next_id;
211        self.next_id += 1;
212
213        let start = nanos(s.start_time_unix_nano);
214        let end = nanos(s.end_time_unix_nano);
215        // A malformed end before start would wrap; a zero end means "still
216        // running" in some exporters. Both become a zero duration rather than a
217        // 584-year one. Measured on the raw wire values, so a span that started
218        // and ended past 2^63 still reports the duration it really had.
219        let duration = s.end_time_unix_nano.saturating_sub(s.start_time_unix_nano);
220        // Only real timestamps set the block's range. OTLP requires a start time,
221        // so a zero is malformed — folding it in would make this block claim to
222        // cover the epoch and match every temporal query ever asked. `nanos`
223        // reads an unrepresentable start as that same zero, so one guard covers
224        // both. The range takes the clamped `end` rather than `start + duration`
225        // for the mirror-image reason: an end that did not fit must contribute
226        // nothing, where the sum would saturate `max_ts` to `i64::MAX` and put
227        // the block past every retention cutoff there will ever be.
228        if start != 0 {
229            self.min_ts = self.min_ts.min(start);
230            self.max_ts = self.max_ts.max(start.max(end));
231        }
232
233        self.id.append_value(id);
234        append_fixed(&mut self.trace_id, &s.trace_id, 16)?;
235        append_fixed(&mut self.span_id, &s.span_id, 8)?;
236        append_fixed(&mut self.parent_span_id, &s.parent_span_id, 8)?;
237        opt_str(&mut self.trace_state, &s.trace_state);
238        self.flags.append_value(s.flags);
239        // Anything outside the enum is a client bug; recording it as
240        // UNSPECIFIED keeps the column a UInt8 and loses nothing real.
241        self.kind
242            .append_value(u8::try_from(s.kind).unwrap_or(0).min(5));
243        self.start.append_value(start);
244        self.duration.append_value(duration);
245        let (code, msg) = s.status.as_ref().map_or((0, ""), |st| {
246            (
247                u8::try_from(st.code).unwrap_or(0).min(2),
248                st.message.as_str(),
249            )
250        });
251        self.status_code.append_value(code);
252        opt_str(&mut self.status_message, msg);
253        self.dropped_attrs.append_value(s.dropped_attributes_count);
254        self.dropped_events.append_value(s.dropped_events_count);
255        self.dropped_links.append_value(s.dropped_links_count);
256        self.resource_id.append_value(rid);
257        self.scope_id.append_value(sid);
258
259        for e in &s.events {
260            self.ev_name.append(&e.name)?;
261            let eid = self.next_event_id;
262            self.next_event_id += 1;
263            self.ev_id.append_value(eid);
264            self.ev_parent.append_value(id);
265            self.ev_time.append_value(nanos(e.time_unix_nano));
266            self.ev_dropped.append_value(e.dropped_attributes_count);
267            self.event_attrs.append_all(eid, &e.attributes)?;
268        }
269
270        for l in &s.links {
271            let lid = self.next_link_id;
272            self.next_link_id += 1;
273            self.ln_id.append_value(lid);
274            self.ln_parent.append_value(id);
275            append_fixed(&mut self.ln_trace_id, &l.trace_id, 16)?;
276            append_fixed(&mut self.ln_span_id, &l.span_id, 8)?;
277            opt_str(&mut self.ln_trace_state, &l.trace_state);
278            self.ln_flags.append_value(l.flags);
279            self.ln_dropped.append_value(l.dropped_attributes_count);
280            self.link_attrs.append_all(lid, &l.attributes)?;
281        }
282
283        self.span_attrs.append_all(id, &s.attributes)?;
284        Ok(())
285    }
286
287    /// Seal and reset, including on the error path — see
288    /// [`SignalBuilder::finish`].
289    pub fn finish(&mut self) -> Result<Sealed> {
290        let out = self.seal(Sidecars::Build);
291        *self = Self::new();
292        out
293    }
294
295    fn seal(&self, sidecars: Sidecars) -> Result<Sealed> {
296        let trace_ids = self.trace_id.finish_cloned();
297        // Built from the finished column rather than accumulated per append:
298        // the ids are already contiguous here, and a filter built alongside the
299        // rows would have to be discarded whenever `finish` fails part-way.
300        let trace_idx = match sidecars {
301            Sidecars::Build => crate::bloom::build(&trace_ids),
302            Sidecars::Skip => None,
303        };
304        let spans: Vec<ArrayRef> = vec![
305            Arc::new(self.id.finish_cloned()),
306            Arc::new(trace_ids),
307            Arc::new(self.span_id.finish_cloned()),
308            Arc::new(self.parent_span_id.finish_cloned()),
309            Arc::new(self.trace_state.finish_cloned()),
310            Arc::new(self.flags.finish_cloned()),
311            self.name.finish(),
312            Arc::new(self.kind.finish_cloned()),
313            Arc::new(self.start.finish_cloned()),
314            Arc::new(self.duration.finish_cloned()),
315            Arc::new(self.status_code.finish_cloned()),
316            Arc::new(self.status_message.finish_cloned()),
317            Arc::new(self.dropped_attrs.finish_cloned()),
318            Arc::new(self.dropped_events.finish_cloned()),
319            Arc::new(self.dropped_links.finish_cloned()),
320            Arc::new(self.resource_id.finish_cloned()),
321            Arc::new(self.scope_id.finish_cloned()),
322        ];
323        let events: Vec<ArrayRef> = vec![
324            Arc::new(self.ev_id.finish_cloned()),
325            Arc::new(self.ev_parent.finish_cloned()),
326            Arc::new(self.ev_time.finish_cloned()),
327            self.ev_name.finish(),
328            Arc::new(self.ev_dropped.finish_cloned()),
329        ];
330        let links: Vec<ArrayRef> = vec![
331            Arc::new(self.ln_id.finish_cloned()),
332            Arc::new(self.ln_parent.finish_cloned()),
333            Arc::new(self.ln_trace_id.finish_cloned()),
334            Arc::new(self.ln_span_id.finish_cloned()),
335            Arc::new(self.ln_trace_state.finish_cloned()),
336            Arc::new(self.ln_flags.finish_cloned()),
337            Arc::new(self.ln_dropped.finish_cloned()),
338        ];
339
340        // Order matches `schema::TRACES_BLOCK_TABLES`; the test pins the two
341        // together so a new table cannot be added to one and forgotten in the
342        // other.
343        let mut tables = vec![
344            ("spans", RecordBatch::try_new(SPANS.clone(), spans)?),
345            ("span_attrs", self.span_attrs.finish()?),
346            (
347                "span_events",
348                RecordBatch::try_new(SPAN_EVENTS.clone(), events)?,
349            ),
350            ("span_event_attrs", self.event_attrs.finish()?),
351            (
352                "span_links",
353                RecordBatch::try_new(SPAN_LINKS.clone(), links)?,
354            ),
355            ("span_link_attrs", self.link_attrs.finish()?),
356        ];
357        tables.extend(self.rs.finish()?);
358        Ok(Sealed::with(
359            sidecars,
360            self.next_id as usize,
361            tables,
362            self.min_ts,
363            self.max_ts,
364        )
365        .with_sidecar(crate::bloom::TRACE_IDX, trace_idx))
366    }
367}
368
369impl SignalBuilder for TracesBuilder {
370    type Request = ExportTraceServiceRequest;
371    const SIGNAL: &'static str = "traces";
372
373    fn has_headroom_for(&self, req: &Self::Request) -> bool {
374        TracesBuilder::has_headroom_for(self, req)
375    }
376    fn append_request(&mut self, req: &Self::Request) -> Result<usize> {
377        TracesBuilder::append_request(self, req)
378    }
379    fn approx_bytes(&self) -> usize {
380        TracesBuilder::approx_bytes(self)
381    }
382    fn is_empty(&self) -> bool {
383        TracesBuilder::is_empty(self)
384    }
385    fn finish(&mut self) -> Result<Sealed> {
386        TracesBuilder::finish(self)
387    }
388    fn snapshot(&self) -> Result<Sealed> {
389        self.seal(Sidecars::Skip)
390    }
391}
392
393/// proto3 gives an unset string and an empty one the same representation, and
394/// for every one of these columns the distinction does not exist. Null costs a
395/// validity bit; an empty string costs an offset entry too.
396fn opt_str(b: &mut StringBuilder, v: &str) {
397    if v.is_empty() {
398        b.append_null();
399    } else {
400        b.append_value(v);
401    }
402}
403
404#[cfg(test)]
405mod tests {
406    use super::*;
407    use arrow_array::TimestampNanosecondArray;
408    use mira_proto::trace::v1::span::Event;
409    use mira_proto::trace::v1::{ResourceSpans, ScopeSpans};
410
411    /// A span time past 2^63 wrapped negative, and a negative `min_ts` becomes a
412    /// block directory name `block::parse_dir_name` refuses — published, acked,
413    /// then invisible to every query and to retention. An SDK scaling its clock
414    /// wrong reaches this today: today's nanoseconds times 1000 overflow.
415    #[test]
416    fn span_times_past_i64_do_not_wrap_the_block_range() {
417        let mut b = TracesBuilder::new();
418        b.append_request(&ExportTraceServiceRequest {
419            resource_spans: vec![ResourceSpans {
420                scope_spans: vec![ScopeSpans {
421                    spans: vec![
422                        Span {
423                            name: "overflowed".into(),
424                            start_time_unix_nano: u64::MAX,
425                            end_time_unix_nano: u64::MAX,
426                            ..Default::default()
427                        },
428                        Span {
429                            name: "half-overflowed".into(),
430                            start_time_unix_nano: 6_000,
431                            end_time_unix_nano: u64::MAX,
432                            events: vec![Event {
433                                time_unix_nano: u64::MAX,
434                                name: "exception".into(),
435                                ..Default::default()
436                            }],
437                            ..Default::default()
438                        },
439                    ],
440                    ..Default::default()
441                }],
442                ..Default::default()
443            }],
444        })
445        .unwrap();
446
447        let sealed = b.finish().unwrap();
448        assert_eq!(
449            sealed.num_rows, 2,
450            "malformed spans are stored, not dropped"
451        );
452        assert_eq!(sealed.min_ts, 6_000, "an unrepresentable start is no start");
453        assert_eq!(
454            sealed.max_ts, 6_000,
455            "an unrepresentable end contributes nothing rather than i64::MAX"
456        );
457
458        let ts = |t: &str, c: &str| {
459            sealed
460                .table(t)
461                .unwrap()
462                .column_by_name(c)
463                .unwrap()
464                .as_any()
465                .downcast_ref::<TimestampNanosecondArray>()
466                .unwrap()
467                .clone()
468        };
469        assert_eq!(ts("spans", "start_time_unix_nano").values(), &[0, 6_000]);
470        assert_eq!(ts("span_events", "time_unix_nano").values(), &[0]);
471        // The duration is measured on the wire values, so a span that both
472        // started and ended past 2^63 still reports the length it really had.
473        let d = sealed.table("spans").unwrap();
474        let d = d
475            .column_by_name("duration_nano")
476            .unwrap()
477            .as_any()
478            .downcast_ref::<arrow_array::UInt64Array>()
479            .unwrap();
480        assert_eq!(d.values(), &[0, u64::MAX - 6_000]);
481    }
482
483    /// The flusher never calls the inherent methods — it holds a
484    /// `dyn SignalBuilder` — so the trait impl is the only path in production.
485    /// A delegate wired to the wrong builder (these three signals are
486    /// copy-pasted from each other) means the flusher asks logs whether traces
487    /// have room, appends anyway, and the dictionary overflows mid-request,
488    /// leaving the span columns different lengths and the block unsealable.
489    ///
490    /// So every answer here is pinned to a value worked out from the request
491    /// rather than to the inherent method's answer: comparing the trait to the
492    /// thing it delegates to compares a function with itself and holds for any
493    /// delegate that compiles. The refusal is a real one — the wide span
494    /// carries more *distinct* keys than the u16 dictionary has slots, and
495    /// appending it really does overflow.
496    #[test]
497    fn the_trait_answers_headroom_and_row_count_the_same_way_the_builder_does() {
498        let req = |attrs: usize| ExportTraceServiceRequest {
499            resource_spans: vec![ResourceSpans {
500                scope_spans: vec![ScopeSpans {
501                    spans: vec![Span {
502                        name: "wide".into(),
503                        start_time_unix_nano: 1,
504                        end_time_unix_nano: 2,
505                        attributes: (0..attrs)
506                            .map(|i| mira_proto::common::v1::KeyValue {
507                                key: format!("k{i}"),
508                                value: None,
509                            })
510                            .collect(),
511                        ..Default::default()
512                    }],
513                    ..Default::default()
514                }],
515                ..Default::default()
516            }],
517        };
518        let mut b = TracesBuilder::new();
519        assert!(SignalBuilder::is_empty(&b) && b.num_rows() == 0);
520        assert_eq!(SignalBuilder::approx_bytes(&b), 0);
521
522        assert!(
523            SignalBuilder::has_headroom_for(&b, &req(1)),
524            "a one-attribute span fits in an empty block"
525        );
526        assert!(
527            !SignalBuilder::has_headroom_for(&b, &req(crate::schema::DICT_CAP + 1)),
528            "one more key than the dictionary has slots does not"
529        );
530
531        assert_eq!(SignalBuilder::append_request(&mut b, &req(1)).unwrap(), 1);
532        assert_eq!(b.num_rows(), 1);
533        assert!(!SignalBuilder::is_empty(&b));
534        // One span row plus one attribute row, so the estimate is a hundred-odd
535        // bytes: the point is the order, not the constants, which are a guess
536        // by construction (`approx_bytes`).
537        let bytes = SignalBuilder::approx_bytes(&b);
538        assert!(
539            (100..500).contains(&bytes),
540            "one span and one attribute, estimated at {bytes} bytes"
541        );
542
543        // And the refusal above is not merely conservative: the request it
544        // refuses is one the appender cannot take, naming the column that
545        // filled so the flusher knows to seal rather than to fail the RPC.
546        assert!(matches!(
547            SignalBuilder::append_request(&mut b, &req(crate::schema::DICT_CAP + 1)),
548            Err(crate::error::Error::DictionaryFull("span_attrs.key"))
549        ));
550    }
551}
mira_core/traces.rs

mira_core/
traces.rs
