diff --git a/config/datanode.example.toml b/config/datanode.example.toml index 70b75d361de9..ebde9c81e3e7 100644 --- a/config/datanode.example.toml +++ b/config/datanode.example.toml @@ -10,6 +10,8 @@ rpc_addr = "127.0.0.1:3001" rpc_hostname = "127.0.0.1" # The number of gRPC server worker threads, 8 by default. rpc_runtime_size = 8 +# Interval for sending heartbeat messages to the Metasrv in milliseconds, 5000 by default. +heartbeat_interval_millis = 5000 # Metasrv client options. [meta_client_options] diff --git a/config/frontend.example.toml b/config/frontend.example.toml index e8020a2b69cc..4ac7e97125c9 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -1,5 +1,9 @@ # Node running mode, see `standalone.example.toml`. mode = "distributed" +# Interval for sending heartbeat task to the Metasrv in milliseconds, 5000 by default. +heartbeat_interval_millis = 5000 +# Interval for retry sending heartbeat task in milliseconds, 5000 by default. +retry_interval_millis = 5000 # HTTP server options, see `standalone.example.toml`. [http_options] diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index d6ad7560710b..fb791f21197d 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -132,6 +132,7 @@ impl StandaloneOptions { prom_options: self.prom_options, meta_client_options: None, logging: self.logging, + ..Default::default() } } diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs index ab8dbde174f9..960d4eb68964 100644 --- a/src/frontend/src/frontend.rs +++ b/src/frontend/src/frontend.rs @@ -27,6 +27,8 @@ use crate::service_config::{ #[serde(default)] pub struct FrontendOptions { pub mode: Mode, + pub heartbeat_interval_millis: u64, + pub retry_interval_millis: u64, pub http_options: Option, pub grpc_options: Option, pub mysql_options: Option, @@ -43,6 +45,8 @@ impl Default for FrontendOptions { fn default() -> Self { Self { mode: Mode::Standalone, + heartbeat_interval_millis: 5000, + retry_interval_millis: 5000, http_options: Some(HttpOptions::default()), grpc_options: Some(GrpcOptions::default()), mysql_options: Some(MysqlOptions::default()), diff --git a/src/frontend/src/heartbeat.rs b/src/frontend/src/heartbeat.rs index 3860f93bcc56..a486f9bfbe24 100644 --- a/src/frontend/src/heartbeat.rs +++ b/src/frontend/src/heartbeat.rs @@ -43,14 +43,14 @@ pub struct HeartbeatTask { impl HeartbeatTask { pub fn new( meta_client: Arc, - report_interval: u64, - retry_interval: u64, + heartbeat_interval_millis: u64, + retry_interval_millis: u64, resp_handler_executor: HeartbeatResponseHandlerExecutorRef, ) -> Self { HeartbeatTask { meta_client, - report_interval, - retry_interval, + report_interval: heartbeat_interval_millis, + retry_interval: retry_interval_millis, resp_handler_executor, } } @@ -92,7 +92,7 @@ impl HeartbeatTask { Err(e) => { error!(e; "Occur error while reading heartbeat response"); capture_self - .start_with_retry(Duration::from_secs(retry_interval)) + .start_with_retry(Duration::from_millis(retry_interval)) .await; break; @@ -136,7 +136,7 @@ impl HeartbeatTask { } } _ = &mut sleep => { - sleep.as_mut().reset(Instant::now() + Duration::from_secs(report_interval)); + sleep.as_mut().reset(Instant::now() + Duration::from_millis(report_interval)); Some(HeartbeatRequest::default()) } }; diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index 57e79eb6897b..db885e65c132 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -136,13 +136,14 @@ impl Instance { let datanode_clients = Arc::new(DatanodeClients::default()); - Self::try_new_distributed_with(meta_client, datanode_clients, plugins).await + Self::try_new_distributed_with(meta_client, datanode_clients, plugins, opts).await } pub async fn try_new_distributed_with( meta_client: Arc, datanode_clients: Arc, plugins: Arc, + opts: &FrontendOptions, ) -> Result { let meta_backend = Arc::new(CachedMetaKvBackend::new(meta_client.clone())); let table_routes = Arc::new(TableRoutes::new(meta_client.clone())); @@ -195,8 +196,8 @@ impl Instance { let heartbeat_task = Some(HeartbeatTask::new( meta_client, - 5, - 5, + opts.heartbeat_interval_millis, + opts.retry_interval_millis, Arc::new(handlers_executor), )); diff --git a/tests-integration/src/cluster.rs b/tests-integration/src/cluster.rs index 1943edc7f659..8e250c398c66 100644 --- a/tests-integration/src/cluster.rs +++ b/tests-integration/src/cluster.rs @@ -29,6 +29,7 @@ use common_test_util::temp_dir::create_temp_dir; use datanode::datanode::{DatanodeOptions, ObjectStoreConfig}; use datanode::heartbeat::HeartbeatTask; use datanode::instance::Instance as DatanodeInstance; +use frontend::frontend::FrontendOptions; use frontend::instance::{FrontendInstance, Instance as FeInstance}; use meta_client::client::MetaClientBuilder; use meta_srv::cluster::MetaPeerClientRef; @@ -221,11 +222,14 @@ impl GreptimeDbClusterBuilder { meta_client.start(&[&meta_srv.server_addr]).await.unwrap(); let meta_client = Arc::new(meta_client); + let frontend_opts = FrontendOptions::default(); + Arc::new( FeInstance::try_new_distributed_with( meta_client, datanode_clients, Arc::new(Plugins::default()), + &frontend_opts, ) .await .unwrap(),