chore(datadog_metrics sink): switch series v2 and sketches to zstd compression (vectordotdev#24956)

* chore(datadog_metrics sink): switch series v2 endpoint to zstd compression

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

* chore(regression): switch statsd_to_datadog_metrics to ingress_throughput

The egress_throughput goal measures compressed bytes received by the
blackhole. Switching v2 series from zlib to zstd produces smaller
compressed payloads (better compression ratio), which registers as a
false regression in egress bytes/sec. ingress_throughput measures how
fast Vector consumes statsd data from the generator, which is
compression-agnostic and reflects actual pipeline performance.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(datadog_metrics sink): prevent division-by-zero in proptest with zero limits

Start proptest ranges at 1 instead of 0 for uncompressed_limit and
compressed_limit. The old validate_payload_size_limits rejected zero
limits, but with_payload_limits is now infallible, so finish() can
panic on division-by-zero when computing recommended_splits.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore(datadog_metrics sink): switch sketches endpoint to zstd compression

Sketches endpoint now uses zstd instead of zlib, matching Series v2.
Only Series v1 remains on zlib.

Validated against real Datadog API: 36/36 correctness checks passed,
all 18 metrics match between v1 and v2.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(datadog_metrics sink): correct changelog to reflect sketches use zstd

The changelog incorrectly stated that Sketches continue to use zlib,
but the code routes Sketches to zstd compression.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(datadog_metrics sink): move compression constants to vector-common

Move zlib and zstd compression-bound constants from inline locals in
DatadogMetricsCompression::max_compressed_size to
lib/vector-common/src/constants.rs with descriptive names and doc
comments linking to their specifications.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(datadog_metrics sink): move inline use statements to module top

Move use statements from inside test helper function bodies to the top
of the test module, as is conventional in Rust.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* cargo fmt

* fix(datadog_metrics sink): update integration test decompression for zstd

The decompress_payload helper in integration_tests.rs hardcoded zlib
decompression, but Series v2 now uses zstd. Auto-detect the compression
format via is_zstd so tests work for both zlib and zstd payloads.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: 3 people

changelog.d/24956_datadog_metrics_zstd_series_v2.enhancement.md

@@ -0,0 +1,3 @@
+The `datadog_metrics` sink now uses zstd compression when submitting metrics to the Series v2 (`/api/v2/series`) and Sketches endpoints. Series v1 continues to use zlib (deflate).
+
+authors: vladimir-dd

lib/vector-common/src/constants.rs

@@ -1,3 +1,19 @@
 pub const GZIP_MAGIC: &[u8] = &[0x1f, 0x8b];
 pub const ZLIB_MAGIC: &[u8] = &[0x78];
 pub const ZSTD_MAGIC: &[u8] = &[0x28, 0xB5, 0x2F, 0xFD];
+
+/// Maximum size of a zlib stored (uncompressed) block in bytes.
+/// See: <https://www.zlib.net/zlib_tech.html>
+pub const ZLIB_STORED_BLOCK_SIZE: usize = 16384;
+
+/// Per-block overhead for zlib stored blocks: 1-byte header + 2-byte length + 2-byte ~length.
+/// See: <https://www.zlib.net/zlib_tech.html>
+pub const ZLIB_STORED_BLOCK_OVERHEAD: usize = 5;
+
+/// Zlib frame overhead: 2-byte header + 4-byte Adler-32 checksum trailer.
+/// See: <https://www.zlib.net/zlib_tech.html>
+pub const ZLIB_FRAME_OVERHEAD: usize = 6;
+
+/// Threshold below which zstd's `ZSTD_compressBound` adds extra margin (128 KiB).
+/// See: <https://github.com/facebook/zstd/blob/dev/lib/zstd.h> (`ZSTD_compressBound`)
+pub const ZSTD_SMALL_INPUT_THRESHOLD: usize = 128 << 10;

regression/cases/statsd_to_datadog_metrics/experiment.yaml

@@ -1,4 +1,4 @@
-optimization_goal: egress_throughput
+optimization_goal: ingress_throughput
 
 target:
   name: vector

src/sinks/datadog/metrics/config.rs

@@ -87,11 +87,6 @@ impl DatadogMetricsEndpoint {
         }
     }
 
-    // Gets whether or not this is a series endpoint.
-    pub const fn is_series(self) -> bool {
-        matches!(self, Self::Series { .. })
-    }
-
     pub(super) const fn payload_limits(self) -> DatadogMetricsPayloadLimits {
         // from https://docs.datadoghq.com/api/latest/metrics/#submit-metrics
         let (uncompressed, compressed) = match self {
@@ -112,6 +107,32 @@ impl DatadogMetricsEndpoint {
             compressed,
         }
     }
+
+    /// Returns the compression scheme used for this endpoint.
+    pub(super) const fn compression(self) -> DatadogMetricsCompression {
+        match self {
+            Self::Series(SeriesApiVersion::V1) => DatadogMetricsCompression::Zlib,
+            _ => DatadogMetricsCompression::Zstd,
+        }
+    }
+}
+
+/// Selects the compressor for a given Datadog metrics endpoint.
+#[derive(Clone, Copy, Debug)]
+pub(super) enum DatadogMetricsCompression {
+    /// zlib (deflate) — used by Series v1.
+    Zlib,
+    /// zstd — used by Series v2 and Sketches.
+    Zstd,
+}
+
+impl DatadogMetricsCompression {
+    pub(super) const fn content_encoding(self) -> &'static str {
+        match self {
+            Self::Zstd => "zstd",
+            Self::Zlib => "deflate",
+        }
+    }
 }
 
 /// Maps Datadog metric endpoints to their actual URI.
@@ -270,7 +291,7 @@ impl DatadogMetricsConfig {
             endpoint_configuration,
             self.default_namespace.clone(),
             self.series_api_version,
-        )?;
+        );
 
         let protocol = self.get_protocol(dd_common);
         let sink = DatadogMetricsSink::new(