feat(codecs): convert native OTLP traces through the typed trace model in otlp serializer

changelog.d/otlp_native_trace_encoding.feature.md

@@ -0,0 +1,6 @@
+The `otlp` codec now converts native trace events from the `opentelemetry` source (decoded without
+`use_otlp_decoding`) back to OTLP spans, so they can be sent with the `opentelemetry` sink. Spans and
+fields that cannot be converted are omitted and reported as errors. A trace whose spans are all
+rejected is left out of the request instead of failing the batch. Datadog traces are still rejected.
+
+authors: thomasqueirozb

lib/codecs/Cargo.toml

@@ -82,7 +82,7 @@ zstd = { workspace = true, features = ["zdict_builder"] }
 [features]
 arrow = ["dep:arrow", "arrow/chrono-tz"]
 parquet = ["dep:parquet", "arrow"]
-opentelemetry = ["dep:opentelemetry-proto"]
+opentelemetry = ["dep:opentelemetry-proto", "opentelemetry-proto/typed-trace", "vector-core/typed-trace"]
 syslog = ["dep:syslog_loose", "dep:strum", "dep:toml"]
 test = []
 

lib/codecs/src/encoding/format/otlp.rs

@@ -1,15 +1,25 @@
 use crate::encoding::ProtobufSerializer;
+use crate::internal_events::OtlpTraceConversionIssue;
 use bytes::BytesMut;
 use opentelemetry_proto::metrics::metric_event_to_export_request;
 use opentelemetry_proto::proto::{
     DESCRIPTOR_BYTES, LOGS_REQUEST_MESSAGE_TYPE, METRICS_REQUEST_MESSAGE_TYPE,
     RESOURCE_LOGS_JSON_FIELD, RESOURCE_METRICS_JSON_FIELD, RESOURCE_SPANS_JSON_FIELD,
-    TRACES_REQUEST_MESSAGE_TYPE,
+    TRACES_REQUEST_MESSAGE_TYPE, collector::trace::v1::ExportTraceServiceRequest,
 };
+use opentelemetry_proto::typed_trace::legacy_to_typed;
 use prost::Message;
 use tokio_util::codec::Encoder;
+use vector_common::internal_event::{ComponentEventsDropped, UNINTENTIONAL, emit};
 use vector_config_macros::configurable_component;
-use vector_core::{config::DataType, event::Event, schema};
+use vector_core::{
+    config::DataType,
+    event::{
+        Event, TraceEvent,
+        typed_trace::{TraceConversionIssue, TraceConversionReporter},
+    },
+    schema,
+};
 use vrl::{event_path, protobuf::encode::Options};
 
 /// Config used to build an `OtlpSerializer`.
