fix(config): migrate endpoints to typed config for provenance awareness

.claude/skills/config-system/SKILL.md

@@ -58,7 +58,11 @@ into `SalukiConfiguration`.
 
 In `SalukiConfiguration`, you may use `ConfigValue<T>` when we need to know the difference between a
 value that was explicitly set by the user, or where the value is a default. The Agent API provides a
-`source` field which we are simplifying into `Provinence`, which is either `Default` or `Explicit`.
+`source` field which we are simplifying into `Provenance`, which is either `Default` or `Explicit`.
+
+Some values must be resolved based on their provenance, for example `dd_url` overrides `site` only
+if it is explicitly set. This sort of resolution should be done in the config layer with a member
+function getter on the relevant struct. The component should store the resolved value.
 
 Paths and type names can move. Notify the user when this skill needs an update.
 
@@ -239,7 +243,9 @@ the witnessed model.
     affected configurations.
 
 A cutover should be behaviorally transparent. If the old behavior conflicts with the source schema
-or typed-system invariants, surface the conflict rather than silently choosing one.
+or typed-system invariants, surface the conflict rather than silently choosing one. After cutover
+artifacts of deserialization logic should not be left behind. For example, no `derive(Deserialize)`
+and if possible a held `Option<SomeType>` should collapse to a held `SomeType` if possible.
 
 ## Review checklist
 

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

@@ -4,7 +4,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use agent_data_plane_config::shared::{Endpoints, MetricsEncoding};
+use agent_data_plane_config::{domains::multi_region_failover, shared::SharedConfiguration};
 use agent_data_plane_config_system::{ConfigurationSystem, LoadedConfiguration};
 use argh::FromArgs;
 use datadog_agent_commons::platform::PlatformSettings;
