diff --git a/changelog.d/25790_datadog_agent_v3_series.feature.md b/changelog.d/25790_datadog_agent_v3_series.feature.md new file mode 100644 index 0000000000000..3c0d435711ba9 --- /dev/null +++ b/changelog.d/25790_datadog_agent_v3_series.feature.md @@ -0,0 +1,3 @@ +The `datadog_agent` source now accepts the v3 series metrics intake format used by default by Datadog Agent 7.81.0 and newer. + +authors: arfa79 diff --git a/lib/datadog-proto/build.rs b/lib/datadog-proto/build.rs index e82587165a6b2..5cf9830adee74 100644 --- a/lib/datadog-proto/build.rs +++ b/lib/datadog-proto/build.rs @@ -7,6 +7,7 @@ use std::{ fn main() -> Result<()> { for path in [ "proto/datadog/agentpayload.proto", + "proto/datadog/metrics/intake_v3.proto", "proto/datadog/trace/agent_payload.proto", "proto/datadog/trace/tracer_payload.proto", "proto/datadog/trace/span.proto", diff --git a/lib/datadog-proto/proto/datadog/agentpayload.proto b/lib/datadog-proto/proto/datadog/agentpayload.proto index 471447d11f801..93e1276bf0892 100644 --- a/lib/datadog-proto/proto/datadog/agentpayload.proto +++ b/lib/datadog-proto/proto/datadog/agentpayload.proto @@ -14,7 +14,11 @@ message CommonMetadata { } message Origin { - reserved 1, 2, 3; + reserved 1, 2; + // The Agent's v2 serializer writes field 3 for no-index series even though the public + // agent-payload schema reserves it. Keep the wire field so Vector can preserve no-index + // metadata when receiving or forwarding v2 series. + int32 metric_type = 3; uint32 origin_product = 4; uint32 origin_category = 5; uint32 origin_service = 6; diff --git a/lib/vector-core/proto/event.proto b/lib/vector-core/proto/event.proto index a85a3052b18c3..f6ece930457e7 100644 --- a/lib/vector-core/proto/event.proto +++ b/lib/vector-core/proto/event.proto @@ -79,6 +79,7 @@ message DatadogOriginMetadata { optional uint32 origin_product = 1; optional uint32 origin_category = 2; optional uint32 origin_service = 3; + optional int32 metric_type = 4; } message Secrets { @@ -106,6 +107,7 @@ message Metadata { Secrets secrets = 6; bytes source_event_id = 7; optional TraceLayout trace_layout = 8; + optional string datadog_metric_unit = 9; } message Metric { diff --git a/lib/vector-core/src/event/metadata.rs b/lib/vector-core/src/event/metadata.rs index 8e8f154dddd85..63f8438f8f74d 100644 --- a/lib/vector-core/src/event/metadata.rs +++ b/lib/vector-core/src/event/metadata.rs @@ -122,6 +122,10 @@ pub(super) struct Inner { #[serde(default)] pub(crate) datadog_origin_metadata: Option, + /// Datadog metric unit received from an upstream Agent payload. + #[serde(default)] + pub(crate) datadog_metric_unit: Option>, + /// Which legacy trace key layout produced this event. `None` means unmarked. /// Meaningful only on traces; always `None` for logs and metrics. #[serde(default)] @@ -135,6 +139,8 @@ pub(super) struct Inner { /// Metric Origin metadata for submission to Datadog. #[derive(Clone, Default, Debug, Deserialize, PartialEq, Serialize)] pub struct DatadogMetricOriginMetadata { + /// Datadog origin metric type. + metric_type: Option, /// `OriginProduct` product: Option, /// `OriginCategory` @@ -152,12 +158,25 @@ impl DatadogMetricOriginMetadata { #[must_use] pub fn new(product: Option, category: Option, service: Option) -> Self { Self { + metric_type: None, product, category, service, } } + /// Returns the Datadog origin metric type. + pub fn metric_type(&self) -> Option { + self.metric_type + } + + /// Sets the Datadog origin metric type. + #[must_use] + pub fn with_metric_type(mut self, metric_type: Option) -> Self { + self.metric_type = metric_type; + self + } + /// Returns a reference to the `OriginProduct`. pub fn product(&self) -> Option { self.product @@ -310,6 +329,21 @@ impl EventMetadata { self.inner.datadog_origin_metadata.as_ref() } + /// Returns the Datadog metric unit. + pub fn datadog_metric_unit(&self) -> Option<&str> { + self.inner.datadog_metric_unit.as_deref() + } + + /// Sets the Datadog metric unit. + pub fn set_datadog_metric_unit(&mut self, unit: impl Into>) { + self.get_mut().datadog_metric_unit = Some(unit.into()); + } + + /// Clears the Datadog metric unit. + pub fn clear_datadog_metric_unit(&mut self) { + self.get_mut().datadog_metric_unit = None; + } + /// Returns a reference to the event id. pub fn source_event_id(&self) -> Option { self.inner.source_event_id @@ -339,6 +373,7 @@ impl Default for Inner { upstream_id: None, dropped_fields: ObjectMap::new(), datadog_origin_metadata: None, + datadog_metric_unit: None, trace_layout: None, source_event_id: Some(Uuid::new_v4()), } @@ -367,10 +402,18 @@ pub(super) fn default_schema_definition() -> Arc { impl ByteSizeOf for EventMetadata { fn allocated_bytes(&self) -> usize { - // NOTE we don't count the `str` here because it's allocated somewhere - // else. We're just moving around the pointer, which is already captured - // by `ByteSizeOf::size_of`. - self.inner.finalizers.allocated_bytes() + // Most Arc-backed metadata points to allocations owned or accounted for elsewhere. The + // Datadog metric unit is retained solely by event metadata, so include its allocation. + // Charge each event conservatively. Arc owner counts are transient: fanout branches can + // disappear after a memory queue has snapshotted this value, leaving that queue as the sole + // owner without updating its reservation. + let datadog_metric_unit_bytes = self + .inner + .datadog_metric_unit + .as_ref() + .map_or(0, |unit| unit.len()); + + self.inner.finalizers.allocated_bytes() + datadog_metric_unit_bytes } } @@ -425,6 +468,13 @@ impl EventMetadata { self } + /// Sets the Datadog metric unit. + #[must_use] + pub fn with_datadog_metric_unit(mut self, unit: impl Into>) -> Self { + self.get_mut().datadog_metric_unit = Some(unit.into()); + self + } + /// Replaces the existing `source_event_id` with the given one. #[must_use] pub fn with_source_event_id(mut self, source_event_id: Option) -> Self { @@ -671,6 +721,32 @@ mod test { assert_eq!(metadata.splunk_hec_token().unwrap().as_ref(), SECRET2); } + #[test] + fn metadata_copy_on_write_keeps_datadog_metric_unit_shared() { + let metadata = EventMetadata::default().with_datadog_metric_unit("byte"); + let mut cloned = metadata.clone(); + + cloned.set_datadog_api_key(Arc::from(SECRET)); + + let original_unit = metadata.inner.datadog_metric_unit.as_ref().unwrap(); + let cloned_unit = cloned.inner.datadog_metric_unit.as_ref().unwrap(); + assert!(Arc::ptr_eq(original_unit, cloned_unit)); + } + + #[test] + fn metadata_byte_size_conservatively_accounts_for_shared_datadog_metric_unit() { + let unit = "x".repeat(1024); + let metadata = EventMetadata::default().with_datadog_metric_unit(unit); + let mut shared_inner = metadata.clone(); + + assert_eq!(metadata.allocated_bytes(), 1024); + assert_eq!(shared_inner.allocated_bytes(), 1024); + + shared_inner.set_datadog_api_key(Arc::from(SECRET)); + assert_eq!(metadata.allocated_bytes(), 1024); + assert_eq!(shared_inner.allocated_bytes(), 1024); + } + #[test] fn secrets_merge() { let mut a = Secrets::new(); diff --git a/lib/vector-core/src/event/proto.rs b/lib/vector-core/src/event/proto.rs index 5bfda738d4df7..f4baefee3b539 100644 --- a/lib/vector-core/src/event/proto.rs +++ b/lib/vector-core/src/event/proto.rs @@ -597,6 +597,7 @@ impl From for super::metadata::Secrets { impl From for DatadogOriginMetadata { fn from(value: super::DatadogMetricOriginMetadata) -> Self { Self { + metric_type: value.metric_type(), origin_product: value.product(), origin_category: value.category(), origin_service: value.service(), @@ -611,6 +612,7 @@ impl From for super::DatadogMetricOriginMetadata { value.origin_category, value.origin_service, ) + .with_metric_type(value.metric_type) } } @@ -638,6 +640,7 @@ impl From for Metadata { source_type, upstream_id, datadog_origin_metadata, + datadog_metric_unit, source_event_id, trace_layout, .. @@ -653,6 +656,7 @@ impl From for Metadata { upstream_id: upstream_id.map(|id| id.as_ref().clone()).map(Into::into), secrets, source_event_id: source_event_id.map_or(vec![], std::convert::Into::into), + datadog_metric_unit: datadog_metric_unit.map(|unit| unit.to_string()), trace_layout: encode_trace_layout(trace_layout), } } @@ -669,6 +673,7 @@ impl TryFrom for EventMetadata { upstream_id, secrets, datadog_origin_metadata, + datadog_metric_unit, source_event_id, trace_layout, } = value; @@ -709,6 +714,7 @@ impl TryFrom for EventMetadata { schema_definition: default_schema_definition(), dropped_fields: ObjectMap::new(), datadog_origin_metadata, + datadog_metric_unit: datadog_metric_unit.map(Arc::from), trace_layout: decode_trace_layout(trace_layout), source_event_id, }), diff --git a/lib/vector-core/src/event/test/serialization.rs b/lib/vector-core/src/event/test/serialization.rs index a2ccdc82eb868..927351f60dbeb 100644 --- a/lib/vector-core/src/event/test/serialization.rs +++ b/lib/vector-core/src/event/test/serialization.rs @@ -109,6 +109,28 @@ fn disk_buffer_preserves_unrecognized_trace_layout() { ); } +#[test] +fn disk_buffer_preserves_datadog_metric_unit() { + let mut metric = Metric::new( + "test", + MetricKind::Absolute, + MetricValue::Gauge { value: 1.0 }, + ); + metric + .metadata_mut() + .set_datadog_metric_unit("byte".to_owned()); + let expected = EventArray::from(Event::Metric(metric)); + + let mut buffer = BytesMut::with_capacity(64); + encode_value(expected, &mut buffer); + let actual = decode_value::(buffer); + + let EventArray::Metrics(metrics) = actual else { + panic!("expected a metrics array"); + }; + assert_eq!(metrics[0].metadata().datadog_metric_unit(), Some("byte")); +} + #[test] fn serialization() { let mut event = LogEvent::from("raw log line"); diff --git a/src/common/datadog.rs b/src/common/datadog.rs index 6467d9ede5f00..861b6d10e6efd 100644 --- a/src/common/datadog.rs +++ b/src/common/datadog.rs @@ -22,6 +22,8 @@ pub const DDTAGS: &str = "ddtags"; /// The datadog message event path. pub const MESSAGE: &str = "message"; pub(crate) const DATADOG_METRIC_RESOURCE_TAG_PREFIX: &str = "resource."; +/// Datadog origin metric-type value for metrics excluded from indexing. +pub(crate) const DATADOG_ORIGIN_METRIC_TYPE_NO_INDEX: i32 = 9; /// Mapping of the semantic meaning of well known Datadog reserved attributes /// to the field name that Datadog intake expects. @@ -65,6 +67,9 @@ pub struct DatadogSeriesMetric { /// device #[serde(skip_serializing_if = "Option::is_none")] pub device: Option, + /// unit + #[serde(skip_serializing_if = "Option::is_none")] + pub unit: Option, /// metadata #[serde(skip_serializing_if = "Option::is_none")] pub metadata: Option, diff --git a/src/sinks/datadog/metrics/encoder.rs b/src/sinks/datadog/metrics/encoder.rs index 13f14fbe87608..fa5c44434daab 100644 --- a/src/sinks/datadog/metrics/encoder.rs +++ b/src/sinks/datadog/metrics/encoder.rs @@ -444,6 +444,7 @@ fn generate_proto_metadata( } ddmetric_proto::Metadata { origin: Some(ddmetric_proto::Origin { + metric_type: origin.metric_type().unwrap_or_default(), origin_product: origin.product().unwrap_or_default(), origin_category: origin.category().unwrap_or_default(), origin_service: origin.service().unwrap_or_default(), @@ -714,8 +715,10 @@ fn series_to_proto_message( tags, points, r#type: metric_type.into(), - // unit is omitted - unit: String::new(), + unit: event_metadata + .datadog_metric_unit() + .unwrap_or_default() + .to_string(), source_type_name, interval: i64::from(maybe_interval.unwrap_or(0)), metadata, @@ -835,11 +838,14 @@ pub(super) fn generate_origin_metadata( // - `log_to_metric` transform set the OriginService in the EventMetadata when it creates // the new metric. if let Some(pass_through) = maybe_pass_through { - Some(DatadogMetricOriginMetadata::new( - pass_through.product().or(Some(origin_product_value)), - pass_through.category().or(Some(ORIGIN_CATEGORY_VALUE)), - pass_through.service().or(Some(no_value)), - )) + Some( + DatadogMetricOriginMetadata::new( + pass_through.product().or(Some(origin_product_value)), + pass_through.category().or(Some(ORIGIN_CATEGORY_VALUE)), + pass_through.service().or(Some(no_value)), + ) + .with_metric_type(pass_through.metric_type()), + ) // No metadata has been set upstream } else { @@ -1293,9 +1299,12 @@ mod tests { let category = 11; let service = 9; - let event_metadata = EventMetadata::default().with_origin_metadata( - DatadogMetricOriginMetadata::new(Some(product), Some(category), Some(service)), - ); + let event_metadata = EventMetadata::default() + .with_origin_metadata( + DatadogMetricOriginMetadata::new(Some(product), Some(category), Some(service)) + .with_metric_type(Some(9)), + ) + .with_datadog_metric_unit("byte".to_string()); let counter = get_simple_counter_with_metadata(event_metadata); // series v2 @@ -1309,9 +1318,11 @@ mod tests { .unwrap(); let generated_origin = series_proto.metadata.unwrap().origin.unwrap(); + assert_eq!(generated_origin.metric_type, 9); assert_eq!(generated_origin.origin_product, product); assert_eq!(generated_origin.origin_category, category); assert_eq!(generated_origin.origin_service, service); + assert_eq!(series_proto.unit, "byte"); } } diff --git a/src/sinks/datadog/metrics/encoder_v3.rs b/src/sinks/datadog/metrics/encoder_v3.rs index a926d68abd1ac..3ec7a7a92c478 100644 --- a/src/sinks/datadog/metrics/encoder_v3.rs +++ b/src/sinks/datadog/metrics/encoder_v3.rs @@ -27,6 +27,7 @@ use super::{ generate_origin_metadata, split_series_tags, }, }; +use crate::common::datadog::DATADOG_ORIGIN_METRIC_TYPE_NO_INDEX; use crate::sinks::util::{ Compression, Compressor, encode_namespace, request_builder::EncodeResult, }; @@ -234,8 +235,6 @@ fn encode_metric_to_v3( // `resource.` tags that the `datadog_agent` source produces from an upstream V2 // payload's resources back into structured resources. // - // `unit` is intentionally never set: V2 always sends it empty (`unit: "".to_string()` in - // `series_to_proto_message`), so a `dd.internal.unit` tag stays an ordinary tag here too. let SeriesTags { tags, resources, @@ -270,7 +269,8 @@ fn encode_metric_to_v3( let product = origin.product().unwrap_or(origin_product_value); let category = origin.category().unwrap_or(ORIGIN_CATEGORY_VALUE); let service = origin.service().unwrap_or(0); - builder.set_origin(product, category, service, false); + let no_index = origin.metric_type() == Some(DATADOG_ORIGIN_METRIC_TYPE_NO_INDEX); + builder.set_origin(product, category, service, no_index); } // Interval — matches V2, which always stamps the interval field on the message @@ -279,8 +279,9 @@ fn encode_metric_to_v3( builder.set_interval(interval.into()); } - // Note: `unit` is intentionally never set — V2 always sends it empty (see - // `series_to_proto_message`'s `unit: "".to_string()`). + if let Some(unit) = event_metadata.datadog_metric_unit() { + builder.set_unit(unit); + } // ── Data points ───────────────────────────────────────────────────────── let timestamp = encode_timestamp(metric.timestamp()); @@ -324,8 +325,13 @@ mod tests { use super::super::config::SeriesApiVersion; use super::*; + use datadog_proto::metrics_v3::MetricData; + use prost::Message; use vector_lib::{ - event::{MetricKind, MetricValue, metric::MetricSketch}, + event::{ + DatadogMetricOriginMetadata, EventMetadata, MetricKind, MetricValue, + metric::MetricSketch, + }, metrics::AgentDDSketch, }; @@ -640,6 +646,43 @@ mod tests { assert_eq!(origin.service(), Some(211)); } + #[test] + fn v3_preserves_unit_and_no_index_metadata() { + const FLAG_NO_INDEX: u64 = 0x100; + const FLAG_HAS_UNIT: u64 = 0x200; + + let metadata = EventMetadata::default() + .with_origin_metadata( + DatadogMetricOriginMetadata::new(Some(10), Some(11), Some(42)) + .with_metric_type(Some(DATADOG_ORIGIN_METRIC_TYPE_NO_INDEX)), + ) + .with_datadog_metric_unit("byte".to_string()); + let metric = Metric::new_with_metadata( + "preserved.metadata", + MetricKind::Absolute, + MetricValue::Gauge { value: 1.0 }, + metadata, + ); + + let mut writer = V3Writer::new(); + encode_metric_to_v3( + &mut writer, + &metric, + &None, + log_schema(), + *ORIGIN_PRODUCT_VALUE, + ) + .unwrap(); + let encoded = writer.finalize().unwrap(); + let data = MetricData::decode(encoded.payload.as_slice()).unwrap(); + + assert_eq!(data.types.len(), 1); + assert_eq!(data.types[0] & FLAG_NO_INDEX, FLAG_NO_INDEX); + assert_eq!(data.types[0] & FLAG_HAS_UNIT, FLAG_HAS_UNIT); + assert_eq!(data.dict_unit_str, b"\x04byte"); + assert_eq!(data.unit_refs, vec![1]); + } + #[test] fn v3_namespace_prepended() { let mut enc = DatadogMetricsV3Encoder::new( diff --git a/src/sources/datadog_agent/metrics.rs b/src/sources/datadog_agent/metrics.rs index f51d598632bf6..da049054d49ab 100644 --- a/src/sources/datadog_agent/metrics.rs +++ b/src/sources/datadog_agent/metrics.rs @@ -30,6 +30,14 @@ use crate::{ sources::util::{extract_tag_key_and_value, http::capped_body}, }; +mod v3; + +pub(crate) use v3::decode_ddseries_v3; +#[cfg(test)] +pub(crate) use v3::decode_v3_metric_data; + +const MAX_DATADOG_SERIES_UNIT_EVENT_BYTES: usize = 16 * 1024 * 1024; + #[derive(Deserialize, Serialize)] pub(crate) struct DatadogSeriesRequest { pub(crate) series: Vec, @@ -41,12 +49,15 @@ pub(super) fn build_warp_filter( ) -> BoxedFilter<(Response,)> { let sketches_service = sketches_service(handler.clone(), source.clone()); let series_v1_service = series_v1_service(handler.clone(), source.clone()); - let series_v2_service = series_v2_service(handler, source); + let series_v2_service = series_v2_service(handler.clone(), source.clone()); + let series_v3_service = series_v3_service(handler, source); sketches_service .or(series_v1_service) .unify() .or(series_v2_service) .unify() + .or(series_v3_service) + .unify() .boxed() } @@ -164,6 +175,42 @@ fn series_v2_service( .boxed() } +fn series_v3_service( + handler: RequestHandler, + source: DatadogAgentSource, +) -> BoxedFilter<(Response,)> { + warp::post() + .and(path!("api" / "intake" / "metrics" / "v3" / "series" / ..).and(warp::path::full())) + .and(warp::header::optional::("content-encoding")) + .and(warp::header::optional::("dd-api-key")) + .and(warp::query::()) + .and(capped_body()) + .and_then({ + move |path: FullPath, + encoding_header: Option, + api_token: Option, + query_params: ApiKeyQueryParams, + body: Bytes| { + let events = source + .decode(&encoding_header, body, path.as_str()) + .and_then(|body| { + decode_datadog_series_v3( + body, + source.api_key_extractor.extract( + path.as_str(), + api_token, + query_params.dd_api_key, + ), + source.split_metric_namespace, + &source.events_received, + ) + }); + handler.clone().handle_request(events, super::METRICS) + } + }) + .boxed() +} + #[expect( clippy::needless_pass_by_value, reason = "Request handlers own payload and token values at this boundary." @@ -226,22 +273,65 @@ fn decode_datadog_series_v2( Ok(metrics) } +#[expect( + clippy::needless_pass_by_value, + reason = "Request handlers own payload and token values at this boundary." +)] +fn decode_datadog_series_v3( + body: Bytes, + api_key: Option>, + split_metric_namespace: bool, + events_received: &Registered, +) -> Result, ErrorMessage> { + if body.is_empty() { + debug!(message = "Empty payload ignored."); + return Ok(Vec::new()); + } + + let metrics = decode_ddseries_v3(body, &api_key, split_metric_namespace).map_err(|error| { + ErrorMessage::new( + StatusCode::UNPROCESSABLE_ENTITY, + format!("Error decoding Datadog v3 series: {error:?}"), + ) + })?; + + events_received.emit(CountByteSize( + metrics.len(), + metrics.estimated_json_encoded_size_of(), + )); + + Ok(metrics) +} + /// Builds Vector's `EventMetadata` from the received metadata. Currently this is only /// utilized for passing through origin metadata set by the Agent. -fn get_event_metadata(metadata: Option<&Metadata>) -> EventMetadata { - metadata - .and_then(|metadata| metadata.origin.as_ref()) - .map_or_else(EventMetadata::default, |origin| { - trace!( - "Deserialized origin_product: `{}` origin_category: `{}` origin_service: `{}`.", - origin.origin_product, origin.origin_category, origin.origin_service, - ); - EventMetadata::default().with_origin_metadata(DatadogMetricOriginMetadata::new( - Some(origin.origin_product), - Some(origin.origin_category), - Some(origin.origin_service), - )) - }) +fn get_event_metadata(metadata: Option<&Metadata>, metric_type: Option) -> EventMetadata { + let origin = metadata.and_then(|metadata| metadata.origin.as_ref()); + let metric_type = metric_type.or_else(|| { + origin + .map(|origin| origin.metric_type) + .filter(|metric_type| *metric_type != 0) + }); + if origin.is_none() && metric_type.is_none() { + EventMetadata::default() + } else { + let (origin_product, origin_category, origin_service) = + origin.map_or((None, None, None), |origin| { + trace!( + "Deserialized origin_product: `{}` origin_category: `{}` origin_service: `{}`.", + origin.origin_product, origin.origin_category, origin.origin_service, + ); + ( + Some(origin.origin_product), + Some(origin.origin_category), + Some(origin.origin_service), + ) + }); + EventMetadata::default().with_origin_metadata( + DatadogMetricOriginMetadata::new(origin_product, origin_category, origin_service) + .with_metric_type(metric_type), + ) + } } #[allow( @@ -263,10 +353,45 @@ pub(crate) fn decode_ddseries_v2( split_metric_namespace: bool, ) -> crate::Result> { let payload = MetricPayload::decode(frame)?; - let decoded_metrics: Vec = payload - .series + decode_ddseries(payload.series, api_key, split_metric_namespace, None) +} + +#[allow( + clippy::ref_option, + reason = "Changing this shared optional metadata signature is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659" +)] +#[allow( + clippy::too_many_lines, + reason = "Splitting this source routine is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659" +)] +#[allow( + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + reason = "Preserve the existing numeric conversion semantics of this source. https://github.com/vectordotdev/vector/issues/23659" +)] +fn decode_ddseries( + series: Vec, + api_key: &Option>, + split_metric_namespace: bool, + metric_types: Option<&[Option]>, +) -> crate::Result> { + validate_unit_event_expansion( + series + .iter() + .map(|series| (series.unit.as_str(), series.points.len())), + )?; + + let decoded_metrics: Vec = series .into_iter() - .flat_map(|serie| { + .enumerate() + .flat_map(|(index, serie)| { + // A series without points cannot emit an event. Skip it before expanding its tags + // into a MetricTags map, since a compressed v3 payload can otherwise make this an + // arbitrarily large allocation that is immediately discarded. + if serie.points.is_empty() { + return Vec::new(); + } + let (namespace, name) = if split_metric_namespace { namespace_name_from_dd_metric(&serie.metric) } else { @@ -274,7 +399,17 @@ pub(crate) fn decode_ddseries_v2( }; let mut tags = into_metric_tags(serie.tags); - let event_metadata = get_event_metadata(serie.metadata.as_ref()); + let metric_type = metric_types + .and_then(|metric_types| metric_types.get(index)) + .copied() + .flatten(); + let mut event_metadata = get_event_metadata(serie.metadata.as_ref(), metric_type); + if !serie.unit.is_empty() { + event_metadata.set_datadog_metric_unit(serie.unit.clone()); + } + if let Some(api_key) = api_key { + event_metadata.set_datadog_api_key(Arc::clone(api_key)); + } // It is possible to receive non-rate metrics from the Agent with an interval set. // That interval can be applied with the `as_rate` function in the Datadog UI. @@ -324,8 +459,6 @@ pub(crate) fn decode_ddseries_v2( }); (!serie.source_type_name.is_empty()) .then(|| tags.replace("source_type_name".into(), serie.source_type_name)); - // As per https://github.com/DataDog/datadog-agent/blob/a62ac9fb13e1e5060b89e731b8355b2b20a07c5b/pkg/serializer/internal/metrics/iterable_series.go#L224 - // serie.unit is omitted match metric_payload::MetricType::try_from(serie.r#type) { Ok(metric_payload::MetricType::Count) => serie .points @@ -402,17 +535,31 @@ pub(crate) fn decode_ddseries_v2( } } }) - .map(|mut metric| { - if let Some(k) = &api_key { - metric.metadata_mut().set_datadog_api_key(Arc::clone(k)); - } - metric.into() - }) + .map(Event::from) .collect(); Ok(decoded_metrics) } +fn validate_unit_event_expansion<'a>( + series: impl IntoIterator, +) -> crate::Result<()> { + let expanded_bytes = series + .into_iter() + .try_fold(0usize, |total, (unit, points)| { + unit.len() + .checked_mul(points) + .and_then(|bytes| total.checked_add(bytes)) + .ok_or("Datadog series unit expansion size overflow") + })?; + + if expanded_bytes > MAX_DATADOG_SERIES_UNIT_EVENT_BYTES { + return Err("Datadog series unit expansion exceeds size limit".into()); + } + + Ok(()) +} + #[expect( clippy::needless_pass_by_value, reason = "Request handlers own payload and token values at this boundary." @@ -437,6 +584,19 @@ fn decode_datadog_series_v1( ) })?; + validate_unit_event_expansion(metrics.series.iter().map(|series| { + ( + series.unit.as_deref().unwrap_or_default(), + series.points.len(), + ) + })) + .map_err(|error| { + ErrorMessage::new( + StatusCode::UNPROCESSABLE_ENTITY, + format!("Error decoding Datadog series: {error:?}"), + ) + })?; + let decoded_metrics: Vec = metrics .series .into_iter() @@ -464,17 +624,32 @@ fn into_metric_tags(tags: Vec) -> MetricTags { .collect() } -#[expect( - clippy::needless_pass_by_value, - reason = "Request handlers own payload and token values at this boundary." -)] fn into_vector_metric( dd_metric: DatadogSeriesMetric, api_key: Option>, schema_definition: &Arc, split_metric_namespace: bool, ) -> Vec { + // Avoid copying large units or expanding tags for a series that cannot emit an event. + if dd_metric.points.is_empty() { + return Vec::new(); + } + let mut tags = into_metric_tags(dd_metric.tags.unwrap_or_default()); + let mut event_metadata = dd_metric + .metadata + .as_ref() + .and_then(|metadata| metadata.origin.as_ref()) + .map_or_else(EventMetadata::default, |origin| { + EventMetadata::default().with_origin_metadata(origin.clone()) + }); + if let Some(unit) = dd_metric.unit.filter(|unit| !unit.is_empty()) { + event_metadata.set_datadog_metric_unit(unit); + } + if let Some(api_key) = api_key { + event_metadata.set_datadog_api_key(api_key); + } + event_metadata.set_schema_definition(schema_definition); if let Some(key) = log_schema().host_key() { dd_metric @@ -500,10 +675,11 @@ fn into_vector_metric( .points .iter() .map(|dd_point| { - Metric::new( + Metric::new_with_metadata( name.to_string(), MetricKind::Incremental, MetricValue::Counter { value: dd_point.1 }, + event_metadata.clone(), ) .with_timestamp(Some( Utc.timestamp_opt(dd_point.0, 0) @@ -518,10 +694,11 @@ fn into_vector_metric( .points .iter() .map(|dd_point| { - Metric::new( + Metric::new_with_metadata( name.to_string(), MetricKind::Absolute, MetricValue::Gauge { value: dd_point.1 }, + event_metadata.clone(), ) .with_timestamp(Some( Utc.timestamp_opt(dd_point.0, 0) @@ -539,12 +716,13 @@ fn into_vector_metric( .iter() .map(|dd_point| { let i = dd_metric.interval.filter(|v| *v != 0).unwrap_or(1); - Metric::new( + Metric::new_with_metadata( name.to_string(), MetricKind::Incremental, MetricValue::Counter { value: dd_point.1 * f64::from(i), }, + event_metadata.clone(), ) .with_timestamp(Some( Utc.timestamp_opt(dd_point.0, 0) @@ -559,17 +737,7 @@ fn into_vector_metric( .collect::>(), } .into_iter() - .map(|mut metric| { - if let Some(k) = &api_key { - metric.metadata_mut().set_datadog_api_key(Arc::clone(k)); - } - - metric - .metadata_mut() - .set_schema_definition(schema_definition); - - metric.into() - }) + .map(Event::from) .collect() } @@ -609,7 +777,7 @@ pub(crate) fn decode_ddsketch( .host_key() .and_then(|key| tags.replace(key.to_string(), sketch_series.host.clone())); - let event_metadata = get_event_metadata(sketch_series.metadata.as_ref()); + let event_metadata = get_event_metadata(sketch_series.metadata.as_ref(), None); sketch_series.dogsketches.into_iter().map(move |sketch| { let k: Vec = sketch.k.iter().map(|k| *k as i16).collect(); diff --git a/src/sources/datadog_agent/metrics/v3.rs b/src/sources/datadog_agent/metrics/v3.rs new file mode 100644 index 0000000000000..443026f306fc7 --- /dev/null +++ b/src/sources/datadog_agent/metrics/v3.rs @@ -0,0 +1,951 @@ +use std::{collections::HashSet, sync::Arc}; + +use bytes::Bytes; +use chrono::{TimeZone, Utc}; +use prost::Message; + +use super::decode_ddseries; +use crate::{ + common::datadog::DATADOG_ORIGIN_METRIC_TYPE_NO_INDEX, + event::{Event, metric::TagValueSet}, + sources::datadog_agent::{ + ddmetric_proto::{ + Metadata, Origin, + metric_payload::{self, MetricPoint, MetricSeries, Resource}, + }, + ddmetric_v3_proto::{ + Metadata as MetricMetadataV3, MetricData as MetricDataV3, Payload as MetricPayloadV3, + }, + }, +}; + +const MAX_V3_EXPANDED_TAGSET_BYTES: usize = 16 * 1024 * 1024; +const MAX_V3_EXPANDED_RESOURCE_BYTES: usize = 16 * 1024 * 1024; +const MAX_V3_EXPANDED_SERIES_BYTES: usize = 16 * 1024 * 1024; +const MAX_V3_EXPANDED_EVENT_BYTES: usize = 16 * 1024 * 1024; +const MAX_V3_DECODED_REPEATED_BYTES: usize = 16 * 1024 * 1024; +// `decode_ddseries` converts every wire tag/resource into a BTreeMap entry. Account for the +// key/value containers plus a conservative three-pointer share of the tree node allocation. +const V3_METRIC_TAG_ENTRY_BYTES: usize = + std::mem::size_of::<(String, TagValueSet)>() + 3 * std::mem::size_of::(); + +// `MetricData.types` packs three independent values into each integer. These masks mirror +// `metricType`, `valueType`, and `metricFlags` in Datadog's `intake_v3.proto`. +const V3_METRIC_TYPE_MASK: u64 = 0x0f; +const V3_VALUE_TYPE_MASK: u64 = 0xf0; +const V3_FLAG_NO_INDEX: u64 = 0x100; +const V3_FLAG_HAS_UNIT: u64 = 0x200; +const V3_METRIC_TYPE_SKETCH: u64 = 4; +const V3_VALUE_TYPE_ZERO: u64 = 0; +const V3_VALUE_TYPE_SINT64: u64 = 0x10; +const V3_VALUE_TYPE_FLOAT32: u64 = 0x20; +const V3_VALUE_TYPE_FLOAT64: u64 = 0x30; + +#[allow( + clippy::ref_option, + reason = "Match the v2 decoder API while the shared signature remains unchanged. https://github.com/vectordotdev/vector/issues/23659" +)] +pub(crate) fn decode_ddseries_v3( + frame: Bytes, + api_key: &Option>, + split_metric_namespace: bool, +) -> crate::Result> { + validate_v3_predecode_allocations(&frame)?; + let payload = MetricPayloadV3::decode(frame)?; + // `types` combines the metric kind, point value encoding, and flags. `flagNoIndex` is not + // represented by the v2 `MetricSeries` used during translation, so carry its Datadog origin + // value alongside each translated series until `decode_ddseries` builds the Vector events. + let metric_types = payload.metric_data.as_ref().map(|metric_data| { + metric_data + .types + .iter() + .map(|packed_type| { + (packed_type & V3_FLAG_NO_INDEX != 0).then_some(DATADOG_ORIGIN_METRIC_TYPE_NO_INDEX) + }) + .collect::>() + }); + let series = payload.metric_data.map_or(Ok(Vec::new()), |metric_data| { + decode_v3_metric_data(&metric_data, payload.metadata.as_ref()) + })?; + decode_ddseries( + series, + api_key, + split_metric_namespace, + metric_types.as_deref(), + ) +} + +fn validate_v3_metric_columns( + data: &MetricDataV3, + metadata: Option<&MetricMetadataV3>, +) -> crate::Result<()> { + if metadata.is_some_and(|metadata| !metadata.resources.len().is_multiple_of(2)) { + return Err("Datadog v3 metadata resources must be pairs".into()); + } + + let series_count = data.types.len(); + if !data.dict_origin_info.len().is_multiple_of(3) + || data.name_refs.len() < series_count + || data.tagset_refs.len() < series_count + || data.resources_refs.len() < series_count + || data.intervals.len() < series_count + || data.num_points.len() < series_count + || data.source_type_name_refs.len() < series_count + || data.origin_info_refs.len() < series_count + { + return Err("invalid Datadog v3 metric columns".into()); + } + Ok(()) +} + +fn v3_string_list_bytes<'a>(values: impl IntoIterator) -> crate::Result { + values + .into_iter() + .try_fold(0, |total, value| checked_v3_string_bytes(total, value)) +} + +fn v3_resource_list_bytes(resources: &[(String, String)]) -> crate::Result { + resources + .iter() + .try_fold(0, |total, (resource_type, name)| { + let total = checked_v3_string_bytes(total, resource_type)?; + checked_v3_string_bytes(total, name) + }) +} + +#[derive(Default)] +struct V3ReferenceDecoder { + name: i64, + tagset: i64, + resources: i64, + source_type: i64, + origin: i64, + unit: i64, + unit_index: usize, +} + +struct V3References { + name: usize, + tagset: usize, + resources: usize, + source_type: usize, + origin: usize, +} + +impl V3ReferenceDecoder { + fn advance_reference( + current: &mut i64, + delta: i64, + dictionary_len: usize, + overflow_error: &'static str, + ) -> crate::Result { + *current = current.checked_add(delta).ok_or(overflow_error)?; + usize::try_from(*current) + .ok() + .filter(|index| *index < dictionary_len) + .ok_or_else(|| "invalid Datadog v3 dictionary reference".into()) + } + + fn advance( + &mut self, + data: &MetricDataV3, + index: usize, + dictionary_lengths: [usize; 5], + ) -> crate::Result { + Ok(V3References { + name: Self::advance_reference( + &mut self.name, + data.name_refs[index], + dictionary_lengths[0], + "Datadog v3 name reference overflow", + )?, + tagset: Self::advance_reference( + &mut self.tagset, + data.tagset_refs[index], + dictionary_lengths[1], + "Datadog v3 tagset reference overflow", + )?, + resources: Self::advance_reference( + &mut self.resources, + data.resources_refs[index], + dictionary_lengths[2], + "Datadog v3 resource reference overflow", + )?, + source_type: Self::advance_reference( + &mut self.source_type, + data.source_type_name_refs[index], + dictionary_lengths[3], + "Datadog v3 source type reference overflow", + )?, + origin: Self::advance_reference( + &mut self.origin, + data.origin_info_refs[index], + dictionary_lengths[4], + "Datadog v3 origin reference overflow", + )?, + }) + } + + fn decode_unit( + &mut self, + data: &MetricDataV3, + units: &[String], + packed_type: u64, + ) -> crate::Result { + if packed_type & V3_FLAG_HAS_UNIT == 0 { + return Ok(String::new()); + } + let delta = *data + .unit_refs + .get(self.unit_index) + .ok_or("invalid Datadog v3 unit column")?; + self.unit_index += 1; + let index = Self::advance_reference( + &mut self.unit, + delta, + units.len(), + "Datadog v3 unit reference overflow", + )?; + Ok(units[index].clone()) + } +} + +struct V3PointDecoder<'a> { + data: &'a MetricDataV3, + timestamp: i64, + timestamp_index: usize, + sint64_index: usize, + float32_index: usize, + float64_index: usize, +} + +impl<'a> V3PointDecoder<'a> { + const fn new(data: &'a MetricDataV3) -> Self { + Self { + data, + timestamp: 0, + timestamp_index: 0, + sint64_index: 0, + float32_index: 0, + float64_index: 0, + } + } + + fn validate_count(&self, value_type: u64, point_count: usize) -> crate::Result<()> { + let timestamp_count = self + .data + .timestamps + .len() + .saturating_sub(self.timestamp_index); + let value_count = match value_type { + V3_VALUE_TYPE_ZERO => timestamp_count, + V3_VALUE_TYPE_SINT64 => self + .data + .vals_sint64 + .len() + .saturating_sub(self.sint64_index), + V3_VALUE_TYPE_FLOAT32 => self + .data + .vals_float32 + .len() + .saturating_sub(self.float32_index), + V3_VALUE_TYPE_FLOAT64 => self + .data + .vals_float64 + .len() + .saturating_sub(self.float64_index), + _ => return Err("invalid Datadog v3 value type".into()), + }; + if point_count > timestamp_count || point_count > value_count { + return Err("invalid Datadog v3 point count".into()); + } + Ok(()) + } + + #[expect( + clippy::cast_precision_loss, + reason = "Vector metrics store protocol integer samples as f64 values." + )] + fn decode_points( + &mut self, + value_type: u64, + point_count: usize, + ) -> crate::Result> { + self.validate_count(value_type, point_count)?; + let mut points = Vec::with_capacity(point_count); + for _ in 0..point_count { + self.timestamp = self + .timestamp + .checked_add(self.data.timestamps[self.timestamp_index]) + .ok_or("Datadog v3 timestamp overflow")?; + if Utc.timestamp_opt(self.timestamp, 0).single().is_none() { + return Err("invalid Datadog v3 timestamp".into()); + } + self.timestamp_index += 1; + let value = match value_type { + V3_VALUE_TYPE_ZERO => 0.0, + V3_VALUE_TYPE_SINT64 => { + let value = self.data.vals_sint64[self.sint64_index] as f64; + self.sint64_index += 1; + value + } + V3_VALUE_TYPE_FLOAT32 => { + let value = f64::from(self.data.vals_float32[self.float32_index]); + self.float32_index += 1; + value + } + V3_VALUE_TYPE_FLOAT64 => { + let value = self.data.vals_float64[self.float64_index]; + self.float64_index += 1; + value + } + _ => return Err("invalid Datadog v3 value type".into()), + }; + points.push(MetricPoint { + value, + timestamp: self.timestamp, + }); + } + Ok(points) + } +} + +#[derive(Default)] +struct V3AllocationBudget { + series_bytes: usize, + event_bytes: usize, +} + +#[derive(Clone, Copy)] +struct V3TagResourceSize { + tag_bytes: usize, + tag_count: usize, + resource_bytes: usize, + resource_count: usize, +} + +impl V3AllocationBudget { + /// Accounts for every clone and vector expansion performed for one series. + /// + /// Callers must invoke this before allocating points, tags, resources, or events for the + /// series. Successful return means both cumulative expansion limits still hold. + fn check_before_allocating( + &mut self, + point_count: usize, + name: &str, + unit: &str, + source_type: &str, + tag_resources: V3TagResourceSize, + ) -> crate::Result<()> { + let tag_resource_bytes = tag_resources + .tag_bytes + .checked_add(source_type.len()) + .and_then(|total| total.checked_add(tag_resources.resource_bytes)) + .ok_or("Datadog v3 expanded series size overflow")?; + let tag_resource_copies = point_count + .checked_add(1) + .ok_or("Datadog v3 expanded series size overflow")?; + let expanded_tag_resource_bytes = tag_resource_bytes + .checked_mul(tag_resource_copies) + .ok_or("Datadog v3 expanded series size overflow")?; + let dictionary_string_bytes = checked_v3_string_bytes(0, name)?; + let dictionary_string_bytes = checked_v3_string_bytes(dictionary_string_bytes, unit)?; + let dictionary_string_bytes = + checked_v3_string_bytes(dictionary_string_bytes, source_type)?; + let series_bytes = std::mem::size_of::() + .checked_add( + point_count + .checked_mul(std::mem::size_of::()) + .ok_or("Datadog v3 expanded series size overflow")?, + ) + .and_then(|total| total.checked_add(expanded_tag_resource_bytes)) + .and_then(|total| total.checked_add(dictionary_string_bytes)) + .ok_or("Datadog v3 expanded series size overflow")?; + self.series_bytes = self + .series_bytes + .checked_add(series_bytes) + .ok_or("Datadog v3 expanded series size overflow")?; + if self.series_bytes > MAX_V3_EXPANDED_SERIES_BYTES { + return Err("Datadog v3 expanded series exceed size limit".into()); + } + + let event_tag_count = tag_resources + .tag_count + .checked_add(tag_resources.resource_count) + .and_then(|total| total.checked_add(usize::from(!source_type.is_empty()))) + .ok_or("Datadog v3 expanded event size overflow")?; + let event_tag_entry_bytes = event_tag_count + .checked_mul(V3_METRIC_TAG_ENTRY_BYTES) + .ok_or("Datadog v3 expanded event size overflow")?; + let event_bytes = std::mem::size_of::() + .checked_add(name.len()) + .and_then(|total| total.checked_add(tag_resources.tag_bytes)) + .and_then(|total| total.checked_add(tag_resources.resource_bytes)) + .and_then(|total| total.checked_add(source_type.len())) + .and_then(|total| total.checked_add(unit.len())) + .and_then(|total| total.checked_add(event_tag_entry_bytes)) + .ok_or("Datadog v3 expanded event size overflow")?; + self.event_bytes = self + .event_bytes + .checked_add( + event_bytes + .checked_mul(point_count) + .ok_or("Datadog v3 expanded event size overflow")?, + ) + .ok_or("Datadog v3 expanded event size overflow")?; + if self.event_bytes > MAX_V3_EXPANDED_EVENT_BYTES { + return Err("Datadog v3 expanded events exceed size limit".into()); + } + Ok(()) + } +} + +#[allow( + clippy::too_many_lines, + reason = "Keep the cursor advances and allocation-budget checks together so their ordering remains auditable." +)] +pub(crate) fn decode_v3_metric_data( + data: &MetricDataV3, + metadata: Option<&MetricMetadataV3>, +) -> crate::Result> { + validate_v3_metric_columns(data, metadata)?; + let mut string_dictionary_bytes = 0; + let names = decode_v3_strings(&data.dict_name_str, false, &mut string_dictionary_bytes)?; + let tag_strings = decode_v3_strings(&data.dict_tag_str, true, &mut string_dictionary_bytes)?; + let units = decode_v3_strings(&data.dict_unit_str, false, &mut string_dictionary_bytes)?; + let tagsets = decode_v3_tagsets(&data.dict_tagsets, &tag_strings, metadata)?; + let resources = decode_v3_resources(data, &mut string_dictionary_bytes)?; + let source_types = decode_v3_strings( + &data.dict_source_type_name, + false, + &mut string_dictionary_bytes, + )?; + let origin_count = data.dict_origin_info.len() / 3 + 1; + + // The size of a dictionary entry is stable, so calculate it once rather than rescanning the + // same tags and resources for every series that references it. + let tagset_bytes = tagsets + .iter() + .map(|tags| v3_string_list_bytes(tags)) + .collect::>>()?; + let tagset_counts = tagsets.iter().map(Vec::len).collect::>(); + let metadata_resource_bytes = metadata + .map(|metadata| v3_string_list_bytes(&metadata.resources)) + .transpose()? + .unwrap_or(0); + let resource_bytes = resources + .iter() + .map(|entry| { + v3_resource_list_bytes(entry)? + .checked_add(metadata_resource_bytes) + .ok_or_else(|| "Datadog v3 expanded resource size overflow".into()) + }) + .collect::>>()?; + let metadata_resource_count = metadata.map_or(0, |metadata| metadata.resources.len() / 2); + let resource_counts = resources + .iter() + .map(|entry| entry.len() + metadata_resource_count) + .collect::>(); + + let dictionary_lengths = [ + names.len(), + tagsets.len(), + resources.len(), + source_types.len(), + origin_count, + ]; + let mut references = V3ReferenceDecoder::default(); + let mut point_decoder = V3PointDecoder::new(data); + let mut allocation_budget = V3AllocationBudget::default(); + let mut series = Vec::new(); + + for index in 0..data.types.len() { + let refs = references.advance(data, index, dictionary_lengths)?; + let packed_type = data.types[index]; + let metric_type = packed_type & V3_METRIC_TYPE_MASK; + let value_type = packed_type & V3_VALUE_TYPE_MASK; + if metric_type == V3_METRIC_TYPE_SKETCH { + return Err("Datadog v3 series payload contains a sketch".into()); + } + let unit = references.decode_unit(data, &units, packed_type)?; + let point_count = usize::try_from(data.num_points[index]) + .map_err(|_| "invalid Datadog v3 point count")?; + let name = &names[refs.name]; + let source_type = &source_types[refs.source_type]; + allocation_budget.check_before_allocating( + point_count, + name, + &unit, + source_type, + V3TagResourceSize { + tag_bytes: tagset_bytes[refs.tagset], + tag_count: tagset_counts[refs.tagset], + resource_bytes: resource_bytes[refs.resources], + resource_count: resource_counts[refs.resources], + }, + )?; + let points = point_decoder.decode_points(value_type, point_count)?; + + let metric_type = match metric_type { + 1 => metric_payload::MetricType::Count, + 2 => metric_payload::MetricType::Rate, + 3 => metric_payload::MetricType::Gauge, + _ => metric_payload::MetricType::Unspecified, + }; + let series_metadata = if refs.origin == 0 { + None + } else { + let offset = (refs.origin - 1) * 3; + Some(Metadata { + origin: Some(Origin { + metric_type: 0, + origin_product: u32::try_from(data.dict_origin_info[offset]) + .map_err(|_| "invalid Datadog v3 origin product")?, + origin_category: u32::try_from(data.dict_origin_info[offset + 1]) + .map_err(|_| "invalid Datadog v3 origin category")?, + origin_service: u32::try_from(data.dict_origin_info[offset + 2]) + .map_err(|_| "invalid Datadog v3 origin service")?, + }), + }) + }; + let mut decoded_resources: Vec = resources[refs.resources] + .iter() + .map(|(resource_type, name)| Resource { + r#type: resource_type.clone(), + name: name.clone(), + }) + .collect(); + if let Some(payload_metadata) = metadata { + decoded_resources.extend(payload_metadata.resources.chunks_exact(2).map(|pair| { + Resource { + r#type: pair[0].clone(), + name: pair[1].clone(), + } + })); + } + let interval = + u32::try_from(data.intervals[index]).map_err(|_| "invalid Datadog v3 interval")?; + interval + .checked_mul(1000) + .ok_or("Datadog v3 interval milliseconds overflow")?; + series.push(MetricSeries { + resources: decoded_resources, + metric: name.clone(), + tags: tagsets[refs.tagset].clone(), + points, + r#type: metric_type as i32, + unit, + source_type_name: source_type.clone(), + interval: i64::from(interval), + metadata: series_metadata, + }); + } + Ok(series) +} + +fn decode_v3_strings( + raw: &[u8], + sanitize: bool, + expanded_bytes: &mut usize, +) -> crate::Result> { + *expanded_bytes = checked_v3_string_bytes(*expanded_bytes, "")?; + if *expanded_bytes > MAX_V3_EXPANDED_SERIES_BYTES { + return Err("Datadog v3 expanded string dictionaries exceed size limit".into()); + } + let mut values = vec![String::new()]; + let mut offset = 0; + while offset < raw.len() { + let (length, consumed) = decode_v3_varint(&raw[offset..])?; + offset += consumed; + let length = usize::try_from(length).map_err(|_| "Datadog v3 string length overflow")?; + let end = offset + .checked_add(length) + .ok_or("Datadog v3 string length overflow")?; + if end > raw.len() { + return Err("truncated Datadog v3 string dictionary".into()); + } + let bytes = &raw[offset..end]; + let valid_string_bytes = (*expanded_bytes) + .checked_add(std::mem::size_of::()) + .and_then(|total| total.checked_add(length)) + .ok_or("Datadog v3 expanded string dictionary size overflow")?; + if valid_string_bytes > MAX_V3_EXPANDED_SERIES_BYTES { + return Err("Datadog v3 expanded string dictionaries exceed size limit".into()); + } + let value = match std::str::from_utf8(bytes) { + Ok(value) => { + *expanded_bytes = valid_string_bytes; + value.to_owned() + } + Err(_) if sanitize => { + let worst_case_length = length + .checked_mul(3) + .ok_or("Datadog v3 expanded string dictionary size overflow")?; + let worst_case_bytes = (*expanded_bytes) + .checked_add(std::mem::size_of::()) + .and_then(|total| total.checked_add(worst_case_length)) + .ok_or("Datadog v3 expanded string dictionary size overflow")?; + if worst_case_bytes > MAX_V3_EXPANDED_SERIES_BYTES { + return Err("Datadog v3 expanded string dictionaries exceed size limit".into()); + } + let value = String::from_utf8_lossy(bytes).into_owned(); + *expanded_bytes = checked_v3_string_bytes(*expanded_bytes, &value)?; + value + } + Err(error) => return Err(error.into()), + }; + values.push(value); + offset = end; + } + Ok(values) +} + +fn decode_v3_varint(raw: &[u8]) -> crate::Result<(u64, usize)> { + let mut value = 0; + for (index, byte) in raw.iter().copied().enumerate() { + if index == 10 { + return Err("Datadog v3 varint overflow".into()); + } + if index == 9 && byte > 1 { + return Err("Datadog v3 varint overflow".into()); + } + value |= u64::from(byte & 0x7f) << (index * 7); + if byte & 0x80 == 0 { + return Ok((value, index + 1)); + } + } + Err("truncated Datadog v3 varint".into()) +} + +fn validate_v3_predecode_allocations(raw: &[u8]) -> crate::Result<()> { + let mut offset = 0; + let mut metadata_bytes = 0usize; + let mut repeated_numeric_bytes = 0usize; + while offset < raw.len() { + let (key, consumed) = decode_v3_varint(&raw[offset..])?; + offset += consumed; + let field = key >> 3; + let wire_type = key & 0x07; + if matches!(field, 2 | 3) && wire_type == 2 { + let (message, end) = decode_v3_length_delimited(raw, offset)?; + if field == 2 { + validate_v3_metadata_allocations(message, &mut metadata_bytes)?; + } else { + validate_v3_metric_data_allocations(message, &mut repeated_numeric_bytes)?; + } + offset = end; + } else { + offset = skip_v3_field(raw, offset, wire_type)?; + } + } + Ok(()) +} + +fn validate_v3_metadata_allocations(raw: &[u8], decoded_bytes: &mut usize) -> crate::Result<()> { + let mut offset = 0; + while offset < raw.len() { + let (key, consumed) = decode_v3_varint(&raw[offset..])?; + offset += consumed; + let field = key >> 3; + let wire_type = key & 0x07; + if matches!(field, 1 | 2) && wire_type == 2 { + let (value, end) = decode_v3_length_delimited(raw, offset)?; + *decoded_bytes = decoded_bytes + .checked_add(std::mem::size_of::()) + .and_then(|total| total.checked_add(value.len())) + .ok_or("Datadog v3 metadata string allocation overflow")?; + if *decoded_bytes > MAX_V3_DECODED_REPEATED_BYTES { + return Err("Datadog v3 metadata strings exceed size limit".into()); + } + offset = end; + } else { + offset = skip_v3_field(raw, offset, wire_type)?; + } + } + Ok(()) +} + +fn validate_v3_metric_data_allocations(raw: &[u8], decoded_bytes: &mut usize) -> crate::Result<()> { + let mut offset = 0; + while offset < raw.len() { + let (key, consumed) = decode_v3_varint(&raw[offset..])?; + offset += consumed; + let field = key >> 3; + let wire_type = key & 0x07; + let element_size = match field { + 3 | 5..=7 | 10..=17 | 19 | 20 | 23 | 24 | 26 => 8, + 9 | 18 | 21 | 22 => 4, + _ => { + offset = skip_v3_field(raw, offset, wire_type)?; + continue; + } + }; + let fixed_width = match field { + 18 => Some(4), + 19 => Some(8), + _ => None, + }; + let count = if wire_type == 2 { + let (packed, end) = decode_v3_length_delimited(raw, offset)?; + offset = end; + match fixed_width { + Some(width) => { + if packed.len() % width != 0 { + return Err("invalid packed Datadog v3 numeric column".into()); + } + packed.len() / width + } + None => count_v3_packed_varints(packed)?, + } + } else if fixed_width.is_some_and(|width| wire_type == if width == 4 { 5 } else { 1 }) + || fixed_width.is_none() && wire_type == 0 + { + offset = skip_v3_field(raw, offset, wire_type)?; + 1 + } else { + offset = skip_v3_field(raw, offset, wire_type)?; + continue; + }; + let allocation = count + .checked_mul(element_size) + .ok_or("Datadog v3 numeric column allocation overflow")?; + *decoded_bytes = decoded_bytes + .checked_add(allocation) + .ok_or("Datadog v3 numeric column allocation overflow")?; + if *decoded_bytes > MAX_V3_DECODED_REPEATED_BYTES { + return Err("Datadog v3 numeric columns exceed size limit".into()); + } + } + Ok(()) +} + +fn count_v3_packed_varints(raw: &[u8]) -> crate::Result { + let mut offset = 0; + let mut count = 0usize; + while offset < raw.len() { + let (_, consumed) = decode_v3_varint(&raw[offset..])?; + offset = offset + .checked_add(consumed) + .ok_or("Datadog v3 packed numeric count overflow")?; + count = count + .checked_add(1) + .ok_or("Datadog v3 packed numeric count overflow")?; + } + Ok(count) +} + +fn decode_v3_length_delimited(raw: &[u8], offset: usize) -> crate::Result<(&[u8], usize)> { + let (length, consumed) = decode_v3_varint( + raw.get(offset..) + .ok_or("truncated Datadog v3 protobuf field")?, + )?; + let length = + usize::try_from(length).map_err(|_| "Datadog v3 protobuf field length overflow")?; + let start = offset + .checked_add(consumed) + .ok_or("Datadog v3 protobuf field length overflow")?; + let end = start + .checked_add(length) + .ok_or("Datadog v3 protobuf field length overflow")?; + let value = raw + .get(start..end) + .ok_or("truncated Datadog v3 protobuf field")?; + Ok((value, end)) +} + +fn skip_v3_field(raw: &[u8], offset: usize, wire_type: u64) -> crate::Result { + match wire_type { + 0 => { + let (_, consumed) = decode_v3_varint( + raw.get(offset..) + .ok_or("truncated Datadog v3 protobuf field")?, + )?; + offset + .checked_add(consumed) + .ok_or_else(|| "Datadog v3 protobuf field length overflow".into()) + } + 1 => offset + .checked_add(8) + .filter(|&end| end <= raw.len()) + .ok_or_else(|| "truncated Datadog v3 protobuf field".into()), + 2 => { + let (length, consumed) = decode_v3_varint( + raw.get(offset..) + .ok_or("truncated Datadog v3 protobuf field")?, + )?; + let length = + usize::try_from(length).map_err(|_| "Datadog v3 protobuf field length overflow")?; + offset + .checked_add(consumed) + .and_then(|start| start.checked_add(length)) + .filter(|&end| end <= raw.len()) + .ok_or_else(|| "truncated Datadog v3 protobuf field".into()) + } + 5 => offset + .checked_add(4) + .filter(|&end| end <= raw.len()) + .ok_or_else(|| "truncated Datadog v3 protobuf field".into()), + _ => Err("invalid Datadog v3 protobuf wire type".into()), + } +} + +fn decode_v3_tagsets( + packed: &[i64], + dictionary: &[String], + metadata: Option<&MetricMetadataV3>, +) -> crate::Result>> { + let mut tagsets: Vec> = vec![Vec::new()]; + let mut expanded_bytes = std::mem::size_of::>(); + let mut offset = 0; + while offset < packed.len() { + expanded_bytes = expanded_bytes + .checked_add(std::mem::size_of::>()) + .ok_or("Datadog v3 expanded tagset size overflow")?; + if expanded_bytes > MAX_V3_EXPANDED_TAGSET_BYTES { + return Err("Datadog v3 expanded tagsets exceed size limit".into()); + } + let size = usize::try_from(packed[offset]).map_err(|_| "invalid Datadog v3 tagset size")?; + offset += 1; + let end = offset + .checked_add(size) + .ok_or("Datadog v3 tagset length overflow")?; + if end > packed.len() { + return Err("truncated Datadog v3 tagset dictionary".into()); + } + let mut tags = Vec::new(); + let mut reference: i64 = 0; + for delta in &packed[offset..end] { + reference = reference + .checked_add(*delta) + .ok_or("Datadog v3 tagset reference overflow")?; + if reference < 0 { + let index = usize::try_from( + reference + .checked_neg() + .ok_or("Datadog v3 tagset reference overflow")?, + ) + .map_err(|_| "invalid Datadog v3 tagset reference")?; + let referenced = tagsets + .get(index) + .ok_or("invalid Datadog v3 tagset reference")?; + let referenced_bytes = referenced.iter().try_fold(0usize, |total, tag| { + total + .checked_add(std::mem::size_of::()) + .and_then(|total| total.checked_add(tag.len())) + .ok_or("Datadog v3 expanded tagset size overflow") + })?; + expanded_bytes = expanded_bytes + .checked_add(referenced_bytes) + .ok_or("Datadog v3 expanded tagset size overflow")?; + if expanded_bytes > MAX_V3_EXPANDED_TAGSET_BYTES { + return Err("Datadog v3 expanded tagsets exceed size limit".into()); + } + tags.extend(referenced.iter().cloned()); + } else { + let index = + usize::try_from(reference).map_err(|_| "invalid Datadog v3 tag reference")?; + let tag = dictionary + .get(index) + .ok_or("invalid Datadog v3 tag reference")?; + expanded_bytes = expanded_bytes + .checked_add(std::mem::size_of::()) + .and_then(|total| total.checked_add(tag.len())) + .ok_or("Datadog v3 expanded tagset size overflow")?; + if expanded_bytes > MAX_V3_EXPANDED_TAGSET_BYTES { + return Err("Datadog v3 expanded tagsets exceed size limit".into()); + } + tags.push(tag.clone()); + } + } + tagsets.push(tags); + offset = end; + } + if let Some(metadata) = metadata { + let mut seen = HashSet::new(); + let metadata_tags = metadata + .tags + .iter() + .filter(|tag| seen.insert(tag.as_str())) + .collect::>(); + for tagset in &mut tagsets { + let mut tagset_members = tagset.iter().map(String::as_str).collect::>(); + let mut additional_tags = Vec::new(); + for &tag in &metadata_tags { + if tagset_members.insert(tag) { + expanded_bytes = checked_v3_string_bytes(expanded_bytes, tag)?; + if expanded_bytes > MAX_V3_EXPANDED_TAGSET_BYTES { + return Err("Datadog v3 expanded tagsets exceed size limit".into()); + } + additional_tags.push(tag.clone()); + } + } + tagset.extend(additional_tags); + } + } + Ok(tagsets) +} + +fn decode_v3_resources( + data: &MetricDataV3, + string_dictionary_bytes: &mut usize, +) -> crate::Result>> { + let dictionary = decode_v3_strings(&data.dict_resource_str, false, string_dictionary_bytes)?; + let mut resources = vec![Vec::new()]; + let mut expanded_bytes = std::mem::size_of::>(); + let mut entry_offset: usize = 0; + for &resource_len in &data.dict_resource_len { + expanded_bytes = expanded_bytes + .checked_add(std::mem::size_of::>()) + .ok_or("Datadog v3 expanded resource size overflow")?; + if expanded_bytes > MAX_V3_EXPANDED_RESOURCE_BYTES { + return Err("Datadog v3 expanded resources exceed size limit".into()); + } + let size = usize::try_from(resource_len).map_err(|_| "invalid Datadog v3 resource size")?; + let end = entry_offset + .checked_add(size) + .ok_or("Datadog v3 resource length overflow")?; + if end > data.dict_resource_type.len() || end > data.dict_resource_name.len() { + return Err("truncated Datadog v3 resource dictionary".into()); + } + let mut set = Vec::new(); + let mut type_ref: i64 = 0; + let mut name_ref: i64 = 0; + for index in entry_offset..end { + type_ref = type_ref + .checked_add(data.dict_resource_type[index]) + .ok_or("Datadog v3 resource reference overflow")?; + name_ref = name_ref + .checked_add(data.dict_resource_name[index]) + .ok_or("Datadog v3 resource reference overflow")?; + if type_ref < 0 || name_ref < 0 { + return Err("invalid Datadog v3 resource reference".into()); + } + let type_index = + usize::try_from(type_ref).map_err(|_| "invalid Datadog v3 resource reference")?; + let name_index = + usize::try_from(name_ref).map_err(|_| "invalid Datadog v3 resource reference")?; + let resource_type = dictionary + .get(type_index) + .ok_or("invalid Datadog v3 resource reference")?; + let resource_name = dictionary + .get(name_index) + .ok_or("invalid Datadog v3 resource reference")?; + expanded_bytes = checked_v3_string_bytes(expanded_bytes, resource_type)?; + expanded_bytes = checked_v3_string_bytes(expanded_bytes, resource_name)?; + if expanded_bytes > MAX_V3_EXPANDED_RESOURCE_BYTES { + return Err("Datadog v3 expanded resources exceed size limit".into()); + } + set.push((resource_type.clone(), resource_name.clone())); + } + resources.push(set); + entry_offset = end; + } + Ok(resources) +} + +fn checked_v3_string_bytes(total: usize, value: &str) -> crate::Result { + total + .checked_add(std::mem::size_of::()) + .and_then(|total| total.checked_add(value.len())) + .ok_or_else(|| "Datadog v3 expanded string size overflow".into()) +} diff --git a/src/sources/datadog_agent/mod.rs b/src/sources/datadog_agent/mod.rs index 13ff9edd4d040..f2e18af0b337e 100644 --- a/src/sources/datadog_agent/mod.rs +++ b/src/sources/datadog_agent/mod.rs @@ -13,6 +13,7 @@ use std::{convert::Infallible, fmt::Debug, net::SocketAddr, sync::Arc, time::Dur use bytes::{Buf, Bytes}; use chrono::{DateTime, Utc, serde::ts_milliseconds}; pub(crate) use datadog_proto::agentpayload as ddmetric_proto; +pub(crate) use datadog_proto::metrics_v3 as ddmetric_v3_proto; pub(crate) use datadog_proto::trace as ddtrace_proto; use futures::FutureExt; use http::StatusCode; diff --git a/src/sources/datadog_agent/tests.rs b/src/sources/datadog_agent/tests.rs index e490087af90f4..cb0e09121a013 100644 --- a/src/sources/datadog_agent/tests.rs +++ b/src/sources/datadog_agent/tests.rs @@ -50,7 +50,7 @@ use crate::{ serde::{default_decoding, default_framing_message_based}, sources::datadog_agent::{ DatadogAgentConfig, DatadogAgentSource, LOGS, LogMsg, METRICS, TRACES, ddmetric_proto, - ddtrace_proto, logs::decode_log_body, metrics::DatadogSeriesRequest, + ddmetric_v3_proto, ddtrace_proto, logs::decode_log_body, metrics::DatadogSeriesRequest, }, test_util::{ addr::{PortGuard, next_addr}, @@ -77,6 +77,7 @@ const DD_API_LOGS_V1_PATH: &str = "/v1/input/"; const DD_API_LOGS_V2_PATH: &str = "/api/v2/logs"; const DD_API_SERIES_V1_PATH: &str = "/api/v1/series"; const DD_API_SERIES_V2_PATH: &str = "/api/v2/series"; +const DD_API_SERIES_V3_PATH: &str = "/api/intake/metrics/v3/series"; const DD_API_SKETCHES_PATH: &str = "/api/beta/sketches"; const DD_API_TRACES_PATH: &str = "/api/v0.2/traces"; const HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(5); @@ -404,6 +405,35 @@ async fn send_and_collect( .await } +async fn send_bytes_and_collect( + address: SocketAddr, + body: Vec, + headers: HeaderMap, + path: &'static str, + rx: impl Stream + Unpin, + expected_count: usize, +) -> Vec { + spawn_collect_n( + async move { + let response = timeout( + HTTP_REQUEST_TIMEOUT, + reqwest::Client::new() + .post(format!("http://{address}{path}")) + .headers(headers) + .body(body) + .send(), + ) + .await + .expect("send_bytes_and_collect request timed out") + .unwrap(); + assert_eq!(200, response.status().as_u16()); + }, + rx, + expected_count, + ) + .await +} + /// Smallest v2 payload the `datadog_agent` traces decoder accepts. #[cfg(all(feature = "sinks-vector", feature = "sources-vector"))] fn minimal_v2_trace_body() -> Vec { @@ -978,6 +1008,7 @@ async fn decode_series_endpoint_v1() { host: Some("random_host".to_string()), source_type_name: None, device: None, + unit: None, metadata: None, }, DatadogSeriesMetric { @@ -989,6 +1020,7 @@ async fn decode_series_endpoint_v1() { host: Some("another_random_host".to_string()), source_type_name: None, device: None, + unit: None, metadata: None, }, DatadogSeriesMetric { @@ -1000,6 +1032,7 @@ async fn decode_series_endpoint_v1() { host: Some("a_host".to_string()), source_type_name: None, device: None, + unit: None, metadata: None, }, DatadogSeriesMetric { @@ -1011,6 +1044,7 @@ async fn decode_series_endpoint_v1() { host: None, source_type_name: None, device: None, + unit: None, metadata: None, }, DatadogSeriesMetric { @@ -1022,6 +1056,7 @@ async fn decode_series_endpoint_v1() { host: None, source_type_name: None, device: None, + unit: None, metadata: None, }, ], @@ -1190,6 +1225,7 @@ async fn decode_sketches() { }], metadata: Some(ddmetric_proto::Metadata { origin: Some(ddmetric_proto::Origin { + metric_type: 0, origin_product: 10, origin_category: 11, origin_service: 9, @@ -1708,6 +1744,7 @@ async fn split_outputs() { host: Some("random_host".to_string()), source_type_name: None, device: None, + unit: None, metadata: None, }], }; @@ -2404,6 +2441,7 @@ async fn decode_series_endpoint_v2() { interval: 0, metadata: Some(ddmetric_proto::Metadata { origin: Some(ddmetric_proto::Origin { + metric_type: 9, origin_product: 10, origin_category: 10, origin_service: 42, @@ -2583,11 +2621,577 @@ async fn decode_series_endpoint_v2() { .unwrap(), 42 ); + assert_eq!( + events[3] + .metadata() + .datadog_origin_metadata() + .unwrap() + .metric_type(), + Some(9) + ); } }) .await; } +#[test] +fn decode_series_v2_rejects_excessive_unit_event_expansion() { + let point = ddmetric_proto::metric_payload::MetricPoint { + value: 1.0, + timestamp: 1_542_182_950, + }; + let payload = ddmetric_proto::MetricPayload { + series: vec![ddmetric_proto::metric_payload::MetricSeries { + metric: "unit.expansion".to_string(), + points: vec![point; 16_385], + r#type: ddmetric_proto::metric_payload::MetricType::Gauge as i32, + unit: "u".repeat(1024), + ..Default::default() + }], + }; + let mut body = Vec::new(); + payload.encode(&mut body).unwrap(); + + assert!(super::metrics::decode_ddseries_v2(Bytes::from(body), &None, false).is_err()); +} + +#[tokio::test] +async fn decode_series_v1_rejects_excessive_unit_event_expansion() { + let (rx, _, _, addr, _guard) = source(EventStatus::Delivered, true, true, false, true).await; + let request = DatadogSeriesRequest { + series: vec![DatadogSeriesMetric { + metric: "unit.expansion".to_string(), + r#type: DatadogMetricType::Gauge, + interval: None, + points: vec![DatadogPoint(1_542_182_950, 1.0); 16_385], + tags: None, + host: None, + source_type_name: None, + device: None, + unit: Some("u".repeat(1024)), + metadata: None, + }], + }; + + assert_eq!( + 422, + send_with_path( + addr, + serde_json::to_vec(&request).unwrap(), + dd_api_key_headers(), + DD_API_SERIES_V1_PATH, + ) + .await + ); + assert!(crate::test_util::collect_ready(rx).is_empty()); +} + +#[tokio::test] +async fn decode_series_v1_skips_unit_materialization_for_zero_point_series() { + let (rx, _, _, addr, _guard) = source(EventStatus::Delivered, true, true, false, true).await; + let request = DatadogSeriesRequest { + series: vec![DatadogSeriesMetric { + metric: "empty".to_string(), + r#type: DatadogMetricType::Gauge, + interval: None, + points: Vec::new(), + tags: None, + host: None, + source_type_name: None, + device: None, + unit: Some("u".repeat(1024 * 1024)), + metadata: None, + }], + }; + + assert_eq!( + 200, + send_with_path( + addr, + serde_json::to_vec(&request).unwrap(), + dd_api_key_headers(), + DD_API_SERIES_V1_PATH, + ) + .await + ); + assert!(crate::test_util::collect_ready(rx).is_empty()); +} + +fn v3_string_dictionary(values: &[&str]) -> Vec { + values + .iter() + .flat_map(|value| { + let mut encoded = Vec::new(); + encode_v3_varint(value.len() as u64, &mut encoded); + encoded.extend_from_slice(value.as_bytes()); + encoded + }) + .collect() +} + +fn encode_v3_varint(mut value: u64, encoded: &mut Vec) { + while value >= 0x80 { + encoded.push((value as u8 & 0x7f) | 0x80); + value >>= 7; + } + encoded.push(value as u8); +} + +fn encode_v3_length_delimited(field: u8, value: &[u8], encoded: &mut Vec) { + encode_v3_varint((u64::from(field) << 3) | 2, encoded); + encode_v3_varint(value.len() as u64, encoded); + encoded.extend_from_slice(value); +} + +fn minimal_v3_metric_data() -> ddmetric_v3_proto::MetricData { + ddmetric_v3_proto::MetricData { + dict_name_str: v3_string_dictionary(&["metric"]), + dict_tag_str: Vec::new(), + dict_tagsets: Vec::new(), + dict_resource_str: Vec::new(), + dict_resource_len: Vec::new(), + dict_resource_type: Vec::new(), + dict_resource_name: Vec::new(), + dict_source_type_name: Vec::new(), + dict_origin_info: Vec::new(), + dict_unit_str: Vec::new(), + types: vec![3 | 0x30], + name_refs: vec![1], + tagset_refs: vec![0], + resources_refs: vec![0], + intervals: vec![0], + num_points: vec![1], + source_type_name_refs: vec![0], + origin_info_refs: vec![0], + unit_refs: Vec::new(), + timestamps: vec![1], + vals_sint64: Vec::new(), + vals_float32: Vec::new(), + vals_float64: vec![1.0], + sketch_num_bins: Vec::new(), + sketch_bin_keys: Vec::new(), + sketch_bin_cnts: Vec::new(), + } +} + +#[test] +fn decode_v3_rejects_point_count_larger_than_columns() { + let mut data = minimal_v3_metric_data(); + data.num_points[0] = u64::MAX; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_negative_origin_values() { + let mut data = minimal_v3_metric_data(); + data.dict_origin_info = vec![-1, 0, 0]; + data.origin_info_refs[0] = 1; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_timestamp_outside_chrono_range() { + let mut data = minimal_v3_metric_data(); + data.timestamps[0] = i64::MAX; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_interval_millisecond_overflow() { + let mut data = minimal_v3_metric_data(); + data.intervals[0] = u64::from(u32::MAX / 1000 + 1); + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_tagset_expansion() { + let mut data = minimal_v3_metric_data(); + data.dict_tag_str = v3_string_dictionary(&[&"x".repeat(1024 * 1024)]); + data.dict_tagsets = vec![1, 1]; + for previous_tagset in 1..=5 { + data.dict_tagsets + .extend([2, -i64::from(previous_tagset), 0]); + } + data.tagset_refs[0] = 6; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_series_tag_cloning() { + let mut data = minimal_v3_metric_data(); + data.dict_tag_str = v3_string_dictionary(&[&"x".repeat(1024 * 1024)]); + data.dict_tagsets = vec![1, 1]; + data.types = vec![3 | 0x30; 17]; + data.name_refs = std::iter::once(1) + .chain(std::iter::repeat_n(0, 16)) + .collect(); + data.tagset_refs = std::iter::once(1) + .chain(std::iter::repeat_n(0, 16)) + .collect(); + data.resources_refs = vec![0; 17]; + data.intervals = vec![0; 17]; + data.num_points = vec![0; 17]; + data.source_type_name_refs = vec![0; 17]; + data.origin_info_refs = vec![0; 17]; + data.timestamps.clear(); + data.vals_float64.clear(); + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_point_tag_cloning() { + let mut data = minimal_v3_metric_data(); + data.dict_tag_str = v3_string_dictionary(&[&"x".repeat(1024 * 1024)]); + data.dict_tagsets = vec![1, 1]; + data.tagset_refs[0] = 1; + data.num_points[0] = 17; + data.timestamps = vec![1; 17]; + data.vals_float64 = vec![1.0; 17]; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_accounts_for_metric_tag_entry_storage() { + let data = minimal_v3_metric_data(); + let metadata = ddmetric_v3_proto::Metadata { + tags: (0..200_000).map(|index| format!("t{index}")).collect(), + resources: Vec::new(), + }; + + assert!(super::metrics::decode_v3_metric_data(&data, Some(&metadata)).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_event_expansion() { + let mut data = minimal_v3_metric_data(); + data.num_points[0] = 1_000_000; + data.timestamps = vec![0; 1_000_000]; + data.vals_float64 = vec![0.0; 1_000_000]; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_source_type_event_expansion() { + let mut data = minimal_v3_metric_data(); + data.dict_source_type_name = v3_string_dictionary(&[&"x".repeat(1024 * 1024)]); + data.source_type_name_refs[0] = 1; + data.num_points[0] = 17; + data.timestamps = vec![1; 17]; + data.vals_float64 = vec![1.0; 17]; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_dictionary_string_cloning() { + let mut data = minimal_v3_metric_data(); + data.dict_name_str = v3_string_dictionary(&[&"x".repeat(1024 * 1024)]); + data.types = vec![3 | 0x30; 17]; + data.name_refs = std::iter::once(1) + .chain(std::iter::repeat_n(0, 16)) + .collect(); + data.tagset_refs = vec![0; 17]; + data.resources_refs = vec![0; 17]; + data.intervals = vec![0; 17]; + data.num_points = vec![0; 17]; + data.source_type_name_refs = vec![0; 17]; + data.origin_info_refs = vec![0; 17]; + data.timestamps.clear(); + data.vals_float64.clear(); + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_empty_string_entries() { + let mut data = minimal_v3_metric_data(); + data.dict_name_str = vec![0; 700_000]; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_accumulates_string_dictionary_expansion_across_fields() { + let mut data = minimal_v3_metric_data(); + data.dict_name_str = vec![0; 300_000]; + data.dict_tag_str = vec![0; 300_000]; + data.dict_unit_str = vec![0; 300_000]; + data.dict_resource_str = vec![0; 300_000]; + data.dict_source_type_name = vec![0; 300_000]; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_empty_tagsets() { + let mut data = minimal_v3_metric_data(); + data.dict_tagsets = vec![0; 700_000]; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_resource_expansion() { + let mut data = minimal_v3_metric_data(); + data.dict_resource_str = v3_string_dictionary(&[&"x".repeat(1024 * 1024)]); + data.dict_resource_len = vec![1; 17]; + data.dict_resource_type = vec![1; 17]; + data.dict_resource_name = vec![0; 17]; + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_deduplicates_metadata_tags() { + let data = minimal_v3_metric_data(); + let metadata = ddmetric_v3_proto::Metadata { + tags: vec!["team:core".to_string(); 10_000], + resources: Vec::new(), + }; + + let series = super::metrics::decode_v3_metric_data(&data, Some(&metadata)).unwrap(); + assert_eq!(series[0].tags, ["team:core"]); +} + +#[test] +fn decode_v3_accepts_many_distinct_metadata_tags() { + let data = minimal_v3_metric_data(); + let metadata = ddmetric_v3_proto::Metadata { + tags: (0..100_000).map(|index| format!("tag:{index}")).collect(), + resources: Vec::new(), + }; + + let series = super::metrics::decode_v3_metric_data(&data, Some(&metadata)).unwrap(); + assert_eq!(series[0].tags.len(), 100_000); +} + +#[test] +fn decode_v3_skips_tag_expansion_for_zero_point_series() { + let mut data = minimal_v3_metric_data(); + data.num_points[0] = 0; + data.timestamps.clear(); + data.vals_float64.clear(); + let payload = ddmetric_v3_proto::Payload { + metadata: Some(ddmetric_v3_proto::Metadata { + tags: (0..100_000).map(|index| format!("tag:{index}")).collect(), + resources: Vec::new(), + }), + metric_data: Some(data), + }; + let mut body = Vec::new(); + payload.encode(&mut body).unwrap(); + + let events = super::metrics::decode_ddseries_v3(Bytes::from(body), &None, false).unwrap(); + + assert!(events.is_empty()); +} + +#[test] +fn decode_v3_rejects_excessive_metadata_string_entries_before_decode() { + let metadata_entry_count = 16 * 1024 * 1024 / std::mem::size_of::() + 1; + let mut metadata = Vec::with_capacity(metadata_entry_count * 2); + for _ in 0..metadata_entry_count { + metadata.extend([0x0a, 0x00]); + } + let mut payload = Vec::new(); + encode_v3_length_delimited(2, &metadata, &mut payload); + + assert!(super::metrics::decode_ddseries_v3(Bytes::from(payload), &None, false).is_err()); +} + +#[test] +fn decode_v3_accumulates_metadata_allocations_across_fields() { + let metadata_entry_count = 16 * 1024 * 1024 / std::mem::size_of::() / 2 + 1; + let mut metadata = Vec::with_capacity(metadata_entry_count * 2); + for _ in 0..metadata_entry_count { + metadata.extend([0x0a, 0x00]); + } + let mut payload = Vec::new(); + encode_v3_length_delimited(2, &metadata, &mut payload); + encode_v3_length_delimited(2, &metadata, &mut payload); + + assert!(super::metrics::decode_ddseries_v3(Bytes::from(payload), &None, false).is_err()); +} + +#[test] +fn decode_v3_rejects_excessive_numeric_columns_before_decode() { + let entry_count = 16 * 1024 * 1024 / std::mem::size_of::() + 1; + let packed_timestamps = vec![0; entry_count]; + let mut metric_data = Vec::new(); + encode_v3_length_delimited(16, &packed_timestamps, &mut metric_data); + let mut payload = Vec::new(); + encode_v3_length_delimited(3, &metric_data, &mut payload); + + assert!(super::metrics::decode_ddseries_v3(Bytes::from(payload), &None, false).is_err()); +} + +#[test] +fn decode_v3_rejects_lossy_tag_expansion_before_allocation() { + let mut data = minimal_v3_metric_data(); + let invalid_utf8 = vec![0xff; 6 * 1024 * 1024]; + encode_v3_varint(invalid_utf8.len() as u64, &mut data.dict_tag_str); + data.dict_tag_str.extend(invalid_utf8); + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_valid_string_expansion_before_allocation() { + let mut data = minimal_v3_metric_data(); + let valid_utf8 = vec![b'x'; 16 * 1024 * 1024]; + encode_v3_varint(valid_utf8.len() as u64, &mut data.dict_tag_str); + data.dict_tag_str.extend(valid_utf8); + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_rejects_reference_accumulator_overflow() { + let mut data = minimal_v3_metric_data(); + data.types = vec![3 | 0x30; 2]; + data.name_refs = vec![1, i64::MAX]; + data.tagset_refs = vec![0; 2]; + data.resources_refs = vec![0; 2]; + data.intervals = vec![0; 2]; + data.num_points = vec![0; 2]; + data.source_type_name_refs = vec![0; 2]; + data.origin_info_refs = vec![0; 2]; + data.timestamps.clear(); + data.vals_float64.clear(); + + assert!(super::metrics::decode_v3_metric_data(&data, None).is_err()); +} + +#[test] +fn decode_v3_does_not_expand_unused_global_resources() { + let mut data = minimal_v3_metric_data(); + data.dict_resource_len = vec![0; 100_000]; + let metadata = ddmetric_v3_proto::Metadata { + tags: Vec::new(), + resources: vec!["container".to_string(), "x".repeat(1024)], + }; + + let series = super::metrics::decode_v3_metric_data(&data, Some(&metadata)).unwrap(); + assert_eq!(series[0].resources.len(), 1); +} + +#[tokio::test] +async fn decode_series_endpoint_v3() { + assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async { + let (rx, _, _, addr, _guard) = + source(EventStatus::Delivered, true, true, false, true).await; + + let payload = ddmetric_v3_proto::Payload { + metadata: Some(ddmetric_v3_proto::Metadata { + tags: vec!["team:core".to_string()], + resources: Vec::new(), + }), + metric_data: Some(ddmetric_v3_proto::MetricData { + dict_name_str: v3_string_dictionary(&["namespace.dd_gauge"]), + dict_tag_str: v3_string_dictionary(&["foo:bar"]), + dict_tagsets: vec![1, 1], + dict_resource_str: v3_string_dictionary(&[ + "host", + "random_host", + "container", + "container_id", + ]), + dict_resource_len: vec![2, 1], + dict_resource_type: vec![1, 2, 3], + dict_resource_name: vec![2, 2, 4], + dict_source_type_name: v3_string_dictionary(&["system"]), + dict_origin_info: vec![10, 10, 42], + dict_unit_str: v3_string_dictionary(&["byte"]), + types: vec![3 | 0x30 | 0x100 | 0x200, 3 | 0x30], + name_refs: vec![1, 0], + tagset_refs: vec![1, 0], + resources_refs: vec![1, 1], + intervals: vec![10, 10], + num_points: vec![1, 1], + source_type_name_refs: vec![1, 0], + origin_info_refs: vec![1, 0], + unit_refs: vec![1], + timestamps: vec![1542182950, 1], + vals_sint64: Vec::new(), + vals_float32: Vec::new(), + vals_float64: vec![3.14, 3.1415], + sketch_num_bins: Vec::new(), + sketch_bin_keys: Vec::new(), + sketch_bin_cnts: Vec::new(), + }), + }; + let mut body = Vec::new(); + payload.encode(&mut body).unwrap(); + let events = send_bytes_and_collect( + addr, + body, + dd_api_key_headers(), + DD_API_SERIES_V3_PATH, + rx, + 2, + ) + .await; + + let metric = events[0].as_metric(); + assert_eq!(metric.name(), "dd_gauge"); + assert_eq!(metric.namespace(), Some("namespace")); + assert_eq!(metric.kind(), MetricKind::Absolute); + assert_eq!(*metric.value(), MetricValue::Gauge { value: 3.14 }); + assert_tags( + metric, + metric_tags!( + "foo" => "bar", + "team" => "core", + "host" => "random_host", + "resource.container" => "container_id", + "source_type_name" => "system" + ), + ); + assert_eq!( + metric + .metadata() + .datadog_origin_metadata() + .unwrap() + .product(), + Some(10) + ); + assert_eq!( + metric + .metadata() + .datadog_origin_metadata() + .unwrap() + .category(), + Some(10) + ); + assert_eq!( + metric + .metadata() + .datadog_origin_metadata() + .unwrap() + .service(), + Some(42) + ); + assert_eq!( + metric + .metadata() + .datadog_origin_metadata() + .unwrap() + .metric_type(), + Some(9) + ); + assert_eq!(metric.metadata().datadog_metric_unit(), Some("byte")); + }) + .await; +} + #[test] fn test_output_schema_definition_json_vector_namespace() { let definition = serde_yaml::from_str::(indoc! { r#" @@ -2816,6 +3420,7 @@ async fn test_series_v1_split_metric_namespace_impl( host: Some("test_host".to_string()), source_type_name: None, device: None, + unit: None, metadata: None, }], }; diff --git a/src/transforms/aggregate/tests.rs b/src/transforms/aggregate/tests.rs index a4de7f841784e..38e51b3cca2cd 100644 --- a/src/transforms/aggregate/tests.rs +++ b/src/transforms/aggregate/tests.rs @@ -15,7 +15,8 @@ use super::{ use crate::{ config::{OutputId, TransformConfig, TransformContext}, event::{ - BatchNotifier, BatchStatus, BatchStatusReceiver, Event, EventStatus, Metric, + BatchNotifier, BatchStatus, BatchStatusReceiver, DatadogMetricOriginMetadata, Event, + EventMetadata, EventStatus, Metric, metric::{MetricKind, MetricValue}, }, schema::Definition, @@ -1534,6 +1535,37 @@ fn event_time_metadata_is_merged_and_diff_finalizers_are_released() { assert_gauge(&out[0], 2.0); } + // The selected sample's absent Datadog attribution must not be filled from a discarded one. + { + let mut agg = + Aggregate::new(&event_time_config(interval_ms, AggregationMode::Auto)).unwrap(); + let attributed_metadata = EventMetadata::default() + .with_origin_metadata(DatadogMetricOriginMetadata::new(Some(1), Some(2), Some(3))) + .with_datadog_metric_unit("byte"); + let attributed = Event::Metric( + Metric::new_with_metadata( + "latest_attribution", + MetricKind::Absolute, + MetricValue::Gauge { value: 1.0 }, + attributed_metadata, + ) + .with_timestamp(Some(later)), + ); + let selected = make_metric_with_timestamp( + "latest_attribution", + MetricKind::Absolute, + MetricValue::Gauge { value: 2.0 }, + latest_ts, + ); + + agg.record(attributed); + agg.record(selected); + let out = flush_final(&mut agg); + assert!(out[0].metadata().datadog_origin_metadata().is_none()); + assert_eq!(out[0].metadata().datadog_metric_unit(), None); + assert_gauge(&out[0], 2.0); + } + // Auto kind switch: retained sample's data and source_type must stay paired. // Inc→Abs: Absolute wins even if older. Abs→Inc: newer Incremental replaces. for (first_kind, first_value, first_src, second_kind, second_value, second_src, expect_kind) in [ diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs index 575129a82c911..1ea32167b1ba3 100644 --- a/src/transforms/log_to_metric.rs +++ b/src/transforms/log_to_metric.rs @@ -380,7 +380,7 @@ fn to_metric_with_config(config: &MetricConfig, event: &Event) -> Result Result=7.81.0, so - # it's dropped in favor of the explicit 7.80.4 pin. Re-add an unpinned v7 - # leg once v3 series support lands - # (see https://github.com/vectordotdev/vector/issues/25790). - agent_version: ["7.80.4", "6"] + # Keep the last Agent version before v3 series became the default, and also + # test the floating v7 tag to cover v3 series intake. + agent_version: ["7.80.4", "7", "6"] + # Validate every series API version still supported by the Datadog metrics sink. series_api_version: ["v2", "v3"] # changes to these files/paths will invoke the integration test in CI diff --git a/tests/e2e/datadog/metrics/series.rs b/tests/e2e/datadog/metrics/series.rs index f6838207bc6d2..99b507794790f 100644 --- a/tests/e2e/datadog/metrics/series.rs +++ b/tests/e2e/datadog/metrics/series.rs @@ -361,9 +361,15 @@ fn compare_intakes(agent_intake: &SeriesIntake, vector_intake: &SeriesIntake) { pub(super) async fn validate() { let api_version = std::env::var("CONFIG_SERIES_API_VERSION") .expect("CONFIG_SERIES_API_VERSION must be set ('v2' or 'v3')"); + let agent_version = std::env::var("CONFIG_AGENT_VERSION") + .expect("CONFIG_AGENT_VERSION must be set ('7.80.4', '7', or '6')"); info!("==== getting series data from agent-only pipeline ==== "); - let agent_intake = get_v2_series_from_pipeline(fake_intake_agent_address()).await; + let agent_intake = match agent_version.as_str() { + "7" => get_v3_series_from_pipeline(fake_intake_agent_address()).await, + "7.80.4" | "6" => get_v2_series_from_pipeline(fake_intake_agent_address()).await, + version => panic!("Unknown CONFIG_AGENT_VERSION: {version}"), + }; info!("==== getting series data from agent-vector pipeline ({api_version}) ===="); let vector_intake = match api_version.as_str() {