fix: datanode start in standalone mode by default (#418)

* fix: datanode start in standalone mode by default

* fix: detech misconfig on startup

* fix: some CR comments and add tests

config/datanode.example.toml

@@ -3,7 +3,7 @@ http_addr = '0.0.0.0:3000'
 rpc_addr = '0.0.0.0:3001'
 wal_dir = '/tmp/greptimedb/wal'
 rpc_runtime_size = 8
-
+mode = "standalone"
 mysql_addr = '0.0.0.0:3306'
 mysql_runtime_size = 4
 

src/cmd/src/datanode.rs

@@ -1,9 +1,9 @@
 use clap::Parser;
 use common_telemetry::logging;
-use datanode::datanode::{Datanode, DatanodeOptions};
+use datanode::datanode::{Datanode, DatanodeOptions, Mode};
 use snafu::ResultExt;
 
-use crate::error::{Error, Result, StartDatanodeSnafu};
+use crate::error::{Error, MissingConfigSnafu, Result, StartDatanodeSnafu};
 use crate::toml_loader;
 
 #[derive(Parser)]
@@ -75,9 +75,6 @@ impl TryFrom<StartCommand> for DatanodeOptions {
             DatanodeOptions::default()
         };
 
-        if let Some(node_id) = cmd.node_id {
-            opts.node_id = node_id;
-        }
         if let Some(addr) = cmd.http_addr {
             opts.http_addr = addr;
         }
@@ -90,8 +87,31 @@ impl TryFrom<StartCommand> for DatanodeOptions {
         if let Some(addr) = cmd.postgres_addr {
             opts.postgres_addr = addr;
         }
-        if let Some(addr) = cmd.metasrv_addr {
-            opts.meta_client_opts.metasrv_addr = addr;
+
+        match (cmd.metasrv_addr, cmd.node_id) {
+            (Some(meta_addr), Some(node_id)) => {
+                // Running mode is only set to Distributed when
+                // both metasrv addr and node id are set in
+                // commandline options
+                opts.meta_client_opts.metasrv_addr = meta_addr;
+                opts.node_id = node_id;
+                opts.mode = Mode::Distributed;
+            }
+            (None, None) => {
+                opts.mode = Mode::Standalone;
+            }
+            (None, Some(_)) => {
+                return MissingConfigSnafu {
+                    msg: "Missing metasrv address option",
+                }
+                .fail();
+            }
+            (Some(_), None) => {
+                return MissingConfigSnafu {
+                    msg: "Missing node id option",
+                }
+                .fail();
+            }
         }
         Ok(opts)
     }
@@ -140,4 +160,58 @@ mod tests {
             }
         };
     }
+
+    #[test]
+    fn test_try_from_cmd() {
+        assert_eq!(
+            Mode::Standalone,
+            DatanodeOptions::try_from(StartCommand {
+                node_id: None,
+                http_addr: None,
+                rpc_addr: None,
+                mysql_addr: None,
+                postgres_addr: None,
+                metasrv_addr: None,
+                config_file: None
+            })
+            .unwrap()
+            .mode
+        );
+
+        assert_eq!(
+            Mode::Distributed,
+            DatanodeOptions::try_from(StartCommand {
+                node_id: Some(42),
+                http_addr: None,
+                rpc_addr: None,
+                mysql_addr: None,
+                postgres_addr: None,
+                metasrv_addr: Some("127.0.0.1:3002".to_string()),
+                config_file: None
+            })
+            .unwrap()
+            .mode
+        );
+
+        assert!(DatanodeOptions::try_from(StartCommand {
+            node_id: None,
+            http_addr: None,
+            rpc_addr: None,
+            mysql_addr: None,
+            postgres_addr: None,
+            metasrv_addr: Some("127.0.0.1:3002".to_string()),
+            config_file: None,
+        })
+        .is_err());
+        assert!(DatanodeOptions::try_from(StartCommand {
+            node_id: Some(42),
+            http_addr: None,
+            rpc_addr: None,
+            mysql_addr: None,
+            postgres_addr: None,
+            metasrv_addr: None,
+            config_file: None,
+        })
+        .is_err());
+    }
 }

src/cmd/src/error.rs

@@ -35,6 +35,9 @@ pub enum Error {
         source: toml::de::Error,
         backtrace: Backtrace,
     },
+
+    #[snafu(display("Missing config, msg: {}", msg))]
+    MissingConfig { msg: String, backtrace: Backtrace },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -45,7 +48,9 @@ impl ErrorExt for Error {
             Error::StartDatanode { source } => source.status_code(),
             Error::StartFrontend { source } => source.status_code(),
             Error::StartMetaServer { source } => source.status_code(),
-            Error::ReadConfig { .. } | Error::ParseConfig { .. } => StatusCode::InvalidArguments,
+            Error::ReadConfig { .. } | Error::ParseConfig { .. } | Error::MissingConfig { .. } => {
+                StatusCode::InvalidArguments
+            }
         }
     }
 

src/datanode/src/datanode.rs

@@ -23,6 +23,13 @@ impl Default for ObjectStoreConfig {
     }
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[serde(rename_all = "lowercase")]
+pub enum Mode {
+    Standalone,
+    Distributed,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct DatanodeOptions {
     pub node_id: u64,
@@ -36,6 +43,7 @@ pub struct DatanodeOptions {
     pub meta_client_opts: MetaClientOpts,
     pub wal_dir: String,
     pub storage: ObjectStoreConfig,
+    pub mode: Mode,
 }
 
 impl Default for DatanodeOptions {
@@ -55,6 +63,7 @@ impl Default for DatanodeOptions {
                 common_time::util::current_time_millis()
             ),
             storage: ObjectStoreConfig::default(),
+            mode: Mode::Standalone,
         }
     }
 }