@@ -129,11 +139,12 @@ impl Encoder<Event> for OtlpSerializer {
             Event::Trace(trace) => {
                 if trace.contains(event_path!(RESOURCE_SPANS_JSON_FIELD)) {
                     self.traces_descriptor.encode(Event::Trace(trace), buffer)
+                } else if let Some(request) = native_trace_to_export_request(trace)? {
+                    request.encode(buffer).map_err(Into::into)
                 } else {
-                    Err(format!(
-                        "Trace event does not contain OTLP top-level field ({RESOURCE_SPANS_JSON_FIELD})",
-                    )
-                        .into())
+                    // Every span was rejected. The conversion already reported that, and
+                    // RFC 25329 treats a conversion that yields no events as success.
+                    Ok(())
                 }
             }
             Event::Metric(metric) => {
@@ -144,6 +155,42 @@ impl Encoder<Event> for OtlpSerializer {
     }
 }
 
+/// Convert a native trace event from the `opentelemetry` source (one flattened span per event)
+/// into an OTLP export request, through the typed trace model (RFC 25329).
+///
+/// Returns `Ok(None)` when every span was rejected. That is a successful conversion with
+/// nothing to send, so one bad span does not fail the rest of the batch. Fails for traces
+/// in other layouts, such as Datadog traces.
+fn native_trace_to_export_request(
+    trace: TraceEvent,
+) -> vector_common::Result<Option<ExportTraceServiceRequest>> {
+    let mut reporter = EmitTraceConversionIssue;
+    let events = legacy_to_typed(trace, &mut reporter)
+        .map_err(|failed| -> vector_common::Error { Box::new(failed.error()) })?;
+    if events.is_empty() {
+        // Links and attributes dropped inside a span that is still encoded are counted by
+        // `OtlpTraceConversionIssue`. This is the case where the Vector event itself
+        // contributed no spans.
+        emit(ComponentEventsDropped::<UNINTENTIONAL> {
+            count: 1,
+            reason: "Trace event has no spans that can be encoded as OTLP.",
+        });
+        return Ok(None);
+    }
+    // The finalizers belong to the input event. Sinks take them before encoding, so this
+    // set is empty and dropping it acknowledges nothing.
+    let (request, _finalizers) = ExportTraceServiceRequest::from_events(events, &mut reporter);
+    Ok(Some(request))
+}
+
+struct EmitTraceConversionIssue;
+
+impl TraceConversionReporter for EmitTraceConversionIssue {
+    fn report(&mut self, issue: TraceConversionIssue<'_>) {
+        emit(OtlpTraceConversionIssue { issue });
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -298,4 +345,155 @@ mod tests {
                 .is_err()
         );
     }
+
+    mod native_traces {
+        use opentelemetry_proto::proto::{
+            common::v1::{AnyValue, KeyValue, any_value::Value as PBValue},
+            resource::v1::Resource,
+            trace::v1::{ResourceSpans, ScopeSpans, Span},
+        };
+        use vector_core::event::TraceLayout;
+
+        use super::*;
+
+        const TRACE_ID: [u8; 16] = [1; 16];
+        const SPAN_ID: [u8; 8] = [2; 8];
+
+        /// Decode one span the way the `opentelemetry` source does without
+        /// `use_otlp_decoding`, giving a flattened native trace event.
+        fn flattened_trace(trace_id: &[u8], end_time_unix_nano: u64) -> Event {
+            let resource_spans = ResourceSpans {
+                resource: Some(Resource {
+                    attributes: vec![KeyValue {
+                        key: "service.name".into(),
+                        value: Some(AnyValue {
+                            value: Some(PBValue::StringValue("checkout".into())),
+                        }),
+                    }],
+                    ..Default::default()
+                }),
+                scope_spans: vec![ScopeSpans {
+                    scope: None,
+                    spans: vec![Span {
+                        trace_id: trace_id.to_vec(),
+                        span_id: SPAN_ID.to_vec(),
+                        name: "GET /cart".into(),
+                        kind: 2,
+                        start_time_unix_nano: 1_000,
+                        end_time_unix_nano,
+                        ..Default::default()
+                    }],
+                    schema_url: String::new(),
+                }],
+                schema_url: String::new(),
+            };
+            let mut events: Vec<Event> = resource_spans.into_event_iter().collect();
+            assert_eq!(events.len(), 1);
+            events.remove(0)
+        }
+
+        fn capture_metrics(encode: impl FnOnce()) -> Vec<vector_core::event::Metric> {
+            vector_core::metrics::init_test();
+            let controller = vector_core::metrics::Controller::get().unwrap();
+            controller.reset();
+            encode();
+            controller.capture_metrics()
+        }
+
+        #[test]
+        fn flattened_trace_is_converted() {
+            let mut buffer = BytesMut::new();
+            OtlpSerializer::new()
+                .unwrap()
+                .encode(flattened_trace(&TRACE_ID, 2_000), &mut buffer)
+                .expect("native trace must be converted, not rejected");
+
+            let request = ExportTraceServiceRequest::decode(buffer.freeze()).unwrap();
+            assert_eq!(request.resource_spans.len(), 1);
+            let resource_spans = &request.resource_spans[0];
+            let resource = resource_spans.resource.as_ref().unwrap();
+            assert_eq!(resource.attributes[0].key, "service.name");
+            let spans = &resource_spans.scope_spans[0].spans;
+            assert_eq!(spans.len(), 1);
+            assert_eq!(spans[0].trace_id, TRACE_ID);
+            assert_eq!(spans[0].span_id, SPAN_ID);
+            assert_eq!(spans[0].name, "GET /cart");
+            assert_eq!(spans[0].kind, 2);
+            assert_eq!(spans[0].start_time_unix_nano, 1_000);
+            assert_eq!(spans[0].end_time_unix_nano, 2_000);
+        }
+
+        #[test]
+        fn trace_with_only_rejected_spans_encodes_nothing() {
+            let mut buffer = BytesMut::new();
+            let metrics = capture_metrics(|| {
+                OtlpSerializer::new()
+                    .unwrap()
+                    .encode(flattened_trace(&[0; 16], 2_000), &mut buffer)
+                    .expect("rejected spans are omitted, not a batch failure");
+            });
+            assert!(buffer.is_empty());
+            assert!(
+                metrics.iter().any(|metric| {
+                    metric.name() == "component_errors_total"
+                        && metric.tag_matches("error_code", "invalid_trace_id")
+                }),
+                "expected invalid_trace_id in {metrics:?}"
+            );
+            assert!(
+                metrics.iter().any(|metric| {
+                    metric.name() == "component_discarded_events_total"
+                        && metric.tag_matches("intentional", "false")
+                }),
+                "expected the empty trace to count as a discarded event in {metrics:?}"
+            );
+        }
+
+        #[test]
+        fn reversed_timestamps_are_reported_and_the_span_is_still_encoded() {
+            let mut buffer = BytesMut::new();
+            let metrics = capture_metrics(|| {
+                OtlpSerializer::new()
+                    .unwrap()
+                    .encode(flattened_trace(&TRACE_ID, 500), &mut buffer)
+                    .expect("a clamped span is still encoded");
+            });
+
+            let request = ExportTraceServiceRequest::decode(buffer.freeze()).unwrap();
+            let span = &request.resource_spans[0].scope_spans[0].spans[0];
+            assert_eq!(span.start_time_unix_nano, 1_000);
+            assert_eq!(span.end_time_unix_nano, 1_000);
+            assert!(
+                metrics.iter().any(|metric| {
+                    metric.name() == "component_errors_total"
+                        && metric.tag_matches("error_code", "reversed_span_timestamps")
+                }),
+                "expected reversed_span_timestamps in {metrics:?}"
+            );
+            assert!(
+                metrics
+                    .iter()
+                    .all(|metric| metric.name() != "component_discarded_events_total"),
+                "a delivered span is not a discarded event: {metrics:?}"
+            );
+        }
+
+        #[test]
+        fn datadog_trace_is_an_error() {
+            let mut trace = TraceEvent::default();
+            trace.insert(event_path!("spans"), vrl::value::Value::Array(Vec::new()));
+            trace.metadata_mut().set_trace_layout(TraceLayout::Datadog);
+
+            let mut buffer = BytesMut::new();
+            let error = OtlpSerializer::new()
+                .unwrap()
+                .encode(Event::Trace(trace), &mut buffer)
+                .expect_err("a Datadog trace is not an OTLP layout");
+            assert!(
+                error.to_string().contains("not a recognized OTLP layout"),
+                "{error}"
+            );
+            assert!(buffer.is_empty());
+        }
+    }
 }

lib/codecs/src/internal_events.rs

@@ -291,3 +291,36 @@ impl InternalEvent for JsonSerializationError<'_> {
         .increment(1);
     }
 }
