chore(traces): Add source layout marker to trace event metadata

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

@@ -8,7 +8,7 @@ use smallvec::{SmallVec, smallvec};
 use vector_config::{configurable_component, indexmap::IndexSet};
 use vector_core::{
     config::{DataType, LogNamespace},
-    event::Event,
+    event::{Event, TRACE_LAYOUT_OTLP, TraceEvent},
     schema,
 };
 use vrl::{event_path, protobuf::parse::Options, value::Kind};
@@ -192,8 +192,9 @@ impl Deserializer for OtlpDeserializer {
                     {
                         // Convert the log event to a trace event by taking ownership
                         if let Some(Event::Log(log)) = events.pop() {
-                            let trace_event = Event::Trace(log.into());
-                            return Ok(smallvec![trace_event]);
+                            let mut trace = TraceEvent::from(log);
+                            trace.metadata_mut().set_trace_layout(TRACE_LAYOUT_OTLP);
+                            return Ok(smallvec![Event::Trace(trace)]);
                         }
                     }
                 }
@@ -380,6 +381,7 @@ mod tests {
             let trace = events[0].as_trace();
             assert!(trace.get(event_path!(field)).is_some());
             validate_trace_ids(trace.value());
+            assert_eq!(events[0].metadata().trace_layout(), Some(TRACE_LAYOUT_OTLP));
         } else {
             assert!(events[0].as_log().get(event_path!(field)).is_some());
         }

lib/opentelemetry-proto/src/spans.rs

@@ -1,7 +1,7 @@
 use std::collections::BTreeMap;
 
 use chrono::{DateTime, TimeZone, Utc};
