fix: configure datanode client gRPC message limits (#8642)

Signed-off-by: evenyag <realevenyag@gmail.com>

Cargo.lock

@@ -2654,6 +2654,7 @@ dependencies = [
 name = "common-options"
 version = "1.2.0"
 dependencies = [
+ "common-base",
  "common-grpc",
  "humantime-serde",
  "serde",

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.<br/><br/>**It's only used when the provider is `kafka`**. |

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]

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.

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;

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 {

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

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);
+    }
+}

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(