fix(meta): set max encode/decode size to 16MB for meta-grpc and raft …

…client/server (databendlabs#11343)

- Fix: databendlabs#11338

Cargo.toml

@@ -124,7 +124,7 @@ thiserror = { version = "1" }
 clap = { version = "3.2.22", features = ["derive", "env"] }
 
 # server
-tonic = { version = "0.9", features = ["transport", "codegen", "prost", "tls-roots", "tls"] }
+tonic = { version = "0.9.2", features = ["transport", "codegen", "prost", "tls-roots", "tls"] }
 
 # serialization
 prost = { version = "0.11.0" }
@@ -133,6 +133,7 @@ prost = { version = "0.11.0" }
 # @xuanwo will address this issue upon his return from Mars.
 serde = { version = "=1.0.152", features = ["derive", "rc"] }
 serde_json = { version = "1.0.85", default-features = false, features = ["preserve_order"] }
+tonic-build = { version = "0.9.2" }
 
 # chrono
 chrono = { version = "0.4.24", features = ["serde"] }

src/meta/app/Cargo.toml

@@ -42,4 +42,4 @@ ignored = ["num-traits"]
 [build-dependencies]
 common-building = { path = "../../common/building" }
 
-tonic-build = "0.8.0"
+tonic-build = { workspace = true }

src/meta/client/src/grpc_client.rs

@@ -51,6 +51,7 @@ use common_meta_types::protobuf::RaftRequest;
 use common_meta_types::protobuf::WatchRequest;
 use common_meta_types::protobuf::WatchResponse;
 use common_meta_types::ConnectionError;
+use common_meta_types::GrpcConfig;
 use common_meta_types::InvalidArgument;
 use common_meta_types::MetaClientError;
 use common_meta_types::MetaError;
@@ -530,7 +531,9 @@ impl MetaGrpcClient {
             let channel = self.make_channel(Some(addr)).await;
             match channel {
                 Ok(c) => {
-                    let mut client = MetaServiceClient::new(c.clone());
+                    let mut client = MetaServiceClient::new(c.clone())
+                        .max_decoding_message_size(GrpcConfig::MAX_DECODING_SIZE)
+                        .max_encoding_message_size(GrpcConfig::MAX_ENCODING_SIZE);
 
                     let new_token = Self::handshake(
                         &mut client,
@@ -542,9 +545,12 @@ impl MetaGrpcClient {
                     .await;
                     match new_token {
                         Ok(token) => {
-                            return Ok(MetaServiceClient::with_interceptor(c, AuthInterceptor {
-                                token,
-                            }));
+                            let client =
+                                MetaServiceClient::with_interceptor(c, AuthInterceptor { token })
+                                    .max_decoding_message_size(GrpcConfig::MAX_DECODING_SIZE)
+                                    .max_encoding_message_size(GrpcConfig::MAX_ENCODING_SIZE);
+
+                            return Ok(client);
                         }
                         Err(handshake_err) => {
                             warn!("handshake error when make client: {:?}", handshake_err);

src/meta/protos/Cargo.toml

@@ -19,7 +19,7 @@ tonic = { workspace = true }
 [build-dependencies]
 prost-build = "0.11.1"
 semver = "1"
-tonic-build = "0.8.0"
+tonic-build = { workspace = true }
 
 [dev-dependencies]
 

src/meta/service/src/api/grpc_server.rs

@@ -23,6 +23,7 @@ use common_base::base::tokio::task::JoinHandle;
 use common_base::base::Stoppable;
 use common_meta_types::protobuf::meta_service_server::MetaServiceServer;
 use common_meta_types::protobuf::FILE_DESCRIPTOR_SET;
+use common_meta_types::GrpcConfig;
 use common_meta_types::MetaNetworkError;
 use futures::future::Either;
 use tonic::transport::Identity;
@@ -93,7 +94,9 @@ impl GrpcServer {
         info!("gRPC addr: {}", addr);
 
         let grpc_impl = MetaServiceImpl::create(meta_node.clone());
-        let grpc_srv = MetaServiceServer::new(grpc_impl);
+        let grpc_srv = MetaServiceServer::new(grpc_impl)
+            .max_decoding_message_size(GrpcConfig::MAX_DECODING_SIZE)
+            .max_encoding_message_size(GrpcConfig::MAX_ENCODING_SIZE);
 
         let j = tokio::spawn(
             async move {

src/meta/service/src/meta_service/raftmeta.rs

@@ -47,6 +47,7 @@ use common_meta_types::ConnectionError;
 use common_meta_types::Endpoint;
 use common_meta_types::ForwardRPCError;
 use common_meta_types::ForwardToLeader;
+use common_meta_types::GrpcConfig;
 use common_meta_types::InvalidReply;
 use common_meta_types::LogEntry;
 use common_meta_types::LogId;
@@ -287,7 +288,9 @@ impl MetaNode {
         let mut rx = mn.running_rx.clone();
 
         let meta_srv_impl = RaftServiceImpl::create(mn.clone());
-        let meta_srv = RaftServiceServer::new(meta_srv_impl);
+        let meta_srv = RaftServiceServer::new(meta_srv_impl)
+            .max_decoding_message_size(GrpcConfig::MAX_DECODING_SIZE)
+            .max_encoding_message_size(GrpcConfig::MAX_ENCODING_SIZE);
 
         let ipv4_addr = host.parse::<Ipv4Addr>();
         let addr = match ipv4_addr {
@@ -1127,7 +1130,7 @@ impl MetaNode {
             .await
             .map_err(|e| MetaNetworkError::GetNodeAddrError(e.to_string()))?;
 
-        let mut client = RaftServiceClient::connect(format!("http://{}", endpoint))
+        let client = RaftServiceClient::connect(format!("http://{}", endpoint))
             .await
             .map_err(|e| {
                 MetaNetworkError::ConnectionError(ConnectionError::new(
@@ -1136,6 +1139,10 @@ impl MetaNode {
                 ))
             })?;
 
+        let mut client = client
+            .max_decoding_message_size(GrpcConfig::MAX_DECODING_SIZE)
+            .max_encoding_message_size(GrpcConfig::MAX_ENCODING_SIZE);
+
         let resp = client.forward(req).await.map_err(|e| {
             MetaNetworkError::from(e)
                 .add_context(format!("target: {}, endpoint: {}", node_id, endpoint))

src/meta/service/src/raft_client.rs

@@ -14,6 +14,7 @@
 
 use common_meta_types::protobuf::raft_service_client::RaftServiceClient;
 use common_meta_types::Endpoint;
+use common_meta_types::GrpcConfig;
 use common_meta_types::NodeId;
 use common_metrics::counter;
 use tonic::transport::channel::Channel;
@@ -52,7 +53,10 @@ impl RaftClientApi for RaftClient {
             target, endpoint_str
         );
 
-        counter::WithCount::new(RaftServiceClient::new(channel), PeerCounter {
+        let cli = RaftServiceClient::new(channel)
+            .max_decoding_message_size(GrpcConfig::MAX_DECODING_SIZE)
+            .max_encoding_message_size(GrpcConfig::MAX_ENCODING_SIZE);
+        counter::WithCount::new(cli, PeerCounter {
             target,
             endpoint,
             endpoint_str,

src/meta/types/Cargo.toml

@@ -29,7 +29,7 @@ tonic = { workspace = true }
 [build-dependencies]
 common-building = { path = "../../common/building" }
 prost-build = "0.11.1"
-tonic-build = "0.8.0"
+tonic-build = { workspace = true }
 
 [dev-dependencies]
 anyhow = { workspace = true }

src/meta/types/src/grpc_config.rs

@@ -0,0 +1,24 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// Grpc default configuration.
+pub struct GrpcConfig {}
+
+impl GrpcConfig {
+    /// The maximum message size the client or server can **send**.
+    pub const MAX_ENCODING_SIZE: usize = 16 * 1024 * 1024;
+
+    /// The maximum message size the client or server can **receive**.
+    pub const MAX_DECODING_SIZE: usize = 16 * 1024 * 1024;
+}

src/meta/types/src/lib.rs

@@ -25,6 +25,7 @@ mod cmd;
 pub mod config;
 mod endpoint;
 pub mod errors;
+mod grpc_config;
 mod log_entry;
 mod match_seq;
 mod message;
@@ -71,6 +72,7 @@ pub use errors::meta_network_errors::MetaNetworkError;
 pub use errors::meta_network_errors::MetaNetworkResult;
 pub use errors::meta_startup_errors::MetaStartupError;
 pub use errors::rpc_errors::ForwardRPCError;
+pub use grpc_config::GrpcConfig;
 pub use log_entry::LogEntry;
 pub use match_seq::MatchSeq;
 pub use match_seq::MatchSeqExt;