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        // OTLP defines string-table indexes only for profiles; other signals treat them as absent.
184        Some(Value::StringValueStrindex(_)) | None => buf.push_str("null"),
185    }
186}
187
188/// Convert an `AnyValue` to a `String` (for promoted fields like service.name).
189fn any_value_to_string(v: Option<&AnyValue>) -> Option<String> {
190    use opentelemetry_proto::tonic::common::v1::any_value::Value;
191    v.and_then(|av| match &av.value {
192        Some(Value::StringValue(s)) => Some(s.clone()),
193        Some(Value::IntValue(i)) => Some(i.to_string()),
194        Some(Value::BoolValue(b)) => Some(b.to_string()),
195        Some(Value::DoubleValue(d)) => Some(d.to_string()),
196        _ => None,
197    })
198}
199
200/// Convert an `ExportTraceServiceRequest` into a `RecordBatch`, or `None` if empty.
201///
202/// # Errors
203///
204/// Returns `ArrowError` if column construction fails.
205#[allow(clippy::too_many_lines)] // 20 columns = inherently long
206pub fn trace_request_to_batch(
207    req: &ExportTraceServiceRequest,
208    schema: &SchemaRef,
209    received_at_nanos: i64,
210) -> Result<Option<RecordBatch>, arrow_schema::ArrowError> {
211    // Count total spans for capacity hints
212    let total_spans: usize = req
213        .resource_spans
214        .iter()
215        .flat_map(|rs| &rs.scope_spans)
216        .map(|ss| ss.spans.len())
217        .sum();
218
219    if total_spans == 0 {
220        return Ok(None);
221    }
222
223    // Column builders
224    let mut trace_ids = FixedSizeBinaryBuilder::with_capacity(total_spans, 16);
225    let mut span_ids = FixedSizeBinaryBuilder::with_capacity(total_spans, 8);
226    let mut parent_span_ids = FixedSizeBinaryBuilder::with_capacity(total_spans, 8);
227    let mut trace_states = StringBuilder::with_capacity(total_spans, total_spans * 8);
228    let mut names = StringBuilder::with_capacity(total_spans, total_spans * 32);
229    let mut kinds = Int32Builder::with_capacity(total_spans);
230    let mut start_times = Int64Builder::with_capacity(total_spans);
231    let mut end_times = Int64Builder::with_capacity(total_spans);
232    let mut durations = Int64Builder::with_capacity(total_spans);
233    let mut status_codes = Int32Builder::with_capacity(total_spans);
234    let mut status_messages = StringBuilder::with_capacity(total_spans, total_spans * 16);
235    let mut res_service_names = StringBuilder::with_capacity(total_spans, total_spans * 16);
236    let mut res_service_versions = StringBuilder::with_capacity(total_spans, total_spans * 8);
237    let mut res_attributes = StringBuilder::with_capacity(total_spans, total_spans * 64);
238    let mut scope_names = StringBuilder::with_capacity(total_spans, total_spans * 16);
239    let mut scope_versions = StringBuilder::with_capacity(total_spans, total_spans * 8);
240    let mut span_attrs = StringBuilder::with_capacity(total_spans, total_spans * 64);
241    let mut events_counts = Int32Builder::with_capacity(total_spans);
242    let mut links_counts = Int32Builder::with_capacity(total_spans);
243    let mut received_at = TimestampNanosecondBuilder::with_capacity(total_spans);
244
245    for resource_spans in &req.resource_spans {
246        let resource = resource_spans.resource.as_ref();
247        let svc_name = extract_resource_attr(resource, "service.name");
248        let svc_version = extract_resource_attr(resource, "service.version");
249        let res_attrs_json = resource_attributes_json(resource);
250
251        for scope_spans in &resource_spans.scope_spans {
252            let scope = scope_spans.scope.as_ref();
253            let scope_n = scope.map(|s| s.name.as_str());
254            let scope_v = scope.map(|s| s.version.as_str());
255
256            for span in &scope_spans.spans {
257                // trace_id: ensure exactly 16 bytes
258                let tid = fixed_bytes(&span.trace_id, 16);
259                trace_ids.append_value(&tid)?;
260
261                // span_id: ensure exactly 8 bytes
262                let sid = fixed_bytes(&span.span_id, 8);
263                span_ids.append_value(&sid)?;
264
265                // parent_span_id: null if all zeros or empty
266                let psid = fixed_bytes(&span.parent_span_id, 8);
267                if span.parent_span_id.is_empty() || is_all_zeros(&psid) {
268                    parent_span_ids.append_null();
269                } else {
270                    parent_span_ids.append_value(&psid)?;
271                }
272
273                // trace_state
274                if span.trace_state.is_empty() {
275                    trace_states.append_null();
276                } else {
277                    trace_states.append_value(&span.trace_state);
278                }
279
280                // name
281                names.append_value(&span.name);
282
283                // kind (SpanKind enum as i32)
284                kinds.append_value(span.kind);
285
286                // timestamps
287                #[allow(clippy::cast_possible_wrap)]
288                let start_ns = span.start_time_unix_nano as i64;
289                #[allow(clippy::cast_possible_wrap)]
290                let end_ns = span.end_time_unix_nano as i64;
291                start_times.append_value(start_ns);
292                end_times.append_value(end_ns);
293                durations.append_value(end_ns.saturating_sub(start_ns));
294
295                // status
296                if let Some(status) = &span.status {
297                    status_codes.append_value(status.code);
298                    if status.message.is_empty() {
299                        status_messages.append_null();
300                    } else {
301                        status_messages.append_value(&status.message);
302                    }
303                } else {
304                    status_codes.append_value(0); // STATUS_CODE_UNSET
305                    status_messages.append_null();
306                }
307
308                // resource/scope/attributes (shared helper)
309                append_context_fields(
310                    svc_name.as_deref(),
311                    res_attrs_json.as_deref(),
312                    scope_n,
313                    &span.attributes,
314                    received_at_nanos,
315                    &mut res_service_names,
316                    &mut res_attributes,
317                    &mut scope_names,
318                    &mut span_attrs,
319                    &mut received_at,
320                );
321
322                // trace-specific extra columns: service_version, scope_version
323                append_nullable_opt(&mut res_service_versions, svc_version.as_deref());
324                append_nullable_opt(&mut scope_versions, scope_v);
325
326                // counts
327                #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
328                let ec = span.events.len() as i32;
329                #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
330                let lc = span.links.len() as i32;
331                events_counts.append_value(ec);
332                links_counts.append_value(lc);
333            }
334        }
335    }
336
337    let batch = RecordBatch::try_new(
338        Arc::clone(schema),
339        vec![
340            Arc::new(trace_ids.finish()),
341            Arc::new(span_ids.finish()),
342            Arc::new(parent_span_ids.finish()),
343            Arc::new(trace_states.finish()),
344            Arc::new(names.finish()),
345            Arc::new(kinds.finish()),
346            Arc::new(start_times.finish()),
347            Arc::new(end_times.finish()),
348            Arc::new(durations.finish()),
349            Arc::new(status_codes.finish()),
350            Arc::new(status_messages.finish()),
351            Arc::new(res_service_names.finish()),
352            Arc::new(res_service_versions.finish()),
353            Arc::new(res_attributes.finish()),
354            Arc::new(scope_names.finish()),
355            Arc::new(scope_versions.finish()),
356            Arc::new(span_attrs.finish()),
357            Arc::new(events_counts.finish()),
358            Arc::new(links_counts.finish()),
359            Arc::new(received_at.finish()),
360        ],
361    )?;
362
363    Ok(Some(batch))
364}
365
366/// Convert an `AnyValue` to a string for log body. Unlike `any_value_to_string`,
367/// this handles complex types (Array, `KvList`, Bytes) by falling back to JSON.
368fn any_value_to_body_string(v: Option<&AnyValue>) -> Option<String> {
369    use opentelemetry_proto::tonic::common::v1::any_value::Value;
370    v.and_then(|av| match &av.value {
371        Some(Value::StringValue(s)) => Some(s.clone()),
372        Some(Value::IntValue(i)) => Some(i.to_string()),
373        Some(Value::BoolValue(b)) => Some(b.to_string()),
374        Some(Value::DoubleValue(d)) => Some(d.to_string()),
375        Some(_) => {
376            let mut buf = String::new();
377            write_any_value_json(&mut buf, av);
378            Some(buf)
379        }
380        None => None,
381    })
382}
383
384// ── Metrics conversion ──
385
386struct MetricBuilders {
387    names: StringBuilder,
388    descs: StringBuilder,
389    units: StringBuilder,
390    types: Int32Builder,
391    timestamps: Int64Builder,
392    value_doubles: Float64Builder,
393    value_ints: Int64Builder,
394    hist_counts: UInt64Builder,
395    hist_sums: Float64Builder,
396    res_svc_names: StringBuilder,
397    res_attrs: StringBuilder,
398    scope_names: StringBuilder,
399    attrs: StringBuilder,
400    received_at: TimestampNanosecondBuilder,
401}
402
403struct MetricContext<'a> {
404    name: &'a str,
405    desc: &'a str,
406    unit: &'a str,
407    metric_type: i32,
408    svc_name: Option<&'a str>,
409    res_json: Option<&'a str>,
410    scope_name: Option<&'a str>,
411    received_at_nanos: i64,
412}
413
414impl MetricBuilders {
415    fn with_capacity(n: usize) -> Self {
416        Self {
417            names: StringBuilder::with_capacity(n, n * 32),
418            descs: StringBuilder::with_capacity(n, n * 32),
419            units: StringBuilder::with_capacity(n, n * 8),
420            types: Int32Builder::with_capacity(n),
421            timestamps: Int64Builder::with_capacity(n),
422            value_doubles: Float64Builder::with_capacity(n),
423            value_ints: Int64Builder::with_capacity(n),
424            hist_counts: UInt64Builder::with_capacity(n),
425            hist_sums: Float64Builder::with_capacity(n),
426            res_svc_names: StringBuilder::with_capacity(n, n * 16),
427            res_attrs: StringBuilder::with_capacity(n, n * 64),
428            scope_names: StringBuilder::with_capacity(n, n * 16),
429            attrs: StringBuilder::with_capacity(n, n * 64),
430            received_at: TimestampNanosecondBuilder::with_capacity(n),
431        }
432    }
433
434    fn append_common(&mut self, ctx: &MetricContext, time_unix_nano: u64, dp_attrs: &[KeyValue]) {
435        self.names.append_value(ctx.name);
436        append_nullable_str(&mut self.descs, ctx.desc);
437        append_nullable_str(&mut self.units, ctx.unit);
438        self.types.append_value(ctx.metric_type);
439        #[allow(clippy::cast_possible_wrap)]
440        self.timestamps.append_value(time_unix_nano as i64);
441
442        append_context_fields(
443            ctx.svc_name,
444            ctx.res_json,
445            ctx.scope_name,
446            dp_attrs,
447            ctx.received_at_nanos,
448            &mut self.res_svc_names,
449            &mut self.res_attrs,
450            &mut self.scope_names,
451            &mut self.attrs,
452            &mut self.received_at,
453        );
454    }
455
456    fn append_number_dp(
457        &mut self,
458        ctx: &MetricContext,
459        time_unix_nano: u64,
460        value: Option<&number_data_point::Value>,
461        dp_attrs: &[KeyValue],
462    ) {
463        self.append_common(ctx, time_unix_nano, dp_attrs);
464        match value {
465            Some(number_data_point::Value::AsDouble(d)) => {
466                self.value_doubles.append_value(*d);
467                self.value_ints.append_null();
468            }
469            Some(number_data_point::Value::AsInt(i)) => {
470                self.value_doubles.append_null();
471                self.value_ints.append_value(*i);
472            }
473            None => {
474                self.value_doubles.append_null();
475                self.value_ints.append_null();
476            }
477        }
478        self.hist_counts.append_null();
479        self.hist_sums.append_null();
480    }
481
482    fn append_histogram_dp(
483        &mut self,
484        ctx: &MetricContext,
485        time_unix_nano: u64,
486        count: u64,
487        sum: Option<f64>,
488        dp_attrs: &[KeyValue],
489    ) {
490        self.append_common(ctx, time_unix_nano, dp_attrs);
491        self.value_doubles.append_null();
492        self.value_ints.append_null();
493        self.hist_counts.append_value(count);
494        match sum {
495            Some(s) => self.hist_sums.append_value(s),
496            None => self.hist_sums.append_null(),
497        }
498    }
499
500    fn finish(mut self, schema: &SchemaRef) -> Result<RecordBatch, arrow_schema::ArrowError> {
501        RecordBatch::try_new(
502            Arc::clone(schema),
503            vec![
504                Arc::new(self.names.finish()),
505                Arc::new(self.descs.finish()),
506                Arc::new(self.units.finish()),
507                Arc::new(self.types.finish()),
508                Arc::new(self.timestamps.finish()),
509                Arc::new(self.value_doubles.finish()),
510                Arc::new(self.value_ints.finish()),
511                Arc::new(self.hist_counts.finish()),
512                Arc::new(self.hist_sums.finish()),
513                Arc::new(self.res_svc_names.finish()),
514                Arc::new(self.res_attrs.finish()),
515                Arc::new(self.scope_names.finish()),
516                Arc::new(self.attrs.finish()),
517                Arc::new(self.received_at.finish()),
518            ],
519        )
520    }
521}
522
523/// Convert an `ExportMetricsServiceRequest` into an Arrow `RecordBatch`.
524///
525/// # Errors
526///
527/// Returns `ArrowError` if column construction fails.
528#[allow(clippy::too_many_lines)] // 5 metric types × dispatch loop; builders already extracted
529pub fn metrics_request_to_batch(
530    req: &ExportMetricsServiceRequest,
531    schema: &SchemaRef,
532    received_at_nanos: i64,
533) -> Result<Option<RecordBatch>, arrow_schema::ArrowError> {
534    let total_points: usize = req
535        .resource_metrics
536        .iter()
537        .flat_map(|rm| &rm.scope_metrics)
538        .flat_map(|sm| &sm.metrics)
539        .map(count_metric_points)
540        .sum();
541
542    if total_points == 0 {
543        return Ok(None);
544    }
545
546    let mut b = MetricBuilders::with_capacity(total_points);
547
548    for rm in &req.resource_metrics {
549        let resource = rm.resource.as_ref();
550        let svc_name = extract_resource_attr(resource, "service.name");
551        let res_json = resource_attributes_json(resource);
552
553        for sm in &rm.scope_metrics {
554            let scope_name = sm.scope.as_ref().map(|s| s.name.as_str());
555
556            for metric in &sm.metrics {
557                let Some(data) = &metric.data else { continue };
558                let base = MetricContext {
559                    name: &metric.name,
560                    desc: &metric.description,
561                    unit: &metric.unit,
562                    metric_type: 0,
563                    svc_name: svc_name.as_deref(),
564                    res_json: res_json.as_deref(),
565                    scope_name,
566                    received_at_nanos,
567                };
568
569                match data {
570                    metric::Data::Gauge(g) => {
571                        for dp in &g.data_points {
572                            b.append_number_dp(
573                                &base,
574                                dp.time_unix_nano,
575                                dp.value.as_ref(),
576                                &dp.attributes,
577                            );
578                        }
579                    }
580                    metric::Data::Sum(s) => {
581                        for dp in &s.data_points {
582                            let ctx = MetricContext {
583                                metric_type: 1,
584                                ..base
585                            };
586                            b.append_number_dp(
587                                &ctx,
588                                dp.time_unix_nano,
589                                dp.value.as_ref(),
590                                &dp.attributes,
591                            );
592                        }
593                    }
594                    metric::Data::Histogram(h) => {
595                        for dp in &h.data_points {
596                            let ctx = MetricContext {
597                                metric_type: 2,
598                                ..base
599                            };
600                            b.append_histogram_dp(
601                                &ctx,
602                                dp.time_unix_nano,
603                                dp.count,
604                                dp.sum,
605                                &dp.attributes,
606                            );
607                        }
608                    }
609                    metric::Data::ExponentialHistogram(eh) => {
610                        for dp in &eh.data_points {
611                            let ctx = MetricContext {
612                                metric_type: 3,
613                                ..base
614                            };
615                            b.append_histogram_dp(
616                                &ctx,
617                                dp.time_unix_nano,
618                                dp.count,
619                                dp.sum,
620                                &dp.attributes,
621                            );
622                        }
623                    }
624                    metric::Data::Summary(s) => {
625                        for dp in &s.data_points {
626                            let ctx = MetricContext {
627                                metric_type: 4,
628                                ..base
629                            };
630                            b.append_histogram_dp(
631                                &ctx,
632                                dp.time_unix_nano,
633                                dp.count,
634                                Some(dp.sum),
635                                &dp.attributes,
636                            );
637                        }
638                    }
639                }
640            }
641        }
642    }
643
644    Ok(Some(b.finish(schema)?))
645}
646
647fn count_metric_points(m: &opentelemetry_proto::tonic::metrics::v1::Metric) -> usize {
648    match &m.data {
649        Some(metric::Data::Gauge(g)) => g.data_points.len(),
650        Some(metric::Data::Sum(s)) => s.data_points.len(),
651        Some(metric::Data::Histogram(h)) => h.data_points.len(),
652        Some(metric::Data::ExponentialHistogram(eh)) => eh.data_points.len(),
653        Some(metric::Data::Summary(s)) => s.data_points.len(),
654        None => 0,
655    }
656}
657
658/// Append resource/scope/attribute/`received_at` fields (shared across all metric types).
659#[allow(clippy::too_many_arguments)]
660fn append_context_fields(
661    svc_name: Option<&str>,
662    res_json: Option<&str>,
663    scope_name: Option<&str>,
664    attrs: &[KeyValue],
665    received_at_nanos: i64,
666    res_svc_names: &mut StringBuilder,
667    res_attrs: &mut StringBuilder,
668    scope_names_col: &mut StringBuilder,
669    attrs_col: &mut StringBuilder,
670    received_at: &mut TimestampNanosecondBuilder,
671) {
672    match svc_name {
673        Some(s) => res_svc_names.append_value(s),
674        None => res_svc_names.append_null(),
675    }
676    match res_json {
677        Some(s) => res_attrs.append_value(s),
678        None => res_attrs.append_null(),
679    }
680    match scope_name {
681        Some(s) if !s.is_empty() => scope_names_col.append_value(s),
682        _ => scope_names_col.append_null(),
683    }
684    match kv_to_json(attrs) {
685        Some(s) => attrs_col.append_value(&s),
686        None => attrs_col.append_null(),
687    }
688    received_at.append_value(received_at_nanos);
689}
690
691/// Append a string value as non-null if non-empty, null otherwise.
692fn append_nullable_str(builder: &mut StringBuilder, s: &str) {
693    if s.is_empty() {
694        builder.append_null();
695    } else {
696        builder.append_value(s);
697    }
698}
699
700/// Append an `Option<&str>` — `Some(non-empty)` → value, else null.
701fn append_nullable_opt(builder: &mut StringBuilder, s: Option<&str>) {
702    match s {
703        Some(v) if !v.is_empty() => builder.append_value(v),
704        _ => builder.append_null(),
705    }
706}
707
708// ── Logs conversion ──
709
710/// Convert an `ExportLogsServiceRequest` into an Arrow `RecordBatch`.
711///
712/// # Errors
713///
714/// Returns `ArrowError` if column construction fails.
715pub fn logs_request_to_batch(
716    req: &ExportLogsServiceRequest,
717    schema: &SchemaRef,
718    received_at_nanos: i64,
719) -> Result<Option<RecordBatch>, arrow_schema::ArrowError> {
720    let total_records: usize = req
721        .resource_logs
722        .iter()
723        .flat_map(|rl| &rl.scope_logs)
724        .map(|sl| sl.log_records.len())
725        .sum();
726
727    if total_records == 0 {
728        return Ok(None);
729    }
730
731    let mut ts = Int64Builder::with_capacity(total_records);
732    let mut observed_ts = Int64Builder::with_capacity(total_records);
733    let mut sev_nums = Int32Builder::with_capacity(total_records);
734    let mut sev_texts = StringBuilder::with_capacity(total_records, total_records * 8);
735    let mut bodies = StringBuilder::with_capacity(total_records, total_records * 64);
736    let mut trace_ids = FixedSizeBinaryBuilder::with_capacity(total_records, 16);
737    let mut span_ids = FixedSizeBinaryBuilder::with_capacity(total_records, 8);
738    let mut res_svc_names = StringBuilder::with_capacity(total_records, total_records * 16);
739    let mut res_attrs = StringBuilder::with_capacity(total_records, total_records * 64);
740    let mut scope_names_col = StringBuilder::with_capacity(total_records, total_records * 16);
741    let mut attrs_col = StringBuilder::with_capacity(total_records, total_records * 64);
742    let mut received_at = TimestampNanosecondBuilder::with_capacity(total_records);
743
744    for rl in &req.resource_logs {
745        let resource = rl.resource.as_ref();
746        let svc_name = extract_resource_attr(resource, "service.name");
747        let res_json = resource_attributes_json(resource);
748
749        for sl in &rl.scope_logs {
750            let scope_name = sl.scope.as_ref().map(|s| s.name.as_str());
751
752            for log in &sl.log_records {
753                #[allow(clippy::cast_possible_wrap)]
754                ts.append_value(log.time_unix_nano as i64);
755
756                if log.observed_time_unix_nano == 0 {
757                    observed_ts.append_null();
758                } else {
759                    #[allow(clippy::cast_possible_wrap)]
760                    observed_ts.append_value(log.observed_time_unix_nano as i64);
761                }
762
763                sev_nums.append_value(log.severity_number);
764
765                if log.severity_text.is_empty() {
766                    sev_texts.append_null();
767                } else {
768                    sev_texts.append_value(&log.severity_text);
769                }
770
771                match any_value_to_body_string(log.body.as_ref()) {
772                    Some(s) => bodies.append_value(&s),
773                    None => bodies.append_null(),
774                }
775
776                // trace_id: null if empty or all-zeros
777                let tid = fixed_bytes(&log.trace_id, 16);
778                if log.trace_id.is_empty() || is_all_zeros(&tid) {
779                    trace_ids.append_null();
780                } else {
781                    trace_ids.append_value(&tid)?;
782                }
783
784                // span_id: null if empty or all-zeros
785                let sid = fixed_bytes(&log.span_id, 8);
786                if log.span_id.is_empty() || is_all_zeros(&sid) {
787                    span_ids.append_null();
788                } else {
789                    span_ids.append_value(&sid)?;
790                }
791
792                append_context_fields(
793                    svc_name.as_deref(),
794                    res_json.as_deref(),
795                    scope_name,
796                    &log.attributes,
797                    received_at_nanos,
798                    &mut res_svc_names,
799                    &mut res_attrs,
800                    &mut scope_names_col,
801                    &mut attrs_col,
802                    &mut received_at,
803                );
804            }
805        }
806    }
807
808    let batch = RecordBatch::try_new(
809        Arc::clone(schema),
810        vec![
811            Arc::new(ts.finish()),
812            Arc::new(observed_ts.finish()),
813            Arc::new(sev_nums.finish()),
814            Arc::new(sev_texts.finish()),
815            Arc::new(bodies.finish()),
816            Arc::new(trace_ids.finish()),
817            Arc::new(span_ids.finish()),
818            Arc::new(res_svc_names.finish()),
819            Arc::new(res_attrs.finish()),
820            Arc::new(scope_names_col.finish()),
821            Arc::new(attrs_col.finish()),
822            Arc::new(received_at.finish()),
823        ],
824    )?;
825
826    Ok(Some(batch))
827}
828
829#[cfg(test)]
830mod tests;
laminar_connectors/otel/convert/mod.rs

laminar_connectors/otel/convert/
mod.rs
