chore(datadog_metrics sink): switch to v2 endpoint (#24842)

* chore(datadog_metrics sink): switch to Datadog Metrics V2 endpoint

* chore(regression): Add statsd_to_datadog_metrics performance test

Add regression test to validate datadog_metrics sink v2 endpoint
performance under realistic high-throughput DogStatsD load.

Test Configuration:
- Load: Default lading dogstatsd settings (realistic ~2KB messages)
- Throughput: 500 Mb/s → ~250k events/sec
- Batch: Default settings (100k max_events, 2s timeout)
- Validates batch splitting when payloads exceed v2 size limits

This test ensures v2 endpoint correctly handles batch splitting
with realistic high-cardinality DogStatsD metrics under load.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix(datadog_metrics sink): apply per-endpoint batch size limit

Different series endpoints have different uncompressed payload limits (v2
is 12x smaller than v1). This ensures each batch fits in a single HTTP
request without splitting, reducing memory overhead.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(datadog_metrics sink): fix test compilation errors in encoder and config

- Remove references to DatadogMetricsCompression and request_compression
  from encoder.rs tests (those symbols don't exist in current codebase;
  they belong to an unmerged compression-options branch)
- Fix batcher_user_max_bytes_is_preserved test to avoid struct update
  syntax with private PhantomData fields in BatchConfig

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore(datadog_metrics sink): warn on deprecated VECTOR_TEMP_USE_DD_METRICS_SERIES_V2_API env var

The old opt-in env var is now a no-op since v2 is the default. Emit a
one-time warning so existing users know they can safely remove it.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* trigger build

* feat(datadog_metrics sink): use separate batch size limits for Series and Sketches

Series v2 has a 5 MiB uncompressed payload limit while Sketches allows 60 MiB.
Previously both used the same (Series-derived) cap, which over-fragmented
sketch-heavy workloads into many small requests.

To support per-partition batch configuration, `PartitionedBatcher` now passes
the partition key to the batch config factory closure (`Fn(&Key) -> C` instead
of `Fn() -> C`). The explicit `timeout: Duration` parameter replaces the
previous extraction via `settings().timeout()`. All existing callers are
updated mechanically.

The `datadog_metrics` sink uses the new capability to select the appropriate
byte size limit per endpoint partition, keeping Series at 5 MiB and Sketches
at 60 MiB.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(datadog_metrics sink): fix CI failures after partitioned batcher refactor

- Add `Batch = B` constraint to `with_timer` test constructor so the
  compiler can infer `B` from the batch config type
- Replace `Box::new(move |_| ...)` with unboxed `move |_: &u8| ...` in
  tests to avoid HRTB inference issues with `Fn(&Key) -> C`
- Run `make fmt` to fix formatting in several sink files

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat(datadog_metrics sink): add series_api_version config field, remove env var support

- Add `series_api_version` config option (`v1` | `v2`, default `v2`) to `DatadogMetricsConfig`
- Make `SeriesApiVersion` a proper `#[configurable_component]` enum with serde support
- Remove `VECTOR_TEMP_USE_DD_METRICS_SERIES_V1_API` and `VECTOR_TEMP_USE_DD_METRICS_SERIES_V2_API` env vars
- Remove `get_api_version_backwards_compatible()` and `DatadogMetricsEndpoint::series()` helpers
- Thread `series_api_version` through config, sink, partitioner, and request builder
- Add `v1_batch_config_uses_v1_size_limit` test; update existing tests to pass explicit version

Rationale: Replace the hidden, process-global env var mechanism with a standard config field so
users can control the series API version per-sink without a full Vector restart, and so the option
is properly documented alongside other sink settings.

* update docs

* mark v1 as deprecated

* fix(datadog_metrics sink): fix gauge interval assertion after normalization strips interval_ms

`into_absolute()` always sets `interval_ms: None`, so the encoded
interval for gauges is always 0. The test expectation of 10 was
dead code until v2 became the default endpoint.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* test(datadog_metrics sink): validate both v1 and v2 series endpoints in e2e tests

- Add series_api_version matrix dimension ['v1', 'v2'] to test.yaml, expanding coverage from 3 to 6 environments
- Pass CONFIG_SERIES_API_VERSION through docker compose environment to the Vector container
- Parameterize series_api_version in vector.toml from the env var
- Refactor series::validate() to dispatch to v1 or v2 fetch function based on CONFIG_SERIES_API_VERSION
- Extract assertions into compare_intakes() helper (no new assertion logic)
- Replace blocking std::thread::sleep with async tokio::time::sleep in the test entry point

Rationale: Vector now supports configurable series_api_version (v1/v2) in the datadog_metrics sink. The e2e test previously hardcoded v1 for the vector pipeline. This change runs the same assertions against both API versions via the CI matrix, ensuring correctness is validated for each version independently without duplicating test logic.

---------

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: vladimir-dd

changelog.d/datadog_metrics_series_v2_default.enhancement.md

@@ -0,0 +1,5 @@
+The `datadog_metrics` sink now defaults to the Datadog series v2 endpoint (`/api/v2/series`) and
+exposes a new `series_api_version` configuration option (`v1` or `v2`) to control which endpoint is
+used. Set `series_api_version: v1` to fall back to the legacy v1 endpoint if needed.
+
+authors: vladimir-dd

lib/vector-stream/src/partitioned_batcher.rs

@@ -211,10 +211,9 @@ where
     Prt::Key: Eq + Hash + Clone,
     Prt::Item: ByteSizeOf,
     C: BatchConfig<Prt::Item>,
-    F: Fn() -> C + Send,
+    F: Fn(&Prt::Key) -> C + Send,
 {
-    pub fn new(stream: St, partitioner: Prt, settings: F) -> Self {
-        let timeout = settings().timeout();
+    pub fn new(stream: St, partitioner: Prt, timeout: Duration, settings: F) -> Self {
         Self {
             state: settings,
             batches: HashMap::default(),
@@ -233,8 +232,8 @@ where
     Prt: Partitioner + Unpin,
     Prt::Key: Eq + Hash + Clone,
     Prt::Item: ByteSizeOf,
-    C: BatchConfig<Prt::Item>,
-    F: Fn() -> C + Send,
+    C: BatchConfig<Prt::Item, Batch = B>,
+    F: Fn(&Prt::Key) -> C + Send,
 {
     pub fn with_timer(stream: St, partitioner: Prt, timer: KT, settings: F) -> Self {
         Self {
@@ -256,7 +255,7 @@ where
     Prt::Item: ByteSizeOf,
     KT: KeyedTimer<Prt::Key>,
     C: BatchConfig<Prt::Item, Batch = B>,
-    F: Fn() -> C + Send,
+    F: Fn(&Prt::Key) -> C + Send,
 {
     type Item = (Prt::Key, B);
 
@@ -307,7 +306,7 @@ where
                     let batch = if let Some(batch) = this.batches.get_mut(&item_key) {
                         batch
                     } else {
-                        let batch = (this.state)();
+                        let batch = (this.state)(&item_key);
                         this.batches.insert(item_key.clone(), batch);
                         this.timer.insert(item_key.clone());
                         this.batches
@@ -479,7 +478,7 @@ mod test {
             let batch_settings = BatcherSettings::new(Duration::from_secs(1), allocation_limit, item_limit);
 
             let batcher = PartitionedBatcher::with_timer(&mut stream, partitioner, timer,
-                                              Box::new(move || batch_settings.as_byte_size_config()));
+                                              move |_: &u8| batch_settings.as_byte_size_config::<u64>());
             let batcher_size_hint = batcher.size_hint();
 
             assert_eq!(stream_size_hint, batcher_size_hint);
@@ -503,7 +502,7 @@ mod test {
             let allocation_limit = NonZeroUsize::new(allocation_limit as usize).unwrap();
             let batch_settings = BatcherSettings::new(Duration::from_secs(1), allocation_limit, item_limit);
             let mut batcher = PartitionedBatcher::with_timer(&mut stream, partitioner,
-                                                  timer, Box::new(move || batch_settings.as_byte_size_config()));
+                                                  timer, move |_: &u8| batch_settings.as_byte_size_config::<u64>());
             let mut batcher = Pin::new(&mut batcher);
 
             loop {
@@ -574,7 +573,7 @@ mod test {
             let allocation_limit = NonZeroUsize::new(allocation_limit as usize).unwrap();
             let batch_settings = BatcherSettings::new(Duration::from_secs(1), allocation_limit, item_limit);
             let mut batcher = PartitionedBatcher::with_timer(&mut stream, partitioner,
-                                                  timer, Box::new(move || batch_settings.as_byte_size_config()));
+                                                  timer, move |_: &u8| batch_settings.as_byte_size_config::<u64>());
             let mut batcher = Pin::new(&mut batcher);
 
             loop {
@@ -618,7 +617,7 @@ mod test {
             let allocation_limit = NonZeroUsize::new(allocation_limit as usize).unwrap();
             let batch_settings = BatcherSettings::new(Duration::from_secs(1), allocation_limit, item_limit);
             let mut batcher = PartitionedBatcher::with_timer(&mut stream, partitioner,
-                                                  timer, Box::new(move || batch_settings.as_byte_size_config()));
+                                                  timer, move |_: &u8| batch_settings.as_byte_size_config::<u64>());
             let mut batcher = Pin::new(&mut batcher);
 
             let mut observed_items = 0;

regression/cases/statsd_to_datadog_metrics/experiment.yaml

@@ -0,0 +1,10 @@
+optimization_goal: egress_throughput
+
+target:
+  name: vector
+  command: /usr/bin/vector
+  cpu_allotment: 6
+  memory_allotment: 8GiB
+
+  environment:
+    VECTOR_THREADS: 4

regression/cases/statsd_to_datadog_metrics/lading/lading.yaml

@@ -0,0 +1,16 @@
+generator:
+  - tcp:
+      seed: [2, 3, 5, 7, 11, 13, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137]
+      addr: "0.0.0.0:8125"
+      variant:
+        dogstatsd: {}
+      bytes_per_second: "500 Mb"
+      maximum_prebuild_cache_size_bytes: "256 Mb"
+
+blackhole:
+  - http:
+      binding_addr: "0.0.0.0:8080"
+
+target_metrics:
+  - prometheus:
+      uri: "http://127.0.0.1:9090/metrics"

regression/cases/statsd_to_datadog_metrics/vector/vector.yaml

@@ -0,0 +1,25 @@
+data_dir: "/var/lib/vector"
+
+sources:
+  internal_metrics:
+    type: "internal_metrics"
+
+  statsd:
+    type: "statsd"
+    address: "0.0.0.0:8125"
+    mode: "tcp"
+
+sinks:
+  prometheus:
+    type: "prometheus_exporter"
+    inputs: ["internal_metrics"]
+    address: "0.0.0.0:9090"
+
+  datadog_metrics:
+    type: "datadog_metrics"
+    inputs: ["statsd"]
+    endpoint: "http://0.0.0.0:8080"
+    default_api_key: "DEADBEEF"
+    default_namespace: "vector"
+    healthcheck:
+      enabled: false

src/sinks/aws_cloudwatch_logs/sink.rs

@@ -49,7 +49,7 @@ where
                 let age_range = start..end;
                 future::ready(age_range.contains(&req.timestamp))
             })
-            .batched_partitioned(CloudwatchPartitioner, || {
+            .batched_partitioned(CloudwatchPartitioner, batcher_settings.timeout, |_| {
                 batcher_settings.as_byte_size_config()
             })
             .map(|(key, events)| {

src/sinks/azure_common/sink.rs

@@ -47,7 +47,9 @@ where
         let request_builder = self.request_builder;
 
         input
-            .batched_partitioned(partitioner, || settings.as_byte_size_config())
+            .batched_partitioned(partitioner, settings.timeout, |_| {
+                settings.as_byte_size_config()
+            })
             .filter_map(|(key, batch)| async move {
                 // We don't need to emit an error here if the event is dropped since this will occur if the template
                 // couldn't be rendered during the partitioning. A `TemplateRenderingError` is already emitted when

src/sinks/clickhouse/sink.rs

@@ -43,7 +43,8 @@ where
         input
             .batched_partitioned(
                 KeyPartitioner::new(self.database, self.table, self.format),
-                || batch_settings.as_byte_size_config(),
+                batch_settings.timeout,
+                |_| batch_settings.as_byte_size_config(),
             )
             .filter_map(|(key, batch)| async move { key.map(move |k| (k, batch)) })
             .request_builder(

src/sinks/datadog/logs/sink.rs

@@ -386,7 +386,7 @@ where
             conforms_as_agent: self.conforms_as_agent,
         });
 
-        let input = input.batched_partitioned(partitioner, || {
+        let input = input.batched_partitioned(partitioner, batch_settings.timeout, |_| {
             batch_settings.as_item_size_config(HttpJsonBatchSizer)
         });
         input

src/sinks/datadog/metrics/config.rs

@@ -1,9 +1,9 @@
-use std::sync::OnceLock;
-
 use http::Uri;
 use snafu::ResultExt;
 use tower::ServiceBuilder;
-use vector_lib::{config::proxy::ProxyConfig, configurable::configurable_component};
+use vector_lib::{
+    config::proxy::ProxyConfig, configurable::configurable_component, stream::BatcherSettings,
+};
 
 use super::{
     request_builder::DatadogMetricsRequestBuilder,
@@ -21,18 +21,13 @@ use crate::{
     },
     tls::{MaybeTlsSettings, TlsEnableableConfig},
 };
-
 #[derive(Clone, Copy, Debug, Default)]
 pub struct DatadogMetricsDefaultBatchSettings;
 
-// This default is centered around "series" data, which should be the lion's share of what we
-// process.  Given that a single series, when encoded, is in the 150-300 byte range, we can fit a
-// lot of these into a single request, something like 150-200K series.  Simply to be a little more
-// conservative, though, we use 100K here.  This will also get a little more tricky when it comes to
-// distributions and sketches, but we're going to have to implement incremental encoding to handle
-// "we've exceeded our maximum payload size, split this batch" scenarios anyways.
 impl SinkBatchSettings for DatadogMetricsDefaultBatchSettings {
     const MAX_EVENTS: Option<usize> = Some(100_000);
+    // No default byte cap here; the appropriate limit (v1: 60 MiB, v2: 5 MiB) is applied at
+    // sink build time based on the active series API version.
     const MAX_BYTES: Option<usize> = None;
     const TIMEOUT_SECS: f64 = 2.0;
 }
@@ -41,9 +36,21 @@ pub(super) const SERIES_V1_PATH: &str = "/api/v1/series";
 pub(super) const SERIES_V2_PATH: &str = "/api/v2/series";
 pub(super) const SKETCHES_PATH: &str = "/api/beta/sketches";
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+/// The API version to use when submitting series metrics to Datadog.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
+#[serde(rename_all = "snake_case")]
 pub enum SeriesApiVersion {
+    /// Use the v1 series endpoint (`/api/v1/series`).
+    ///
+    /// This is a legacy endpoint. Prefer `v2` unless you have a specific reason to use v1.
+    #[configurable(deprecated)]
     V1,
+
+    /// Use the v2 series endpoint (`/api/v2/series`).
+    ///
+    /// This is the recommended and default endpoint.
+    #[default]
     V2,
 }
 
@@ -54,15 +61,6 @@ impl SeriesApiVersion {
             Self::V2 => SERIES_V2_PATH,
         }
     }
-    fn get_api_version() -> Self {
-        static API_VERSION: OnceLock<SeriesApiVersion> = OnceLock::new();
-        *API_VERSION.get_or_init(|| {
-            match std::env::var("VECTOR_TEMP_USE_DD_METRICS_SERIES_V2_API") {
-                Ok(_) => Self::V2,
-                Err(_) => Self::V1,
-            }
-        })
-    }
 }
 
 /// Various metric type-specific API types.
@@ -94,14 +92,8 @@ impl DatadogMetricsEndpoint {
         matches!(self, Self::Series { .. })
     }
 
-    // Creates an instance of the `Series` variant with the default API version.
-    pub fn series() -> Self {
-        Self::Series(SeriesApiVersion::get_api_version())
-    }
-
     pub(super) const fn payload_limits(self) -> DatadogMetricsPayloadLimits {
         // from https://docs.datadoghq.com/api/latest/metrics/#submit-metrics
-
         let (uncompressed, compressed) = match self {
             // Sketches use the same payload size limits as v1 series
             DatadogMetricsEndpoint::Series(SeriesApiVersion::V1)
@@ -162,6 +154,13 @@ pub struct DatadogMetricsConfig {
     #[serde(default)]
     pub default_namespace: Option<String>,
 
+    /// 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.
+    #[serde(default)]
+    pub series_api_version: SeriesApiVersion,
+
     #[configurable(derived)]
     #[serde(default)]
     pub batch: BatchConfig<DatadogMetricsDefaultBatchSettings>,
@@ -222,7 +221,7 @@ impl DatadogMetricsConfig {
     ) -> crate::Result<DatadogMetricsEndpointConfiguration> {
         let base_uri = self.get_base_agent_endpoint(dd_common);
 
-        let series_endpoint = build_uri(&base_uri, SeriesApiVersion::get_api_version().get_path())?;
+        let series_endpoint = build_uri(&base_uri, self.series_api_version.get_path())?;
         let sketches_endpoint = build_uri(&base_uri, SKETCHES_PATH)?;
 
         Ok(DatadogMetricsEndpointConfiguration::new(
@@ -253,7 +252,8 @@ impl DatadogMetricsConfig {
         dd_common: &DatadogCommonConfig,
         client: HttpClient,
     ) -> crate::Result<VectorSink> {
-        let batcher_settings = self.batch.into_batcher_settings()?;
+        let (batcher_settings, sketches_batcher_settings) =
+            resolve_endpoint_batch_settings(self.batch, self.series_api_version)?;
 
         // TODO: revisit our concurrency and batching defaults
         let request_limits = self.request.into_settings();
@@ -269,10 +269,18 @@ impl DatadogMetricsConfig {
         let request_builder = DatadogMetricsRequestBuilder::new(
             endpoint_configuration,
             self.default_namespace.clone(),
+            self.series_api_version,
         )?;
 
         let protocol = self.get_protocol(dd_common);
-        let sink = DatadogMetricsSink::new(service, request_builder, batcher_settings, protocol);
+        let sink = DatadogMetricsSink::new(
+            service,
+            request_builder,
+            batcher_settings,
+            sketches_batcher_settings,
+            protocol,
+            self.series_api_version,
+        );
 
         Ok(VectorSink::from_event_streamsink(sink))
     }
@@ -287,6 +295,28 @@ impl DatadogMetricsConfig {
     }
 }
 
+/// Returns `(series_settings, sketches_settings)`.
+///
+/// When the user has not set an explicit `max_bytes`, each endpoint is capped to its own
+/// uncompressed payload limit (5 MiB for Series v2, 60 MiB for Sketches). When an explicit
+/// limit is configured, both endpoints share it.
+fn resolve_endpoint_batch_settings(
+    batch: BatchConfig<DatadogMetricsDefaultBatchSettings>,
+    series_version: SeriesApiVersion,
+) -> crate::Result<(BatcherSettings, BatcherSettings)> {
+    let mut series = batch.into_batcher_settings()?;
+    let mut sketches = series;
+    if series.size_limit == usize::MAX {
+        series.size_limit = DatadogMetricsEndpoint::Series(series_version)
+            .payload_limits()
+            .uncompressed;
+        sketches.size_limit = DatadogMetricsEndpoint::Sketches
+            .payload_limits()
+            .uncompressed;
+    }
+    Ok((series, sketches))
+}
+
 fn build_uri(host: &str, endpoint: &str) -> crate::Result<Uri> {
     let result = format!("{host}{endpoint}")
         .parse::<Uri>()
@@ -302,4 +332,36 @@ mod tests {
     fn generate_config() {
         crate::test_util::test_generate_config::<DatadogMetricsConfig>();
     }
+
+    // When max_bytes is unset, each endpoint gets its own API payload limit.
+    #[test]
+    fn default_batch_config_uses_endpoint_specific_size_limits() {
+        let (series, sketches) =
+            resolve_endpoint_batch_settings(BatchConfig::default(), SeriesApiVersion::V2).unwrap();
+
+        assert_eq!(series.size_limit, 5_242_880); // 5 MiB — Series v2 limit
+        assert_eq!(sketches.size_limit, 62_914_560); // 60 MiB — Sketches limit
+    }
+
+    #[test]
+    fn v1_batch_config_uses_v1_size_limit() {
+        let (series, sketches) =
+            resolve_endpoint_batch_settings(BatchConfig::default(), SeriesApiVersion::V1).unwrap();
+
+        assert_eq!(series.size_limit, 62_914_560); // 60 MiB — Series v1 limit
+        assert_eq!(sketches.size_limit, 62_914_560); // 60 MiB — Sketches limit
+    }
+
+    // When the user sets max_bytes, both endpoints share that limit unchanged.
+    #[test]
+    fn explicit_max_bytes_applies_to_both_endpoints() {
+        let mut config = BatchConfig::<DatadogMetricsDefaultBatchSettings>::default();
+        config.max_bytes = Some(1_000_000);
+
+        let (series, sketches) =
+            resolve_endpoint_batch_settings(config, SeriesApiVersion::V2).unwrap();
+
+        assert_eq!(series.size_limit, 1_000_000);
+        assert_eq!(sketches.size_limit, 1_000_000);
+    }
 }