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"]

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,14 @@
+# `datadog_metrics` sink now dual-writes a V3 shadow payload by default
+
+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). 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.
+
+Only series metrics are dual-written. Sketches (distributions and histograms) are never shadowed,
+because the V3 sketches intake endpoints do not exist.
+
+If you don't want this additional traffic, set `dual_write.enabled: false` on your `datadog_metrics`
+sink configuration.
+
+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 = false,
+        );
+    }
+}