@@ -359,8 +359,7 @@ async fn create_topology(
     let dp = DataPlaneConfiguration::from_configuration(&config);
     let mut blueprint = TopologyBlueprint::new("primary", component_registry);
     blueprint.with_shutdown_timeout(dp.stop_timeout());
-    let metrics_encoding = config.shared.metrics_encoding.clone();
-    let endpoints = config.shared.endpoints.clone();
+    let shared = config.shared.clone();
 
     let mut control_surfaces = TopologyControlSurfaces::default();
 
@@ -385,24 +384,12 @@ async fn create_topology(
         || dp.service_checks_pipeline_required()
         || dp.traces_pipeline_required()
     {
-        let dd_forwarder_config = DatadogForwarderConfiguration::from_configuration_with_metrics_routing(
-            &config_system.raw_map(),
-            &metrics_encoding,
-            &endpoints,
-        )
-        .error_context("Failed to configure Datadog forwarder.")?;
+        let dd_forwarder_config = DatadogForwarderConfiguration::from_configuration(&shared, &config_system.raw_map());
         blueprint.add_forwarder("dd_out", dd_forwarder_config)?;
     }
 
     if dp.metrics_pipeline_required() {
-        add_baseline_metrics_pipeline_to_blueprint(
-            &mut blueprint,
-            config_system,
-            &metrics_encoding,
-            &endpoints,
-            env_provider,
-        )
-        .await?;
+        add_baseline_metrics_pipeline_to_blueprint(&mut blueprint, config_system, &shared, env_provider).await?;
     }
 
     if dp.logs_pipeline_required() {
@@ -483,8 +470,8 @@ async fn add_checks_pipeline_to_blueprint(
 }
 
 async fn add_baseline_metrics_pipeline_to_blueprint(
-    blueprint: &mut TopologyBlueprint, config_system: &ConfigurationSystem, metrics: &MetricsEncoding,
-    endpoints: &Endpoints, env_provider: &ADPEnvironmentProvider,
+    blueprint: &mut TopologyBlueprint, config_system: &ConfigurationSystem, shared: &SharedConfiguration,
+    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());
@@ -500,12 +487,7 @@ async fn add_baseline_metrics_pipeline_to_blueprint(
         }
     }
 
-    let dd_metrics_config = DatadogMetricsConfiguration::from_configuration_with_metrics_routing(
-        &config_system.raw_map(),
-        metrics,
-        endpoints,
-    )
-    .error_context("Failed to configure Datadog Metrics encoder.")?;
+    let dd_metrics_config = DatadogMetricsConfiguration::from_configuration(shared);
 
     blueprint
         // Components.
@@ -514,17 +496,22 @@ async fn add_baseline_metrics_pipeline_to_blueprint(
         // Metrics, then forwarding.
         .connect_components_in_order(["metrics_enrich", "dd_metrics_encode", "dd_out"])?;
 
-    add_mrf_metrics_pipeline_to_blueprint(blueprint, &config_system.raw_map(), metrics, endpoints)?;
-    add_autoscaling_failover_metrics_pipeline_to_blueprint(blueprint, &config_system.raw_map(), metrics, endpoints)?;
+    add_mrf_metrics_pipeline_to_blueprint(
+        blueprint,
+        &config_system.raw_map(),
+        shared,
+        &config.domains.multi_region_failover,
+    )?;
+    add_autoscaling_failover_metrics_pipeline_to_blueprint(blueprint, &config_system.raw_map(), shared)?;
 
     Ok(())
 }
 
 fn add_mrf_metrics_pipeline_to_blueprint(
-    blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, metrics: &MetricsEncoding, endpoints: &Endpoints,
+    blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, shared: &SharedConfiguration,
+    mrf: &multi_region_failover::Domain,
 ) -> Result<(), GenericError> {
-    let mrf_config = MrfConfiguration::from_configuration(config)
-        .error_context("Failed to configure Multi-Region Failover metrics pipeline.")?;
+    let mrf_config = MrfConfiguration::from_configuration(mrf);
 
     let Some((mrf_dd_url, mrf_api_key)) = mrf_config.metrics_endpoint_override() else {
         if mrf_config.is_enabled() {
@@ -541,20 +528,15 @@ 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_with_metrics_routing(config, metrics, endpoints)
-            .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_with_metrics_routing(config, metrics, endpoints)
-            .map(|config| {
-                config.with_endpoint_override_and_api_key_refresh_config_path(
-                    mrf_dd_url,
-                    mrf_api_key,
-                    "multi_region_failover.api_key",
-                )
-            })
-            .error_context("Failed to configure Multi-Region Failover Datadog forwarder.")?;
+        DatadogMetricsConfiguration::from_configuration(shared).with_metrics_endpoint_override(mrf_dd_url.clone());
+
+    let mrf_forwarder_config = DatadogForwarderConfiguration::for_endpoint_override(
+        shared,
+        config,
+        mrf_dd_url,
+        mrf_api_key,
+        "multi_region_failover.api_key",
+    );
 
     blueprint
         .add_transform("mrf_metrics_gateway", mrf_gateway_config)?
@@ -571,7 +553,7 @@ fn add_mrf_metrics_pipeline_to_blueprint(
 }
 
 fn add_autoscaling_failover_metrics_pipeline_to_blueprint(
-    blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, metrics: &MetricsEncoding, endpoints: &Endpoints,
+    blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, shared: &SharedConfiguration,
 ) -> Result<(), GenericError> {
     let af_config = AutoscalingFailoverConfiguration::from_configuration(config)
         .error_context("Failed to configure autoscaling failover metrics pipeline.")?;
@@ -596,14 +578,10 @@ fn add_autoscaling_failover_metrics_pipeline_to_blueprint(
     }
 
     let af_gateway_config = AutoscalingFailoverGatewayConfiguration::new(af_config);
-    let af_metrics_config =
-        DatadogMetricsConfiguration::from_configuration_with_metrics_routing(config, metrics, endpoints)
-            .error_context("Failed to configure autoscaling failover metrics encoder.")?
-            .with_v2_series_only();
-    let cluster_agent_forwarder_config = ClusterAgentForwarderConfiguration::from_configuration_with_metrics_routing(
-        config, metrics, endpoints, ca_url, ca_token,
-    )
-    .error_context("Failed to configure Cluster Agent forwarder.")?;
+    let af_metrics_config = DatadogMetricsConfiguration::from_configuration(shared).with_v2_series_only();
+    let cluster_agent_forwarder_config =
+        ClusterAgentForwarderConfiguration::from_configuration(shared, config, ca_url, ca_token)
+            .error_context("Failed to configure Cluster Agent forwarder.")?;
 
     blueprint
         .add_transform("af_metrics_gateway", af_gateway_config)?

lib/agent-data-plane-config-system/src/loaded.rs

@@ -18,7 +18,7 @@ use tokio::sync::mpsc;
 use crate::env_provider::EnvironmentProvider;
 use crate::saluki_env_overlay;
 use crate::source::SourceTree;
-use crate::system::{translate_strict, ConfigurationSystem, Error};
+use crate::system::{translate_strict, validate, ConfigurationSystem, Error};
 
 // The environment-variable prefix ADP reads (`DD_`). Mirrors
 // `PlatformSettings::get_env_var_prefix()`; hardcoded so the configuration system need not depend on
@@ -62,6 +62,11 @@ pub struct LoadedConfiguration {
 impl LoadedConfiguration {
     /// Loads and strictly translates the local file and environment using the requested precedence.
     ///
+    /// The snapshot is translated but not validated: it is not yet authoritative. Under the Datadog
+    /// Agent the stream still has settings to contribute, so a local snapshot that could not run on
+    /// its own is normal here. Whichever authority [`run`](Self::run) or
+    /// [`standalone`](Self::standalone) selects applies validation.
+    ///
     /// # Errors
     ///
     /// Returns an error if a local source cannot be read, decoded, deserialized, or translated.
@@ -107,12 +112,16 @@ impl LoadedConfiguration {
 
     /// Uses the translated local configuration as the runtime authority.
     ///
-    /// No configuration stream or update task is created.
+    /// No configuration stream or update task is created. Because nothing further will be layered on,
+    /// the local snapshot is validated here, where [`run`](Self::run) instead validates the merged
+    /// result of the Agent's initial snapshot.
     ///
     /// # Errors
     ///
-    /// Returns an error if the compatibility map cannot be built from the local sources.
+    /// Returns an error if the compatibility map cannot be built from the local sources, or if the
+    /// local configuration fails validation.
     pub async fn standalone(self) -> Result<ConfigurationSystem, Error> {
+        validate(&self.local)?;
         let compat_map = self.loader.into_generic().await?;
         Ok(ConfigurationSystem::standalone(compat_map, self.local))
     }
@@ -257,6 +266,26 @@ mod tests {
         assert!(matches!(result, Err(Error::Translate { .. })));
     }
 
+    #[tokio::test]
+    async fn load_leaves_an_incomplete_local_snapshot_to_the_selected_authority() {
+        // A local snapshot with no API key is normal: under the Datadog Agent the key arrives over the
+        // configuration stream, and the CLI subcommands read this snapshot without submitting
+        // anything. Validation therefore belongs to whichever authority the caller then selects.
+        let path = std::env::temp_dir().join(format!("adp_no_api_key_{}.yaml", std::process::id()));
+        std::fs::write(&path, "log_level: warn\n").unwrap();
+
+        let loaded = LoadedConfiguration::load(&path, EnvPrecedence::Disabled)
+            .await
+            .expect("a local snapshot without an API key loads");
+        assert_eq!("", loaded.local().shared.endpoints.api_key);
+
+        // Standalone mode makes that same snapshot authoritative, so the missing key is fatal there.
+        let result = loaded.standalone().await;
+
+        std::fs::remove_file(&path).ok();
+        assert!(matches!(result, Err(Error::MissingApiKey)));
+    }
+
     // `LoadedConfiguration::load` is `async` only for symmetry with the rest of the API; it awaits
     // nothing. The environment tests below drive it on a local runtime rather than with
     // `#[tokio::test]`, so the blocking environment guard is never held across an await point.