chore(datadog_metrics sink): support and migrate to the v2 series API endpoint

docs/DEPRECATION.md

@@ -78,12 +78,12 @@ When introducing a deprecation into Vector, the pull request introducing the dep
   the new name will be appended with the text `(formerly OldName)`.
 - Add a log message to Vector that is logged at the `WARN` level starting with the word `DEPRECATION` if Vector detects
   the deprecated configuration or feature being used (when possible).
-- Add the deprecation to [DEPRECATIONS.md](docs/DEPRECATIONS.md) to track migration (if applicable) and removal
+- Add the deprecation to [DEPRECATIONS.md](DEPRECATIONS.md) to track migration (if applicable) and removal
 
 When removing a deprecation in a subsequent release, the pull request should:
 
 - Indicate that it is a breaking change by including `!` in the title after the type/scope
 - Remove the deprecation from the documentation
 - Add a note to the Breaking Changes section of the upgrade guide for the next release with a description and directions
   for transitioning if applicable.
-- Remove the deprecation from [DEPRECATIONS.md](docs/DEPRECATIONS.md)
+- Remove the deprecation from [DEPRECATIONS.md](DEPRECATIONS.md)

docs/DEPRECATIONS.md

@@ -6,6 +6,7 @@ See [DEPRECATION.md](docs/DEPRECATION.md#process) for the process for updating t
 
 ## To be removed
 
+* Support for `v1` series endpoint in the `datadog_metrics` sink should be removed.
 * legacy_openssl_provider v0.34.0 OpenSSL legacy provider flag should be removed
 * armv7_rpm v0.34.0 The armv7 RPM packages should be removed (replaced by armv7hl)
 * yaml_migration v0.34.0 Prefer loading `/etc/vector/vector.yaml` first

lib/vector-core/src/event/mod.rs

@@ -319,6 +319,13 @@ impl Event {
         self
     }
 
+    /// Sets the `source_type` in the event metadata to the provided value.
+    #[must_use]
+    pub fn with_source_type(mut self, source_type: &'static str) -> Self {
+        self.metadata_mut().set_source_type(source_type);
+        self
+    }
+
     /// Sets the `upstream_id` in the event metadata to the provided value.
     #[must_use]
     pub fn with_upstream_id(mut self, upstream_id: Arc<OutputId>) -> Self {

src/sinks/datadog/metrics/config.rs

@@ -1,3 +1,5 @@
+use std::sync::OnceLock;
+
 use http::Uri;
 use snafu::ResultExt;
 use tower::ServiceBuilder;
@@ -42,27 +44,63 @@ impl SinkBatchSettings for DatadogMetricsDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = 2.0;
 }
 
+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";
+
+// TODO: the series V1 endpoint support is considered deprecated and should be removed in a future release.
+// At that time when the V1 support is removed, the SeriesApiVersion stops being useful and can be removed.
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum SeriesApiVersion {
+    V1,
+    V2,
+}
+
+impl SeriesApiVersion {
+    pub const fn get_path(self) -> &'static str {
+        match self {
+            Self::V1 => SERIES_V1_PATH,
+            Self::V2 => SERIES_V2_PATH,
+        }
+    }
+    fn get_api_version_backwards_compatible() -> Self {
+        static API_VERSION: OnceLock<SeriesApiVersion> = OnceLock::new();
+        *API_VERSION.get_or_init(
+            || match option_env!("VECTOR_TEMP_USE_DD_METRICS_SERIES_V1_API") {
+                Some(_) => Self::V1,
+                None => Self::V2,
+            },
+        )
+    }
+}
+
 /// Various metric type-specific API types.
 ///
 /// Each of these corresponds to a specific request path when making a request to the agent API.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum DatadogMetricsEndpoint {
-    Series,
+    Series(SeriesApiVersion),
     Sketches,
 }
 
 impl DatadogMetricsEndpoint {
     /// Gets the content type associated with the specific encoder for a given metric endpoint.
     pub const fn content_type(self) -> &'static str {
         match self {
-            DatadogMetricsEndpoint::Series => "application/json",
-            DatadogMetricsEndpoint::Sketches => "application/x-protobuf",
+            Self::Series(SeriesApiVersion::V1) => "application/json",
+            Self::Sketches | Self::Series(SeriesApiVersion::V2) => "application/x-protobuf",
         }
     }
 
     // Gets whether or not this is a series endpoint.
     pub const fn is_series(self) -> bool {
-        matches!(self, Self::Series)
+        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_backwards_compatible())
     }
 }
 
@@ -84,7 +122,7 @@ impl DatadogMetricsEndpointConfiguration {
     /// Gets the URI for the given Datadog metrics endpoint.
     pub fn get_uri_for_endpoint(&self, endpoint: DatadogMetricsEndpoint) -> Uri {
         match endpoint {
-            DatadogMetricsEndpoint::Series => self.series_endpoint.clone(),
+            DatadogMetricsEndpoint::Series { .. } => self.series_endpoint.clone(),
             DatadogMetricsEndpoint::Sketches => self.sketches_endpoint.clone(),
         }
     }
@@ -169,8 +207,14 @@ impl DatadogMetricsConfig {
         &self,
     ) -> crate::Result<DatadogMetricsEndpointConfiguration> {
         let base_uri = self.get_base_agent_endpoint();
-        let series_endpoint = build_uri(&base_uri, "/api/v1/series")?;
-        let sketches_endpoint = build_uri(&base_uri, "/api/beta/sketches")?;
+
+        // TODO: the V1 endpoint support is considered deprecated and should be removed in a future release.
+        // At that time, the get_api_version_backwards_compatible() should be replaced with statically using the v2.
+        let series_endpoint = build_uri(
+            &base_uri,
+            SeriesApiVersion::get_api_version_backwards_compatible().get_path(),
+        )?;
+        let sketches_endpoint = build_uri(&base_uri, SKETCHES_PATH)?;
 
         Ok(DatadogMetricsEndpointConfiguration::new(
             series_endpoint,