fix(metrics): preserve v2 for overridden endpoints

Author: rayz

bin/agent-data-plane/src/cli/run.rs

@@ -518,7 +518,8 @@ fn add_mrf_metrics_pipeline_to_blueprint(
 
     let mrf_gateway_config = MrfMetricsGatewayConfiguration::new(mrf_config.clone(), config.clone());
     let mrf_metrics_config = DatadogMetricsConfiguration::from_configuration(config)
-        .error_context("Failed to configure Multi-Region Failover Datadog Metrics encoder.")?;
+        .error_context("Failed to configure Multi-Region Failover Datadog Metrics encoder.")?
+        .with_metrics_endpoint_override(mrf_dd_url.clone());
 
     let mrf_forwarder_config = DatadogForwarderConfiguration::from_configuration(config)
         .map(|config| {
@@ -571,7 +572,8 @@ fn add_autoscaling_failover_metrics_pipeline_to_blueprint(
 
     let af_gateway_config = AutoscalingFailoverGatewayConfiguration::new(af_config);
     let af_metrics_config = DatadogMetricsConfiguration::from_configuration(config)
-        .error_context("Failed to configure autoscaling failover metrics encoder.")?;
+        .error_context("Failed to configure autoscaling failover metrics encoder.")?
+        .with_v2_series_only();
     let cluster_agent_forwarder_config =
         ClusterAgentForwarderConfiguration::from_configuration(config, ca_url, ca_token)
             .error_context("Failed to configure Cluster Agent forwarder.")?;

lib/saluki-components/src/common/datadog/config.rs

@@ -395,6 +395,12 @@ impl ForwarderConfiguration {
         self.opw_metrics = OpwMetricsConfiguration::default();
     }
 
+    /// Forces series metrics routing to accept only V2 payloads.
+    pub(crate) fn force_v2_series(&mut self) {
+        self.data_plane_metrics_v3_series_enabled = false;
+        self.v3_api.series.shadow_sites.clear();
+    }
+
     /// Builds resolved endpoints with routing metadata.
     ///
     /// The normal primary and OPW metrics primary endpoints share the same dynamic API key source.

lib/saluki-components/src/encoders/datadog/metrics/mod.rs

@@ -181,16 +181,17 @@ fn selected_metrics_primary_endpoint<'a>(
     vector_use_v3_series: bool,
 ) -> Option<(&'a str, bool)> {
     if opw_enabled {
+        let opw_url = opw_url.trim();
         metrics_primary_url_can_resolve(opw_url).then_some((opw_url, opw_use_v3_series))
     } else if vector_enabled {
+        let vector_url = vector_url.trim();
         metrics_primary_url_can_resolve(vector_url).then_some((vector_url, vector_use_v3_series))
     } else {
         None
     }
 }
 
 fn metrics_primary_url_can_resolve(url: &str) -> bool {
-    let url = url.trim();
     if url.is_empty() {
         return false;
     }
@@ -403,10 +404,14 @@ pub struct DatadogMetricsConfiguration {
     vector_metrics_use_v3_api_series: bool,
 
     /// The Datadog site used to resolve the primary metrics endpoint.
+    ///
+    /// Defaults to `datadoghq.com`.
     #[serde(default = "default_site")]
     site: String,
 
     /// The optional explicit primary metrics endpoint.
+    ///
+    /// Defaults to unset, in which case `site` determines the endpoint.
     #[serde(default, alias = "url", deserialize_with = "deserialize_dd_url")]
     dd_url: Option<String>,
 
@@ -427,6 +432,27 @@ impl DatadogMetricsConfiguration {
         self
     }
 
+    /// Restricts endpoint-aware protocol selection to a single overridden metrics endpoint.
+    ///
+    /// This mirrors a forwarder branch that replaces the normal primary endpoint and removes additional and
+    /// OPW/Vector endpoints, such as Multi-Region Failover.
+    pub fn with_metrics_endpoint_override(mut self, dd_url: String) -> Self {
+        self.dd_url = Some(dd_url);
+        self.additional_endpoints = AdditionalEndpoints::default();
+        self.observability_pipelines_worker_metrics_enabled = false;
+        self.vector_metrics_enabled = false;
+        self
+    }
+
+    /// Forces series metrics to use V2 without producing V3 shadow payloads.
+    ///
+    /// This is used for local destinations that only accept the V2 series protocol, such as the Cluster Agent.
+    pub fn with_v2_series_only(mut self) -> Self {
+        self.data_plane_metrics_v3_series_enabled = false;
+        self.v3_api.series.shadow_sample_rate = 0.0;
+        self
+    }
+
     fn v3_payload_limits(&self) -> V3PayloadLimits {
         V3PayloadLimits::new(
             self.max_series_payload_size,
@@ -2057,6 +2083,52 @@ serializer_experimental_use_v3_api:
         assert!(!config.requires_v2_series(false).expect("endpoints should resolve"));
     }
 
+    #[test]
+    fn endpoint_override_uses_the_overridden_endpoint_protocol() {
+        let config = v3_series_config(
+            r#"
+dd_url: https://primary.example.com
+data_plane_metrics_v3_series_enabled: true
+use_v3_api_series_enabled: "false"
+serializer_experimental_use_v3_api:
+  series:
+    endpoints:
+      - https://primary.example.com
+      - https://v3-mrf.example.com
+"#,
+        );
+
+        let v2_mrf_config = config
+            .clone()
+            .with_metrics_endpoint_override("https://v2-mrf.example.com".to_string());
+        let v3_mrf_config = config.with_metrics_endpoint_override("https://v3-mrf.example.com".to_string());
+
+        assert!(v2_mrf_config
+            .requires_v2_series(false)
+            .expect("V2 MRF endpoint should resolve"));
+        assert!(!v3_mrf_config
+            .requires_v2_series(false)
+            .expect("V3 MRF endpoint should resolve"));
+    }
+
+    #[test]
+    fn v2_series_only_override_keeps_v2_and_disables_shadowing() {
+        let config = v3_series_config(
+            r#"
+data_plane_metrics_v3_series_enabled: true
+use_v3_api_series_enabled: "true"
+serializer_experimental_use_v3_api:
+  series:
+    shadow_sample_rate: 1.0
+"#,
+        )
+        .with_v2_series_only();
+
+        assert!(!config.data_plane_metrics_v3_series_enabled);
+        assert_eq!(0.0, config.v3_api.series.shadow_sample_rate);
+        assert!(config.requires_v2_series(false).expect("endpoint should resolve"));
+    }
+
     #[test]
     fn agent_default_v3_does_not_enable_opw_only_encoder_mode() {
         let series_config = UseV3ApiSeriesConfig::default();

lib/saluki-components/src/forwarders/cluster_agent/mod.rs

@@ -55,6 +55,7 @@ impl ClusterAgentForwarderConfiguration {
         endpoint.set_dd_url(endpoint_url);
         endpoint.set_api_key(auth_token);
         forwarder_config.clear_opw_metrics_endpoint();
+        forwarder_config.force_v2_series();
 
         Ok(Self {
             forwarder_config,
@@ -250,13 +251,32 @@ mod tests {
                         "enabled": true,
                         "url": "https://opw.example.com"
                     }
+                },
+                "data_plane_metrics_v3_series_enabled": true,
+                "use_v3_api": {
+                    "series": {
+                        "enabled": "true"
+                    }
+                },
+                "serializer_experimental_use_v3_api": {
+                    "series": {
+                        "shadow_sites": ["example.com"]
+                    }
                 }
             })),
             None,
             false,
         )
         .await;
 
+        let unmodified_forwarder =
+            ForwarderConfiguration::from_configuration(&config).expect("forwarder configuration should parse");
+        assert!(unmodified_forwarder.data_plane_metrics_v3_series_enabled());
+        assert_eq!(
+            &["example.com".to_string()],
+            unmodified_forwarder.v3_api().series.shadow_sites.as_slice()
+        );
+
         let config = ClusterAgentForwarderConfiguration::from_configuration(
             &config,
             "https://cluster-agent.example.com".to_string(),
@@ -275,5 +295,7 @@ mod tests {
             "https://cluster-agent.example.com/"
         );
         assert_eq!(endpoints[0].endpoint().cached_api_key(), "secret-token");
+        assert!(!config.forwarder_config.data_plane_metrics_v3_series_enabled());
+        assert!(config.forwarder_config.v3_api().series.shadow_sites.is_empty());
     }
 }