feat(remote_wal): add skeleton for remote wal related to datanode

config/datanode.example.toml

@@ -29,16 +29,36 @@ connect_timeout = "1s"
 # `TCP_NODELAY` option for accepted connections, true by default.
 tcp_nodelay = true
 
-# WAL options, see `standalone.example.toml`.
+# WAL options.
+# Uncomments all other wal options except the one you chose.
 [wal]
 # WAL data directory
+provider = "raft-engine"
+
+# Raft-engine wal options, see `standalone.example.toml`
 # dir = "/tmp/greptimedb/wal"
 file_size = "256MB"
 purge_threshold = "4GB"
 purge_interval = "10m"
 read_batch_size = 128
 sync_write = false
 
+# Kafka wal options.
+# The broker endpoints of the Kafka cluster. ["127.0.0.1:9090"] by default.
+broker_endpoints = ["127.0.0.1:9090"]
+# Number of topics shall be created beforehand.
+num_topics = 64
+# Topic name prefix.
+topic_name_prefix = "greptimedb_wal_kafka_topic"
+# Number of partitions per topic.
+num_partitions = 1
+# The maximum log size an rskafka batch producer could buffer.
+max_batch_size = "4MB"
+# The linger duration of an rskafka batch producer.
+linger = "200ms"
+# The maximum amount of time (in milliseconds) to wait for Kafka records to be returned.
+max_wait_time = "100ms"
+
 # Storage options, see `standalone.example.toml`.
 [storage]
 # The working home directory.

config/standalone.example.toml

@@ -82,6 +82,13 @@ enable = true
 
 # WAL options.
 [wal]
+# Available wal providers:
+# - "RaftEngine" (default)
+# - "Kafka"
+provider = "raft-engine"
+
+# There's no kafka wal config for standalone mode.
+
 # WAL data directory
 # dir = "/tmp/greptimedb/wal"
 # WAL file size in bytes.

src/cmd/src/datanode.rs

@@ -18,6 +18,7 @@ use std::time::Duration;
 use async_trait::async_trait;
 use catalog::kvbackend::MetaKvBackend;
 use clap::Parser;
+use common_config::WalConfig;
 use common_telemetry::logging;
 use datanode::config::DatanodeOptions;
 use datanode::datanode::{Datanode, DatanodeBuilder};
