feat: expose wal config (#852)

* feat: wal config

* fix: use human-readable string in wal config

* feat: copy ReadableSize and humanize size config items in toml files

* fix: clippy

src/datanode/Cargo.toml

@@ -30,6 +30,7 @@ datafusion.workspace = true
 datatypes = { path = "../datatypes" }
 futures = "0.3"
 hyper = { version = "0.14", features = ["full"] }
+humantime-serde = "1.1"
 log-store = { path = "../log-store" }
 meta-client = { path = "../meta-client" }
 meta-srv = { path = "../meta-srv", features = ["mock"] }

src/datanode/src/datanode.rs

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Duration;
 
+use common_base::readable_size::ReadableSize;
 use common_telemetry::info;
 use meta_client::MetaClientOpts;
 use serde::{Deserialize, Serialize};
@@ -47,6 +49,36 @@ impl Default for ObjectStoreConfig {
     }
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WalConfig {
+    // wal directory
+    pub dir: 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: "/tmp/greptimedb/wal".to_string(),
+            file_size: ReadableSize::gb(1),        // log file size 1G
+            purge_threshold: ReadableSize::gb(50), // purge threshold 50G
+            purge_interval: Duration::from_secs(600),
+            read_batch_size: 128,
+            sync_write: false,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(default)]
 pub struct DatanodeOptions {
@@ -56,7 +88,7 @@ pub struct DatanodeOptions {
     pub mysql_addr: String,
     pub mysql_runtime_size: usize,
     pub meta_client_opts: Option<MetaClientOpts>,
-    pub wal_dir: String,
+    pub wal: WalConfig,
     pub storage: ObjectStoreConfig,
     pub enable_memory_catalog: bool,
     pub mode: Mode,
@@ -71,7 +103,7 @@ impl Default for DatanodeOptions {
             mysql_addr: "127.0.0.1:4406".to_string(),
             mysql_runtime_size: 2,
             meta_client_opts: None,
-            wal_dir: "/tmp/greptimedb/wal".to_string(),
+            wal: WalConfig::default(),
             storage: ObjectStoreConfig::default(),
             enable_memory_catalog: false,
             mode: Mode::Standalone,

src/datanode/src/instance.rs

@@ -38,7 +38,7 @@ use storage::config::EngineConfig as StorageEngineConfig;
 use storage::EngineImpl;
 use table::table::TableIdProviderRef;
 
-use crate::datanode::{DatanodeOptions, ObjectStoreConfig};
+use crate::datanode::{DatanodeOptions, ObjectStoreConfig, WalConfig};
 use crate::error::{
     self, CatalogSnafu, MetaClientInitSnafu, MissingMetasrvOptsSnafu, MissingNodeIdSnafu,
     NewCatalogSnafu, OpenLogStoreSnafu, Result,
@@ -68,7 +68,7 @@ pub type InstanceRef = Arc<Instance>;
 impl Instance {
     pub async fn new(opts: &DatanodeOptions) -> Result<Self> {
         let object_store = new_object_store(&opts.storage).await?;
-        let logstore = Arc::new(create_log_store(&opts.wal_dir).await?);
+        let logstore = Arc::new(create_log_store(&opts.wal).await?);
 
         let meta_client = match opts.mode {
             Mode::Standalone => None,
@@ -289,16 +289,19 @@ async fn new_metasrv_client(node_id: u64, meta_config: &MetaClientOpts) -> Resul
     Ok(meta_client)
 }
 
-pub(crate) async fn create_log_store(path: impl AsRef<str>) -> Result<RaftEngineLogStore> {
-    let path = path.as_ref();
+pub(crate) async fn create_log_store(wal_config: &WalConfig) -> Result<RaftEngineLogStore> {
     // create WAL directory
-    fs::create_dir_all(path::Path::new(path)).context(error::CreateDirSnafu { dir: path })?;
-
-    info!("The WAL directory is: {}", path);
-
+    fs::create_dir_all(path::Path::new(&wal_config.dir)).context(error::CreateDirSnafu {
+        dir: &wal_config.dir,
+    })?;
+    info!("Creating logstore with config: {:?}", wal_config);
     let log_config = LogConfig {
-        log_file_dir: path.to_string(),
-        ..Default::default()
+        file_size: wal_config.file_size.0,
+        log_file_dir: wal_config.dir.clone(),
+        purge_interval: wal_config.purge_interval,
+        purge_threshold: wal_config.purge_threshold.0,
+        read_batch_size: wal_config.read_batch_size,
+        sync_write: wal_config.sync_write,
     };
 
     let logstore = RaftEngineLogStore::try_new(log_config)

src/datanode/src/mock.rs

@@ -41,7 +41,7 @@ impl Instance {
 
     pub async fn with_mock_meta_server(opts: &DatanodeOptions, meta_srv: MockInfo) -> Result<Self> {
         let object_store = new_object_store(&opts.storage).await?;
-        let logstore = Arc::new(create_log_store(&opts.wal_dir).await?);
+        let logstore = Arc::new(create_log_store(&opts.wal).await?);
         let meta_client = Arc::new(mock_meta_client(meta_srv, opts.node_id.unwrap_or(42)).await);
         let table_engine = Arc::new(DefaultEngine::new(
             TableEngineConfig::default(),

src/datanode/src/tests/test_util.rs

@@ -27,7 +27,7 @@ use table::engine::{EngineContext, TableEngineRef};
 use table::requests::CreateTableRequest;
 use tempdir::TempDir;
 
-use crate::datanode::{DatanodeOptions, ObjectStoreConfig};
+use crate::datanode::{DatanodeOptions, ObjectStoreConfig, WalConfig};
 use crate::error::{CreateTableSnafu, Result};
 use crate::instance::Instance;
 use crate::sql::SqlHandler;
@@ -61,7 +61,10 @@ fn create_tmp_dir_and_datanode_opts(name: &str) -> (DatanodeOptions, TestGuard)
     let wal_tmp_dir = TempDir::new(&format!("gt_wal_{name}")).unwrap();
     let data_tmp_dir = TempDir::new(&format!("gt_data_{name}")).unwrap();
     let opts = DatanodeOptions {
-        wal_dir: wal_tmp_dir.path().to_str().unwrap().to_string(),
+        wal: WalConfig {
+            dir: wal_tmp_dir.path().to_str().unwrap().to_string(),
+            ..Default::default()
+        },
         storage: ObjectStoreConfig::File {
             data_dir: data_tmp_dir.path().to_str().unwrap().to_string(),
         },