chore(datadog_metrics sink): incrementally encode sketches

Cargo.toml

@@ -200,8 +200,9 @@ serde_yaml = { version = "0.9.22", default-features = false }
 rmp-serde = { version = "1.1.1", default-features = false, optional = true }
 rmpv = { version = "1.0.0", default-features = false, features = ["with-serde"], optional = true }
 
-# Prost
+# Prost / Protocol Buffers
 prost = { version = "0.11", default-features = false, features = ["std"] }
+prost-reflect = { version = "0.11", default-features = false, optional = true }
 prost-types = { version = "0.11", default-features = false, optional = true }
 
 # GCP
@@ -673,7 +674,7 @@ sinks-databend = []
 sinks-datadog_archives = ["sinks-aws_s3", "sinks-azure_blob", "sinks-gcp"]
 sinks-datadog_events = []
 sinks-datadog_logs = []
-sinks-datadog_metrics = ["protobuf-build"]
+sinks-datadog_metrics = ["protobuf-build", "dep:prost-reflect"]
 sinks-datadog_traces = ["protobuf-build", "dep:rmpv", "dep:rmp-serde", "dep:serde_bytes"]
 sinks-elasticsearch = ["aws-core", "transforms-metric_to_log"]
 sinks-file = ["dep:async-compression"]

LICENSE-3rdparty.csv

@@ -394,6 +394,7 @@ proc-macro2,https://github.com/dtolnay/proc-macro2,MIT OR Apache-2.0,"David Toln
 proptest,https://github.com/proptest-rs/proptest,MIT OR Apache-2.0,Jason Lingle
 prost,https://github.com/tokio-rs/prost,Apache-2.0,"Dan Burkert <dan@danburkert.com>, Lucio Franco <luciofranco14@gmail.com, Tokio Contributors <team@tokio.rs>"
 prost-derive,https://github.com/tokio-rs/prost,Apache-2.0,"Dan Burkert <dan@danburkert.com>, Lucio Franco <luciofranco14@gmail.com>, Tokio Contributors <team@tokio.rs>"
+prost-reflect,https://github.com/andrewhickman/prost-reflect,MIT OR Apache-2.0,Andrew Hickman <andrew.hickman1@sky.com>
 ptr_meta,https://github.com/djkoloski/ptr_meta,MIT,David Koloski <djkoloski@gmail.com>
 pulsar,https://github.com/streamnative/pulsar-rs,MIT OR Apache-2.0,"Colin Stearns <cstearns@developers.wyyerd.com>, Kevin Stenerson <kstenerson@developers.wyyerd.com>, Geoffroy Couprie <contact@geoffroycouprie.com>"
 quad-rand,https://github.com/not-fl3/quad-rand,MIT,not-fl3 <not.fl3@gmail.com>

build.rs

@@ -1,4 +1,11 @@
-use std::{collections::HashSet, env, fs::File, io::Write, path::Path, process::Command};
+use std::{
+    collections::HashSet,
+    env,
+    fs::File,
+    io::Write,
+    path::{Path, PathBuf},
+    process::Command,
+};
 
 struct TrackedEnv {
     tracked: HashSet<String>,
@@ -124,8 +131,19 @@ fn main() {
         println!("cargo:rerun-if-changed=proto/google/rpc/status.proto");
         println!("cargo:rerun-if-changed=proto/vector.proto");
 
+        // Create and store the "file descriptor set" from the compiled Protocol Buffers packages.
+        // This allows us to use runtime reflection to manually build Protocol Buffers payloads
+        // in a type-safe way, which is necessary for incrementally building certain payloads, like
+        // the ones generated in the `datadog_metrics` sink.
+        let protobuf_fds_path = std::env::var("OUT_DIR")
+            .map(PathBuf::from)
+            .map(|path| path.join("protobuf-fds.bin"))
+            .expect("OUT_DIR environment variable not set");
+
         let mut prost_build = prost_build::Config::new();
-        prost_build.btree_map(["."]);
+        prost_build
+            .btree_map(["."])
+            .file_descriptor_set_path(protobuf_fds_path);
 
         tonic_build::configure()
             .protoc_arg("--experimental_allow_proto3_optional")

src/internal_events/datadog_metrics.rs

@@ -7,19 +7,17 @@ use vector_common::internal_event::{
 };
 
 #[derive(Debug)]
-pub struct DatadogMetricsEncodingError {
-    pub error_message: &'static str,
+pub struct DatadogMetricsEncodingError<'a> {
+    pub reason: &'a str,
     pub error_code: &'static str,
     pub dropped_events: usize,
 }
 
-impl InternalEvent for DatadogMetricsEncodingError {
+impl<'a> InternalEvent for DatadogMetricsEncodingError<'a> {
     fn emit(self) {
-        let reason = "Failed to encode Datadog metrics.";
         error!(
-            message = reason,
-            error = %self.error_message,
-            error_code = %self.error_code,
+            message = self.reason,
+            error_code = self.error_code,
             error_type = error_type::ENCODER_FAILED,
             intentional = "false",
             stage = error_stage::PROCESSING,
@@ -35,7 +33,7 @@ impl InternalEvent for DatadogMetricsEncodingError {
         if self.dropped_events > 0 {
             emit!(ComponentEventsDropped::<UNINTENTIONAL> {
                 count: self.dropped_events,
-                reason,
+                reason: self.reason,
             });
         }
     }

src/proto/mod.rs

@@ -0,0 +1,20 @@
+#[cfg(any(feature = "sources-vector", feature = "sinks-vector"))]
+use crate::event::proto as event;
+
+#[cfg(any(feature = "sources-vector", feature = "sinks-vector"))]
+pub mod vector;
+
+#[cfg(feature = "sinks-datadog_metrics")]
+pub mod fds {
+    use once_cell::sync::OnceCell;
+    use prost_reflect::DescriptorPool;
+
+    static PROTOBUF_FDS: OnceCell<DescriptorPool> = OnceCell::new();
+
+    pub fn get_protobuf_descriptors() -> &'static DescriptorPool {
+        PROTOBUF_FDS.get_or_init(|| {
+            DescriptorPool::decode(include_bytes!(concat!(env!("OUT_DIR"), "/protobuf-fds.bin")).as_ref())
+                .expect("should not fail to decode protobuf file descriptor set generated from build script")
+        })
+    }
+}

src/sinks/datadog/metrics/config.rs

@@ -59,6 +59,11 @@ impl DatadogMetricsEndpoint {
             DatadogMetricsEndpoint::Sketches => "application/x-protobuf",
         }
     }
+
+    // Gets whether or not this is a series endpoint.
+    pub const fn is_series(self) -> bool {
+        matches!(self, Self::Series)
+    }
 }
 
 /// Maps Datadog metric endpoints to their actual URI.