diff --git a/lib/codecs/src/decoding/format/otlp.rs b/lib/codecs/src/decoding/format/otlp.rs index 65b1a76a5a904..a0bee79b4baea 100644 --- a/lib/codecs/src/decoding/format/otlp.rs +++ b/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()); } diff --git a/lib/opentelemetry-proto/src/spans.rs b/lib/opentelemetry-proto/src/spans.rs index 40644138fe9dc..0d662bf21d05f 100644 --- a/lib/opentelemetry-proto/src/spans.rs +++ b/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) -> 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), diff --git a/lib/vector-core/src/event/metadata.rs b/lib/vector-core/src/event/metadata.rs index 114c1c1e22308..55935c9240989 100644 --- a/lib/vector-core/src/event/metadata.rs +++ b/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) { 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(); diff --git a/lib/vector-core/src/event/mod.rs b/lib/vector-core/src/event/mod.rs index f5a5a4c7c0c70..b6ee53eeab4e1 100644 --- a/lib/vector-core/src/event/mod.rs +++ b/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}; diff --git a/lib/vector-core/src/event/test/serialization.rs b/lib/vector-core/src/event/test/serialization.rs index bc875032a887c..0caab7789cbff 100644 --- a/lib/vector-core/src/event/test/serialization.rs +++ b/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::(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"); diff --git a/src/sources/datadog_agent/tests.rs b/src/sources/datadog_agent/tests.rs index 3bd90b15b543b..982aaa69b4c94 100644 --- a/src/sources/datadog_agent/tests.rs +++ b/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 { + 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::(&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::(&format!("address: \"{relay_addr}\"")).unwrap(), + ); + config.add_sink( + "to_relay", + &["dd.traces"], + serde_yaml::from_str::(&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 { diff --git a/src/sources/datadog_agent/traces.rs b/src/sources/datadog_agent/traces.rs index a14f874b56e1c..89eeaa7a5e217 100644 --- a/src/sources/datadog_agent/traces.rs +++ b/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 { 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 diff --git a/src/sources/opentelemetry/tests.rs b/src/sources/opentelemetry/tests.rs index a00ce3148f28e..b89179b6c3f10 100644 --- a/src/sources/opentelemetry/tests.rs +++ b/src/sources/opentelemetry/tests.rs @@ -9,7 +9,7 @@ use crate::{ config::{OutputId, SourceConfig, SourceContext}, event::{ Event, EventStatus, LogEvent, Metric as MetricEvent, MetricKind, MetricTags, MetricValue, - ObjectMap, Value, into_event_stream, + ObjectMap, TRACE_LAYOUT_OPENTELEMETRY, TRACE_LAYOUT_OTLP, Value, into_event_stream, metric::{Bucket, Quantile}, }, sources::opentelemetry::config::{ @@ -38,6 +38,7 @@ use vector_lib::{ metrics::v1::{ ExportMetricsServiceRequest, metrics_service_client::MetricsServiceClient, }, + trace::v1::trace_service_client::TraceServiceClient, }, common::v1::{AnyValue, InstrumentationScope, KeyValue, any_value::Value::StringValue}, logs::v1::{LogRecord, ResourceLogs, ScopeLogs}, @@ -1482,6 +1483,10 @@ async fn http_headers_traces_use_otlp_decoding_false() { .unwrap(), &value!("Test") ); + assert_eq!( + event.metadata().trace_layout(), + Some(TRACE_LAYOUT_OPENTELEMETRY) + ); }) .await; } @@ -1517,10 +1522,45 @@ async fn http_headers_traces_use_otlp_decoding_true() { .unwrap(), &value!("Test") ); + assert_eq!(event.metadata().trace_layout(), Some(TRACE_LAYOUT_OTLP)); }) .await; } +async fn assert_grpc_trace_layout_marker(use_otlp_decoding: bool) { + assert_source_compliance(&SOURCE_TAGS, async { + let env = build_otlp_test_env_with(TRACES, None, use_otlp_decoding).await; + let mut client = TraceServiceClient::connect(format!("http://{}", env.grpc_addr)) + .await + .unwrap(); + _ = client + .export(Request::new(create_test_traces_request())) + .await; + let mut events = test_util::collect_ready(env.output).await; + assert_eq!(events.len(), 1); + let expected = if use_otlp_decoding { + TRACE_LAYOUT_OTLP + } else { + TRACE_LAYOUT_OPENTELEMETRY + }; + assert_eq!( + events.pop().unwrap().metadata().trace_layout(), + Some(expected) + ); + }) + .await; +} + +#[tokio::test] +async fn grpc_traces_use_otlp_decoding_false_sets_layout_marker() { + assert_grpc_trace_layout_marker(false).await; +} + +#[tokio::test] +async fn grpc_traces_use_otlp_decoding_true_sets_layout_marker() { + assert_grpc_trace_layout_marker(true).await; +} + pub struct OTelTestEnv { pub grpc_addr: String, pub config: OpentelemetryConfig, @@ -1530,6 +1570,14 @@ pub struct OTelTestEnv { pub async fn build_otlp_test_env( event_name: &'static str, log_namespace: Option, +) -> OTelTestEnv { + build_otlp_test_env_with(event_name, log_namespace, false).await +} + +async fn build_otlp_test_env_with( + event_name: &'static str, + log_namespace: Option, + use_otlp_decoding: bool, ) -> OTelTestEnv { let (_guard_0, grpc_addr) = next_addr(); let (_guard_1, http_addr) = next_addr(); @@ -1548,7 +1596,7 @@ pub async fn build_otlp_test_env( }, acknowledgements: Default::default(), log_namespace, - use_otlp_decoding: false.into(), + use_otlp_decoding: use_otlp_decoding.into(), }; let (sender, output, _) = new_source(EventStatus::Delivered, event_name.to_string()); diff --git a/src/transforms/remap.rs b/src/transforms/remap.rs index 45631ad838b9a..021aff3603d39 100644 --- a/src/transforms/remap.rs +++ b/src/transforms/remap.rs @@ -699,7 +699,7 @@ mod tests { use crate::{ config::{ConfigBuilder, build_unit_tests}, event::{ - LogEvent, Metric, Value, + LogEvent, Metric, TRACE_LAYOUT_DATADOG, TRACE_LAYOUT_KEY, Value, metric::{MetricKind, MetricValue}, }, metrics::Controller, @@ -747,6 +747,51 @@ mod tests { crate::test_util::test_generate_config::(); } + #[test] + fn remap_cannot_overwrite_trace_layout() { + let config = RemapConfig { + source: Some(format!("%vector.{TRACE_LAYOUT_KEY} = \"forged\"")), + ..Default::default() + }; + let err = remap(config).unwrap_err().to_string(); + assert!( + err.contains("mutation of read-only value"), + "unexpected compile error: {err}" + ); + } + + #[test] + fn remap_cannot_delete_trace_layout() { + let config = RemapConfig { + source: Some(format!("del(%vector.{TRACE_LAYOUT_KEY})")), + ..Default::default() + }; + let err = remap(config).unwrap_err().to_string(); + assert!(err.contains("read-only"), "unexpected compile error: {err}"); + } + + #[test] + fn remap_can_read_trace_layout() { + let event = { + let mut event = LogEvent::from("input"); + event.metadata_mut().set_trace_layout(TRACE_LAYOUT_DATADOG); + Event::from(event) + }; + let conf = RemapConfig { + source: Some(format!(".layout = %vector.{TRACE_LAYOUT_KEY}")), + drop_on_error: true, + drop_on_abort: false, + ..Default::default() + }; + let mut tform = remap(conf).unwrap(); + let result = transform_one(&mut tform, event).unwrap(); + assert_eq!(get_field_string(&result, "layout"), TRACE_LAYOUT_DATADOG); + assert_eq!( + result.as_log().metadata().trace_layout(), + Some(TRACE_LAYOUT_DATADOG) + ); + } + #[test] fn config_missing_source_and_file() { let config = RemapConfig { diff --git a/src/transforms/trace_to_log.rs b/src/transforms/trace_to_log.rs index eca35a0418a7c..8e50d0909bc8f 100644 --- a/src/transforms/trace_to_log.rs +++ b/src/transforms/trace_to_log.rs @@ -1,10 +1,11 @@ use vector_lib::config::clone_input_definitions; use vector_lib::configurable::configurable_component; +use vector_lib::lookup::metadata_path; use crate::config::OutputId; use crate::{ config::{DataType, GenerateConfig, Input, TransformConfig, TransformContext, TransformOutput}, - event::{Event, LogEvent}, + event::{Event, LogEvent, TRACE_LAYOUT_KEY}, schema::Definition, transforms::{FunctionTransform, OutputBuffer, Transform}, }; @@ -66,7 +67,12 @@ pub struct TraceToLog; impl FunctionTransform for TraceToLog { fn transform(&mut self, output: &mut OutputBuffer, event: Event) { if let Event::Trace(trace) = event { - output.push(Event::Log(LogEvent::from(trace))); + let mut log = LogEvent::from(trace); + // The layout marker is only meaningful to components that consume + // traces as traces. Drop it so a converted log is not classified + // as Vector-namespaced solely because of `%vector.trace_layout`. + log.remove_prune(metadata_path!("vector", TRACE_LAYOUT_KEY), true); + output.push(Event::Log(log)); } } } @@ -78,7 +84,10 @@ mod tests { use crate::transforms::test::create_topology; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; - use vector_lib::event::TraceEvent; + use vector_lib::{ + config::LogNamespace, + event::{TRACE_LAYOUT_DATADOG, TraceEvent}, + }; #[test] fn generate_config() { @@ -131,4 +140,23 @@ mod tests { "Trace data fields should be preserved" ); } + + #[tokio::test] + async fn drops_trace_layout_marker() { + use vrl::btreemap; + + let mut trace = TraceEvent::from(btreemap! { + "host" => "a_hostname", + "span_id" => "abc123", + }); + trace.metadata_mut().set_trace_layout(TRACE_LAYOUT_DATADOG); + + let log = do_transform(trace).await.unwrap(); + assert_eq!(log.namespace(), LogNamespace::Legacy); + assert_eq!(log.metadata().trace_layout(), None); + assert_eq!( + log.get(vrl::event_path!("host")), + Some(&"a_hostname".into()) + ); + } }