feat(metrics): support Datadog metric origin metadata

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

@@ -163,6 +163,15 @@ impl EventArray {
         }
     }
 
+    /// Sets the `source_type` in the metadata for all metric events in this array.
+    pub fn set_source_type(&mut self, source_type: &'static str) {
+        if let EventArray::Metrics(metrics) = self {
+            for metric in metrics {
+                metric.metadata_mut().set_source_type(source_type);
+            }
+        }
+    }
+
     /// Iterate over references to this array's events.
     pub fn iter_events(&self) -> impl Iterator<Item = EventRef> {
         match self {

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

@@ -33,6 +33,9 @@ pub struct EventMetadata {
     /// The id of the source
     source_id: Option<Arc<ComponentKey>>,
 
+    /// The type of the source
+    source_type: Option<&'static str>,
+
     /// The id of the component this event originated from. This is used to
     /// determine which schema definition to attach to an event in transforms.
     /// This should always have a value set for events in transforms. It will always be `None`
@@ -54,6 +57,60 @@ pub struct EventMetadata {
     /// This field could almost be keyed by `&'static str`, but because it needs to be deserializable
     /// we have to use `String`.
     dropped_fields: BTreeMap<String, Value>,
+
+    /// Metadata to track the origin of metrics. This is always `None` for log and trace events.
+    /// Only a small set of Vector sources and transforms explicitly set this field.
+    #[serde(default)]
+    datadog_origin_metadata: Option<DatadogMetricOriginMetadata>,
+}
+
+/// Metric Origin metadata for submission to Datadog.
+#[derive(Clone, Default, Debug, Deserialize, PartialEq, Serialize)]
+pub struct DatadogMetricOriginMetadata {
+    /// OriginProduct
+    product: Option<u32>,
+    /// OriginCategory
+    category: Option<u32>,
+    /// OriginService
+    service: Option<u32>,
+}
+
+impl DatadogMetricOriginMetadata {
+    /// Replaces the `OriginProduct`.
+    #[must_use]
+    pub fn with_product(mut self, product: u32) -> Self {
+        self.product = Some(product);
+        self
+    }
+
+    /// Replaces the `OriginCategory`.
+    #[must_use]
+    pub fn with_category(mut self, category: u32) -> Self {
+        self.category = Some(category);
+        self
+    }
+
+    /// Replaces the `OriginService`.
+    #[must_use]
+    pub fn with_service(mut self, service: u32) -> Self {
+        self.service = Some(service);
+        self
+    }
+
+    /// Returns a reference to the `OriginProduct`
+    pub fn product(&self) -> Option<u32> {
+        self.product
+    }
+
+    /// Returns a reference to the `OriginCategory`
+    pub fn category(&self) -> Option<u32> {
+        self.category
+    }
+
+    /// Returns a reference to the `OriginService`
+    pub fn service(&self) -> Option<u32> {
+        self.service
+    }
 }
 
 fn default_metadata_value() -> Value {
@@ -95,6 +152,12 @@ impl EventMetadata {
         self.source_id.as_ref()
     }
 
+    /// Returns a reference to the metadata source type.
+    #[must_use]
+    pub fn source_type(&self) -> Option<&'static str> {
+        self.source_type
+    }
+
     /// Returns a reference to the metadata parent id. This is the `OutputId`
     /// of the previous component the event was sent through (if any).
     #[must_use]
@@ -107,6 +170,11 @@ impl EventMetadata {
         self.source_id = Some(source_id);
     }
 
+    /// Sets the `source_type` in the metadata to the provided value.
+    pub fn set_source_type(&mut self, source_type: &'static str) {
+        self.source_type = Some(source_type);
+    }
+
     /// Sets the `upstream_id` in the metadata to the provided value.
     pub fn set_upstream_id(&mut self, upstream_id: Arc<OutputId>) {
         self.upstream_id = Some(upstream_id);
@@ -144,6 +212,11 @@ impl EventMetadata {
     pub fn dropped_field(&self, meaning: impl AsRef<str>) -> Option<&Value> {
         self.dropped_fields.get(meaning.as_ref())
     }
+
+    /// Returns a reference to the `DatadogMetricOriginMetadata`.
+    pub fn datadog_origin_metadata(&self) -> Option<&DatadogMetricOriginMetadata> {
+        self.datadog_origin_metadata.as_ref()
+    }
 }
 
 impl Default for EventMetadata {
@@ -154,8 +227,10 @@ impl Default for EventMetadata {
             finalizers: Default::default(),
             schema_definition: default_schema_definition(),
             source_id: None,
+            source_type: None,
             upstream_id: None,
             dropped_fields: BTreeMap::new(),
+            datadog_origin_metadata: None,
         }
     }
 }
@@ -213,6 +288,20 @@ impl EventMetadata {
         self
     }
 
+    /// Replaces the existing `source_type` with the given one.
+    #[must_use]
+    pub fn with_source_type(mut self, source_type: &'static str) -> Self {
+        self.source_type = Some(source_type);
+        self
+    }
+
+    /// Replaces the existing `DatadogMetricOriginMetadata` with the given one.
+    #[must_use]
+    pub fn with_origin_metadata(mut self, origin_metadata: DatadogMetricOriginMetadata) -> Self {
+        self.datadog_origin_metadata = Some(origin_metadata);
+        self
+    }
+
     /// Merge the other `EventMetadata` into this.
     /// If a Datadog API key is not set in `self`, the one from `other` will be used.
     /// If a Splunk HEC token is not set in `self`, the one from `other` will be used.

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

@@ -9,7 +9,7 @@ pub use finalization::{
     Finalizable,
 };
 pub use log_event::LogEvent;
-pub use metadata::{EventMetadata, WithMetadata};
+pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, WithMetadata};
 pub use metric::{Metric, MetricKind, MetricTags, MetricValue, StatisticKind};
 pub use r#ref::{EventMutRef, EventRef};
 use serde::{Deserialize, Serialize};
@@ -307,6 +307,11 @@ impl Event {
         self.metadata_mut().set_upstream_id(upstream_id);
     }
 
+    /// Sets the `source_type` in the event metadata to the provided value.
+    pub fn set_source_type(&mut self, source_type: &'static str) {
+        self.metadata_mut().set_source_type(source_type);
+    }
+
     /// Sets the `source_id` in the event metadata to the provided value.
     #[must_use]
     pub fn with_source_id(mut self, source_id: Arc<ComponentKey>) -> Self {

proto/dd_metric.proto

@@ -13,6 +13,17 @@ message CommonMetadata {
 	string api_key = 6;
 }
 
+message Origin {
+		reserved 1,2,3;
+    uint32 origin_product = 4;
+    uint32 origin_category = 5;
+    uint32 origin_service = 6;
+}
+
+message Metadata {
+    Origin origin = 1;
+}
+
 message MetricPayload {
     enum MetricType {
         UNSPECIFIED = 0;
@@ -40,7 +51,7 @@ message MetricPayload {
         string unit = 6;
         string source_type_name = 7;
         int64 interval = 8;
-        reserved 9;
+        Metadata metadata = 9;
     }
     repeated MetricSeries series = 1;
 }
@@ -76,6 +87,7 @@ message SketchPayload {
                 reserved 5, 6;
 				reserved "distributionsK", "distributionsC";
 		repeated Dogsketch dogsketches = 7;
+    Metadata metadata = 8;
 	}
 	repeated Sketch sketches = 1;
 	CommonMetadata metadata = 2;

src/common/datadog.rs

@@ -4,6 +4,7 @@
 #![allow(unreachable_pub)]
 use serde::{Deserialize, Serialize};
 use vector_config::configurable_component;
+use vector_core::event::DatadogMetricOriginMetadata;
 
 pub const DD_US_SITE: &str = "datadoghq.com";
 pub const DD_EU_SITE: &str = "datadoghq.eu";
@@ -21,6 +22,14 @@ pub(crate) struct DatadogSeriesMetric {
     pub(crate) source_type_name: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) device: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) metadata: Option<DatadogSeriesMetricMetadata>,
+}
+
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub(crate) struct DatadogSeriesMetricMetadata {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) origin: Option<DatadogMetricOriginMetadata>,
 }
 
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]