feat(datadog metrics sink): V3 encoder based on v0.57.0

Cargo.toml

@@ -146,6 +146,8 @@ await_holding_lock = "warn"
 let_underscore_must_use = "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"] }
@@ -348,6 +350,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
@@ -950,7 +956,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 = ["protobuf-build", "dep:prost", "dep:prost-reflect"]
+sinks-datadog_metrics = ["protobuf-build", "dep:prost", "dep:prost-reflect", "dep:datadog-agent-metrics-v3", "dep:protobuf"]
 sinks-datadog_traces = ["protobuf-build", "dep:prost", "dep:rmpv", "dep:rmp-serde", "dep:serde_bytes"]
 sinks-doris = ["sqlx/mysql"]
 sinks-elasticsearch = ["transforms-metric_to_log"]

LICENSE-3rdparty.csv

@@ -220,6 +220,7 @@ data-encoding,https://github.com/ia0/data-encoding,MIT,Julien Cretin <git@ia0.eu
 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>
@@ -598,6 +599,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/1_datadog_metrics_v3.enhancement.md

@@ -0,0 +1,8 @@
+Adds a new encoder to the Datadog metrics sink to encode metrics with v3 of
+the payload protocol. An additional option `dual_write` will make Vector send
+duplicate series payloads to the given endpoint encoded with the configured
+protocol. This allows the Datadog backend to validate that the metrics sent via
+both protocols specify the exact same metrics. Only series metrics are
+dual-written; sketches are never shadowed.
+
+authors: stephenwakely

changelog.d/datadog_metrics_sketches_v3_disabled.fix.md

@@ -0,0 +1,13 @@
+The `datadog_metrics` sink's `sketches_api_version: v3` option can no longer be configured; it
+is rejected at config-load time with an `unknown variant` error.
+
+Datadog's V3 sketches intake routes don't currently exist (both `/api/intake/metrics/v3/sketches`
+and its beta counterpart return `404`), and a `404` response is treated as retriable, so a sink
+configured this way would retry every sketches flush forever without ever delivering it.
+
+`sketches_api_version: v2` (the default) is unaffected. `series_api_version: v3` is unaffected;
+this only restricts the sketches endpoint. The V3 sketches encoder and its plumbing remain in the
+codebase and are exercised by tests directly, so re-enabling it later is a small change once the
+intake side is ready.
+
+authors: stephenwakely

changelog.d/datadog_metrics_v3_dual_write_default.breaking.md

@@ -0,0 +1,22 @@
+# `datadog_metrics` sink now dual-writes a V3 shadow payload by default when submitting directly to Datadog
+
+The `datadog_metrics` sink's `dual_write` V3 shadow option is now enabled by default (with
+`shadow_every: 1000`, sampling 1 in every 1000 legacy series flushes), but only when submitting
+directly to Datadog (no custom `endpoint` configured). This means Vector now sends an additional,
+sampled V3-encoded payload to Datadog's shadow intake endpoint alongside the normal legacy
+payload, without any configuration required.
+
+If a custom `endpoint` is configured (for example, a Datadog Agent, relay, or test collector),
+dual-write defaults to **disabled** instead. The shadow route
+(`/api/intake/metrics/v3beta/series`) is only guaranteed to exist on Datadog's own intake; hitting
+it on a custom endpoint that doesn't implement it returns a `404`, which is treated as retriable,
+so every sampled flush would otherwise add a request that retries forever.
+
+Only series metrics are dual-written. Sketches (distributions and histograms) are never shadowed,
+because the V3 sketches intake endpoints do not exist.
+
+`dual_write.enabled` can be set explicitly to override either default in either direction: `true`
+to opt in to shadow traffic against a custom endpoint, or `false` to disable it even when
+submitting directly to Datadog.
+
+authors: stephenwakely

changelog.d/datadog_metrics_v3_shadow_timestamp.fix.md

@@ -0,0 +1,11 @@
+The `datadog_metrics` sink now resolves the "no timestamp" fallback once per flush instead of once
+per metric per encoder.
+
+Metrics from sources that don't set a timestamp (such as `statsd`) had their timestamp filled in
+with `Utc::now()` independently by the V2 and the V3 shadow encoder. Because the two payloads are
+encoded one after the other, any flush whose encoding straddled a second boundary produced
+different timestamps in each payload, which made the intake's V2/V3 comparison report large
+numbers of series as present on only one side. It also meant a single flush's points could be
+split across two seconds within the V2 payload on its own.
+
+authors: stephenwakely

src/internal_events/datadog_metrics.rs

@@ -35,3 +35,35 @@ impl InternalEvent for DatadogMetricsEncodingError<'_> {
         }
     }
 }
+
+/// Fired on every failed attempt to send a Datadog metrics request (including ones that
+/// will be retried), tagged with the request's `batch_id` and target `uri` so a specific
+/// failure can be correlated with a Datadog-side error such as
+/// `[<batch_id>] MISMATCH (timeout): incomplete payloads` from V3 shadow-write validation.
+///
+/// This is diagnostic logging only — it does not increment `component_errors_total`, since
+/// the generic request driver already counts the final, post-retry failure via `CallError`.
+/// Deliberately not named `...Error`: `cargo vdev check events` treats any event ending in
+/// `Error` as a terminal component error that MUST log at `error!` and increment
+/// `component_errors_total`. This event fires once per retry attempt, so doing that would
+/// inflate `component_errors_total` by the retry count instead of by 1 per failed flush.
+#[derive(Debug, NamedInternalEvent)]
+pub struct DatadogMetricsRequestFailed<'a> {
+    pub error: &'a str,
+    pub batch_id: Option<&'a str>,
+    pub uri: &'a http::Uri,
+}
+
+impl InternalEvent for DatadogMetricsRequestFailed<'_> {
+    fn emit(self) {
+        warn!(
+            message = "Failed to send Datadog metrics request.",
+            error = self.error,
+            error_type = error_type::REQUEST_FAILED,
+            stage = error_stage::SENDING,
+            batch_id = self.batch_id.unwrap_or("none"),
+            uri = %self.uri,
+            internal_log_rate_limit = true,
+        );
+    }
+}