-use vector_core::event::{Event, TraceEvent};
+use vector_core::event::{Event, TRACE_LAYOUT_OPENTELEMETRY, TraceEvent};
 use vrl::{
     event_path,
     value::{KeyString, Value},
@@ -52,6 +52,9 @@ struct ResourceSpan {
 impl ResourceSpan {
     fn into_event(self, now: DateTime<Utc>) -> Event {
         let mut trace = TraceEvent::default();
+        trace
+            .metadata_mut()
+            .set_trace_layout(TRACE_LAYOUT_OPENTELEMETRY);
         let span = self.span;
         trace.insert(
             event_path!(TRACE_ID_KEY),

lib/vector-core/src/event/metadata.rs

@@ -3,7 +3,7 @@
 use std::{borrow::Cow, collections::BTreeMap, fmt, sync::Arc, time::Instant};
 
 use derivative::Derivative;
-use lookup::OwnedTargetPath;
+use lookup::{OwnedTargetPath, path};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 use vector_common::{EventDataEq, byte_size_of::ByteSizeOf, config::ComponentKey};
@@ -20,6 +20,27 @@ use crate::{
 
 const DATADOG_API_KEY: &str = "datadog_api_key";
 const SPLUNK_HEC_TOKEN: &str = "splunk_hec_token";
+const VECTOR_METADATA_NAMESPACE: &str = "vector";
+
+/// Field name of the internal, unstable trace-layout marker under the read-only
+/// `vector` metadata namespace (`%vector.trace_layout`).
+///
+/// This marker records which trace key layout an event was produced with so
+/// Vector's own transforms and sinks can dispatch on layout. It lives under
+/// `%vector` so it is already reserved and locked to VRL. Log namespacing does
+/// not apply to traces; `trace_to_log` drops the field when converting to a
+/// log. It is not documented for users, is not a compatibility contract, and
+/// may change or be removed without a deprecation cycle.
+pub const TRACE_LAYOUT_KEY: &str = "trace_layout";
+
+/// Layout marker value written by the `datadog_agent` source.
+pub const TRACE_LAYOUT_DATADOG: &str = "datadog";
+
+/// Layout marker value for flattened native traces from the `opentelemetry` source.
+pub const TRACE_LAYOUT_OPENTELEMETRY: &str = "opentelemetry";
+
+/// Layout marker value for raw OTLP `resourceSpans` batches from the `opentelemetry` source.
+pub const TRACE_LAYOUT_OTLP: &str = "otlp";
 
 /// The event metadata structure is a `Arc` wrapper around the actual metadata to avoid cloning the
 /// underlying data until it becomes necessary to provide a `mut` copy.
@@ -204,6 +225,21 @@ impl EventMetadata {
         self.get_mut().source_type = Some(source_type.into());
     }
 
+    /// Writes the internal, unstable trace-layout marker under `%vector.trace_layout`.
+    pub fn set_trace_layout(&mut self, layout: &'static str) {
+        self.value_mut()
+            .insert(path!(VECTOR_METADATA_NAMESPACE, TRACE_LAYOUT_KEY), layout);
+    }
+
+    /// Returns the internal, unstable trace-layout marker, if present.
+    #[must_use]
+    pub fn trace_layout(&self) -> Option<&str> {
+        self.value()
+            .get(path!(VECTOR_METADATA_NAMESPACE, TRACE_LAYOUT_KEY))
+            .and_then(Value::as_bytes)
+            .and_then(|bytes| std::str::from_utf8(bytes).ok())
+    }
+
     /// Sets the `upstream_id` in the metadata to the provided value.
     pub fn set_upstream_id(&mut self, upstream_id: Arc<OutputId>) {
         self.get_mut().upstream_id = Some(upstream_id);
@@ -562,6 +598,18 @@ mod test {
     const SECRET: &str = "secret";
     const SECRET2: &str = "secret2";
 
+    #[test]
+    fn trace_layout_round_trip() {
+        let mut metadata = EventMetadata::default();
+        assert_eq!(metadata.trace_layout(), None);
+        metadata.set_trace_layout(TRACE_LAYOUT_DATADOG);
+        assert_eq!(metadata.trace_layout(), Some(TRACE_LAYOUT_DATADOG));
+        metadata.set_trace_layout(TRACE_LAYOUT_OPENTELEMETRY);
+        assert_eq!(metadata.trace_layout(), Some(TRACE_LAYOUT_OPENTELEMETRY));
+        metadata.set_trace_layout(TRACE_LAYOUT_OTLP);
+        assert_eq!(metadata.trace_layout(), Some(TRACE_LAYOUT_OTLP));
+    }
+
     #[test]
     fn metadata_hardcoded_secrets_get_set() {
         let mut metadata = EventMetadata::default();

lib/vector-core/src/event/mod.rs

@@ -7,7 +7,10 @@ pub use finalization::{
     EventFinalizers, EventStatus, Finalizable, GroupedFinalizable, MergeFinalizable,
 };
 pub use log_event::LogEvent;
-pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, Secrets, WithMetadata};
+pub use metadata::{
+    DatadogMetricOriginMetadata, EventMetadata, Secrets, TRACE_LAYOUT_DATADOG, TRACE_LAYOUT_KEY,
+    TRACE_LAYOUT_OPENTELEMETRY, TRACE_LAYOUT_OTLP, WithMetadata,
+};
 pub use metric::{Metric, MetricKind, MetricTags, MetricValue, StatisticKind};
 pub use r#ref::{EventMutRef, EventRef};
 pub use ser::{MAX_VALUE_NESTING_FRAMES, event_exceeds_max_nesting_cost};

lib/vector-core/src/event/test/serialization.rs

@@ -69,6 +69,27 @@ fn back_and_forth_through_bytes() {
         .quickcheck(inner as fn(EventArray) -> TestResult);
 }
 
+#[test]
+fn disk_buffer_preserves_trace_layout_metadata() {
+    let mut trace = TraceEvent::default();
+    trace
+        .metadata_mut()
+        .set_trace_layout(crate::event::TRACE_LAYOUT_DATADOG);
+    let expected = EventArray::from(Event::Trace(trace));
+
+    let mut buffer = BytesMut::with_capacity(64);
+    encode_value(expected, &mut buffer);
+    let actual = decode_value::<EventArray, _>(buffer);
+
+    let EventArray::Traces(traces) = actual else {
+        panic!("expected a traces array");
+    };
+    assert_eq!(
+        traces[0].metadata().trace_layout(),
+        Some(crate::event::TRACE_LAYOUT_DATADOG)
+    );
+}
+
 #[test]
 fn serialization() {
     let mut event = LogEvent::from("raw log line");

src/sources/datadog_agent/tests.rs

@@ -59,6 +59,15 @@ use crate::{
     },
 };
 
+#[cfg(all(feature = "sinks-vector", feature = "sources-vector"))]
+use crate::{
+    config::Config,
+    event::TRACE_LAYOUT_DATADOG,
+    sinks::vector::VectorConfig as VectorSinkConfig,
+    sources::vector::VectorConfig as VectorSourceConfig,
+    test_util::{mock::basic_sink, start_topology},
+};
+
 use crate::sources::datadog_agent::llmobs::decode_llmobs_body;
 
 const DD_API_KEY: &str = "12345678abcdefgh12345678abcdefgh";
@@ -376,6 +385,25 @@ async fn send_and_collect(
     .await
 }
 
+/// Smallest v2 payload the `datadog_agent` traces decoder accepts.
+#[cfg(all(feature = "sinks-vector", feature = "sources-vector"))]
+fn minimal_v2_trace_body() -> Vec<u8> {
+    let mut buf = Vec::new();
+    ddtrace_proto::TracePayload {
+        tracer_payloads: vec![ddtrace_proto::TracerPayload {
+            chunks: vec![ddtrace_proto::TraceChunk {
+                spans: vec![ddtrace_proto::Span::default()],
+                ..Default::default()
+            }],
+            ..Default::default()
+        }],
+        ..Default::default()
+    }
+    .encode(&mut buf)
+    .unwrap();
+    buf
+}
+
 fn dd_api_key_headers() -> HeaderMap {
     let mut headers = HeaderMap::new();
     headers.insert("dd-api-key", DD_API_KEY.parse().unwrap());
@@ -1328,6 +1356,10 @@ async fn decode_traces() {
 
         {
             let trace_v1 = events[0].as_trace();
+            assert_eq!(
+                events[0].metadata().trace_layout(),
+                Some(vector_lib::event::TRACE_LAYOUT_DATADOG)
+            );
             assert_eq!(trace_v1.as_map()["host"], "a_hostname".into());
             assert_eq!(trace_v1.as_map()["env"], "an_environment".into());
             assert_eq!(trace_v1.as_map()["language_name"], "ada".into());
@@ -1367,6 +1399,10 @@ async fn decode_traces() {
             );
 
             let apm_event = events[1].as_trace();
+            assert_eq!(
+                events[1].metadata().trace_layout(),
+                Some(vector_lib::event::TRACE_LAYOUT_DATADOG)
+            );
             assert!(apm_event.contains(event_path!("spans")));
             assert_eq!(apm_event.as_map()["host"], "a_hostname".into());
             assert_eq!(apm_event.as_map()["env"], "an_environment".into());
@@ -1385,6 +1421,10 @@ async fn decode_traces() {
             );
 
             let trace_v2 = events[2].as_trace();
+            assert_eq!(
+                events[2].metadata().trace_layout(),
+                Some(vector_lib::event::TRACE_LAYOUT_DATADOG)
+            );
             assert_eq!(trace_v2.as_map()["host"], "a_hostname".into());
             assert_eq!(trace_v2.as_map()["env"], "env".into());
 
@@ -1450,6 +1490,77 @@ async fn decode_traces() {
     .await;
 }
 
+#[cfg(all(feature = "sinks-vector", feature = "sources-vector"))]
+#[tokio::test]
+async fn trace_layout_survives_vector_hop_unlike_source_type() {
+    trace_init();
+
+    let (_dd_guard, dd_addr) = next_addr();
+    let (_relay_guard, relay_addr) = next_addr();
+    let (out_rx, out_sink) = basic_sink(10);
+
+    let mut config = Config::builder();
+    config.add_source(
+        "dd",
+        serde_yaml::from_str::<DatadogAgentConfig>(&format!(
+            r#"
+                address: "{dd_addr}"
+                disable_logs: true
+                disable_metrics: true
+                disable_llmobs: true
+                multiple_outputs: true
+            "#
+        ))
+        .unwrap(),
+    );
+    config.add_source(
+        "relay",
+        serde_yaml::from_str::<VectorSourceConfig>(&format!("address: \"{relay_addr}\"")).unwrap(),
+    );
+    config.add_sink(
+        "to_relay",
+        &["dd.traces"],
+        serde_yaml::from_str::<VectorSinkConfig>(&format!(
+            r#"
+                address: "{relay_addr}"
+                batch:
+                  max_events: 1
+            "#
+        ))
+        .unwrap(),
+    );
+    config.add_sink("out", &["relay"], out_sink);
+
+    let (topology, _crash) = start_topology(config.build().unwrap(), false).await;
+    wait_for_tcp(dd_addr).await;
+    wait_for_tcp(relay_addr).await;
+
+    let body = minimal_v2_trace_body();
+    assert_eq!(
+        200,
+        send_with_path(
+            dd_addr,
+            unsafe { str::from_utf8_unchecked(&body) },
+            HeaderMap::new(),
+            DD_API_TRACES_PATH
+        )
+        .await
+    );
+
+    let event = timeout(
+        Duration::from_secs(10),
+        out_rx.flat_map(into_event_stream).next(),
+    )
+    .await
+    .expect("timed out waiting for relayed trace")
+    .expect("relay produced no event");
+
+    assert_eq!(event.metadata().trace_layout(), Some(TRACE_LAYOUT_DATADOG));
+    assert_eq!(event.metadata().source_type(), Some("vector"));
+
+    topology.stop().await;
+}
+
 #[tokio::test]
 async fn split_outputs() {
     assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {

src/sources/datadog_agent/traces.rs

@@ -8,6 +8,7 @@ use ordered_float::NotNan;
 use prost::Message;
 use vector_lib::{
     EstimatedJsonEncodedSizeOf,
+    event::TRACE_LAYOUT_DATADOG,
     internal_event::{CountByteSize, InternalEventHandle as _},
 };
 use vrl::event_path;
@@ -164,13 +165,21 @@ fn handle_dd_trace_payload_v1(
     Ok(enriched_events)
 }
 
+fn new_trace_event() -> TraceEvent {
+    let mut trace_event = TraceEvent::default();
+    trace_event
+        .metadata_mut()
+        .set_trace_layout(TRACE_LAYOUT_DATADOG);
+    trace_event
+}
+
 fn convert_dd_tracer_payload(payload: ddtrace_proto::TracerPayload) -> Vec<TraceEvent> {
     let tags = convert_tags(payload.tags);
     payload
         .chunks
         .into_iter()
         .map(|trace| {
-            let mut trace_event = TraceEvent::default();
+            let mut trace_event = new_trace_event();
             trace_event.insert(event_path!("priority"), trace.priority as i64);
             trace_event.insert(event_path!("origin"), trace.origin);
             trace_event.insert(event_path!("dropped"), trace.dropped_trace);
@@ -220,7 +229,7 @@ fn handle_dd_trace_payload_v0(
         .traces
         .into_iter()
         .map(|dd_trace| {
-            let mut trace_event = TraceEvent::default();
+            let mut trace_event = new_trace_event();
 
             // TODO trace_id is being forced into an i64 but
             // the incoming payload is u64. This is a bug and needs to be fixed per:
@@ -240,7 +249,7 @@ fn handle_dd_trace_payload_v0(
         })
         //... and each APM event is also mapped into its own event
         .chain(decoded_payload.transactions.into_iter().map(|s| {
-            let mut trace_event = TraceEvent::default();
+            let mut trace_event = new_trace_event();
             trace_event.insert(event_path!("spans"), vec![Value::from(convert_span(s))]);
             trace_event.insert(event_path!("dropped"), true);
             trace_event