refactor: meta-service: simplify raft store and state machine (#18749)

- Remove `is_opened` flag from `RaftStore`

- Remove obsolete config no_sync, which is only used by sled tree store

- Add config to MetaRaftLog

- Remove `RaftStoreInner`

Author: drmingdrmer

src/meta/control/src/export_from_disk.rs

@@ -14,10 +14,9 @@
 
 use std::fs::File;
 use std::io::Write;
-use std::sync::Arc;
 
 use databend_common_meta_raft_store::config::RaftConfig;
-use databend_meta::store::RaftStoreInner;
+use databend_meta::store::RaftStore;
 use futures::TryStreamExt;
 
 use crate::args::ExportArgs;
@@ -36,8 +35,8 @@ pub async fn export_from_dir(args: &ExportArgs) -> anyhow::Result<()> {
     eprintln!();
     eprintln!("Export:");
 
-    let sto_inn = RaftStoreInner::open(&raft_config).await?;
-    let mut lines = Arc::new(sto_inn).export();
+    let sto = RaftStore::open(&raft_config).await?;
+    let mut lines = sto.clone().export();
 
     eprintln!("    From: {}", raft_config.raft_dir);
 

src/meta/control/src/import.rs

@@ -237,7 +237,7 @@ async fn init_new_cluster(
             payload: EntryPayload::Membership(membership),
         };
 
-        let mut log = sto.log.clone();
+        let mut log = sto.log().clone();
         log.blocking_append([entry]).await?;
 
         // insert AddNodes logs
@@ -261,7 +261,7 @@ async fn init_new_cluster(
 
     // Reset node id
     {
-        let mut log = sto.log.write().await;
+        let mut log = sto.log().write().await;
         log.save_user_data(Some(raft_log_v004::LogStoreMeta {
             node_id: Some(args.id),
         }))?;

src/meta/raft-store/src/config.rs

@@ -53,11 +53,6 @@ pub struct RaftConfig {
     /// The dir to store persisted meta state, including raft logs, state machine etc.
     pub raft_dir: String,
 
-    /// Whether to fsync meta to disk for every meta write(raft log, state machine etc).
-    /// No-sync brings risks of data loss during a crash.
-    /// You should only use this in a testing environment, unless YOU KNOW WHAT YOU ARE DOING.
-    pub no_sync: bool,
-
     /// Maximum log entries to cache in memory.
     ///
     /// Higher values improve read performance but use more memory.
@@ -188,7 +183,6 @@ impl Default for RaftConfig {
             raft_advertise_host: get_default_raft_advertise_host(),
             raft_api_port: 28004,
             raft_dir: "./.databend/meta".to_string(),
-            no_sync: false,
 
             log_cache_max_items: 1_000_000,
             log_cache_capacity: 1024 * 1024 * 1024,
@@ -283,11 +277,6 @@ impl RaftConfig {
         }
     }
 
-    /// Returns true to fsync after a write operation to meta.
-    pub fn is_sync(&self) -> bool {
-        !self.no_sync
-    }
-
     /// Returns the min and max election timeout, in milli seconds.
     ///
     /// Raft will choose a random timeout in this range for next election.

src/meta/raft-store/src/ondisk/mod.rs

@@ -160,7 +160,7 @@ impl OnDisk {
 
         let db = init_get_sled_db(config.raft_dir.clone(), 1024 * 1024 * 1024);
 
-        let tree = SledTree::open(&db, TREE_HEADER, config.is_sync())?;
+        let tree = SledTree::open(&db, TREE_HEADER)?;
         let ks = tree.key_space::<DataHeader>();
 
         let header = ks.get(&Self::KEY_HEADER.to_string()).map_err(|e| {

src/meta/raft-store/src/ondisk/upgrade_to_v004.rs

@@ -63,7 +63,7 @@ impl OnDisk {
 
         // Read the purged index
         let first_log_index = {
-            let tree = SledTree::open(&db, "raft_log", self.config.is_sync())?;
+            let tree = SledTree::open(&db, "raft_log")?;
             let ks = tree.key_space::<LogMeta>();
             let purged = ks.get(&LogMetaKey::LastPurged).map_err(|e| {
                 io::Error::other(format!(
@@ -77,7 +77,7 @@ impl OnDisk {
 
         // import logs
         {
-            let tree = SledTree::open(&db, "raft_log", self.config.is_sync())?;
+            let tree = SledTree::open(&db, "raft_log")?;
             let it = tree.tree.iter();
 
             for (i, rkv) in it.enumerate() {
@@ -111,7 +111,7 @@ impl OnDisk {
 
         // import raft_state
         {
-            let tree = SledTree::open(&db, "raft_state", self.config.is_sync())?;
+            let tree = SledTree::open(&db, "raft_state")?;
             let kvs = tree.export()?;
             for kv in kvs {
                 let ent = RaftStoreEntry::deserialize(&kv[0], &kv[1])?;

src/meta/service/src/api/grpc/grpc_service.rs

@@ -413,7 +413,7 @@ impl MetaService for MetaServiceImpl {
 
         let meta_node = self.try_get_meta_node()?;
 
-        let strm = meta_node.raft_store.inner().export();
+        let strm = meta_node.raft_store.clone().export();
 
         let chunk_size = 32;
         // - Chunk up upto 32 Ok items inside a Vec<String>;
@@ -446,7 +446,7 @@ impl MetaService for MetaServiceImpl {
 
         let meta_node = self.try_get_meta_node()?;
 
-        let strm = meta_node.raft_store.inner().export();
+        let strm = meta_node.raft_store.clone().export();
 
         let chunk_size = request.get_ref().chunk_size.unwrap_or(32) as usize;
         // - Chunk up upto `chunk_size` Ok items inside a Vec<String>;

src/meta/service/src/configs/outer_v0.rs

@@ -290,7 +290,6 @@ pub struct ConfigViaEnv {
     pub kvsrv_advertise_host: String,
     pub kvsrv_api_port: u16,
     pub kvsrv_raft_dir: String,
-    pub kvsrv_no_sync: bool,
 
     pub kvsrv_log_cache_max_items: u64,
     pub kvsrv_log_cache_capacity: u64,
@@ -348,7 +347,6 @@ impl From<Config> for ConfigViaEnv {
             kvsrv_advertise_host: cfg.raft_config.raft_advertise_host,
             kvsrv_api_port: cfg.raft_config.raft_api_port,
             kvsrv_raft_dir: cfg.raft_config.raft_dir,
-            kvsrv_no_sync: cfg.raft_config.no_sync,
 
             kvsrv_log_cache_max_items: 1_000_000,
             kvsrv_log_cache_capacity: 1024 * 1024 * 1024,
@@ -385,7 +383,6 @@ impl Into<Config> for ConfigViaEnv {
             raft_advertise_host: self.kvsrv_advertise_host,
             raft_api_port: self.kvsrv_api_port,
             raft_dir: self.kvsrv_raft_dir,
-            no_sync: self.kvsrv_no_sync,
 
             log_cache_max_items: self.kvsrv_log_cache_max_items,
             log_cache_capacity: self.kvsrv_log_cache_capacity,
@@ -482,12 +479,6 @@ pub struct RaftConfig {
     #[clap(long, default_value = "./.databend/meta")]
     pub raft_dir: String,
 
-    /// Whether to fsync meta to disk for every meta write(raft log, state machine etc).
-    /// No-sync brings risks of data loss during a crash.
-    /// You should only use this in a testing environment, unless YOU KNOW WHAT YOU ARE DOING.
-    #[clap(long)]
-    pub no_sync: bool,
-
     /// The maximum number of log entries for log entries cache. Default value is 1_000_000.
     #[clap(long, default_value = "1000000")]
     pub log_cache_max_items: u64,
@@ -614,7 +605,6 @@ impl From<RaftConfig> for InnerRaftConfig {
             raft_advertise_host: x.raft_advertise_host,
             raft_api_port: x.raft_api_port,
             raft_dir: x.raft_dir,
-            no_sync: x.no_sync,
 
             log_cache_max_items: x.log_cache_max_items,
             log_cache_capacity: x.log_cache_capacity,
@@ -651,7 +641,6 @@ impl From<InnerRaftConfig> for RaftConfig {
             raft_advertise_host: inner.raft_advertise_host,
             raft_api_port: inner.raft_api_port,
             raft_dir: inner.raft_dir,
-            no_sync: inner.no_sync,
 
             log_cache_max_items: inner.log_cache_max_items,
             log_cache_capacity: inner.log_cache_capacity,

src/meta/service/src/meta_service/meta_node.rs

@@ -172,8 +172,8 @@ impl MetaNodeBuilder {
 
         let net = NetworkFactory::new(sto.clone());
 
-        let log_store = sto.log.clone();
-        let sm_store = sto.state_machine.clone();
+        let log_store = sto.log().clone();
+        let sm_store = sto.state_machine().clone();
 
         let raft = MetaRaft::new(node_id, Arc::new(config), net, log_store, sm_store)
             .await
@@ -374,25 +374,12 @@ impl MetaNode {
     ) -> Result<Arc<MetaNode>, MetaStartupError> {
         info!("MetaNode::open, config: {:?}", config);
 
-        let mut config = config.clone();
-
-        // Always disable fsync on mac.
-        // Because there are some integration tests running on mac VM.
-        //
-        // On mac File::sync_all() takes 10 ms ~ 30 ms, 500 ms at worst, which very likely to fail a test.
-        if cfg!(target_os = "macos") {
-            warn!("Disabled fsync for meta data tests. fsync on mac is quite slow");
-            config.no_sync = true;
-        }
+        let config = config.clone();
 
         let log_store = RaftStore::open(&config).await?;
 
         // config.id only used for the first time
-        let self_node_id = if log_store.is_opened {
-            log_store.id
-        } else {
-            config.id
-        };
+        let self_node_id = log_store.id;
 
         let builder = MetaNode::builder(&config)
             .sto(log_store.clone())
@@ -1264,11 +1251,11 @@ impl MetaNode {
 
     /// Get the size in bytes of the on disk files of the raft log storage.
     async fn get_raft_log_size(&self) -> u64 {
-        self.raft_store.log.read().await.on_disk_size()
+        self.raft_store.log().read().await.on_disk_size()
     }
 
     async fn get_raft_log_stat(&self) -> RaftLogStat {
-        self.raft_store.log.read().await.stat()
+        self.raft_store.log().read().await.stat()
     }
 
     async fn get_snapshot_key_count(&self) -> u64 {

src/meta/service/src/store/meta_raft_log/mod.rs

@@ -15,6 +15,7 @@
 use std::ops::Deref;
 use std::sync::Arc;
 
+use databend_common_meta_raft_store::config::RaftConfig;
 use databend_common_meta_raft_store::raft_log_v004::RaftLogV004;
 use databend_common_meta_types::raft_types::NodeId;
 use tokio::sync::RwLock;
@@ -25,6 +26,8 @@ mod impl_raft_log_storage;
 #[derive(Debug, Clone)]
 pub struct MetaRaftLog {
     pub(crate) id: NodeId,
+    #[allow(dead_code)]
+    pub(crate) config: RaftConfig,
     inner: Arc<RwLock<RaftLogV004>>,
 }
 
@@ -37,9 +40,10 @@ impl Deref for MetaRaftLog {
 }
 
 impl MetaRaftLog {
-    pub fn new(id: NodeId, inner: RaftLogV004) -> Self {
+    pub fn new(id: NodeId, config: RaftConfig, inner: RaftLogV004) -> Self {
         Self {
             id,
+            config,
             inner: Arc::new(RwLock::new(inner)),
         }
     }

src/meta/service/src/store/mod.rs

@@ -16,7 +16,5 @@ pub mod meta_raft_log;
 pub mod meta_raft_state_machine;
 #[allow(clippy::module_inception)]
 mod store;
-mod store_inner;
 
 pub use store::RaftStore;
-pub use store_inner::RaftStoreInner;