gRPC: Implement max_receive_message_length in envoy gRPC

api/envoy/config/core/v3/grpc_service.proto

@@ -49,6 +49,10 @@ message GrpcService {
     // Currently only supported for xDS gRPC streams.
     // If not set, xDS gRPC streams default base interval:500ms, maximum interval:30s will be applied.
     RetryPolicy retry_policy = 3;
+
+    // Maximum receive message length.
+    // Defaults to 0, which means unlimited.
+    google.protobuf.UInt32Value max_receive_message_length = 4;
   }
 
   // [#next-free-field: 9]

source/common/grpc/async_client_impl.cc

@@ -17,7 +17,9 @@ namespace Grpc {
 AsyncClientImpl::AsyncClientImpl(Upstream::ClusterManager& cm,
                                  const envoy::config::core::v3::GrpcService& config,
                                  TimeSource& time_source)
-    : cm_(cm), remote_cluster_name_(config.envoy_grpc().cluster_name()),
+    : max_recv_message_length_(
+          PROTOBUF_GET_WRAPPED_OR_DEFAULT(config.envoy_grpc(), max_receive_message_length, 0)),
+      cm_(cm), remote_cluster_name_(config.envoy_grpc().cluster_name()),
       host_name_(config.envoy_grpc().authority()), time_source_(time_source),
       metadata_parser_(Router::HeaderParser::configure(
           config.initial_metadata(),
@@ -81,6 +83,8 @@ AsyncStreamImpl::AsyncStreamImpl(AsyncClientImpl& parent, absl::string_view serv
   if (!options.retry_policy.has_value() && parent_.retryPolicy().has_value()) {
     options_.setRetryPolicy(*parent_.retryPolicy());
   }
+  // Configure the maximum frame length
+  decoder_.setMaxFrameLength(parent_.max_recv_message_length_);
 }
 
 void AsyncStreamImpl::initialize(bool buffer_body_for_retry) {

source/common/grpc/async_client_impl.h

@@ -43,6 +43,7 @@ class AsyncClientImpl final : public RawAsyncClient {
   }
 
 private:
+  const uint32_t max_recv_message_length_;
   Upstream::ClusterManager& cm_;
   const std::string remote_cluster_name_;
   // The host header value in the http transport.

source/common/grpc/codec.cc

@@ -38,7 +38,6 @@ bool Decoder::decode(Buffer::Instance& input, std::vector<Frame>& output) {
   output_ = &output;
   inspect(input);
   output_ = nullptr;
-
   if (decoding_error_ || is_frame_oversized_) {
     return false;
   }

test/common/grpc/grpc_client_integration_test.cc

@@ -225,6 +225,29 @@ TEST_P(GrpcClientIntegrationTest, BadReplyGrpcFraming) {
   dispatcher_helper_.runDispatcher();
 }
 
+// Validate that a reply that exceeds gRPC maximum frame size is handled as an RESOURCE_EXHAUSTED
+// gRPC error.
+TEST_P(GrpcClientIntegrationTest, BadReplyOverGrpcFrameLimit) {
+  // Only testing behavior of Envoy client, since `max_receive_message_length` configuration is
+  // added to Envoy-gRPC only.
+  SKIP_IF_GRPC_CLIENT(ClientType::GoogleGrpc);
+
+  helloworld::HelloReply reply;
+  reply.set_message("HelloWorld");
+
+  initialize(/*envoy_grpc_max_recv_msg_length=*/2);
+
+  auto stream = createStream(empty_metadata_);
+  stream->sendRequest();
+  stream->sendServerInitialMetadata(empty_metadata_);
+  stream->expectTrailingMetadata(empty_metadata_);
+  // TODO(tyxia) Change the error status to RESOURCE_EXHAUSTED.
+  stream->expectGrpcStatus(Status::WellKnownGrpcStatus::Internal);
+  auto serialized_response = Grpc::Common::serializeToGrpcFrame(reply);
+  stream->fake_stream_->encodeData(*serialized_response, true);
+  dispatcher_helper_.runDispatcher();
+}
+
 // Validate that custom channel args can be set on the Google gRPC client.
 //
 TEST_P(GrpcClientIntegrationTest, CustomChannelArgs) {

test/common/grpc/grpc_client_integration_test_harness.h

@@ -278,14 +278,14 @@ class GrpcClientIntegrationTest : public GrpcClientIntegrationParamTest {
         dispatcher_(api_->allocateDispatcher("test_thread")),
         http_context_(stats_store_.symbolTable()), router_context_(stats_store_.symbolTable()) {}
 
-  virtual void initialize() {
+  virtual void initialize(uint32_t envoy_grpc_max_recv_msg_length = 0) {
     if (fake_upstream_ == nullptr) {
       fake_upstream_config_.upstream_protocol_ = Http::CodecType::HTTP2;
       fake_upstream_ = std::make_unique<FakeUpstream>(0, ipVersion(), fake_upstream_config_);
     }
     switch (clientType()) {
     case ClientType::EnvoyGrpc:
-      grpc_client_ = createAsyncClientImpl();
+      grpc_client_ = createAsyncClientImpl(envoy_grpc_max_recv_msg_length);
       break;
     case ClientType::GoogleGrpc: {
       grpc_client_ = createGoogleAsyncClientImpl();
@@ -321,7 +321,7 @@ class GrpcClientIntegrationTest : public GrpcClientIntegrationParamTest {
 
   // Create a Grpc::AsyncClientImpl instance backed by enough fake/mock
   // infrastructure to initiate a loopback TCP connection to fake_upstream_.
-  RawAsyncClientPtr createAsyncClientImpl() {
+  RawAsyncClientPtr createAsyncClientImpl(uint32_t envoy_grpc_max_recv_msg_length = 0) {
     client_connection_ = std::make_unique<Network::ClientConnectionImpl>(
         *dispatcher_, fake_upstream_->localAddress(), nullptr,
         std::move(async_client_transport_socket_), nullptr, nullptr);
@@ -347,6 +347,11 @@ class GrpcClientIntegrationTest : public GrpcClientIntegrationParamTest {
         .WillRepeatedly(ReturnRef(*http_async_client_));
     envoy::config::core::v3::GrpcService config;
     config.mutable_envoy_grpc()->set_cluster_name("fake_cluster");
+    if (envoy_grpc_max_recv_msg_length != 0) {
+      config.mutable_envoy_grpc()->mutable_max_receive_message_length()->set_value(
+          envoy_grpc_max_recv_msg_length);
+    }
+
     fillServiceWideInitialMetadata(config);
     return std::make_unique<AsyncClientImpl>(cm_, config, dispatcher_->timeSource());
   }
@@ -548,7 +553,7 @@ class GrpcSslClientIntegrationTest : public GrpcClientIntegrationTest {
     return config;
   }
 
-  void initialize() override {
+  void initialize(uint32_t envoy_grpc_max_recv_msg_length = 0) override {
     envoy::extensions::transport_sockets::tls::v3::UpstreamTlsContext tls_context;
     auto* common_tls_context = tls_context.mutable_common_tls_context();
     auto* validation_context = common_tls_context->mutable_validation_context();
@@ -575,7 +580,7 @@ class GrpcSslClientIntegrationTest : public GrpcClientIntegrationTest {
     fake_upstream_ =
         std::make_unique<FakeUpstream>(createUpstreamSslContext(), 0, ipVersion(), config);
 
-    GrpcClientIntegrationTest::initialize();
+    GrpcClientIntegrationTest::initialize(envoy_grpc_max_recv_msg_length);
   }
 
   Network::DownstreamTransportSocketFactoryPtr createUpstreamSslContext() {