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

## Human Summary

This migrates all components affected by #1965 (we were unable to detect
explicitly set to default vs default config values) to typed config to
pick up the fix provided in #2279.

After implementation I ran a clean-room Opus audit with this prompt:

> please do a clean-room audit of this commit
`f4ae503e807b398ca55ee3f6275f554889cff645`
>
> Our goal is to correct a previous incorrect behavior in which we were
not sensitive to whether a configuration value was set by the Agent as a
default or whether the customer explicitly set the value (and happened
to set it to the default value).
>
> Other than that specific intentional bug fix, no other behavioral
changes are intended.
>
> Validate that only the desired behavioral change has occurred and that
configuration defaults are in-tact.
>
> Read AGENTS.md and /config-system

It only found two actionable items, which I fixed. One was that we were
previously rejecting empty API keys, so I retained that behavior. The
other was that an MRF endpoint consisting only of whitespace could be
treated as a valid setting. Also fixed. All defaults stood up to
scrutiny.

Edit: @aqian01 found a flaw that went undetected which became #2325.

### Review Guide

I tried breaking this up, which is why there are a bunch of subissues
being closed all at once, but it did not divide very well. As such, the
diff a bit large:

Focus on these files:
- `lib/saluki-components/*`
- `bin/agent-data-plane/src/cli/run.rs`
- `lib/agent-data-plane-config*`

Deprioritize generated code and inventory churn in:
- `lib/datadog-agent/*`


## AI Summary

Migrate Datadog endpoint, retry, proxy, forwarder, metrics encoder, MRF,
and Cluster Agent configuration consumers from raw
`GenericConfiguration` deserialization to resolved typed configuration.

Primary endpoint resolution now happens once in the typed configuration
layer:

- A default-sourced `dd_url` no longer shadows `site`.
- An explicitly configured `dd_url`, including the schema-default URL,
remains an override.
- MRF and Cluster Agent destination overrides cannot be overwritten by
global endpoint settings.
- Retry queue size precedence uses configuration provenance, so explicit
zero values are preserved.
- Raw configuration access remains only for live API-key refresh,
secrets retry behavior, and the `run_path` compatibility fallback whose
schema default is still an unresolved placeholder.

The schema inventories, generated registries, smoke-test metadata, and
component tests are updated for the typed consumers.

## Change Type

- [x] Non-functional (chore, refactoring, docs)

## How did you test this PR?

- unit tests updated and created throughout
- integration test modified to test this behavior. this was proven to
have caught the original bug (`red -> green`)

## References

- Closes: #1965
- Closes: #2310
- Closes: #2312
- Closes: #2313
- Closes: #2314
- Closes: #2315
- Closes: #2316
- Related to: #2279

Author: webern

.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
 

Cargo.lock

@@ -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;
@@ -394,8 +394,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();
 
@@ -420,24 +419,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() {
@@ -518,8 +505,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());
@@ -535,12 +522,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.
@@ -549,17 +531,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() {
@@ -576,20 +563,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)?
@@ -606,7 +588,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.")?;
@@ -631,14 +613,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)?

bin/agent-data-plane/src/cli/run.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.