apollo_config_manager: implement config updates from runner to manager (#9349)

Author: nadin-Starkware

Cargo.lock

@@ -14,6 +14,7 @@ workspace = true
 [dependencies]
 apollo_config_manager_config.workspace = true
 apollo_config_manager_types.workspace = true
+apollo_consensus_config.workspace = true
 apollo_infra.workspace = true
 apollo_metrics.workspace = true
 apollo_node_config.workspace = true

crates/apollo_config_manager/Cargo.toml

@@ -1,57 +1,41 @@
-use std::sync::Arc;
-
 use apollo_config_manager_config::config::ConfigManagerConfig;
 use apollo_config_manager_types::communication::{ConfigManagerRequest, ConfigManagerResponse};
 use apollo_config_manager_types::config_manager_types::ConfigManagerResult;
+use apollo_consensus_config::config::ConsensusDynamicConfig;
 use apollo_infra::component_definitions::{ComponentRequestHandler, ComponentStarter};
 use apollo_infra::component_server::{ConcurrentLocalComponentServer, RemoteComponentServer};
 use apollo_node_config::node_config::NodeDynamicConfig;
 use async_trait::async_trait;
-use serde_json::Value;
 use tracing::{info, instrument};
 
-/// Internal state management for the ConfigManager.
-#[derive(Clone)]
-pub struct ConfigManagerState {
-    pub node_dynamic_config: Arc<NodeDynamicConfig>,
-}
+#[cfg(test)]
+#[path = "config_manager_tests.rs"]
+pub mod config_manager_tests;
 
-impl Default for ConfigManagerState {
-    fn default() -> Self {
-        Self { node_dynamic_config: Arc::new(NodeDynamicConfig::default()) }
-    }
-}
-
-// TODO(Nadin): remove dead_code once we have actual config manager logic
-#[allow(dead_code)]
 #[derive(Clone)]
 pub struct ConfigManager {
-    config: ConfigManagerConfig,
-    state: ConfigManagerState,
+    _config: ConfigManagerConfig,
+    latest_node_dynamic_config: NodeDynamicConfig,
 }
 
 impl ConfigManager {
-    pub fn new(config: ConfigManagerConfig) -> Self {
-        let state = ConfigManagerState::default();
-        info!("ConfigManager initialized with default configuration");
-        Self { config, state }
+    pub fn new(config: ConfigManagerConfig, node_dynamic_config: NodeDynamicConfig) -> Self {
+        Self { _config: config, latest_node_dynamic_config: node_dynamic_config }
     }
 
-    pub async fn update_config(&mut self) -> ConfigManagerResult<()> {
-        // TODO(Nadin): Implement actual config update logic
-        info!("ConfigManager: updating configuration");
-
+    pub(crate) fn set_node_dynamic_config(
+        &mut self,
+        node_dynamic_config: NodeDynamicConfig,
+    ) -> ConfigManagerResult<()> {
+        info!("ConfigManager: updating node dynamic config");
+        self.latest_node_dynamic_config = node_dynamic_config;
         Ok(())
     }
 
-    pub fn get_current_config(&self) -> Arc<Value> {
-        let config_json = serde_json::to_value(&*self.state.node_dynamic_config)
-            .unwrap_or_else(|_| serde_json::json!({}));
-        Arc::new(config_json)
-    }
-
-    pub fn get_node_dynamic_config(&self) -> Arc<NodeDynamicConfig> {
-        self.state.node_dynamic_config.clone()
+    pub(crate) fn get_consensus_dynamic_config(
+        &self,
+    ) -> ConfigManagerResult<ConsensusDynamicConfig> {
+        Ok(self.latest_node_dynamic_config.consensus_dynamic_config.as_ref().unwrap().clone())
     }
 }
 
@@ -65,17 +49,17 @@ impl ComponentRequestHandler<ConfigManagerRequest, ConfigManagerResponse> for Co
     #[instrument(skip(self), ret)]
     async fn handle_request(&mut self, request: ConfigManagerRequest) -> ConfigManagerResponse {
         match request {
-            ConfigManagerRequest::ReadConfig => {
-                info!("ConfigManager: handling ReadConfig request");
-                let config_data = self.get_current_config();
-                let result = Ok((*config_data).clone());
-                ConfigManagerResponse::ReadConfig(result)
+            // TODO(Nadin/Tsabary): consider using a macro to generate the responses for each type
+            // of request.
+            ConfigManagerRequest::GetConsensusDynamicConfig => {
+                ConfigManagerResponse::GetConsensusDynamicConfig(
+                    self.get_consensus_dynamic_config(),
+                )
             }
-            ConfigManagerRequest::GetNodeDynamicConfig => {
-                info!("ConfigManager: handling GetNodeDynamicConfig request");
-                let node_config = self.get_node_dynamic_config();
-                let result = Ok((*node_config).clone());
-                ConfigManagerResponse::GetNodeDynamicConfig(result)
+            ConfigManagerRequest::SetNodeDynamicConfig(new_config) => {
+                ConfigManagerResponse::SetNodeDynamicConfig(
+                    self.set_node_dynamic_config(new_config),
+                )
             }
         }
     }

crates/apollo_config_manager/src/config_manager.rs

@@ -56,17 +56,26 @@ impl ConfigManagerRunner {
             .expect("consensus_manager_config must be present");
 
         let node_dynamic_config = NodeDynamicConfig {
-            consensus_dynamic_config: consensus_manager_config
-                .consensus_manager_config
-                .dynamic_config
-                .clone(),
+            consensus_dynamic_config: Some(
+                consensus_manager_config.consensus_manager_config.dynamic_config.clone(),
+            ),
         };
 
         info!("Extracted NodeDynamicConfig: {:?}", node_dynamic_config);
 
-        // TODO(Nadin): Send the new config to the config manager through the client.
-
-        Ok(node_dynamic_config)
+        // TODO(Nadin/Tsabary): Store the last loaded config, compare for changes and only send the
+        // changes to the config manager.
+        match self.config_manager_client.set_node_dynamic_config(node_dynamic_config.clone()).await
+        {
+            Ok(()) => {
+                info!("Successfully updated dynamic config");
+                Ok(node_dynamic_config)
+            }
+            Err(e) => {
+                error!("Failed to update dynamic config: {:?}", e);
+                Err(format!("Failed to update dynamic config: {:?}", e).into())
+            }
+        }
     }
 }
 

crates/apollo_config_manager/src/config_manager_runner.rs

@@ -85,8 +85,11 @@ fn update_config_file(temp_file: &NamedTempFile) -> String {
 }
 
 #[tokio::test]
-async fn test_update_config_with_changed_values() {
-    let config_manager_client: SharedConfigManagerClient = Arc::new(MockConfigManagerClient::new());
+async fn test_config_manager_runner_update_config_with_changed_values() {
+    // Set a mock config manager client to expect the update dynamic config request.
+    let mut mock_client = MockConfigManagerClient::new();
+    mock_client.expect_set_node_dynamic_config().times(1..).return_const(Ok(()));
+    let config_manager_client: SharedConfigManagerClient = Arc::new(mock_client);
 
     // Create a temporary config file and get the validator id value.
     let (temp_file, cli_args, validator_id_value) = create_temp_config_file_and_args();
@@ -107,9 +110,11 @@ async fn test_update_config_with_changed_values() {
     let first_dynamic_config =
         first_update_config_result.expect("First update_config should succeed");
     assert_eq!(
-        first_dynamic_config.consensus_dynamic_config.validator_id, expected_validator_id,
+        first_dynamic_config.consensus_dynamic_config.as_ref().unwrap().validator_id,
+        expected_validator_id,
         "First update_config: Validator id mismatch: {} != {}",
-        first_dynamic_config.consensus_dynamic_config.validator_id, expected_validator_id
+        first_dynamic_config.consensus_dynamic_config.as_ref().unwrap().validator_id,
+        expected_validator_id
     );
 
     // Edit the config file and then trigger a config update, expecting the new validator id.
@@ -120,8 +125,10 @@ async fn test_update_config_with_changed_values() {
     let second_dynamic_config =
         second_update_config_result.expect("Second update_config should succeed");
     assert_eq!(
-        second_dynamic_config.consensus_dynamic_config.validator_id, expected_validator_id,
+        second_dynamic_config.consensus_dynamic_config.as_ref().unwrap().validator_id,
+        expected_validator_id,
         "Second update_config: Validator id mismatch: {} != {}",
-        second_dynamic_config.consensus_dynamic_config.validator_id, expected_validator_id
+        second_dynamic_config.consensus_dynamic_config.as_ref().unwrap().validator_id,
+        expected_validator_id
     );
 }

crates/apollo_config_manager/src/config_manager_runner_tests.rs

@@ -0,0 +1,53 @@
+use apollo_config_manager_config::config::ConfigManagerConfig;
+use apollo_consensus_config::config::ConsensusDynamicConfig;
+use apollo_consensus_config::ValidatorId;
+use apollo_node_config::node_config::NodeDynamicConfig;
+
+use crate::config_manager::ConfigManager;
+
+#[tokio::test]
+async fn test_config_manager_update_config() {
+    // Set a config manager.
+    let config = ConfigManagerConfig::default();
+
+    let consensus_dynamic_config = ConsensusDynamicConfig::default();
+    let node_dynamic_config =
+        NodeDynamicConfig { consensus_dynamic_config: Some(consensus_dynamic_config) };
+    let mut config_manager = ConfigManager::new(config, node_dynamic_config.clone());
+
+    // Get the consensus dynamic config and assert it is the expected one.
+    let consensus_dynamic_config = config_manager
+        .get_consensus_dynamic_config()
+        .expect("Failed to get consensus dynamic config");
+    assert_eq!(
+        &consensus_dynamic_config,
+        node_dynamic_config.consensus_dynamic_config.as_ref().unwrap(),
+        "Consensus dynamic config mismatch: {consensus_dynamic_config:#?} != {:#?}",
+        node_dynamic_config.consensus_dynamic_config
+    );
+
+    // Set a new dynamic config by creating a new consensus dynamic config. For simplicity, we
+    // create an arbitrary one and assert it's not the default one.
+    let new_consensus_dynamic_config =
+        ConsensusDynamicConfig { validator_id: ValidatorId::from(1_u8) };
+    assert_ne!(
+        consensus_dynamic_config, new_consensus_dynamic_config,
+        "Consensus dynamic config should be different: {consensus_dynamic_config:#?} != {:#?}",
+        new_consensus_dynamic_config
+    );
+    config_manager
+        .set_node_dynamic_config(NodeDynamicConfig {
+            consensus_dynamic_config: Some(new_consensus_dynamic_config.clone()),
+        })
+        .expect("Failed to set node dynamic config");
+
+    // Get the post-change consensus dynamic config and assert it is the expected one.
+    let consensus_dynamic_config = config_manager
+        .get_consensus_dynamic_config()
+        .expect("Failed to get consensus dynamic config");
+    assert_eq!(
+        consensus_dynamic_config, new_consensus_dynamic_config,
+        "Consensus dynamic config mismatch: {consensus_dynamic_config:#?} != {:#?}",
+        new_consensus_dynamic_config
+    );
+}

crates/apollo_config_manager/src/config_manager_tests.rs

@@ -9,15 +9,17 @@ repository.workspace = true
 workspace = true
 
 [dependencies]
+apollo_consensus_config.workspace = true
 apollo_infra.workspace = true
 apollo_metrics.workspace = true
 apollo_node_config.workspace = true
+apollo_proc_macros.workspace = true
 async-trait.workspace = true
 mockall = { workspace = true, optional = true }
 serde = { workspace = true, features = ["derive"] }
-serde_json.workspace = true
 strum = { workspace = true, features = ["derive"] }
 strum_macros.workspace = true
+thiserror.workspace = true
 
 [dev-dependencies]
 mockall.workspace = true

crates/apollo_config_manager_types/Cargo.toml

@@ -1,17 +1,20 @@
 use std::sync::Arc;
 
-use apollo_infra::component_client::{LocalComponentClient, RemoteComponentClient};
+use apollo_consensus_config::config::ConsensusDynamicConfig;
+use apollo_infra::component_client::{ClientError, LocalComponentClient, RemoteComponentClient};
 use apollo_infra::component_definitions::{ComponentClient, PrioritizedRequest, RequestWrapper};
 use apollo_infra::{impl_debug_for_infra_requests_and_responses, impl_labeled_request};
 use apollo_metrics::generate_permutation_labels;
 use apollo_node_config::node_config::NodeDynamicConfig;
+use apollo_proc_macros::handle_all_response_variants;
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 use strum::{EnumVariantNames, VariantNames};
 use strum_macros::{AsRefStr, EnumDiscriminants, EnumIter, IntoStaticStr};
+use thiserror::Error;
 
 use crate::config_manager_types::ConfigManagerResult;
+use crate::errors::ConfigManagerError;
 
 pub type LocalConfigManagerClient =
     LocalComponentClient<ConfigManagerRequest, ConfigManagerResponse>;
@@ -23,7 +26,16 @@ pub type SharedConfigManagerClient = Arc<dyn ConfigManagerClient>;
 
 #[cfg_attr(any(feature = "testing", test), mockall::automock)]
 #[async_trait]
-pub trait ConfigManagerClient: Send + Sync {}
+pub trait ConfigManagerClient: Send + Sync {
+    async fn get_consensus_manager_dynamic_config(
+        &self,
+    ) -> ConfigManagerClientResult<ConsensusDynamicConfig>;
+
+    async fn set_node_dynamic_config(
+        &self,
+        config: NodeDynamicConfig,
+    ) -> ConfigManagerClientResult<()>;
+}
 
 #[derive(Serialize, Deserialize, Clone, AsRefStr, EnumDiscriminants)]
 #[strum_discriminants(
@@ -32,8 +44,8 @@ pub trait ConfigManagerClient: Send + Sync {}
     strum(serialize_all = "snake_case")
 )]
 pub enum ConfigManagerRequest {
-    ReadConfig,
-    GetNodeDynamicConfig,
+    GetConsensusDynamicConfig,
+    SetNodeDynamicConfig(NodeDynamicConfig),
 }
 impl_debug_for_infra_requests_and_responses!(ConfigManagerRequest);
 impl_labeled_request!(ConfigManagerRequest, ConfigManagerRequestLabelValue);
@@ -48,16 +60,48 @@ generate_permutation_labels! {
 
 #[derive(Clone, Serialize, Deserialize, AsRefStr)]
 pub enum ConfigManagerResponse {
-    ReadConfig(ConfigManagerResult<Value>),
-    GetNodeDynamicConfig(ConfigManagerResult<NodeDynamicConfig>),
+    GetConsensusDynamicConfig(ConfigManagerResult<ConsensusDynamicConfig>),
+    SetNodeDynamicConfig(ConfigManagerResult<()>),
 }
 impl_debug_for_infra_requests_and_responses!(ConfigManagerResponse);
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-pub enum ConfigManagerClientError {}
+#[derive(Clone, Debug, Error)]
+pub enum ConfigManagerClientError {
+    #[error(transparent)]
+    ClientError(#[from] ClientError),
+    #[error(transparent)]
+    ConfigManagerError(#[from] ConfigManagerError),
+}
 
 #[async_trait]
-impl<ComponentClientType> ConfigManagerClient for ComponentClientType where
-    ComponentClientType: Send + Sync + ComponentClient<ConfigManagerRequest, ConfigManagerResponse>
+impl<ComponentClientType> ConfigManagerClient for ComponentClientType
+where
+    ComponentClientType: Send + Sync + ComponentClient<ConfigManagerRequest, ConfigManagerResponse>,
 {
+    async fn get_consensus_manager_dynamic_config(
+        &self,
+    ) -> ConfigManagerClientResult<ConsensusDynamicConfig> {
+        let request = ConfigManagerRequest::GetConsensusDynamicConfig;
+        handle_all_response_variants!(
+            ConfigManagerResponse,
+            GetConsensusDynamicConfig,
+            ConfigManagerClientError,
+            ConfigManagerError,
+            Direct
+        )
+    }
+
+    async fn set_node_dynamic_config(
+        &self,
+        config: NodeDynamicConfig,
+    ) -> ConfigManagerClientResult<()> {
+        let request = ConfigManagerRequest::SetNodeDynamicConfig(config);
+        handle_all_response_variants!(
+            ConfigManagerResponse,
+            SetNodeDynamicConfig,
+            ConfigManagerClientError,
+            ConfigManagerError,
+            Direct
+        )
+    }
 }