feat(datadog_agent): support v3 series metrics intake

build.rs

@@ -134,6 +134,7 @@ fn main() {
         println!("cargo:rerun-if-changed=proto/third-party/google/pubsub/v1/pubsub.proto");
         println!("cargo:rerun-if-changed=proto/third-party/google/rpc/status.proto");
         println!("cargo:rerun-if-changed=proto/vector/dd_metric.proto");
+        println!("cargo:rerun-if-changed=proto/vector/dd_metric_v3.proto");
         println!("cargo:rerun-if-changed=proto/vector/dd_trace.proto");
         println!("cargo:rerun-if-changed=proto/vector/ddsketch_full.proto");
         println!("cargo:rerun-if-changed=proto/vector/vector.proto");
@@ -162,6 +163,7 @@ fn main() {
                     "lib/vector-core/proto/event.proto",
                     "proto/vector/ddsketch_full.proto",
                     "proto/vector/dd_metric.proto",
+                    "proto/vector/dd_metric_v3.proto",
                     "proto/vector/dd_trace.proto",
                     "proto/third-party/google/pubsub/v1/pubsub.proto",
                     "proto/third-party/google/rpc/status.proto",

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/vector-core/proto/event.proto

@@ -75,6 +75,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 {
@@ -94,6 +95,7 @@ message Metadata {
   OutputId upstream_id = 5;
   Secrets secrets = 6;
   bytes source_event_id = 7;
+  optional string datadog_metric_unit = 8;
 }
 
 message Metric {

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

@@ -84,6 +84,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<String>,
+
     /// An internal vector id that can be used to identify this event across all components.
     #[derivative(PartialEq = "ignore")]
     pub(crate) source_event_id: Option<Uuid>,
@@ -92,6 +96,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`
@@ -109,12 +115,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
@@ -247,6 +266,16 @@ 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: String) {
+        self.get_mut().datadog_metric_unit = Some(unit);
+    }
+
     /// Returns a reference to the event id.
     pub fn source_event_id(&self) -> Option<Uuid> {
         self.inner.source_event_id
@@ -276,6 +305,7 @@ impl Default for Inner {
             upstream_id: None,
             dropped_fields: ObjectMap::new(),
             datadog_origin_metadata: None,
+            datadog_metric_unit: None,
             source_event_id: Some(Uuid::new_v4()),
         }
     }
@@ -357,6 +387,13 @@ impl EventMetadata {
         self
     }
 
+    /// Sets the Datadog metric unit.
+    #[must_use]
+    pub fn with_datadog_metric_unit(mut self, unit: String) -> Self {
+        self.get_mut().datadog_metric_unit = Some(unit);
+        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 {
@@ -373,6 +410,12 @@ impl EventMetadata {
         let other = other.into_owned();
         inner.finalizers.merge(other.finalizers);
         inner.secrets.merge(other.secrets);
+        if inner.datadog_origin_metadata.is_none() {
+            inner.datadog_origin_metadata = other.datadog_origin_metadata;
+        }
+        if inner.datadog_metric_unit.is_none() {
+            inner.datadog_metric_unit = other.datadog_metric_unit;
+        }
 
         // Update `source_event_id` if necessary.
         if inner.source_event_id.is_none() {

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

@@ -590,6 +590,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(),
@@ -604,6 +605,7 @@ impl From<DatadogOriginMetadata> for super::DatadogMetricOriginMetadata {
             value.origin_category,
             value.origin_service,
         )
+        .with_metric_type(value.metric_type)
     }
 }
 
@@ -631,6 +633,7 @@ impl From<EventMetadata> for Metadata {
             source_type,
             upstream_id,
             datadog_origin_metadata,
+            datadog_metric_unit,
             source_event_id,
             ..
         } = value.into_owned();
@@ -645,6 +648,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,
         }
     }
 }
@@ -658,6 +662,7 @@ impl From<Metadata> for EventMetadata {
             upstream_id,
             secrets,
             datadog_origin_metadata,
+            datadog_metric_unit,
             source_event_id,
         } = value;
 
@@ -694,6 +699,7 @@ impl From<Metadata> for EventMetadata {
                 schema_definition: default_schema_definition(),
                 dropped_fields: ObjectMap::new(),
                 datadog_origin_metadata,
+                datadog_metric_unit,
                 source_event_id,
             }),
             last_transform_timestamp: None,

proto/vector/dd_metric.proto

@@ -14,7 +14,8 @@ message CommonMetadata {
 }
 
 message Origin {
-    reserved 1,2,3;
+    reserved 1,2;
+    int32 metric_type = 3;
     uint32 origin_product = 4;
     uint32 origin_category = 5;
     uint32 origin_service = 6;

proto/vector/dd_metric_v3.proto

@@ -0,0 +1,74 @@
+// Extracted from https://github.com/DataDog/agent-payload/blob/master/proto/metrics/intake_v3.proto
+
+syntax = "proto3";
+
+package datadoghq.api.metrics.v3;
+
+message Payload {
+  reserved 1;
+  Metadata metadata = 2;
+  MetricData metricData = 3;
+}
+
+message Metadata {
+  repeated string tags = 1;
+  repeated string resources = 2;
+}
+
+message MetricData {
+  bytes dictNameStr = 1;
+  bytes dictTagStr = 2;
+  repeated sint64 dictTagsets = 3;
+
+  bytes dictResourceStr = 4;
+  repeated int64 dictResourceLen = 5;
+  repeated sint64 dictResourceType = 6;
+  repeated sint64 dictResourceName = 7;
+
+  bytes dictSourceTypeName = 8;
+  repeated int32 dictOriginInfo = 9;
+  bytes dictUnitStr = 25;
+
+  repeated uint64 types = 10;
+  repeated sint64 nameRefs = 11;
+  repeated sint64 tagsetRefs = 12;
+  repeated sint64 resourcesRefs = 13;
+  repeated uint64 intervals = 14;
+  repeated uint64 numPoints = 15;
+  repeated sint64 sourceTypeNameRefs = 23;
+  repeated sint64 originInfoRefs = 24;
+  repeated sint64 unitRefs = 26;
+
+  repeated sint64 timestamps = 16;
+  repeated sint64 valsSint64 = 17;
+  repeated float valsFloat32 = 18;
+  repeated double valsFloat64 = 19;
+  repeated uint64 sketchNumBins = 20;
+  repeated sint32 sketchBinKeys = 21;
+  repeated uint32 sketchBinCnts = 22;
+}
+
+enum metricType {
+  UNUSED = 0;
+  Count = 1;
+  Rate = 2;
+  Gauge = 3;
+  Sketch = 4;
+}
+
+enum valueType {
+  Zero = 0x00;
+  Sint64 = 0x10;
+  Float32 = 0x20;
+  Float64 = 0x30;
+}
+
+enum metricFlags {
+  flagNone = 0;
+  flagNoIndex = 0x100;
+  flagHasUnit = 0x200;
+}
+
+message Response {
+  string error = 1;
+}

src/common/datadog.rs

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

@@ -491,6 +491,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(),
@@ -718,8 +719,10 @@ fn series_to_proto_message(
         tags,
         points,
         r#type: metric_type.into(),
-        // unit is omitted
-        unit: "".to_string(),
+        unit: event_metadata
+            .datadog_metric_unit()
+            .unwrap_or_default()
+            .to_string(),
         source_type_name,
         interval: maybe_interval.unwrap_or(0) as i64,
         metadata,
@@ -829,11 +832,14 @@ 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 {
@@ -929,6 +935,7 @@ fn generate_series_metrics(
         host,
         source_type_name,
         device,
+        unit: event_metadata.datadog_metric_unit().map(ToOwned::to_owned),
         metadata,
     }])
 }
@@ -1529,9 +1536,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 v1
@@ -1553,6 +1563,8 @@ mod tests {
             assert_eq!(generated_origin.product().unwrap(), product);
             assert_eq!(generated_origin.category().unwrap(), category);
             assert_eq!(generated_origin.service().unwrap(), service);
+            assert_eq!(generated_origin.metric_type(), Some(9));
+            assert_eq!(actual.unit.as_deref(), Some("byte"));
         }
         // series v2
         {
@@ -1568,6 +1580,8 @@ mod tests {
             assert_eq!(generated_origin.origin_product, product);
             assert_eq!(generated_origin.origin_category, category);
             assert_eq!(generated_origin.origin_service, service);
+            assert_eq!(generated_origin.metric_type, 9);
+            assert_eq!(series_proto.unit, "byte");
         }
     }