feat!: improve region manifest service

config/datanode.example.toml

@@ -37,11 +37,19 @@ type = "File"
 data_dir = "/tmp/greptimedb/data/"
 
 # Compaction options, see `standalone.example.toml`.
-[compaction]
+[storage.compaction]
 max_inflight_tasks = 4
 max_files_in_level0 = 8
 max_purge_tasks = 32
 
+# Storage manifest options
+[storage.manifest]
+# Region checkpoint actions margin.
+# Create a checkpoint every <checkpoint_margin> actions.
+checkpoint_margin = 10
+# Region manifest logs and checkpoints gc execution duration
+gc_duration = '30s'
+
 # Procedure storage options, see `standalone.example.toml`.
 # [procedure.store]
 # type = "File"

config/standalone.example.toml

@@ -99,14 +99,23 @@ type = "File"
 data_dir = "/tmp/greptimedb/data/"
 
 # Compaction options.
-[compaction]
+[storage.compaction]
 # Max task number that can concurrently run.
 max_inflight_tasks = 4
 # Max files in level 0 to trigger compaction.
 max_files_in_level0 = 8
 # Max task number for SST purge task after compaction.
 max_purge_tasks = 32
 
+# Storage manifest options
+[storage.manifest]
+# Region checkpoint actions margin.
+# Create a checkpoint every <checkpoint_margin> actions.
+checkpoint_margin = 10
+# Region manifest logs and checkpoints gc execution duration
+gc_duration = '30s'
+
+
 # Procedure storage options.
 # Uncomment to enable.
 # [procedure.store]

src/cmd/src/datanode.rs

@@ -146,7 +146,7 @@ impl TryFrom<StartCommand> for DatanodeOptions {
         }
 
         if let Some(data_dir) = cmd.data_dir {
-            opts.storage = ObjectStoreConfig::File(FileConfig { data_dir });
+            opts.storage.store = ObjectStoreConfig::File(FileConfig { data_dir });
         }
 
         if let Some(wal_dir) = cmd.wal_dir {
@@ -167,7 +167,7 @@ mod tests {
     use std::time::Duration;
 
     use common_test_util::temp_dir::create_named_temp_file;
-    use datanode::datanode::{CompactionConfig, ObjectStoreConfig};
+    use datanode::datanode::{CompactionConfig, ObjectStoreConfig, RegionManifestConfig};
     use servers::Mode;
 
     use super::*;
@@ -203,10 +203,14 @@ mod tests {
             type = "File"
             data_dir = "/tmp/greptimedb/data/"
 
-            [compaction]
-            max_inflight_tasks = 4
-            max_files_in_level0 = 8
+            [storage.compaction]
+            max_inflight_tasks = 3
+            max_files_in_level0 = 7
             max_purge_tasks = 32
+
+            [storage.manifest]
+            checkpoint_margin = 9
+            gc_duration = '7s'
         "#;
         write!(file, "{}", toml_str).unwrap();
 
@@ -237,21 +241,28 @@ mod tests {
         assert_eq!(3000, timeout_millis);
         assert!(tcp_nodelay);
 
-        match options.storage {
-            ObjectStoreConfig::File(FileConfig { data_dir }) => {
-                assert_eq!("/tmp/greptimedb/data/".to_string(), data_dir)
+        match &options.storage.store {
+            ObjectStoreConfig::File(FileConfig { data_dir, .. }) => {
+                assert_eq!("/tmp/greptimedb/data/", data_dir)
             }
             ObjectStoreConfig::S3 { .. } => unreachable!(),
             ObjectStoreConfig::Oss { .. } => unreachable!(),
         };
 
         assert_eq!(
             CompactionConfig {
-                max_inflight_tasks: 4,
-                max_files_in_level0: 8,
+                max_inflight_tasks: 3,
+                max_files_in_level0: 7,
                 max_purge_tasks: 32,
             },
-            options.compaction
+            options.storage.compaction,
+        );
+        assert_eq!(
+            RegionManifestConfig {
+                checkpoint_margin: Some(9),
+                gc_duration: Some(Duration::from_secs(7)),
+            },
+            options.storage.manifest,
         );
     }
 

src/cmd/src/standalone.rs

@@ -18,7 +18,7 @@ use clap::Parser;
 use common_base::Plugins;
 use common_telemetry::info;
 use datanode::datanode::{
-    CompactionConfig, Datanode, DatanodeOptions, ObjectStoreConfig, ProcedureConfig, WalConfig,
+    CompactionConfig, Datanode, DatanodeOptions, ProcedureConfig, StorageConfig, WalConfig,
 };
 use datanode::instance::InstanceRef;
 use frontend::frontend::FrontendOptions;
@@ -82,7 +82,7 @@ pub struct StandaloneOptions {
     pub prometheus_options: Option<PrometheusOptions>,
     pub prom_options: Option<PromOptions>,
     pub wal: WalConfig,
-    pub storage: ObjectStoreConfig,
+    pub storage: StorageConfig,
     pub compaction: CompactionConfig,
     pub procedure: Option<ProcedureConfig>,
 }
@@ -101,7 +101,7 @@ impl Default for StandaloneOptions {
             prometheus_options: Some(PrometheusOptions::default()),
             prom_options: Some(PromOptions::default()),
             wal: WalConfig::default(),
-            storage: ObjectStoreConfig::default(),
+            storage: StorageConfig::default(),
             compaction: CompactionConfig::default(),
             procedure: None,
         }
@@ -129,7 +129,6 @@ impl StandaloneOptions {
             enable_memory_catalog: self.enable_memory_catalog,
             wal: self.wal,
             storage: self.storage,
-            compaction: self.compaction,
             procedure: self.procedure,
             ..Default::default()
         }

src/common/runtime/Cargo.toml

@@ -5,13 +5,15 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+async-trait.workspace = true
 common-error = { path = "../error" }
 common-telemetry = { path = "../telemetry" }
 metrics = "0.20"
 once_cell = "1.12"
 paste.workspace = true
 snafu.workspace = true
 tokio.workspace = true
+tokio-util.workspace = true
 
 [dev-dependencies]
 tokio-test = "0.4"

src/common/runtime/src/error.rs

@@ -15,6 +15,7 @@
 use std::any::Any;
 
 use common_error::prelude::*;
+use tokio::task::JoinError;
 
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -26,6 +27,19 @@ pub enum Error {
         source: std::io::Error,
         backtrace: Backtrace,
     },
+    #[snafu(display("Repeated task {} not started yet", name))]
+    IllegalState { name: String, backtrace: Backtrace },
+
+    #[snafu(display(
+        "Failed to wait for repeated task {} to stop, source: {}",
+        name,
+        source
+    ))]
+    WaitGcTaskStop {
+        name: String,
+        source: JoinError,
+        backtrace: Backtrace,
+    },
 }
 
 impl ErrorExt for Error {

src/common/runtime/src/lib.rs

@@ -15,6 +15,7 @@
 pub mod error;
 mod global;
 pub mod metric;
+mod repeated_task;
 pub mod runtime;
 
 pub use global::{
@@ -23,4 +24,5 @@ pub use global::{
     spawn_read, spawn_write, write_runtime,
 };
 
+pub use crate::repeated_task::{RepeatedTask, TaskFunction, TaskFunctionRef};
 pub use crate::runtime::{Builder, JoinError, JoinHandle, Runtime};