feat(datadog_agent): support v3 series metrics intake

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

lib/datadog-proto/build.rs

@@ -9,6 +9,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",

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;

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 {

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

@@ -116,6 +116,10 @@ pub(super) struct Inner {
     #[serde(default)]
     pub(crate) datadog_origin_metadata: Option<DatadogMetricOriginMetadata>,
 
+    /// Datadog metric unit received from an upstream Agent payload.
+    #[serde(default)]
+    pub(crate) datadog_metric_unit: Option<Arc<str>>,
+
     /// Which legacy trace key layout produced this event. `None` means unmarked.
     /// Meaningful only on traces; always `None` for logs and metrics.
     #[serde(default)]
@@ -129,6 +133,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<i32>,
     /// `OriginProduct`
     product: Option<u32>,
     /// `OriginCategory`
@@ -146,12 +152,25 @@ impl DatadogMetricOriginMetadata {
     #[must_use]
     pub fn new(product: Option<u32>, category: Option<u32>, service: Option<u32>) -> Self {
         Self {
+            metric_type: None,
             product,
             category,
             service,
         }
     }
 
+    /// Returns the Datadog origin metric type.
+    pub fn metric_type(&self) -> Option<i32> {
+        self.metric_type
+    }
+
+    /// Sets the Datadog origin metric type.
+    #[must_use]
+    pub fn with_metric_type(mut self, metric_type: Option<i32>) -> Self {
+        self.metric_type = metric_type;
+        self
+    }
+
     /// Returns a reference to the `OriginProduct`.
     pub fn product(&self) -> Option<u32> {
         self.product
@@ -304,6 +323,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<Arc<str>>) {
+        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<Uuid> {
         self.inner.source_event_id
@@ -333,6 +367,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()),
         }
@@ -357,10 +392,18 @@ pub(super) fn default_schema_definition() -> Arc<schema::Definition> {
 
 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
     }
 }
 
@@ -415,6 +458,13 @@ impl EventMetadata {
         self
     }
 
+    /// Sets the Datadog metric unit.
+    #[must_use]
+    pub fn with_datadog_metric_unit(mut self, unit: impl Into<Arc<str>>) -> 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<Uuid>) -> Self {
@@ -661,6 +711,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();

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

@@ -597,6 +597,7 @@ impl From<Secrets> for super::metadata::Secrets {
 impl From<super::DatadogMetricOriginMetadata> 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<DatadogOriginMetadata> for super::DatadogMetricOriginMetadata {
             value.origin_category,
             value.origin_service,
         )
+        .with_metric_type(value.metric_type)
     }
 }
 
@@ -638,6 +640,7 @@ impl From<EventMetadata> for Metadata {
             source_type,
             upstream_id,
             datadog_origin_metadata,
+            datadog_metric_unit,
             source_event_id,
             trace_layout,
             ..
@@ -653,6 +656,7 @@ impl From<EventMetadata> 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<Metadata> for EventMetadata {
             upstream_id,
             secrets,
             datadog_origin_metadata,
+            datadog_metric_unit,
             source_event_id,
             trace_layout,
         } = value;
@@ -709,6 +714,7 @@ impl TryFrom<Metadata> 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,
             }),

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::<EventArray, _>(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");

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<String>,
+    /// unit
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub unit: Option<String>,
     /// metadata
     #[serde(skip_serializing_if = "Option::is_none")]
     pub metadata: Option<DatadogSeriesMetricMetadata>,

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");
         }
     }