chore(config): migrate autoscaling failover to typed config

Build the autoscaling failover component from its translated shared configuration and keep source parsing at the configuration boundary.

(cherry picked from commit 80c8c20)

Author: webern

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

@@ -410,7 +410,8 @@ async fn create_topology(
     }
 
     if dp_config.metrics_pipeline_required() {
-        add_baseline_metrics_pipeline_to_blueprint(&mut blueprint, config, dp_config, env_provider).await?;
+        add_baseline_metrics_pipeline_to_blueprint(&mut blueprint, config, config_system, dp_config, env_provider)
+            .await?;
     }
 
     if dp_config.logs_pipeline_required() {
@@ -464,8 +465,8 @@ async fn add_checks_pipeline_to_blueprint(
 }
 
 async fn add_baseline_metrics_pipeline_to_blueprint(
-    blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, dp_config: &DataPlaneConfiguration,
-    env_provider: &ADPEnvironmentProvider,
+    blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, config_system: &ConfigurationSystem,
+    dp_config: &DataPlaneConfiguration, env_provider: &ADPEnvironmentProvider,
 ) -> Result<(), GenericError> {
     // Create the back half of the metrics processing pipeline.
     let host_enrichment_config = HostEnrichmentConfiguration::from_environment_provider(env_provider.clone());
@@ -490,7 +491,8 @@ async fn add_baseline_metrics_pipeline_to_blueprint(
         .connect_components_in_order(["metrics_enrich", "dd_metrics_encode", "dd_out"])?;
 
     add_mrf_metrics_pipeline_to_blueprint(blueprint, config)?;
-    add_autoscaling_failover_metrics_pipeline_to_blueprint(blueprint, config)?;
+    let saluki = config_system.config();
+    add_autoscaling_failover_metrics_pipeline_to_blueprint(blueprint, config, &saluki.shared.autoscaling_failover)?;
 
     Ok(())
 }
@@ -544,8 +546,9 @@ fn add_mrf_metrics_pipeline_to_blueprint(
 
 fn add_autoscaling_failover_metrics_pipeline_to_blueprint(
     blueprint: &mut TopologyBlueprint, config: &GenericConfiguration,
+    autoscaling_failover: &agent_data_plane_config::shared::AutoscalingFailover,
 ) -> Result<(), GenericError> {
-    let af_config = AutoscalingFailoverConfiguration::from_configuration(config)
+    let af_config = AutoscalingFailoverConfiguration::from_configuration(autoscaling_failover)
         .error_context("Failed to configure autoscaling failover metrics pipeline.")?;
     let ca_config = ClusterAgentConfiguration::from_configuration(config)
         .error_context("Failed to configure Cluster Agent metrics forwarding.")?;

lib/saluki-components/src/config/autoscaling_failover.rs

@@ -1,12 +1,8 @@
 //! Autoscaling failover configuration.
 
-use saluki_config::GenericConfiguration;
+use agent_data_plane_config::shared::AutoscalingFailover;
 use saluki_error::GenericError;
 
-fn default_metrics() -> Vec<String> {
-    vec!["container.memory.usage".to_string(), "container.cpu.usage".to_string()]
-}
-
 /// Autoscaling failover configuration for the metrics pipeline.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct AutoscalingFailoverConfiguration {
@@ -15,13 +11,11 @@ pub struct AutoscalingFailoverConfiguration {
 }
 
 impl AutoscalingFailoverConfiguration {
-    /// Creates a new `AutoscalingFailoverConfiguration` from the given configuration.
-    pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
+    /// Builds an `AutoscalingFailoverConfiguration` from the translated failover settings.
+    pub fn from_configuration(config: &AutoscalingFailover) -> Result<Self, GenericError> {
         Ok(Self {
-            enabled: config.try_get_typed("autoscaling.failover.enabled")?.unwrap_or(false),
-            metrics: config
-                .try_get_typed("autoscaling.failover.metrics")?
-                .unwrap_or_else(default_metrics),
+            enabled: config.enabled,
+            metrics: config.metrics.clone(),
         })
     }
 
@@ -35,60 +29,3 @@ impl AutoscalingFailoverConfiguration {
         &self.metrics
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use saluki_config::ConfigurationLoader;
-    use serde_json::json;
-
-    use super::*;
-
-    async fn autoscaling_config_from(value: serde_json::Value) -> AutoscalingFailoverConfiguration {
-        let (config, _) = ConfigurationLoader::for_tests(Some(value), None, false).await;
-        AutoscalingFailoverConfiguration::from_configuration(&config)
-            .expect("autoscaling failover configuration should deserialize")
-    }
-
-    #[tokio::test]
-    async fn defaults_to_disabled_with_default_metric_allowlist() {
-        let config = autoscaling_config_from(json!({})).await;
-
-        assert!(!config.is_branch_requested());
-        assert_eq!(
-            config.metrics(),
-            ["container.memory.usage".to_string(), "container.cpu.usage".to_string()]
-        );
-    }
-
-    #[tokio::test]
-    async fn branch_is_requested_when_enabled_with_non_empty_metrics() {
-        let config = autoscaling_config_from(json!({
-            "autoscaling": {
-                "failover": {
-                    "enabled": true,
-                    "metrics": ["custom.metric"]
-                }
-            }
-        }))
-        .await;
-
-        assert!(config.is_branch_requested());
-        assert_eq!(config.metrics(), ["custom.metric".to_string()]);
-    }
-
-    #[tokio::test]
-    async fn empty_metric_allowlist_disables_branch() {
-        let config = autoscaling_config_from(json!({
-            "autoscaling": {
-                "failover": {
-                    "enabled": true,
-                    "metrics": []
-                }
-            }
-        }))
-        .await;
-
-        assert!(!config.is_branch_requested());
-        assert!(config.metrics().is_empty());
-    }
-}

lib/saluki-components/src/transforms/autoscaling_failover_gateway/mod.rs

@@ -169,60 +169,40 @@ impl Transform for AutoscalingFailoverGateway {
 mod tests {
     use std::time::Duration;
 
-    use saluki_config::ConfigurationLoader;
+    use agent_data_plane_config::shared::AutoscalingFailover;
     use saluki_core::data_model::event::{metric::Metric, Event};
-    use serde_json::json;
 
     use super::*;
 
-    async fn gateway_from_config(value: serde_json::Value) -> AutoscalingFailoverGateway {
-        let (config, _) = ConfigurationLoader::for_tests(Some(value), None, false).await;
-        let failover_config = AutoscalingFailoverConfiguration::from_configuration(&config)
-            .expect("autoscaling failover configuration should deserialize");
+    fn gateway_from_config(enabled: bool, metrics: Vec<&str>) -> AutoscalingFailoverGateway {
+        let failover_config = AutoscalingFailoverConfiguration::from_configuration(&AutoscalingFailover {
+            enabled,
+            metrics: metrics.into_iter().map(String::from).collect(),
+        })
+        .expect("autoscaling failover configuration should build");
         AutoscalingFailoverGateway::new(failover_config)
     }
 
-    #[tokio::test]
-    async fn inactive_gateway_drops_everything() {
-        let gw = gateway_from_config(json!({
-            "autoscaling": {
-                "failover": {
-                    "enabled": false,
-                    "metrics": ["allowed.metric"]
-                }
-            }
-        }))
-        .await;
+    #[test]
+    fn inactive_gateway_drops_everything() {
+        let gw = gateway_from_config(false, vec!["allowed.metric"]);
 
         assert!(!gw.should_forward(&Event::Metric(Metric::counter("allowed.metric", 1.0))));
     }
 
-    #[tokio::test]
-    async fn empty_metric_allowlist_drops_everything() {
-        let gw = gateway_from_config(json!({
-            "autoscaling": {
-                "failover": {
-                    "enabled": true,
-                    "metrics": []
-                }
-            }
-        }))
-        .await;
+    #[test]
+    fn empty_metric_allowlist_drops_everything() {
+        let gw = gateway_from_config(true, vec![]);
 
         assert!(!gw.should_forward(&Event::Metric(Metric::counter("allowed.metric", 1.0))));
     }
 
-    #[tokio::test]
-    async fn active_gateway_forwards_only_allowed_series_metrics() {
-        let gw = gateway_from_config(json!({
-            "autoscaling": {
-                "failover": {
-                    "enabled": true,
-                    "metrics": ["allowed.counter", "allowed.gauge", "allowed.rate", "allowed.set"]
-                }
-            }
-        }))
-        .await;
+    #[test]
+    fn active_gateway_forwards_only_allowed_series_metrics() {
+        let gw = gateway_from_config(
+            true,
+            vec!["allowed.counter", "allowed.gauge", "allowed.rate", "allowed.set"],
+        );
 
         assert!(gw.should_forward(&Event::Metric(Metric::counter("allowed.counter", 1.0))));
         assert!(gw.should_forward(&Event::Metric(Metric::gauge("allowed.gauge", 1.0))));
@@ -235,17 +215,9 @@ mod tests {
         assert!(!gw.should_forward(&Event::Metric(Metric::counter("blocked.counter", 1.0))));
     }
 
-    #[tokio::test]
-    async fn active_gateway_drops_sketch_metrics_even_when_allowed() {
-        let gw = gateway_from_config(json!({
-            "autoscaling": {
-                "failover": {
-                    "enabled": true,
-                    "metrics": ["allowed.histogram", "allowed.distribution"]
-                }
-            }
-        }))
-        .await;
+    #[test]
+    fn active_gateway_drops_sketch_metrics_even_when_allowed() {
+        let gw = gateway_from_config(true, vec!["allowed.histogram", "allowed.distribution"]);
 
         assert!(!gw.should_forward(&Event::Metric(Metric::histogram("allowed.histogram", [1.0, 2.0, 3.0]))));
         assert!(!gw.should_forward(&Event::Metric(Metric::distribution(