enhancement(datadog metrics sink): implement V3

Cargo.toml

@@ -157,6 +157,8 @@ rc_buffer = "warn"
 todo = "warn"
 
 [workspace.dependencies]
+datadog-agent-metrics-v3 = { git = "https://github.com/DataDog/saluki.git", tag = "1.3.0" }
+protobuf = { version = "3.7", default-features = false, features = ["with-bytes"] }
 antithesis-instrumentation = { version = "0.1", default-features = false, features = [] }
 antithesis_sdk = { version = "0.2", default-features = false, features = [] }
 anyhow = { version = "1.0.102", default-features = false, features = ["std"] }
@@ -366,6 +368,10 @@ rmpv = { version = "1.3.0", default-features = false, features = ["with-serde"],
 # Prost / Protocol Buffers
 prost = { workspace = true, optional = true }
 prost-reflect = { workspace = true, optional = true }
+
+# Datadog metrics V3 columnar codec
+datadog-agent-metrics-v3 = { workspace = true, optional = true }
+protobuf = { workspace = true, optional = true }
 prost-types = { workspace = true, optional = true }
 
 # Databricks Zerobus
@@ -971,7 +977,7 @@ sinks-databend = ["dep:databend-client"]
 sinks-databricks_zerobus = ["dep:databricks-zerobus-ingest-sdk", "codecs-arrow", "arrow/ipc_compression"]
 sinks-datadog_events = []
 sinks-datadog_logs = []
-sinks-datadog_metrics = ["dep:datadog-proto", "dep:prost", "dep:prost-reflect"]
+sinks-datadog_metrics = ["dep:datadog-proto", "dep:prost", "dep:prost-reflect", "dep:datadog-agent-metrics-v3", "dep:protobuf"]
 sinks-datadog_traces = ["protobuf-build", "dep:datadog-proto", "dep:prost", "dep:rmpv", "dep:rmp-serde", "dep:serde_bytes"]
 sinks-doris = ["sqlx/mysql"]
 sinks-elasticsearch = ["transforms-metric_to_log"]

LICENSE-3rdparty.csv

@@ -225,6 +225,7 @@ data-encoding,https://github.com/ia0/data-encoding,MIT,The data-encoding Authors
 data-url,https://github.com/servo/rust-url,MIT OR Apache-2.0,Simon Sapin <simon.sapin@exyr.org>
 databend-client,https://github.com/databendlabs/bendsql,Apache-2.0,Databend Authors <opensource@databend.com>
 databricks-zerobus-ingest-sdk,https://github.com/databricks/zerobus-sdk,Apache-2.0,Databricks
+datadog-agent-metrics-v3,https://github.com/DataDog/saluki,Apache-2.0,The datadog-agent-metrics-v3 Authors
 dbl,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developers
 deadpool,https://github.com/deadpool-rs/deadpool,MIT OR Apache-2.0,Michael P. Jung <michael.jung@terreon.de>
 deadpool-runtime,https://github.com/deadpool-rs/deadpool,MIT OR Apache-2.0,Michael P. Jung <michael.jung@terreon.de>
@@ -624,6 +625,8 @@ prost,https://github.com/tokio-rs/prost,Apache-2.0,"Dan Burkert <dan@danburkert.
 prost-derive,https://github.com/tokio-rs/prost,Apache-2.0,"Dan Burkert <dan@danburkert.com>, Lucio Franco <luciofranco14@gmail.com>, Casper Meijn <casper@meijn.net>, Tokio Contributors <team@tokio.rs>"
 prost-reflect,https://github.com/andrewhickman/prost-reflect,MIT OR Apache-2.0,Andrew Hickman <andrew.hickman1@sky.com>
 prost-types,https://github.com/tokio-rs/prost,Apache-2.0,"Dan Burkert <dan@danburkert.com>, Lucio Franco <luciofranco14@gmail.com>, Casper Meijn <casper@meijn.net>, Tokio Contributors <team@tokio.rs>"
+protobuf,https://github.com/stepancheg/rust-protobuf,MIT,Stepan Koltsov <stepan.koltsov@gmail.com>
+protobuf-support,https://github.com/stepancheg/rust-protobuf,MIT,Stepan Koltsov <stepan.koltsov@gmail.com>
 psl,https://github.com/addr-rs/psl,MIT OR Apache-2.0,rushmorem <rushmore@webenchanter.com>
 psl-types,https://github.com/addr-rs/psl-types,MIT OR Apache-2.0,rushmorem <rushmore@webenchanter.com>
 ptr_meta,https://github.com/djkoloski/ptr_meta,MIT,David Koloski <djkoloski@gmail.com>

changelog.d/datadog_metrics_series_v3_default.breaking.md

@@ -0,0 +1,32 @@
+# Datadog metrics series submitted to the V3 intake by default {#datadog-metrics-series-v3-default}
+
+## Summary
+
+The `datadog_metrics` sink now submits series metrics to `/api/intake/metrics/v3/series` by
+default, using Datadog's columnar protobuf format. This format uses dictionary-based string
+deduplication and delta encoding, making it more efficient than `v2` for workloads with many
+metrics that share common names or tags.
+
+Sketch metrics (distributions and histograms) are unaffected and continue to be submitted to
+`/api/beta/sketches`.
+
+## Migration
+
+No configuration change is required when sending directly to Datadog's managed intake.
+
+If your `datadog_metrics` sink forwards to another Vector instance's `datadog_agent`
+source, the receiving instance must support V3 before the sender switches to the new
+default. Upgrading the sender first causes series metric requests to fail.
+
+Upgrade receiving instances first, or explicitly configure sending sinks to continue
+using V2 until all receivers support V3. The same workaround applies to proxies or
+other endpoints that do not accept the V3 intake route:
+
+```yaml
+sinks:
+  my_sink:
+    type: datadog_metrics
+    series_api_version: v2
+```
+
+authors: stephenwakely

src/sinks/datadog/metrics/config.rs

@@ -35,6 +35,7 @@ impl SinkBatchSettings for DatadogMetricsDefaultBatchSettings {
 
 pub(super) const SERIES_V1_PATH: &str = "/api/v1/series";
 pub(super) const SERIES_V2_PATH: &str = "/api/v2/series";
+pub(super) const SERIES_V3_PATH: &str = "/api/intake/metrics/v3/series";
 pub(super) const SKETCHES_PATH: &str = "/api/beta/sketches";
 
 /// The API version to use when submitting series metrics to Datadog.
@@ -49,19 +50,32 @@ pub enum SeriesApiVersion {
     V1,
 
     /// Use the v2 series endpoint (`/api/v2/series`).
+    V2,
+
+    /// Use the v3 series endpoint (`/api/intake/metrics/v3/series`).
+    ///
+    /// Columnar protobuf format with dictionary-based string deduplication and delta
+    /// encoding. More efficient than v2 for workloads with many metrics that share
+    /// common tags or names.
     ///
     /// This is the recommended and default endpoint.
     #[default]
-    V2,
+    V3,
 }
 
 impl SeriesApiVersion {
     pub const fn get_path(self) -> &'static str {
         match self {
             Self::V1 => SERIES_V1_PATH,
             Self::V2 => SERIES_V2_PATH,
+            Self::V3 => SERIES_V3_PATH,
         }
     }
+
+    /// Returns true if this version uses the V3 columnar encoding format.
+    pub const fn is_v3_format(self) -> bool {
+        matches!(self, Self::V3)
+    }
 }
 
 /// Various metric type-specific API types.
@@ -84,7 +98,9 @@ impl DatadogMetricsEndpoint {
     pub const fn content_type(self) -> &'static str {
         match self {
             Self::Series(SeriesApiVersion::V1) => "application/json",
-            Self::Sketches | Self::Series(SeriesApiVersion::V2) => "application/x-protobuf",
+            Self::Sketches | Self::Series(SeriesApiVersion::V2 | SeriesApiVersion::V3) => {
+                "application/x-protobuf"
+            }
         }
     }
 
@@ -97,7 +113,7 @@ impl DatadogMetricsEndpoint {
                 62_914_560, // 60 MiB
                 3_200_000,  // 3.2 MB
             ),
-            DatadogMetricsEndpoint::Series(SeriesApiVersion::V2) => (
+            DatadogMetricsEndpoint::Series(SeriesApiVersion::V2 | SeriesApiVersion::V3) => (
                 5_242_880, // 5 MiB
                 512_000,   // 512 KB
             ),
@@ -178,8 +194,8 @@ pub struct DatadogMetricsConfig {
 
     /// Controls which Datadog series API endpoint is used to submit metrics.
     ///
-    /// Defaults to `v2` (`/api/v2/series`). Set to `v1` (`/api/v1/series`) only if you need to
-    /// fall back to the legacy endpoint.
+    /// Defaults to `v3` (`/api/intake/metrics/v3/series`). Set to `v2` (`/api/v2/series`) or
+    /// to `v1` (`/api/v1/series`) only if you need to fall back to the legacy endpoint.
     #[serde(default)]
     pub series_api_version: SeriesApiVersion,
 
@@ -387,7 +403,7 @@ mod tests {
     fn validate_produces_endpoint_specific_batch_settings() {
         let config = DatadogMetricsConfig::default();
         let validated = config.validate().expect("validation should succeed");
-        assert_eq!(validated.batcher_settings.size_limit, 5_242_880); // 5 MiB — Series v2 limit
+        assert_eq!(validated.batcher_settings.size_limit, 5_242_880); // 5 MiB — Series v3 limit
         assert_eq!(validated.sketches_batcher_settings.size_limit, 62_914_560); // 60 MiB — Sketches limit
     }
 
@@ -448,4 +464,44 @@ mod tests {
         assert_eq!(series.size_limit, 1_000_000);
         assert_eq!(sketches.size_limit, 1_000_000);
     }
+
+    // `v1`, `v2`, `v3` -- and the unset default, which is now `v3` -- must all still parse.
+    #[test]
+    fn series_api_version_v1_v2_v3_and_default_are_configurable() {
+        for (toml, expected) in [
+            (r#"default_api_key = "unused""#, SeriesApiVersion::V3),
+            (
+                r#"default_api_key = "unused"
+            series_api_version = "v1""#,
+                SeriesApiVersion::V1,
+            ),
+            (
+                r#"default_api_key = "unused"
+            series_api_version = "v2""#,
+                SeriesApiVersion::V2,
+            ),
+            (
+                r#"default_api_key = "unused"
+            series_api_version = "v3""#,
+                SeriesApiVersion::V3,
+            ),
+        ] {
+            let config = toml::from_str::<DatadogMetricsConfig>(toml)
+                .expect("v1, v2, v3, and the unset default must all parse");
+            assert_eq!(config.series_api_version, expected);
+        }
+    }
+
+    // Each configurable series version must resolve to its own intake path, and only `v3` uses
+    // the columnar wire format.
+    #[test]
+    fn series_api_version_paths_and_formats() {
+        assert_eq!(SeriesApiVersion::V1.get_path(), SERIES_V1_PATH);
+        assert_eq!(SeriesApiVersion::V2.get_path(), SERIES_V2_PATH);
+        assert_eq!(SeriesApiVersion::V3.get_path(), SERIES_V3_PATH);
+
+        assert!(!SeriesApiVersion::V1.is_v3_format());
+        assert!(!SeriesApiVersion::V2.is_v3_format());
+        assert!(SeriesApiVersion::V3.is_v3_format());
+    }
 }