diff --git a/Cargo.lock b/Cargo.lock index e68996fa14..b632005409 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2654,6 +2654,7 @@ dependencies = [ name = "common-options" version = "1.2.0" dependencies = [ + "common-base", "common-grpc", "humantime-serde", "serde", diff --git a/config/config.md b/config/config.md index 22fec7e709..860aa446b4 100644 --- a/config/config.md +++ b/config/config.md @@ -336,6 +336,8 @@ | `datanode.client` | -- | -- | Datanode client options. | | `datanode.client.connect_timeout` | String | `10s` | -- | | `datanode.client.tcp_nodelay` | Bool | `true` | -- | +| `datanode.client.max_recv_message_size` | String | `512MB` | The maximum receive message size for the gRPC client. | +| `datanode.client.max_send_message_size` | String | `512MB` | The maximum send message size for the gRPC client. | | `logging` | -- | -- | The logging options. | | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. | | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. | @@ -420,6 +422,8 @@ | `datanode.client.timeout` | String | `10s` | Operation timeout. | | `datanode.client.connect_timeout` | String | `10s` | Connect server timeout. | | `datanode.client.tcp_nodelay` | Bool | `true` | `TCP_NODELAY` option for accepted connections. | +| `datanode.client.max_recv_message_size` | String | `512MB` | The maximum receive message size for the gRPC client. | +| `datanode.client.max_send_message_size` | String | `512MB` | The maximum send message size for the gRPC client. | | `wal` | -- | -- | -- | | `wal.provider` | String | `raft_engine` | -- | | `wal.broker_endpoints` | Array | -- | The broker endpoints of the Kafka cluster.

**It's only used when the provider is `kafka`**. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index dd10005f62..a40d1818ec 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -305,6 +305,10 @@ memory_pool_size = "50%" [datanode.client] connect_timeout = "10s" tcp_nodelay = true +## The maximum receive message size for the gRPC client. +max_recv_message_size = "512MB" +## The maximum send message size for the gRPC client. +max_send_message_size = "512MB" ## The logging options. [logging] diff --git a/config/metasrv.example.toml b/config/metasrv.example.toml index 18023cafbc..a80091f762 100644 --- a/config/metasrv.example.toml +++ b/config/metasrv.example.toml @@ -209,6 +209,12 @@ connect_timeout = "10s" ## `TCP_NODELAY` option for accepted connections. tcp_nodelay = true +## The maximum receive message size for the gRPC client. +max_recv_message_size = "512MB" + +## The maximum send message size for the gRPC client. +max_send_message_size = "512MB" + [wal] # Available wal providers: # - `raft_engine` (default): there're none raft-engine wal config since metasrv only involves in remote wal currently. diff --git a/src/cmd/src/frontend.rs b/src/cmd/src/frontend.rs index cbbd797eed..3a314734c9 100644 --- a/src/cmd/src/frontend.rs +++ b/src/cmd/src/frontend.rs @@ -31,7 +31,6 @@ use client::client_manager::NodeClients; use common_base::Plugins; use common_config::{Configurable, DEFAULT_DATA_HOME}; use common_error::ext::BoxedError; -use common_grpc::channel_manager::ChannelConfig; use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder}; use common_meta::heartbeat::handler::HandlerGroupExecutor; use common_meta::heartbeat::handler::invalidate_table_cache::InvalidateCacheHandler; @@ -435,12 +434,8 @@ impl StartCommand { // frontend to datanode need not timeout. // Some queries are expected to take long time. - let mut channel_config = ChannelConfig { - timeout: None, - tcp_nodelay: opts.datanode.client.tcp_nodelay, - connect_timeout: Some(opts.datanode.client.connect_timeout), - ..Default::default() - }; + let mut channel_config = opts.datanode.client.channel_config(); + channel_config.timeout = None; if opts.grpc.flight_compression.transport_compression() { channel_config.accept_compression = true; channel_config.send_compression = true; diff --git a/src/cmd/tests/load_config_test.rs b/src/cmd/tests/load_config_test.rs index e07eadf9f4..bfb890e745 100644 --- a/src/cmd/tests/load_config_test.rs +++ b/src/cmd/tests/load_config_test.rs @@ -216,6 +216,7 @@ fn test_load_metasrv_example_config() { timeout: Duration::from_secs(10), connect_timeout: Duration::from_secs(10), tcp_nodelay: true, + ..Default::default() }, }, backend_tls: Some(TlsOption { diff --git a/src/common/options/Cargo.toml b/src/common/options/Cargo.toml index b3ed3661e2..54c916fa5e 100644 --- a/src/common/options/Cargo.toml +++ b/src/common/options/Cargo.toml @@ -5,6 +5,7 @@ edition.workspace = true license.workspace = true [dependencies] +common-base.workspace = true common-grpc.workspace = true humantime-serde.workspace = true serde.workspace = true diff --git a/src/common/options/src/datanode.rs b/src/common/options/src/datanode.rs index 7d361247c3..e0692f4f45 100644 --- a/src/common/options/src/datanode.rs +++ b/src/common/options/src/datanode.rs @@ -14,7 +14,8 @@ use std::time::Duration; -use common_grpc::channel_manager; +use common_base::readable_size::ReadableSize; +use common_grpc::channel_manager::{self, ChannelConfig}; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] @@ -23,12 +24,17 @@ pub struct DatanodeClientOptions { } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] pub struct ClientOptions { #[serde(with = "humantime_serde")] pub timeout: Duration, #[serde(with = "humantime_serde")] pub connect_timeout: Duration, pub tcp_nodelay: bool, + /// Maximum size of a message received from a datanode. + pub max_recv_message_size: ReadableSize, + /// Maximum size of a message sent to a datanode. + pub max_send_message_size: ReadableSize, } impl Default for ClientOptions { @@ -39,6 +45,64 @@ impl Default for ClientOptions { channel_manager::DEFAULT_GRPC_CONNECT_TIMEOUT_SECS, ), tcp_nodelay: true, + max_recv_message_size: channel_manager::DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE, + max_send_message_size: channel_manager::DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE, } } } + +impl ClientOptions { + /// Creates a gRPC [`ChannelConfig`] from these datanode client options. + pub fn channel_config(&self) -> ChannelConfig { + ChannelConfig { + timeout: Some(self.timeout), + connect_timeout: Some(self.connect_timeout), + tcp_nodelay: self.tcp_nodelay, + max_recv_message_size: self.max_recv_message_size, + max_send_message_size: self.max_send_message_size, + ..Default::default() + } + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use common_base::readable_size::ReadableSize; + use serde_json::json; + + use super::ClientOptions; + + #[test] + fn test_client_options_backward_compatibility() { + let options: ClientOptions = serde_json::from_value(json!({ + "timeout": "10s", + "connect_timeout": "5s", + "tcp_nodelay": false + })) + .unwrap(); + + assert_eq!(ReadableSize::mb(512), options.max_recv_message_size); + assert_eq!(ReadableSize::mb(512), options.max_send_message_size); + } + + #[test] + fn test_client_options_channel_config() { + let options: ClientOptions = serde_json::from_value(json!({ + "timeout": "20s", + "connect_timeout": "8s", + "tcp_nodelay": false, + "max_recv_message_size": "1GB", + "max_send_message_size": "2GB" + })) + .unwrap(); + + let channel_config = options.channel_config(); + assert_eq!(Some(Duration::from_secs(20)), channel_config.timeout); + assert_eq!(Some(Duration::from_secs(8)), channel_config.connect_timeout); + assert!(!channel_config.tcp_nodelay); + assert_eq!(ReadableSize::gb(1), channel_config.max_recv_message_size); + assert_eq!(ReadableSize::gb(2), channel_config.max_send_message_size); + } +} diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs index 64e99f540a..a528583ba6 100644 --- a/src/meta-srv/src/metasrv/builder.rs +++ b/src/meta-srv/src/metasrv/builder.rs @@ -22,7 +22,6 @@ use client::inserter::InsertOptions; use common_base::Plugins; use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID}; use common_event_recorder::{DEFAULT_COMPACTION_TIME_WINDOW, EventRecorderImpl, EventRecorderRef}; -use common_grpc::channel_manager::ChannelConfig; use common_meta::ddl::flow_meta::FlowMetadataAllocator; use common_meta::ddl::table_meta::{TableMetadataAllocator, TableMetadataAllocatorRef}; use common_meta::ddl::{ @@ -325,10 +324,7 @@ impl MetasrvBuilder { let memory_region_keeper = Arc::new(MemoryRegionKeeper::default()); let node_manager = node_manager.unwrap_or_else(|| { - let datanode_client_channel_config = ChannelConfig::new() - .timeout(Some(options.datanode.client.timeout)) - .connect_timeout(options.datanode.client.connect_timeout) - .tcp_nodelay(options.datanode.client.tcp_nodelay); + let datanode_client_channel_config = options.datanode.client.channel_config(); Arc::new(NodeClients::new(datanode_client_channel_config)) }); let cache_invalidator = Arc::new(MetasrvCacheInvalidator::new(