@@ -167,7 +168,13 @@ impl StartCommand {
         }
 
         if let Some(wal_dir) = &self.wal_dir {
-            opts.wal.dir = Some(wal_dir.clone());
+            // `wal_dir` only affects raft-engine config.
+            match &mut opts.wal {
+                WalConfig::RaftEngine(raft_engine_config) => {
+                    raft_engine_config.dir.replace(wal_dir.clone());
+                }
+                WalConfig::Kafka(_) => {}
+            }
         }
 
         if let Some(http_addr) = &self.http_addr {
@@ -256,6 +263,7 @@ mod tests {
             tcp_nodelay = true
 
             [wal]
+            provider = "raft-engine"
             dir = "/other/wal"
             file_size = "1GB"
             purge_threshold = "50GB"
@@ -293,12 +301,18 @@ mod tests {
 
         assert_eq!("127.0.0.1:3001".to_string(), options.rpc_addr);
         assert_eq!(Some(42), options.node_id);
-        assert_eq!("/other/wal", options.wal.dir.unwrap());
 
-        assert_eq!(Duration::from_secs(600), options.wal.purge_interval);
-        assert_eq!(1024 * 1024 * 1024, options.wal.file_size.0);
-        assert_eq!(1024 * 1024 * 1024 * 50, options.wal.purge_threshold.0);
-        assert!(!options.wal.sync_write);
+        let WalConfig::RaftEngine(raft_engine_config) = options.wal else {
+            unreachable!()
+        };
+        assert_eq!("/other/wal", raft_engine_config.dir.unwrap());
+        assert_eq!(Duration::from_secs(600), raft_engine_config.purge_interval);
+        assert_eq!(1024 * 1024 * 1024, raft_engine_config.file_size.0);
+        assert_eq!(
+            1024 * 1024 * 1024 * 50,
+            raft_engine_config.purge_threshold.0
+        );
+        assert!(!raft_engine_config.sync_write);
 
         let HeartbeatOptions {
             interval: heart_beat_interval,
@@ -412,9 +426,10 @@ mod tests {
             tcp_nodelay = true
 
             [wal]
+            provider = "raft-engine"
             file_size = "1GB"
             purge_threshold = "50GB"
-            purge_interval = "10m"
+            purge_interval = "5m"
             sync_write = false
 
             [storage]
@@ -475,7 +490,10 @@ mod tests {
                 };
 
                 // Should be read from env, env > default values.
-                assert_eq!(opts.wal.read_batch_size, 100,);
+                let WalConfig::RaftEngine(raft_engine_config) = opts.wal else {
+                    unreachable!()
+                };
+                assert_eq!(raft_engine_config.read_batch_size, 100);
                 assert_eq!(
                     opts.meta_client.unwrap().metasrv_addrs,
                     vec![
@@ -486,10 +504,13 @@ mod tests {
                 );
 
                 // Should be read from config file, config file > env > default values.
-                assert_eq!(opts.wal.purge_interval, Duration::from_secs(60 * 10));
+                assert_eq!(
+                    raft_engine_config.purge_interval,
+                    Duration::from_secs(60 * 5)
+                );
 
                 // Should be read from cli, cli > config file > env > default values.
-                assert_eq!(opts.wal.dir.unwrap(), "/other/wal/dir");
+                assert_eq!(raft_engine_config.dir.unwrap(), "/other/wal/dir");
 
                 // Should be default value.
                 assert_eq!(opts.http.addr, DatanodeOptions::default().http.addr);

src/cmd/src/options.rs

@@ -171,6 +171,7 @@ impl Options {
 mod tests {
     use std::io::Write;
 
+    use common_config::WalConfig;
     use common_test_util::temp_dir::create_named_temp_file;
     use datanode::config::{DatanodeOptions, ObjectStoreConfig};
 
@@ -194,6 +195,7 @@ mod tests {
             tcp_nodelay = true
 
             [wal]
+            provider = "raft-engine"
             dir = "/tmp/greptimedb/wal"
             file_size = "1GB"
             purge_threshold = "50GB"
@@ -277,7 +279,10 @@ mod tests {
                 );
 
                 // Should be the values from config file, not environment variables.
-                assert_eq!(opts.wal.dir.unwrap(), "/tmp/greptimedb/wal");
+                let WalConfig::RaftEngine(raft_engine_config) = opts.wal else {
+                    unreachable!()
+                };
+                assert_eq!(raft_engine_config.dir.unwrap(), "/tmp/greptimedb/wal");
 
                 // Should be default values.
                 assert_eq!(opts.node_id, None);

src/cmd/src/standalone.rs

@@ -482,6 +482,7 @@ mod tests {
             enable_memory_catalog = true
 
             [wal]
+            provider = "raft-engine"
             dir = "/tmp/greptimedb/test/wal"
             file_size = "1GB"
             purge_threshold = "50GB"
@@ -544,7 +545,10 @@ mod tests {
         assert_eq!(None, fe_opts.mysql.reject_no_database);
         assert!(fe_opts.influxdb.enable);
 
-        assert_eq!("/tmp/greptimedb/test/wal", dn_opts.wal.dir.unwrap());
+        let WalConfig::RaftEngine(raft_engine_config) = dn_opts.wal else {
+            unreachable!()
+        };
+        assert_eq!("/tmp/greptimedb/test/wal", raft_engine_config.dir.unwrap());
 
         assert!(matches!(
             &dn_opts.storage.store,

src/common/config/Cargo.toml

@@ -7,4 +7,9 @@ license.workspace = true
 [dependencies]
 common-base.workspace = true
 humantime-serde.workspace = true
+itertools.workspace = true
+rskafka = "0.5"
 serde.workspace = true
+serde_json.workspace = true
+serde_with = "3"
+toml.workspace = true

src/common/config/src/lib.rs

@@ -12,41 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::time::Duration;
+pub mod wal;
 
 use common_base::readable_size::ReadableSize;
 use serde::{Deserialize, Serialize};
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
-#[serde(default)]
-pub struct WalConfig {
-    // wal directory
-    pub dir: Option<String>,
-    // wal file size in bytes
-    pub file_size: ReadableSize,
-    // wal purge threshold in bytes
-    pub purge_threshold: ReadableSize,
-    // purge interval in seconds
-    #[serde(with = "humantime_serde")]
-    pub purge_interval: Duration,
-    // read batch size
-    pub read_batch_size: usize,
-    // whether to sync log file after every write
-    pub sync_write: bool,
-}
-
-impl Default for WalConfig {
-    fn default() -> Self {
-        Self {
-            dir: None,
-            file_size: ReadableSize::mb(256), // log file size 256MB
-            purge_threshold: ReadableSize::gb(4), // purge threshold 4GB
-            purge_interval: Duration::from_secs(600),
-            read_batch_size: 128,
-            sync_write: false,
-        }
-    }
-}
+pub use crate::wal::WalConfig;
 
 pub fn metadata_store_dir(store_dir: &str) -> String {
     format!("{store_dir}/metadata")