RSCBC-28: Add support for cluster map known versions

Author: chvck

sdk/couchbase-core/src/agent.rs

@@ -27,10 +27,10 @@ use crate::collection_resolver_cached::{
 };
 use crate::collection_resolver_memd::{CollectionResolverMemd, CollectionResolverMemdOptions};
 use crate::compressionmanager::{CompressionManager, StdCompressor};
-use crate::configparser::ConfigParser;
-use crate::configwatcher::{
-    ConfigWatcher, ConfigWatcherMemd, ConfigWatcherMemdConfig, ConfigWatcherMemdOptions,
+use crate::configmanager::{
+    ConfigManager, ConfigManagerMemd, ConfigManagerMemdConfig, ConfigManagerMemdOptions,
 };
+use crate::configparser::ConfigParser;
 use crate::crudcomponent::CrudComponent;
 use crate::errmapcomponent::ErrMapComponent;
 use crate::error::{Error, ErrorKind, Result};
@@ -72,7 +72,6 @@ struct AgentState {
     num_pool_connections: usize,
     // http_transport:
     last_clients: HashMap<String, KvClientConfig>,
-    latest_config: ParsedConfig,
     network_type: String,
 
     client_name: String,
@@ -84,7 +83,7 @@ type AgentCollectionResolver = CollectionResolverCached<CollectionResolverMemd<A
 pub(crate) struct AgentInner {
     state: Arc<Mutex<AgentState>>,
 
-    cfg_watcher: Arc<dyn ConfigWatcher>,
+    config_manager: Arc<dyn ConfigManager>,
     conn_mgr: Arc<AgentClientManager>,
     vb_router: Arc<StdVbucketRouter>,
     collections: Arc<AgentCollectionResolver>,
@@ -111,11 +110,11 @@ pub(crate) struct AgentInner {
 pub struct Agent {
     pub(crate) inner: Arc<AgentInner>,
 
-    config_watcher_shutdown_tx: Sender<()>,
+    config_manager_shutdown_tx: Sender<()>,
 }
 
 struct AgentComponentConfigs {
-    pub config_watcher_memd_config: ConfigWatcherMemdConfig,
+    pub config_manager_memd_config: ConfigManagerMemdConfig,
     pub kv_client_manager_client_configs: HashMap<String, KvClientConfig>,
     pub vbucket_routing_info: VbucketRoutingInfo,
     pub query_config: QueryComponentConfig,
@@ -129,18 +128,13 @@ impl AgentInner {
     pub async fn apply_config(&self, config: ParsedConfig) {
         let mut state = self.state.lock().await;
 
-        if !Self::can_update_config(&config, &state.latest_config) {
-            return;
-        }
-
         info!("Applying updated config");
-        state.latest_config = config;
 
-        self.update_state(&mut state).await;
+        self.update_state(config, &mut state).await;
     }
 
-    async fn update_state(&self, state: &mut AgentState) {
-        let agent_component_configs = Self::gen_agent_component_configs(state);
+    async fn update_state(&self, config: ParsedConfig, state: &mut AgentState) {
+        let agent_component_configs = Self::gen_agent_component_configs(&config, state);
 
         // In order to avoid race conditions between operations selecting the
         // endpoint they need to send the request to, and fetching an actual
@@ -178,8 +172,8 @@ impl AgentInner {
             .update_vbucket_info(agent_component_configs.vbucket_routing_info);
 
         if let Err(e) = self
-            .cfg_watcher
-            .reconfigure(agent_component_configs.config_watcher_memd_config)
+            .config_manager
+            .reconfigure(agent_component_configs.config_manager_memd_config)
         {
             error!("Failed to reconfigure memd config watcher component; {}", e);
         }
@@ -213,27 +207,11 @@ impl AgentInner {
         self.mgmt.reconfigure(agent_component_configs.mgmt_config);
     }
 
-    fn can_update_config(new_config: &ParsedConfig, old_config: &ParsedConfig) -> bool {
-        if new_config.bucket != old_config.bucket {
-            debug!("Switching config due to changed bucket type (bucket takeover)");
-            return true;
-        } else if let Some(cmp) = new_config.partial_cmp(old_config) {
-            if cmp == Ordering::Less {
-                debug!("Skipping config due to new config being an older revision")
-            } else if cmp == Ordering::Equal {
-                debug!("Skipping config due to matching revisions")
-            } else {
-                return true;
-            }
-        }
-
-        false
-    }
-
-    fn gen_agent_component_configs(state: &mut AgentState) -> AgentComponentConfigs {
-        let network_info = state
-            .latest_config
-            .addresses_group_for_network_type(&state.network_type);
+    fn gen_agent_component_configs(
+        config: &ParsedConfig,
+        state: &mut AgentState,
+    ) -> AgentComponentConfigs {
+        let network_info = config.addresses_group_for_network_type(&state.network_type);
 
         let mut kv_data_node_ids = Vec::new();
         let mut kv_data_hosts: HashMap<String, String> = HashMap::new();
@@ -311,7 +289,7 @@ impl AgentInner {
             clients.insert(node_id, config);
         }
 
-        let vbucket_routing_info = if let Some(info) = &state.latest_config.bucket {
+        let vbucket_routing_info = if let Some(info) = &config.bucket {
             VbucketRoutingInfo {
                 // TODO: Clone
                 vbucket_info: info.vbucket_map.clone(),
@@ -327,7 +305,7 @@ impl AgentInner {
         };
 
         AgentComponentConfigs {
-            config_watcher_memd_config: ConfigWatcherMemdConfig {
+            config_manager_memd_config: ConfigManagerMemdConfig {
                 endpoints: kv_data_node_ids,
             },
             kv_client_manager_client_configs: clients,
@@ -339,8 +317,7 @@ impl AgentInner {
             search_config: SearchComponentConfig {
                 endpoints: search_endpoints,
                 authenticator: state.authenticator.clone(),
-                vector_search_enabled: state
-                    .latest_config
+                vector_search_enabled: config
                     .features
                     .contains(&ParsedConfigFeature::FtsVectorSearch),
             },
@@ -358,11 +335,8 @@ impl AgentInner {
         }
     }
 
-    // TODO: This really shouldn't be async
-    pub async fn bucket_features(&self) -> Result<Vec<BucketFeature>> {
-        let guard = self.state.lock().await;
-
-        if let Some(bucket) = &guard.latest_config.bucket {
+    pub fn bucket_features(&self) -> Result<Vec<BucketFeature>> {
+        if let Some(bucket) = self.config_manager.bucket() {
             let mut features = vec![];
 
             for feature in &bucket.features {
@@ -442,7 +416,6 @@ impl Agent {
             authenticator: Arc::new(opts.authenticator),
             num_pool_connections: 1,
             last_clients: Default::default(),
-            latest_config: ParsedConfig::default(),
             network_type: "".to_string(),
             client_name: client_name.clone(),
             tls_config: opts.tls_config,
@@ -471,12 +444,11 @@ impl Agent {
         )
         .await?;
 
-        state.latest_config = first_config;
-
-        let network_type = NetworkTypeHeuristic::identify(&state.latest_config);
+        let network_type = NetworkTypeHeuristic::identify(&first_config);
         state.network_type = network_type;
 
-        let agent_component_configs = AgentInner::gen_agent_component_configs(&mut state);
+        let agent_component_configs =
+            AgentInner::gen_agent_component_configs(&first_config, &mut state);
 
         let err_map_component_conn_mgr = err_map_component.clone();
 
@@ -501,11 +473,12 @@ impl Agent {
             .await?,
         );
 
-        let cfg_watcher = Arc::new(ConfigWatcherMemd::new(
-            agent_component_configs.config_watcher_memd_config,
-            ConfigWatcherMemdOptions {
+        let cfg_watcher = Arc::new(ConfigManagerMemd::new(
+            agent_component_configs.config_manager_memd_config,
+            ConfigManagerMemdOptions {
                 polling_period: opts.config_poller_config.poll_interval,
                 kv_client_manager: conn_mgr.clone(),
+                first_config,
             },
         ));
         let vb_router = Arc::new(StdVbucketRouter::new(
@@ -575,7 +548,7 @@ impl Agent {
 
         let inner = Arc::new(AgentInner {
             state: Arc::new(Mutex::new(state)),
-            cfg_watcher: cfg_watcher.clone(),
+            config_manager: cfg_watcher.clone(),
             conn_mgr,
             vb_router,
             crud,
@@ -596,20 +569,20 @@ impl Agent {
 
         let agent = Agent {
             inner,
-            config_watcher_shutdown_tx: shutdown_tx,
+            config_manager_shutdown_tx: shutdown_tx,
         };
 
-        agent.start_config_watcher(cfg_watcher, shutdown_rx);
+        agent.start_config_manager(cfg_watcher, shutdown_rx);
 
         Ok(agent)
     }
 
-    fn start_config_watcher(
+    fn start_config_manager(
         &self,
-        config_watcher: Arc<impl ConfigWatcher>,
+        config_manager: Arc<impl ConfigManager>,
         shutdown_rx: Receiver<()>,
     ) {
-        let mut watch_rx = config_watcher.watch(shutdown_rx);
+        let mut watch_rx = config_manager.watch(shutdown_rx);
 
         let inner = self.inner.clone();
         tokio::spawn(async move {
@@ -668,7 +641,12 @@ impl Agent {
                     Err(_e) => continue,
                 };
 
-                let raw_config = match client.get_cluster_config(GetClusterConfigRequest {}).await {
+                let raw_config = match client
+                    .get_cluster_config(GetClusterConfigRequest {
+                        known_version: None,
+                    })
+                    .await
+                {
                     Ok(resp) => resp.config,
                     Err(_e) => continue,
                 };
@@ -821,7 +799,7 @@ impl Agent {
     }
 
     pub async fn close(&mut self) {
-        self.config_watcher_shutdown_tx.send(()).unwrap_or_default();
+        self.config_manager_shutdown_tx.send(()).unwrap_or_default();
 
         self.inner.conn_mgr.close().await.unwrap_or_default();
     }
@@ -837,7 +815,7 @@ struct FirstHttpConfig {
 
 // impl Drop for Agent {
 //     fn drop(&mut self) {
-//         self.config_watcher_shutdown_tx.send(()).unwrap_or_default();
+//         self.config_manager_shutdown_tx.send(()).unwrap_or_default();
 //
 //         block_on(async { self.inner.conn_mgr.close().await }).unwrap_or_default();
 //     }

sdk/couchbase-core/src/agent_ops.rs

@@ -47,8 +47,8 @@ use crate::searchx::document_analysis::DocumentAnalysis;
 use crate::{searchmgmt_options, searchx};
 
 impl Agent {
-    pub async fn bucket_features(&self) -> Result<Vec<BucketFeature>> {
-        self.inner.bucket_features().await
+    pub fn bucket_features(&self) -> Result<Vec<BucketFeature>> {
+        self.inner.bucket_features()
     }
 
     pub async fn upsert(&self, opts: UpsertOptions<'_>) -> Result<UpsertResult> {