+
+#[cfg(feature = "opentelemetry")]
+#[derive(Debug, NamedInternalEvent)]
+/// Emitted when OTLP trace conversion drops or normalizes part of a trace.
+///
+/// The surrounding event may still be encoded. RFC 25329 reports every such issue
+/// through the instrumentation spec, with a bounded `error_code`.
+pub(crate) struct OtlpTraceConversionIssue<'a> {
+    pub issue: vector_core::event::typed_trace::TraceConversionIssue<'a>,
+}
+
+#[cfg(feature = "opentelemetry")]
+impl InternalEvent for OtlpTraceConversionIssue<'_> {
+    fn emit(self) {
+        let error_code = self.issue.error_code();
+        let message = self.issue.message();
+        error!(
+            message = message,
+            error_code,
+            error_type = error_type::ENCODER_FAILED,
+            stage = error_stage::SENDING,
+            issue = ?self.issue,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            CounterName::ComponentErrorsTotal,
+            "error_code" => error_code,
+            "error_type" => error_type::ENCODER_FAILED,
+            "stage" => error_stage::SENDING,
+        )
+        .increment(1);
+    }
+}

lib/vector-core/src/event/typed_trace/conversion.rs

@@ -89,6 +89,95 @@ pub enum TraceConversionIssue<'a> {
     ClampedTimestamp,
 }
 
