laminar_connectors/otel/convert/
mod.rs

1//! Protobuf-to-Arrow conversion for OTel traces, metrics, and logs.
2//!
3//! Flattens the nested protobuf hierarchies into flat Arrow `RecordBatch` rows.
4
5use std::sync::Arc;
6
7use arrow_array::builder::{
8    FixedSizeBinaryBuilder, Float64Builder, Int32Builder, Int64Builder, StringBuilder,
9    TimestampNanosecondBuilder, UInt64Builder,
10};
11use arrow_array::RecordBatch;
12use arrow_schema::SchemaRef;
13
14use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
15use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
16use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
17use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue};
18use opentelemetry_proto::tonic::metrics::v1::{metric, number_data_point};
19use opentelemetry_proto::tonic::resource::v1::Resource;
20
21/// Pad or truncate a byte slice to exactly `len` bytes.
22/// Returns the input directly if it's already the right size.
23fn fixed_bytes(src: &[u8], len: usize) -> std::borrow::Cow<'_, [u8]> {
24    if src.len() == len {
25        std::borrow::Cow::Borrowed(src)
26    } else {
27        let mut out = vec![0u8; len];
28        let copy_len = src.len().min(len);
29        out[..copy_len].copy_from_slice(&src[..copy_len]);
30        std::borrow::Cow::Owned(out)
31    }
32}
33
34/// Returns `true` if every byte in the slice is zero.
35fn is_all_zeros(bytes: &[u8]) -> bool {
36    bytes.iter().all(|&b| b == 0)
37}
38
39/// Extract a string attribute value from resource attributes by key.
40fn extract_resource_attr(resource: Option<&Resource>, key: &str) -> Option<String> {
41    resource.and_then(|r| {
42        r.attributes.iter().find_map(|kv| {
43            if kv.key == key {
44                any_value_to_string(kv.value.as_ref())
45            } else {
46                None
47            }
48        })
49    })
50}
51
52/// Serialize resource attributes to JSON, excluding promoted fields.
53fn resource_attributes_json(resource: Option<&Resource>) -> Option<String> {
54    let attrs: Vec<&KeyValue> = resource
55        .map(|r| {
56            r.attributes
57                .iter()
58                .filter(|kv| kv.key != "service.name" && kv.key != "service.version")
59                .collect()
60        })
61        .unwrap_or_default();
62
63    if attrs.is_empty() {
64        return None;
65    }
66
67    Some(key_values_to_json(&attrs))
68}
69
70/// Serialize a `KeyValue` slice to a JSON object string, or `None` if empty.
71fn kv_to_json(attrs: &[KeyValue]) -> Option<String> {
72    if attrs.is_empty() {
73        return None;
74    }
75    let refs: Vec<&KeyValue> = attrs.iter().collect();
76    Some(key_values_to_json(&refs))
77}
78
79/// Convert a list of `KeyValue` to a JSON object string.
80fn key_values_to_json(kvs: &[&KeyValue]) -> String {
81    let mut buf = String::with_capacity(kvs.len() * 32);
82    buf.push('{');
83    for (i, kv) in kvs.iter().enumerate() {
84        if i > 0 {
85            buf.push(',');
86        }
87        // Key is always a JSON string
88        write_json_string(&mut buf, &kv.key);
89        buf.push(':');
90        // Value
91        if let Some(v) = &kv.value {
92            write_any_value_json(&mut buf, v);
93        } else {
94            buf.push_str("null");
95        }
96    }
97    buf.push('}');
98    buf
99}
100
101/// Write a JSON-escaped string.
102fn write_json_string(buf: &mut String, s: &str) {
103    buf.push('"');
104    for c in s.chars() {
105        match c {
106            '"' => buf.push_str("\\\""),
107            '\\' => buf.push_str("\\\\"),
108            '\n' => buf.push_str("\\n"),
109            '\r' => buf.push_str("\\r"),
110            '\t' => buf.push_str("\\t"),
111            c if c.is_control() => {
112                use std::fmt::Write;
113                let _ = write!(buf, "\\u{:04x}", c as u32);
114            }
115            c => buf.push(c),
116        }
117    }
118    buf.push('"');
119}
120
121/// Maximum nesting depth for `AnyValue` JSON serialization.
122/// Protects against stack overflow from malicious OTLP payloads.
123const MAX_JSON_DEPTH: usize = 32;
124
125/// Write an `AnyValue` as JSON with bounded recursion depth.
126fn write_any_value_json(buf: &mut String, v: &AnyValue) {
127    write_any_value_json_depth(buf, v, 0);
128}
129
130fn write_any_value_json_depth(buf: &mut String, v: &AnyValue, depth: usize) {
131    use opentelemetry_proto::tonic::common::v1::any_value::Value;
132    if depth >= MAX_JSON_DEPTH {
133        buf.push_str("null");
134        return;
135    }
136    match &v.value {
137        Some(Value::StringValue(s)) => write_json_string(buf, s),
138        Some(Value::BoolValue(b)) => buf.push_str(if *b { "true" } else { "false" }),
139        Some(Value::IntValue(i)) => {
140            use std::fmt::Write;
141            let _ = write!(buf, "{i}");
142        }
143        Some(Value::DoubleValue(d)) => {
144            if d.is_finite() {
145                use std::fmt::Write;
146                let _ = write!(buf, "{d}");
147            } else {
148                buf.push_str("null");
149            }
150        }
151        Some(Value::ArrayValue(arr)) => {
152            buf.push('[');
153            for (i, val) in arr.values.iter().enumerate() {
154                if i > 0 {
155                    buf.push(',');
156                }
157                write_any_value_json_depth(buf, val, depth + 1);
158            }
159            buf.push(']');
160        }
161        Some(Value::KvlistValue(kvl)) => {
162            buf.push('{');
163            for (i, kv) in kvl.values.iter().enumerate() {
164                if i > 0 {
165                    buf.push(',');
166                }
167                write_json_string(buf, &kv.key);
168                buf.push(':');
169                if let Some(val) = &kv.value {
170                    write_any_value_json_depth(buf, val, depth + 1);
171                } else {
172                    buf.push_str("null");
173                }
174            }
175            buf.push('}');
176        }
177        Some(Value::BytesValue(b)) => {
178            use base64::Engine;
179            buf.push('"');
180            buf.push_str(&base64::engine::general_purpose::STANDARD.encode(b));
181            buf.push('"');
182        }
183        None => buf.push_str("null"),
184    }
185}
186
187/// Convert an `AnyValue` to a `String` (for promoted fields like service.name).
188fn any_value_to_string(v: Option<&AnyValue>) -> Option<String> {
189    use opentelemetry_proto::tonic::common::v1::any_value::Value;
190    v.and_then(|av| match &av.value {
191        Some(Value::StringValue(s)) => Some(s.clone()),
192        Some(Value::IntValue(i)) => Some(i.to_string()),
193        Some(Value::BoolValue(b)) => Some(b.to_string()),
194        Some(Value::DoubleValue(d)) => Some(d.to_string()),
195        _ => None,
196    })
197}
198
199/// Convert an `ExportTraceServiceRequest` into a `RecordBatch`, or `None` if empty.
200///
201/// # Errors
202///
203/// Returns `ArrowError` if column construction fails.
204#[allow(clippy::too_many_lines)] // 20 columns = inherently long
205pub fn trace_request_to_batch(
206    req: &ExportTraceServiceRequest,
207    schema: &SchemaRef,
208    received_at_nanos: i64,
209) -> Result<Option<RecordBatch>, arrow_schema::ArrowError> {
210    // Count total spans for capacity hints
211    let total_spans: usize = req
212        .resource_spans
213        .iter()
214        .flat_map(|rs| &rs.scope_spans)
215        .map(|ss| ss.spans.len())
216        .sum();
217
218    if total_spans == 0 {
219        return Ok(None);
220    }
221
222    // Column builders
223    let mut trace_ids = FixedSizeBinaryBuilder::with_capacity(total_spans, 16);
224    let mut span_ids = FixedSizeBinaryBuilder::with_capacity(total_spans, 8);
225    let mut parent_span_ids = FixedSizeBinaryBuilder::with_capacity(total_spans, 8);
226    let mut trace_states = StringBuilder::with_capacity(total_spans, total_spans * 8);
227    let mut names = StringBuilder::with_capacity(total_spans, total_spans * 32);
228    let mut kinds = Int32Builder::with_capacity(total_spans);
229    let mut start_times = Int64Builder::with_capacity(total_spans);
230    let mut end_times = Int64Builder::with_capacity(total_spans);
231    let mut durations = Int64Builder::with_capacity(total_spans);
232    let mut status_codes = Int32Builder::with_capacity(total_spans);
233    let mut status_messages = StringBuilder::with_capacity(total_spans, total_spans * 16);
234    let mut res_service_names = StringBuilder::with_capacity(total_spans, total_spans * 16);
235    let mut res_service_versions = StringBuilder::with_capacity(total_spans, total_spans * 8);
236    let mut res_attributes = StringBuilder::with_capacity(total_spans, total_spans * 64);
237    let mut scope_names = StringBuilder::with_capacity(total_spans, total_spans * 16);
238    let mut scope_versions = StringBuilder::with_capacity(total_spans, total_spans * 8);
239    let mut span_attrs = StringBuilder::with_capacity(total_spans, total_spans * 64);
240    let mut events_counts = Int32Builder::with_capacity(total_spans);
241    let mut links_counts = Int32Builder::with_capacity(total_spans);
242    let mut received_at = TimestampNanosecondBuilder::with_capacity(total_spans);
243
244    for resource_spans in &req.resource_spans {
245        let resource = resource_spans.resource.as_ref();
246        let svc_name = extract_resource_attr(resource, "service.name");
247        let svc_version = extract_resource_attr(resource, "service.version");
248        let res_attrs_json = resource_attributes_json(resource);
249
250        for scope_spans in &resource_spans.scope_spans {
251            let scope = scope_spans.scope.as_ref();
252            let scope_n = scope.map(|s| s.name.as_str());
253            let scope_v = scope.map(|s| s.version.as_str());
254
255            for span in &scope_spans.spans {
256                // trace_id: ensure exactly 16 bytes
257                let tid = fixed_bytes(&span.trace_id, 16);
258                trace_ids.append_value(&tid)?;
259
260                // span_id: ensure exactly 8 bytes
261                let sid = fixed_bytes(&span.span_id, 8);
262                span_ids.append_value(&sid)?;
263
264                // parent_span_id: null if all zeros or empty
265                let psid = fixed_bytes(&span.parent_span_id, 8);
266                if span.parent_span_id.is_empty() || is_all_zeros(&psid) {
267                    parent_span_ids.append_null();
268                } else {
269                    parent_span_ids.append_value(&psid)?;
270                }
271
272                // trace_state
273                if span.trace_state.is_empty() {
274                    trace_states.append_null();
275                } else {
276                    trace_states.append_value(&span.trace_state);
277                }
278
279                // name
280                names.append_value(&span.name);
281
282                // kind (SpanKind enum as i32)
283                kinds.append_value(span.kind);
284
285                // timestamps
286                #[allow(clippy::cast_possible_wrap)]
287                let start_ns = span.start_time_unix_nano as i64;
288                #[allow(clippy::cast_possible_wrap)]
289                let end_ns = span.end_time_unix_nano as i64;
290                start_times.append_value(start_ns);
291                end_times.append_value(end_ns);
292                durations.append_value(end_ns.saturating_sub(start_ns));
293
294                // status
295                if let Some(status) = &span.status {
296                    status_codes.append_value(status.code);
297                    if status.message.is_empty() {
298                        status_messages.append_null();
299                    } else {
300                        status_messages.append_value(&status.message);
301                    }
302                } else {
303                    status_codes.append_value(0); // STATUS_CODE_UNSET
304                    status_messages.append_null();
305                }
306
307                // resource/scope/attributes (shared helper)
308                append_context_fields(
309                    svc_name.as_deref(),
310                    res_attrs_json.as_deref(),
311                    scope_n,
312                    &span.attributes,
313                    received_at_nanos,
314                    &mut res_service_names,
315                    &mut res_attributes,
316                    &mut scope_names,
317                    &mut span_attrs,
318                    &mut received_at,
319                );
320
321                // trace-specific extra columns: service_version, scope_version
322                append_nullable_opt(&mut res_service_versions, svc_version.as_deref());
323                append_nullable_opt(&mut scope_versions, scope_v);
324
325                // counts
326                #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
327                let ec = span.events.len() as i32;
328                #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
329                let lc = span.links.len() as i32;
330                events_counts.append_value(ec);
331                links_counts.append_value(lc);
332            }
333        }
334    }
335
336    let batch = RecordBatch::try_new(
337        Arc::clone(schema),
338        vec![
339            Arc::new(trace_ids.finish()),
340            Arc::new(span_ids.finish()),
341            Arc::new(parent_span_ids.finish()),
342            Arc::new(trace_states.finish()),
343            Arc::new(names.finish()),
344            Arc::new(kinds.finish()),
345            Arc::new(start_times.finish()),
346            Arc::new(end_times.finish()),
347            Arc::new(durations.finish()),
348            Arc::new(status_codes.finish()),
349            Arc::new(status_messages.finish()),
350            Arc::new(res_service_names.finish()),
351            Arc::new(res_service_versions.finish()),
352            Arc::new(res_attributes.finish()),
353            Arc::new(scope_names.finish()),
354            Arc::new(scope_versions.finish()),
355            Arc::new(span_attrs.finish()),
356            Arc::new(events_counts.finish()),
357            Arc::new(links_counts.finish()),
358            Arc::new(received_at.finish()),
359        ],
360    )?;
361
362    Ok(Some(batch))
363}
364
365/// Convert an `AnyValue` to a string for log body. Unlike `any_value_to_string`,
366/// this handles complex types (Array, `KvList`, Bytes) by falling back to JSON.
367fn any_value_to_body_string(v: Option<&AnyValue>) -> Option<String> {
368    use opentelemetry_proto::tonic::common::v1::any_value::Value;
369    v.and_then(|av| match &av.value {
370        Some(Value::StringValue(s)) => Some(s.clone()),
371        Some(Value::IntValue(i)) => Some(i.to_string()),
372        Some(Value::BoolValue(b)) => Some(b.to_string()),
373        Some(Value::DoubleValue(d)) => Some(d.to_string()),
374        Some(_) => {
375            let mut buf = String::new();
376            write_any_value_json(&mut buf, av);
377            Some(buf)
378        }
379        None => None,
380    })
381}
382
383// ── Metrics conversion ──
384
385struct MetricBuilders {
386    names: StringBuilder,
387    descs: StringBuilder,
388    units: StringBuilder,
389    types: Int32Builder,
390    timestamps: Int64Builder,
391    value_doubles: Float64Builder,
392    value_ints: Int64Builder,
393    hist_counts: UInt64Builder,
394    hist_sums: Float64Builder,
395    res_svc_names: StringBuilder,
396    res_attrs: StringBuilder,
397    scope_names: StringBuilder,
398    attrs: StringBuilder,
399    received_at: TimestampNanosecondBuilder,
400}
401
402struct MetricContext<'a> {
403    name: &'a str,
404    desc: &'a str,
405    unit: &'a str,
406    metric_type: i32,
407    svc_name: Option<&'a str>,
408    res_json: Option<&'a str>,
409    scope_name: Option<&'a str>,
410    received_at_nanos: i64,
411}
412
413impl MetricBuilders {
414    fn with_capacity(n: usize) -> Self {
415        Self {
416            names: StringBuilder::with_capacity(n, n * 32),
417            descs: StringBuilder::with_capacity(n, n * 32),
418            units: StringBuilder::with_capacity(n, n * 8),
419            types: Int32Builder::with_capacity(n),
420            timestamps: Int64Builder::with_capacity(n),
421            value_doubles: Float64Builder::with_capacity(n),
422            value_ints: Int64Builder::with_capacity(n),
423            hist_counts: UInt64Builder::with_capacity(n),
424            hist_sums: Float64Builder::with_capacity(n),
425            res_svc_names: StringBuilder::with_capacity(n, n * 16),
426            res_attrs: StringBuilder::with_capacity(n, n * 64),
427            scope_names: StringBuilder::with_capacity(n, n * 16),
428            attrs: StringBuilder::with_capacity(n, n * 64),
429            received_at: TimestampNanosecondBuilder::with_capacity(n),
430        }
431    }
432
433    fn append_common(&mut self, ctx: &MetricContext, time_unix_nano: u64, dp_attrs: &[KeyValue]) {
434        self.names.append_value(ctx.name);
435        append_nullable_str(&mut self.descs, ctx.desc);
436        append_nullable_str(&mut self.units, ctx.unit);
437        self.types.append_value(ctx.metric_type);
438        #[allow(clippy::cast_possible_wrap)]
439        self.timestamps.append_value(time_unix_nano as i64);
440
441        append_context_fields(
442            ctx.svc_name,
443            ctx.res_json,
444            ctx.scope_name,
445            dp_attrs,
446            ctx.received_at_nanos,
447            &mut self.res_svc_names,
448            &mut self.res_attrs,
449            &mut self.scope_names,
450            &mut self.attrs,
451            &mut self.received_at,
452        );
453    }
454
455    fn append_number_dp(
456        &mut self,
457        ctx: &MetricContext,
458        time_unix_nano: u64,
459        value: Option<&number_data_point::Value>,
460        dp_attrs: &[KeyValue],
461    ) {
462        self.append_common(ctx, time_unix_nano, dp_attrs);
463        match value {
464            Some(number_data_point::Value::AsDouble(d)) => {
465                self.value_doubles.append_value(*d);
466                self.value_ints.append_null();
467            }
468            Some(number_data_point::Value::AsInt(i)) => {
469                self.value_doubles.append_null();
470                self.value_ints.append_value(*i);
471            }
472            None => {
473                self.value_doubles.append_null();
474                self.value_ints.append_null();
475            }
476        }
477        self.hist_counts.append_null();
478        self.hist_sums.append_null();
479    }
480
481    fn append_histogram_dp(
482        &mut self,
483        ctx: &MetricContext,
484        time_unix_nano: u64,
485        count: u64,
486        sum: Option<f64>,
487        dp_attrs: &[KeyValue],
488    ) {
489        self.append_common(ctx, time_unix_nano, dp_attrs);
490        self.value_doubles.append_null();
491        self.value_ints.append_null();
492        self.hist_counts.append_value(count);
493        match sum {
494            Some(s) => self.hist_sums.append_value(s),
495            None => self.hist_sums.append_null(),
496        }
497    }
498
499    fn finish(mut self, schema: &SchemaRef) -> Result<RecordBatch, arrow_schema::ArrowError> {
500        RecordBatch::try_new(
501            Arc::clone(schema),
502            vec![
503                Arc::new(self.names.finish()),
504                Arc::new(self.descs.finish()),
505                Arc::new(self.units.finish()),
506                Arc::new(self.types.finish()),
507                Arc::new(self.timestamps.finish()),
508                Arc::new(self.value_doubles.finish()),
509                Arc::new(self.value_ints.finish()),
510                Arc::new(self.hist_counts.finish()),
511                Arc::new(self.hist_sums.finish()),
512                Arc::new(self.res_svc_names.finish()),
513                Arc::new(self.res_attrs.finish()),
514                Arc::new(self.scope_names.finish()),
515                Arc::new(self.attrs.finish()),
516                Arc::new(self.received_at.finish()),
517            ],
518        )
519    }
520}
521
522/// Convert an `ExportMetricsServiceRequest` into an Arrow `RecordBatch`.
523///
524/// # Errors
525///
526/// Returns `ArrowError` if column construction fails.
527#[allow(clippy::too_many_lines)] // 5 metric types × dispatch loop; builders already extracted
528pub fn metrics_request_to_batch(
529    req: &ExportMetricsServiceRequest,
530    schema: &SchemaRef,
531    received_at_nanos: i64,
532) -> Result<Option<RecordBatch>, arrow_schema::ArrowError> {
533    let total_points: usize = req
534        .resource_metrics
535        .iter()
536        .flat_map(|rm| &rm.scope_metrics)
537        .flat_map(|sm| &sm.metrics)
538        .map(count_metric_points)
539        .sum();
540
541    if total_points == 0 {
542        return Ok(None);
543    }
544
545    let mut b = MetricBuilders::with_capacity(total_points);
546
547    for rm in &req.resource_metrics {
548        let resource = rm.resource.as_ref();
549        let svc_name = extract_resource_attr(resource, "service.name");
550        let res_json = resource_attributes_json(resource);
551
552        for sm in &rm.scope_metrics {
553            let scope_name = sm.scope.as_ref().map(|s| s.name.as_str());
554
555            for metric in &sm.metrics {
556                let Some(data) = &metric.data else { continue };
557                let base = MetricContext {
558                    name: &metric.name,
559                    desc: &metric.description,
560                    unit: &metric.unit,
561                    metric_type: 0,
562                    svc_name: svc_name.as_deref(),
563                    res_json: res_json.as_deref(),
564                    scope_name,
565                    received_at_nanos,
566                };
567
568                match data {
569                    metric::Data::Gauge(g) => {
570                        for dp in &g.data_points {
571                            b.append_number_dp(
572                                &base,
573                                dp.time_unix_nano,
574                                dp.value.as_ref(),
575                                &dp.attributes,
576                            );
577                        }
578                    }
579                    metric::Data::Sum(s) => {
580                        for dp in &s.data_points {
581                            let ctx = MetricContext {
582                                metric_type: 1,
583                                ..base
584                            };
585                            b.append_number_dp(
586                                &ctx,
587                                dp.time_unix_nano,
588                                dp.value.as_ref(),
589                                &dp.attributes,
590                            );
591                        }
592                    }
593                    metric::Data::Histogram(h) => {
594                        for dp in &h.data_points {
595                            let ctx = MetricContext {
596                                metric_type: 2,
597                                ..base
598                            };
599                            b.append_histogram_dp(
600                                &ctx,
601                                dp.time_unix_nano,
602                                dp.count,
603                                dp.sum,
604                                &dp.attributes,
605                            );
606                        }
607                    }
608                    metric::Data::ExponentialHistogram(eh) => {
609                        for dp in &eh.data_points {
610                            let ctx = MetricContext {
611                                metric_type: 3,
612                                ..base
613                            };
614                            b.append_histogram_dp(
615                                &ctx,
616                                dp.time_unix_nano,
617                                dp.count,
618                                dp.sum,
619                                &dp.attributes,
620                            );
621                        }
622                    }
623                    metric::Data::Summary(s) => {
624                        for dp in &s.data_points {
625                            let ctx = MetricContext {
626                                metric_type: 4,
627                                ..base
628                            };
629                            b.append_histogram_dp(
630                                &ctx,
631                                dp.time_unix_nano,
632                                dp.count,
633                                Some(dp.sum),
634                                &dp.attributes,
635                            );
636                        }
637                    }
638                }
639            }
640        }
641    }
642
643    Ok(Some(b.finish(schema)?))
644}
645
646fn count_metric_points(m: &opentelemetry_proto::tonic::metrics::v1::Metric) -> usize {
647    match &m.data {
648        Some(metric::Data::Gauge(g)) => g.data_points.len(),
649        Some(metric::Data::Sum(s)) => s.data_points.len(),
650        Some(metric::Data::Histogram(h)) => h.data_points.len(),
651        Some(metric::Data::ExponentialHistogram(eh)) => eh.data_points.len(),
652        Some(metric::Data::Summary(s)) => s.data_points.len(),
653        None => 0,
654    }
655}
656
657/// Append resource/scope/attribute/`received_at` fields (shared across all metric types).
658#[allow(clippy::too_many_arguments)]
659fn append_context_fields(
660    svc_name: Option<&str>,
661    res_json: Option<&str>,
662    scope_name: Option<&str>,
663    attrs: &[KeyValue],
664    received_at_nanos: i64,
665    res_svc_names: &mut StringBuilder,
666    res_attrs: &mut StringBuilder,
667    scope_names_col: &mut StringBuilder,
668    attrs_col: &mut StringBuilder,
669    received_at: &mut TimestampNanosecondBuilder,
670) {
671    match svc_name {
672        Some(s) => res_svc_names.append_value(s),
673        None => res_svc_names.append_null(),
674    }
675    match res_json {
676        Some(s) => res_attrs.append_value(s),
677        None => res_attrs.append_null(),
678    }
679    match scope_name {
680        Some(s) if !s.is_empty() => scope_names_col.append_value(s),
681        _ => scope_names_col.append_null(),
682    }
683    match kv_to_json(attrs) {
684        Some(s) => attrs_col.append_value(&s),
685        None => attrs_col.append_null(),
686    }
687    received_at.append_value(received_at_nanos);
688}
689
690/// Append a string value as non-null if non-empty, null otherwise.
691fn append_nullable_str(builder: &mut StringBuilder, s: &str) {
692    if s.is_empty() {
693        builder.append_null();
694    } else {
695        builder.append_value(s);
696    }
697}
698
699/// Append an `Option<&str>` — `Some(non-empty)` → value, else null.
700fn append_nullable_opt(builder: &mut StringBuilder, s: Option<&str>) {
701    match s {
702        Some(v) if !v.is_empty() => builder.append_value(v),
703        _ => builder.append_null(),
704    }
705}
706
707// ── Logs conversion ──
708
709/// Convert an `ExportLogsServiceRequest` into an Arrow `RecordBatch`.
710///
711/// # Errors
712///
713/// Returns `ArrowError` if column construction fails.
714pub fn logs_request_to_batch(
715    req: &ExportLogsServiceRequest,
716    schema: &SchemaRef,
717    received_at_nanos: i64,
718) -> Result<Option<RecordBatch>, arrow_schema::ArrowError> {
719    let total_records: usize = req
720        .resource_logs
721        .iter()
722        .flat_map(|rl| &rl.scope_logs)
723        .map(|sl| sl.log_records.len())
724        .sum();
725
726    if total_records == 0 {
727        return Ok(None);
728    }
729
730    let mut ts = Int64Builder::with_capacity(total_records);
731    let mut observed_ts = Int64Builder::with_capacity(total_records);
732    let mut sev_nums = Int32Builder::with_capacity(total_records);
733    let mut sev_texts = StringBuilder::with_capacity(total_records, total_records * 8);
734    let mut bodies = StringBuilder::with_capacity(total_records, total_records * 64);
735    let mut trace_ids = FixedSizeBinaryBuilder::with_capacity(total_records, 16);
736    let mut span_ids = FixedSizeBinaryBuilder::with_capacity(total_records, 8);
737    let mut res_svc_names = StringBuilder::with_capacity(total_records, total_records * 16);
738    let mut res_attrs = StringBuilder::with_capacity(total_records, total_records * 64);
739    let mut scope_names_col = StringBuilder::with_capacity(total_records, total_records * 16);
740    let mut attrs_col = StringBuilder::with_capacity(total_records, total_records * 64);
741    let mut received_at = TimestampNanosecondBuilder::with_capacity(total_records);
742
743    for rl in &req.resource_logs {
744        let resource = rl.resource.as_ref();
745        let svc_name = extract_resource_attr(resource, "service.name");
746        let res_json = resource_attributes_json(resource);
747
748        for sl in &rl.scope_logs {
749            let scope_name = sl.scope.as_ref().map(|s| s.name.as_str());
750
751            for log in &sl.log_records {
752                #[allow(clippy::cast_possible_wrap)]
753                ts.append_value(log.time_unix_nano as i64);
754
755                if log.observed_time_unix_nano == 0 {
756                    observed_ts.append_null();
757                } else {
758                    #[allow(clippy::cast_possible_wrap)]
759                    observed_ts.append_value(log.observed_time_unix_nano as i64);
760                }
761
762                sev_nums.append_value(log.severity_number);
763
764                if log.severity_text.is_empty() {
765                    sev_texts.append_null();
766                } else {
767                    sev_texts.append_value(&log.severity_text);
768                }
769
770                match any_value_to_body_string(log.body.as_ref()) {
771                    Some(s) => bodies.append_value(&s),
772                    None => bodies.append_null(),
773                }
774
775                // trace_id: null if empty or all-zeros
776                let tid = fixed_bytes(&log.trace_id, 16);
777                if log.trace_id.is_empty() || is_all_zeros(&tid) {
778                    trace_ids.append_null();
779                } else {
780                    trace_ids.append_value(&tid)?;
781                }
782
783                // span_id: null if empty or all-zeros
784                let sid = fixed_bytes(&log.span_id, 8);
785                if log.span_id.is_empty() || is_all_zeros(&sid) {
786                    span_ids.append_null();
787                } else {
788                    span_ids.append_value(&sid)?;
789                }
790
791                append_context_fields(
792                    svc_name.as_deref(),
793                    res_json.as_deref(),
794                    scope_name,
795                    &log.attributes,
796                    received_at_nanos,
797                    &mut res_svc_names,
798                    &mut res_attrs,
799                    &mut scope_names_col,
800                    &mut attrs_col,
801                    &mut received_at,
802                );
803            }
804        }
805    }
806
807    let batch = RecordBatch::try_new(
808        Arc::clone(schema),
809        vec![
810            Arc::new(ts.finish()),
811            Arc::new(observed_ts.finish()),
812            Arc::new(sev_nums.finish()),
813            Arc::new(sev_texts.finish()),
814            Arc::new(bodies.finish()),
815            Arc::new(trace_ids.finish()),
816            Arc::new(span_ids.finish()),
817            Arc::new(res_svc_names.finish()),
818            Arc::new(res_attrs.finish()),
819            Arc::new(scope_names_col.finish()),
820            Arc::new(attrs_col.finish()),
821            Arc::new(received_at.finish()),
822        ],
823    )?;
824
825    Ok(Some(batch))
826}
827
828#[cfg(test)]
829mod tests;
laminar_connectors/otel/convert/mod.rs

laminar_connectors/otel/convert/
mod.rs
