GRPC: truncate error status message to make sure size of error messag…

…e se… (envoyproxy#9861)

* truncate protobuf debug message to make sure size of error message sent via GRPC trailers is within the size limit (default 8KiB)

Signed-off-by: Xin Zhuang <[email protected]>

source/common/config/BUILD

@@ -87,6 +87,7 @@ envoy_cc_library(
     deps = [
         ":api_version_lib",
         ":pausable_ack_queue_lib",
+        ":utility_lib",
         "//include/envoy/config:subscription_interface",
         "//include/envoy/event:dispatcher_interface",
         "//source/common/common:assert_lib",

source/common/config/delta_subscription_state.cc

@@ -4,6 +4,7 @@
 
 #include "common/common/assert.h"
 #include "common/common/hash.h"
+#include "common/config/utility.h"
 
 namespace Envoy {
 namespace Config {
@@ -117,7 +118,7 @@ void DeltaSubscriptionState::handleGoodResponse(
 void DeltaSubscriptionState::handleBadResponse(const EnvoyException& e, UpdateAck& ack) {
   // Note that error_detail being set is what indicates that a DeltaDiscoveryRequest is a NACK.
   ack.error_detail_.set_code(Grpc::Status::WellKnownGrpcStatus::Internal);
-  ack.error_detail_.set_message(e.what());
+  ack.error_detail_.set_message(Config::Utility::truncateGrpcStatusMessage(e.what()));
   disableInitFetchTimeoutTimer();
   ENVOY_LOG(warn, "delta config for {} rejected: {}", type_url_, e.what());
   callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::UpdateRejected, &e);

source/common/config/grpc_mux_impl.cc

@@ -206,7 +206,7 @@ void GrpcMuxImpl::onDiscoveryResponse(
     }
     ::google::rpc::Status* error_detail = api_state_[type_url].request_.mutable_error_detail();
     error_detail->set_code(Grpc::Status::WellKnownGrpcStatus::Internal);
-    error_detail->set_message(e.what());
+    error_detail->set_message(Config::Utility::truncateGrpcStatusMessage(e.what()));
   }
   api_state_[type_url].request_.set_response_nonce(message->nonce());
   queueDiscoveryRequest(type_url);

source/common/config/utility.cc

@@ -27,6 +27,15 @@
 namespace Envoy {
 namespace Config {
 
+std::string Utility::truncateGrpcStatusMessage(absl::string_view error_message) {
+  // GRPC sends error message via trailers, which by default has a 8KB size limit(see
+  // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests). Truncates the
+  // error message if it's too long.
+  constexpr uint32_t kProtobufErrMsgLen = 4096;
+  return fmt::format("{}{}", error_message.substr(0, kProtobufErrMsgLen),
+                     error_message.length() > kProtobufErrMsgLen ? "...(truncated)" : "");
+}
+
 void Utility::translateApiConfigSource(
     const std::string& cluster, uint32_t refresh_delay_ms, const std::string& api_type,
     envoy::config::core::v3::ApiConfigSource& api_config_source) {

source/common/config/utility.h

@@ -269,6 +269,11 @@ class Utility {
     return config;
   }
 
+  /**
+   * Truncates the message to a length less than default GRPC trailers size limit (by default 8KiB).
+   */
+  static std::string truncateGrpcStatusMessage(absl::string_view error_message);
+
   /**
    * Create TagProducer instance. Check all tag names for conflicts to avoid
    * unexpected tag name overwriting.

test/common/config/delta_subscription_state_test.cc

@@ -12,6 +12,7 @@
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
+using testing::Eq;
 using testing::NiceMock;
 using testing::Throw;
 using testing::UnorderedElementsAre;
@@ -52,13 +53,13 @@ class DeltaSubscriptionStateTest : public testing::Test {
   UpdateAck deliverBadDiscoveryResponse(
       const Protobuf::RepeatedPtrField<envoy::service::discovery::v3::Resource>& added_resources,
       const Protobuf::RepeatedPtrField<std::string>& removed_resources,
-      const std::string& version_info, std::string nonce) {
+      const std::string& version_info, std::string nonce, std::string error_message) {
     envoy::service::discovery::v3::DeltaDiscoveryResponse message;
     *message.mutable_resources() = added_resources;
     *message.mutable_removed_resources() = removed_resources;
     message.set_system_version_info(version_info);
     message.set_nonce(nonce);
-    EXPECT_CALL(callbacks_, onConfigUpdate(_, _, _)).WillOnce(Throw(EnvoyException("oh no")));
+    EXPECT_CALL(callbacks_, onConfigUpdate(_, _, _)).WillOnce(Throw(EnvoyException(error_message)));
     return state_.handleResponse(message);
   }
 
@@ -196,7 +197,7 @@ TEST_F(DeltaSubscriptionStateTest, AckGenerated) {
     Protobuf::RepeatedPtrField<envoy::service::discovery::v3::Resource> added_resources =
         populateRepeatedResource(
             {{"name1", "version1C"}, {"name2", "version2C"}, {"name3", "version3B"}});
-    UpdateAck ack = deliverBadDiscoveryResponse(added_resources, {}, "debug3", "nonce3");
+    UpdateAck ack = deliverBadDiscoveryResponse(added_resources, {}, "debug3", "nonce3", "oh no");
     EXPECT_EQ("nonce3", ack.nonce_);
     EXPECT_NE(Grpc::Status::WellKnownGrpcStatus::Ok, ack.error_detail_.code());
   }
@@ -209,6 +210,19 @@ TEST_F(DeltaSubscriptionStateTest, AckGenerated) {
     EXPECT_EQ("nonce4", ack.nonce_);
     EXPECT_EQ(Grpc::Status::WellKnownGrpcStatus::Ok, ack.error_detail_.code());
   }
+  // Bad response error detail is truncated if it's too large.
+  {
+    const std::string very_large_error_message(1 << 20, 'A');
+    Protobuf::RepeatedPtrField<envoy::service::discovery::v3::Resource> added_resources =
+        populateRepeatedResource(
+            {{"name1", "version1D"}, {"name2", "version2D"}, {"name3", "version3D"}});
+    UpdateAck ack = deliverBadDiscoveryResponse(added_resources, {}, "debug5", "nonce5",
+                                                very_large_error_message);
+    EXPECT_EQ("nonce5", ack.nonce_);
+    EXPECT_NE(Grpc::Status::WellKnownGrpcStatus::Ok, ack.error_detail_.code());
+    EXPECT_TRUE(absl::EndsWith(ack.error_detail_.message(), "AAAAAAA...(truncated)"));
+    EXPECT_LT(ack.error_detail_.message().length(), very_large_error_message.length());
+  }
 }
 
 // Tests population of the initial_resource_versions map in the first request of a new stream.
@@ -290,7 +304,8 @@ TEST_F(DeltaSubscriptionStateTest, SubscribeAndUnsubscribeAfterReconnect) {
   // name1: do not include: we lost interest.
   // name2: yes do include: we're interested and we have a version of it.
   // name3: yes do include: even though we don't have a version of it, we are interested.
-  // name4: yes do include: we are newly interested. (If this wasn't a stream reconnect, only name4
+  // name4: yes do include: we are newly interested. (If this wasn't a stream reconnect, only
+  // name4
   //                        would belong in this subscribe field).
   EXPECT_THAT(cur_request.resource_names_subscribe(),
               UnorderedElementsAre("name2", "name3", "name4"));
@@ -345,8 +360,8 @@ TEST_F(DeltaSubscriptionStateTest, CheckUpdatePending) {
 }
 
 // The next three tests test that duplicate resource names (whether additions or removals) cause
-// DeltaSubscriptionState to reject the update without even trying to hand it to the consuming API's
-// onConfigUpdate().
+// DeltaSubscriptionState to reject the update without even trying to hand it to the consuming
+// API's onConfigUpdate().
 TEST_F(DeltaSubscriptionStateTest, DuplicatedAdd) {
   Protobuf::RepeatedPtrField<envoy::service::discovery::v3::Resource> additions =
       populateRepeatedResource({{"name1", "version1A"}, {"name1", "sdfsdfsdfds"}});

test/common/config/grpc_mux_impl_test.cc

@@ -237,6 +237,36 @@ TEST_F(GrpcMuxImplTest, TypeUrlMismatch) {
   expectSendMessage("foo", {}, "");
 }
 
+TEST_F(GrpcMuxImplTest, RpcErrorMessageTruncated) {
+  setup();
+  auto invalid_response = std::make_unique<envoy::service::discovery::v3::DiscoveryResponse>();
+  InSequence s;
+  auto foo_sub = grpc_mux_->addWatch("foo", {"x", "y"}, callbacks_, std::chrono::milliseconds(0));
+
+  EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_));
+  expectSendMessage("foo", {"x", "y"}, "", true);
+  grpc_mux_->start();
+
+  { // Large error message sent back to management server is truncated.
+    const std::string very_large_type_url(1 << 20, 'A');
+    invalid_response->set_type_url("foo");
+    invalid_response->mutable_resources()->Add()->set_type_url(very_large_type_url);
+    EXPECT_CALL(callbacks_, onConfigUpdateFailed(_, _))
+        .WillOnce(Invoke([&very_large_type_url](Envoy::Config::ConfigUpdateFailureReason,
+                                                const EnvoyException* e) {
+          EXPECT_TRUE(IsSubstring(
+              "", "",
+              fmt::format("{} does not match the message-wide type URL foo in DiscoveryResponse",
+                          very_large_type_url), // Local error message is not truncated.
+              e->what()));
+        }));
+    expectSendMessage("foo", {"x", "y"}, "", false, "", Grpc::Status::WellKnownGrpcStatus::Internal,
+                      fmt::format("{}...(truncated)", std::string(4096, 'A')));
+    grpc_mux_->grpcStreamForTest().onReceiveMessage(std::move(invalid_response));
+  }
+  expectSendMessage("foo", {}, "");
+}
+
 // Validate behavior when watches has an unknown resource name.
 TEST_F(GrpcMuxImplTest, WildcardWatch) {
   setup();