+impl TraceConversionIssue<'_> {
+    /// Bounded error code for component-error metrics.
+    ///
+    /// The set is fixed so it can be a metric tag. Attribute keys and field names stay
+    /// out of it.
+    #[must_use]
+    pub const fn error_code(self) -> &'static str {
+        match self {
+            Self::RejectedSpan {
+                field: IdField::TraceId,
+                ..
+            }
+            | Self::DroppedLink {
+                field: IdField::TraceId,
+                ..
+            } => "invalid_trace_id",
+            Self::RejectedSpan {
+                field: IdField::SpanId | IdField::ParentSpanId,
+                ..
+            }
+            | Self::DroppedLink {
+                field: IdField::SpanId | IdField::ParentSpanId,
+                ..
+            } => "invalid_span_id",
+            Self::MalformedGrouping | Self::DiscardedGrouping => "discarded_span",
+            Self::EmptyEvent => "discarded_empty_chunk",
+            Self::UnrepresentableChunkContext => "unrepresentable_chunk_context",
+            Self::MalformedField { .. } | Self::MalformedAttribute => "invalid_attribute_value",
+            Self::UnmappedField { .. } => "legacy_conversion_error",
+            Self::MalformedEvent => "discarded_event",
+            Self::DuplicateAttribute { .. } => "duplicate_attribute",
+            Self::AttributeCollision { .. } => "attribute_collision",
+            Self::InvalidReservedAttribute { .. } => "invalid_reserved_attribute",
+            Self::ConflictingChunkContext { .. } => "conflicting_chunk_context",
+            Self::NonconformingStatus => "nonconforming_status",
+            Self::ReversedSpanTimestamps => "reversed_span_timestamps",
+            Self::ClampedTimestamp => "timestamp_clamped",
+        }
+    }
+
+    /// Operator-facing description of this issue.
+    #[must_use]
+    pub const fn message(self) -> &'static str {
+        match self {
+            Self::RejectedSpan { .. } => {
+                "Trace span was dropped because an identifier was invalid."
+            }
+            Self::DroppedLink { .. } => "Trace link was dropped because an identifier was invalid.",
+            Self::MalformedGrouping => "Trace grouping was dropped because it was malformed.",
+            Self::DiscardedGrouping => {
+                "Trace grouping was dropped because it had no accepted spans."
+            }
+            Self::EmptyEvent => "Empty trace event was encoded without its trace ID.",
+            Self::UnrepresentableChunkContext => {
+                "Datadog chunk context could not be distinguished from an absent one."
+            }
+            Self::MalformedField { .. } => {
+                "Trace field had the wrong type, so its default was used."
+            }
+            Self::UnmappedField { .. } => {
+                "Trace field was discarded because the typed model has no place for it."
+            }
+            Self::MalformedEvent => "Trace span event was dropped because it was malformed.",
+            Self::MalformedAttribute => "Trace attribute was dropped because it was malformed.",
+            Self::DuplicateAttribute { .. } => {
+                "Trace attribute was replaced by a later duplicate of its key."
+            }
+            Self::AttributeCollision { .. } => {
+                "Trace attribute was discarded because another field supplies its value."
+            }
+            Self::InvalidReservedAttribute { .. } => {
+                "Reserved Datadog bridge attribute was malformed."
+            }
+            Self::ConflictingChunkContext { .. } => {
+                "Datadog chunk context conflicted with an earlier span."
+            }
+            Self::NonconformingStatus => {
+                "Trace status message was discarded from an unset or ok status."
+            }
+            Self::ReversedSpanTimestamps => {
+                "Trace span ended before it started, so its duration was clamped to zero."
+            }
+            Self::ClampedTimestamp => {
+                "Trace timestamp was clamped to the destination field's domain."
+            }
+        }
+    }
+}
+
 /// Receives conversion issues so the caller can report them.
 pub trait TraceConversionReporter {
     /// Records one issue.