src/datanode/src/instance.rs

@@ -14,8 +14,8 @@ use storage::{config::EngineConfig as StorageEngineConfig, EngineImpl};
 use table_engine::config::EngineConfig as TableEngineConfig;
 use table_engine::engine::MitoEngine;
 
-use crate::datanode::{DatanodeOptions, MetaClientOpts, ObjectStoreConfig};
-use crate::error::{self, MetaClientInitSnafu, NewCatalogSnafu, Result};
+use crate::datanode::{DatanodeOptions, MetaClientOpts, Mode, ObjectStoreConfig};
+use crate::error::{self, CatalogSnafu, MetaClientInitSnafu, NewCatalogSnafu, Result};
 use crate::heartbeat::HeartbeatTask;
 use crate::script::ScriptExecutor;
 use crate::server::grpc::plan::PhysicalPlanner;
@@ -34,8 +34,8 @@ pub struct Instance {
     pub(crate) physical_planner: PhysicalPlanner,
     pub(crate) script_executor: ScriptExecutor,
     #[allow(unused)]
-    pub(crate) meta_client: MetaClient,
-    pub(crate) heartbeat_task: HeartbeatTask,
+    pub(crate) meta_client: Option<MetaClient>,
+    pub(crate) heartbeat_task: Option<HeartbeatTask>,
 }
 
 pub type InstanceRef = Arc<Instance>;
@@ -44,7 +44,13 @@ impl Instance {
     pub async fn new(opts: &DatanodeOptions) -> Result<Self> {
         let object_store = new_object_store(&opts.storage).await?;
         let log_store = create_local_file_log_store(opts).await?;
-        let meta_client = new_metasrv_client(opts.node_id, &opts.meta_client_opts).await?;
+
+        let meta_client = match opts.mode {
+            Mode::Standalone => None,
+            Mode::Distributed => {
+                Some(new_metasrv_client(opts.node_id, &opts.meta_client_opts).await?)
+            }
+        };
 
         let table_engine = Arc::new(DefaultEngine::new(
             TableEngineConfig::default(),
@@ -57,24 +63,42 @@ impl Instance {
         ));
 
         // create remote catalog manager
-        let catalog_manager = Arc::new(catalog::remote::RemoteCatalogManager::new(
-            table_engine.clone(),
-            opts.node_id,
-            Arc::new(MetaKvBackend {
-                client: meta_client.clone(),
-            }),
-        ));
+        let (catalog_manager, factory) = match opts.mode {
+            Mode::Standalone => {
+                let catalog = Arc::new(
+                    catalog::local::LocalCatalogManager::try_new(table_engine.clone())
+                        .await
+                        .context(CatalogSnafu)?,
+                );
+                let factory = QueryEngineFactory::new(catalog.clone());
+                (catalog as CatalogManagerRef, factory)
+            }
+
+            Mode::Distributed => {
+                let catalog = Arc::new(catalog::remote::RemoteCatalogManager::new(
+                    table_engine.clone(),
+                    opts.node_id,
+                    Arc::new(MetaKvBackend {
+                        client: meta_client.as_ref().unwrap().clone(),
+                    }),
+                ));
+                let factory = QueryEngineFactory::new(catalog.clone());
+                (catalog as CatalogManagerRef, factory)
+            }
+        };
 
-        let factory = QueryEngineFactory::new(catalog_manager.clone());
         let query_engine = factory.query_engine().clone();
         let script_executor =
             ScriptExecutor::new(catalog_manager.clone(), query_engine.clone()).await?;
 
-        let heartbeat_task = HeartbeatTask::new(
-            opts.node_id, /*node id not set*/
-            opts.rpc_addr.clone(),
-            meta_client.clone(),
-        );
+        let heartbeat_task = match opts.mode {
+            Mode::Standalone => None,
+            Mode::Distributed => Some(HeartbeatTask::new(
+                opts.node_id, /*node id not set*/
+                opts.rpc_addr.clone(),
+                meta_client.as_ref().unwrap().clone(),
+            )),
+        };
         Ok(Self {
             query_engine: query_engine.clone(),
             sql_handler: SqlHandler::new(table_engine, catalog_manager.clone()),
@@ -91,7 +115,9 @@ impl Instance {
             .start()
             .await
             .context(NewCatalogSnafu)?;
-        self.heartbeat_task.start().await?;
+        if let Some(task) = &self.heartbeat_task {
+            task.start().await?;
+        }
         Ok(())
     }
 

src/datanode/src/mock.rs

@@ -23,7 +23,7 @@ impl Instance {
         use table_engine::table::test_util::MockEngine;
         use table_engine::table::test_util::MockMitoEngine;
 
-        let meta_client = mock_meta_client().await;
+        let meta_client = Some(mock_meta_client().await);
         let (_dir, object_store) = new_test_object_store("setup_mock_engine_and_table").await;
         let mock_engine = Arc::new(MockMitoEngine::new(
             TableEngineConfig::default(),
@@ -46,8 +46,11 @@ impl Instance {
             .await
             .unwrap();
 
-        let heartbeat_task =
-            HeartbeatTask::new(0, "127.0.0.1:3302".to_string(), meta_client.clone());
+        let heartbeat_task = Some(HeartbeatTask::new(
+            0,
+            "127.0.0.1:3302".to_string(),
+            meta_client.as_ref().unwrap().clone(),
+        ));
         Ok(Self {
             query_engine,
             sql_handler,
@@ -95,8 +98,8 @@ impl Instance {
             catalog_manager,
             physical_planner: PhysicalPlanner::new(query_engine),
             script_executor,
-            meta_client,
-            heartbeat_task,
+            meta_client: Some(meta_client),
+            heartbeat_task: Some(heartbeat_task),
         })
     }
 }