chore(config): integrate and test the config system

bin/agent-data-plane/Cargo.toml

@@ -14,7 +14,10 @@ fips = ["saluki-app/tls-fips", "saluki-components/fips"]
 antithesis = ["saluki-antithesis/antithesis", "dep:antithesis-instrumentation"]
 
 [dependencies]
+agent-data-plane-config = { workspace = true }
+agent-data-plane-config-system = { workspace = true }
 antithesis-instrumentation = { workspace = true, optional = true }
+arc-swap = { workspace = true }
 argh = { workspace = true, features = ["help"] }
 async-trait = { workspace = true }
 bytesize = { workspace = true }

bin/agent-data-plane/src/components/dogstatsd_prefix_filter/mod.rs

@@ -703,9 +703,9 @@ mod tests {
 
 #[cfg(test)]
 mod config_smoke {
+    use datadog_agent_config::{DatadogRemapper, KEY_ALIASES};
     use datadog_agent_config_testing::config_registry::structs;
     use datadog_agent_config_testing::run_config_smoke_tests;
-    use saluki_components::config::{DatadogRemapper, KEY_ALIASES};
     use serde_json::json;
 
     use super::DogStatsDPrefixFilterConfiguration;

bin/agent-data-plane/src/config.rs

@@ -9,8 +9,6 @@ use saluki_io::net::ListenAddress;
 pub struct DataPlaneConfiguration {
     enabled: bool,
     standalone_mode: bool,
-    use_new_config_stream_endpoint: bool,
-    remote_agent_enabled: bool,
     stop_timeout: Duration,
     api_listen_address: ListenAddress,
     secure_api_listen_address: ListenAddress,
@@ -36,10 +34,6 @@ impl DataPlaneConfiguration {
         Ok(Self {
             enabled: config.try_get_typed("data_plane.enabled")?.unwrap_or(false),
             standalone_mode: config.try_get_typed("data_plane.standalone_mode")?.unwrap_or(false),
-            use_new_config_stream_endpoint: config
-                .try_get_typed("data_plane.use_new_config_stream_endpoint")?
-                .unwrap_or(true),
-            remote_agent_enabled: config.try_get_typed("data_plane.remote_agent_enabled")?.unwrap_or(true),
             stop_timeout: topology_stop_timeout_from_configuration(config)?,
             api_listen_address: config
                 .try_get_typed("data_plane.api_listen_address")?
@@ -63,16 +57,6 @@ impl DataPlaneConfiguration {
         self.standalone_mode
     }
 
-    /// Returns `true` if the new config stream endpoint should be used.
-    pub const fn use_new_config_stream_endpoint(&self) -> bool {
-        self.use_new_config_stream_endpoint
-    }
-
-    /// Returns `true` if the data plane should register as a remote agent.
-    pub const fn remote_agent_enabled(&self) -> bool {
-        self.remote_agent_enabled
-    }
-
     /// Returns the topology shutdown timeout.
     pub const fn stop_timeout(&self) -> Duration {
         self.stop_timeout

bin/agent-data-plane/src/internal/config_internal.rs

@@ -0,0 +1,103 @@
+//! Internal configuration API handler.
+
+use std::sync::Arc;
+
+use agent_data_plane_config::SalukiConfiguration;
+use arc_swap::ArcSwap;
+use async_trait::async_trait;
+use http::StatusCode;
+use saluki_api::{
+    extract::State,
+    response::IntoResponse,
+    routing::{get, Router},
+    APIHandler, DynamicRoute, EndpointType,
+};
+use saluki_common::sync::shutdown::ShutdownHandle;
+use saluki_core::runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture};
+use saluki_error::generic_error;
+
+/// State used for the internal configuration API handler.
+#[derive(Clone)]
+pub struct ConfigInternalState {
+    current: Arc<ArcSwap<SalukiConfiguration>>,
+}
+
+/// An API handler for returning the translated runtime configuration.
+///
+/// This handler exposes a single route -- `/config/internal` -- that returns the current
+/// translated [`SalukiConfiguration`] in its serialized JSON form. It reflects any dynamic updates
+/// applied to the configuration since startup.
+pub struct ConfigInternalAPIHandler {
+    state: ConfigInternalState,
+}
+
+impl ConfigInternalAPIHandler {
+    fn new(current: Arc<ArcSwap<SalukiConfiguration>>) -> Self {
+        Self {
+            state: ConfigInternalState { current },
+        }
+    }
+
+    async fn config_handler(State(state): State<ConfigInternalState>) -> impl IntoResponse {
+        let config = state.current.load();
+        match serde_json::to_string(&**config) {
+            Ok(body) => (StatusCode::OK, body).into_response(),
+            Err(e) => (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                format!("Failed to serialize configuration: {}", e),
+            )
+                .into_response(),
+        }
+    }
+}
+
+impl APIHandler for ConfigInternalAPIHandler {
+    type State = ConfigInternalState;
+
+    fn generate_initial_state(&self) -> Self::State {
+        self.state.clone()
+    }
+
+    fn generate_routes(&self) -> Router<Self::State> {
+        Router::new().route("/config/internal", get(Self::config_handler))
+    }
+}
+
+/// A worker for exposing an endpoint that returns the translated runtime configuration.
+///
+/// When running, the worker asserts a route (based on [`ConfigInternalAPIHandler`]) that returns
+/// the current translated configuration. As the configuration may contain sensitive data, this
+/// route is only present on the privileged API endpoint.
+pub struct ConfigInternalWorker {
+    handler: ConfigInternalAPIHandler,
+}
+
+impl ConfigInternalWorker {
+    /// Creates a new [`ConfigInternalWorker`] serving the given configuration.
+    pub fn new(current: Arc<ArcSwap<SalukiConfiguration>>) -> Self {
+        Self {
+            handler: ConfigInternalAPIHandler::new(current),
+        }
+    }
+}
+
+#[async_trait]
+impl Supervisable for ConfigInternalWorker {
+    fn name(&self) -> &str {
+        "config-internal-api"
+    }
+
+    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
+        let config_route = DynamicRoute::http(EndpointType::Privileged, &self.handler);
+
+        Ok(Box::pin(async move {
+            let dataspace =
+                DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
+
+            dataspace.assert(config_route, "config-internal-api");
+
+            process_shutdown.await;
+            Ok(())
+        }))
+    }
+}

bin/agent-data-plane/src/internal/control_plane.rs

@@ -1,10 +1,14 @@
+use std::sync::Arc;
+
+use agent_data_plane_config::SalukiConfiguration;
+use agent_data_plane_config_system::ConfigurationSystem;
+use arc_swap::ArcSwap;
 use datadog_agent_commons::ipc::{config::IpcAuthConfiguration, tls::build_ipc_server_tls_config};
 use saluki_api::EndpointType;
 use saluki_app::{
     accounting::ResourceTelemetryWorker, config::ConfigWorker, dynamic_api::DynamicAPIBuilder,
     logging::LoggingOverrideController,
 };
-use saluki_config::GenericConfiguration;
 use saluki_core::accounting::ComponentRegistry;
 use saluki_core::{
     health::HealthRegistry,
@@ -15,8 +19,8 @@ use saluki_error::GenericError;
 use crate::{
     config::DataPlaneConfiguration,
     internal::{
-        logging::DynamicLogLevelWorker, remote_agent::RemoteAgentBootstrap, telemetry::InternalTelemetryAPIWorker,
-        TopologyControlSurfaces,
+        config_internal::ConfigInternalWorker, logging::DynamicLogLevelWorker, remote_agent::RemoteAgentBootstrap,
+        telemetry::InternalTelemetryAPIWorker, TopologyControlSurfaces,
     },
 };
 
@@ -31,9 +35,10 @@ use crate::{
 ///
 /// If the supervisor can't be created, an error is returned.
 pub async fn create_control_plane_supervisor(
-    config: &GenericConfiguration, dp_config: &DataPlaneConfiguration, component_registry: &ComponentRegistry,
+    config: &ConfigurationSystem, dp_config: &DataPlaneConfiguration, component_registry: &ComponentRegistry,
     health_registry: HealthRegistry, control_surfaces: TopologyControlSurfaces,
     ra_bootstrap: Option<RemoteAgentBootstrap>, logging_controller: LoggingOverrideController,
+    current_config: Arc<ArcSwap<SalukiConfiguration>>,
 ) -> Result<Supervisor, GenericError> {
     let mut supervisor = Supervisor::new("ctrl-pln")?
         .with_dedicated_runtime(RuntimeConfiguration::single_threaded())
@@ -42,14 +47,15 @@ pub async fn create_control_plane_supervisor(
     supervisor.add_worker(health_registry.worker());
     supervisor.add_worker(ResourceTelemetryWorker::new(component_registry));
     supervisor.add_worker(InternalTelemetryAPIWorker::new());
-    supervisor.add_worker(DynamicLogLevelWorker::new(config, logging_controller));
-    supervisor.add_worker(ConfigWorker::new(config.clone()));
+    supervisor.add_worker(DynamicLogLevelWorker::new(&config.raw_map(), logging_controller));
+    supervisor.add_worker(ConfigWorker::new(config.raw_map()));
+    supervisor.add_worker(ConfigInternalWorker::new(current_config));
 
     supervisor.add_worker(DynamicAPIBuilder::new(
         EndpointType::Unprivileged,
         dp_config.api_listen_address().clone(),
     ));
-    let ipc_config = IpcAuthConfiguration::from_configuration(config)?;
+    let ipc_config = IpcAuthConfiguration::from_configuration(&config.raw_map())?;
     let tls_config = build_ipc_server_tls_config(ipc_config.ipc_cert_file_path()).await?;
 
     let mut privileged_api =

bin/agent-data-plane/src/internal/env/mod.rs

@@ -1,6 +1,6 @@
 use std::future::Future;
 
-use saluki_config::GenericConfiguration;
+use agent_data_plane_config_system::ConfigurationSystem;
 use saluki_core::accounting::ComponentRegistry;
 use saluki_core::health::HealthRegistry;
 use saluki_core::runtime::Supervisor;
@@ -56,15 +56,17 @@ impl ADPEnvironmentProvider {
     /// In standalone mode, no supervisor is returned as all behavior/functionality is either provided via
     /// fixed configuration or operates in a no-op fashion.
     pub async fn from_configuration(
-        config: &GenericConfiguration, dp_config: &DataPlaneConfiguration, component_registry: &ComponentRegistry,
+        config: &ConfigurationSystem, dp_config: &DataPlaneConfiguration, component_registry: &ComponentRegistry,
         health_registry: &HealthRegistry,
     ) -> Result<(Self, Option<Supervisor>), GenericError> {
         // When we're in standalone mode, all of our functionality is either fixed or a no-op.
         if dp_config.standalone_mode() {
             warn!("Running in standalone mode. Origin detection/enrichment and other features dependent upon the Datadog Agent will not be available.");
 
             let env = Self {
-                host_provider: BoxedHostProvider::from_provider(FixedHostProvider::from_configuration(config)?),
+                host_provider: BoxedHostProvider::from_provider(FixedHostProvider::from_configuration(
+                    &config.raw_map(),
+                )?),
                 workload_provider: None,
                 autodiscovery_provider: None,
                 health_registry: health_registry.clone(),
@@ -75,14 +77,15 @@ impl ADPEnvironmentProvider {
         // Otherwise, construct our real providers that will interact directly with the Datadog Agent.
         let mut env_supervisor = Supervisor::new("env-provider")?;
 
-        let host_provider = RemoteAgentHostProvider::from_configuration(config, component_registry).await?;
+        let host_provider = RemoteAgentHostProvider::from_configuration(&config.raw_map(), component_registry).await?;
 
         let (workload_provider, workload_supervisor) =
-            RemoteAgentWorkloadProvider::from_configuration(config, component_registry, health_registry).await?;
+            RemoteAgentWorkloadProvider::from_configuration(&config.raw_map(), component_registry, health_registry)
+                .await?;
         env_supervisor.add_worker(workload_supervisor);
 
         let (autodiscovery_provider, autodiscovery_supervisor) =
-            RemoteAgentAutodiscoveryProvider::from_configuration(config).await?;
+            RemoteAgentAutodiscoveryProvider::from_configuration(&config.raw_map()).await?;
         env_supervisor.add_worker(autodiscovery_supervisor);
 
         let env = Self {

bin/agent-data-plane/src/internal/mod.rs

@@ -1,12 +1,18 @@
+use std::sync::Arc;
+
+use agent_data_plane_config::SalukiConfiguration;
+use agent_data_plane_config_system::ConfigurationSystem;
+use arc_swap::ArcSwap;
 use saluki_app::logging::LoggingOverrideController;
-use saluki_config::GenericConfiguration;
 use saluki_core::accounting::ComponentRegistry;
 use saluki_core::health::HealthRegistry;
 use saluki_core::runtime::Supervisor;
 use saluki_error::GenericError;
 
 use crate::config::DataPlaneConfiguration;
 
+mod config_internal;
+
 mod control_plane;
 pub use self::control_plane::create_control_plane_supervisor;
 
@@ -35,9 +41,10 @@ mod telemetry;
 ///
 /// If the supervisor can't be created, an error is returned.
 pub async fn create_internal_supervisor(
-    config: &GenericConfiguration, dp_config: &DataPlaneConfiguration, component_registry: &ComponentRegistry,
+    config: &ConfigurationSystem, dp_config: &DataPlaneConfiguration, component_registry: &ComponentRegistry,
     health_registry: HealthRegistry, control_surfaces: TopologyControlSurfaces,
     ra_bootstrap: Option<RemoteAgentBootstrap>, logging_controller: LoggingOverrideController,
+    current_config: Arc<ArcSwap<SalukiConfiguration>>,
 ) -> Result<Supervisor, GenericError> {
     // The root supervisor runs in ambient mode (caller's runtime) since its children each have their own
     // dedicated runtimes. The default restart strategy (one-for-one, 1 restart per 5s) applies to the child
@@ -54,6 +61,7 @@ pub async fn create_internal_supervisor(
             control_surfaces,
             ra_bootstrap,
             logging_controller,
+            current_config,
         )
         .await?,
     );