gRPC retries, assuming gRPC status code is in the returned headers

include/envoy/http/header_map.h

@@ -205,6 +205,7 @@ class HeaderEntry {
   HEADER_FUNC(EnvoyMaxRetries)                                                                     \
   HEADER_FUNC(EnvoyOriginalPath)                                                                   \
   HEADER_FUNC(EnvoyRetryOn)                                                                        \
+  HEADER_FUNC(EnvoyRetryGrpcOn)                                                                        \
   HEADER_FUNC(EnvoyUpstreamAltStatName)                                                            \
   HEADER_FUNC(EnvoyUpstreamCanary)                                                                 \
   HEADER_FUNC(EnvoyUpstreamHealthCheckedCluster)                                                   \

include/envoy/router/router.h

@@ -37,10 +37,13 @@ class RedirectEntry {
 class RetryPolicy {
 public:
   // clang-format off
-  static const uint32_t RETRY_ON_5XX             = 0x1;
-  static const uint32_t RETRY_ON_CONNECT_FAILURE = 0x2;
-  static const uint32_t RETRY_ON_RETRIABLE_4XX   = 0x4;
-  static const uint32_t RETRY_ON_REFUSED_STREAM  = 0x8;
+  static const uint32_t RETRY_ON_5XX                     = 0x1;
+  static const uint32_t RETRY_ON_CONNECT_FAILURE         = 0x2;
+  static const uint32_t RETRY_ON_RETRIABLE_4XX           = 0x4;
+  static const uint32_t RETRY_ON_REFUSED_STREAM          = 0x8;
+  static const uint32_t RETRY_ON_GRPC_CANCELLED          = 0x16;
+  static const uint32_t RETRY_ON_GRPC_DEADLINE_EXCEEDED  = 0x32;
+  static const uint32_t RETRY_ON_GRPC_RESOURCE_EXHAUSTED = 0x64;
   // clang-format on
 
   virtual ~RetryPolicy() {}

source/common/grpc/common.cc

@@ -24,6 +24,20 @@ namespace Grpc {
 
 const std::string Common::GRPC_CONTENT_TYPE{"application/grpc"};
 
+Common::GrpcStatus Common::getGrpcStatus(const Http::HeaderMap& trailers) {
+  const Http::HeaderEntry* grpc_status_header = trailers.GrpcStatus();
+
+  uint64_t grpc_status_code;
+  if (!grpc_status_header ||
+      !StringUtil::atoul(grpc_status_header->value().c_str(), grpc_status_code)) {
+    return GrpcStatus::InvalidCode;
+  }
+  if (grpc_status_code > InvalidCode) {
+    return GrpcStatus::InvalidCode;
+  }
+  return static_cast<GrpcStatus>(grpc_status_code);
+}
+
 void Common::chargeStat(const Upstream::ClusterInfo& cluster, const std::string& grpc_service,
                         const std::string& grpc_method, bool success) {
   cluster.statsScope()

source/common/grpc/common.h

@@ -25,6 +25,34 @@ class Exception : public EnvoyException {
 
 class Common {
 public:
+  enum GrpcStatus {
+    Ok = 0,
+    Canceled = 1,
+    Unknown = 2,
+    InvalidArgument = 3,
+    DeadlineExceeded = 4,
+    NotFound = 5,
+    AlreadyExists = 6,
+    PermissionDenied = 7,
+    Unauthenticated = 16,
+    ResourceExhausted = 8,
+    FailedPrecondition = 9,
+    Aborted = 10,
+    OutOfRange = 11,
+    Unimplemented = 12,
+    Internal = 13,
+    Unavailable = 14,
+    DataLoss = 15,
+    InvalidCode = 16,
+  };
+
+  /**
+   * Returns the GrpcStatus code from a given set of headers, if present.
+   * @headers the headers to parse.
+   * @returns the parsed status code or InvalidCode if no valid status is found.
+   */
+  static GrpcStatus getGrpcStatus(const Http::HeaderMap& headers);
+
   /**
    * Charge a success/failure stat to a cluster/service/method.
    * @param cluster supplies the target cluster.

source/common/http/headers.h

@@ -30,6 +30,7 @@ class HeaderValues {
   const LowerCaseString EnvoyMaxRetries{"x-envoy-max-retries"};
   const LowerCaseString EnvoyOriginalPath{"x-envoy-original-path"};
   const LowerCaseString EnvoyRetryOn{"x-envoy-retry-on"};
+  const LowerCaseString EnvoyRetryGrpcOn{"x-envoy-retry-grpc-on"};
   const LowerCaseString EnvoyUpstreamAltStatName{"x-envoy-upstream-alt-stat-name"};
   const LowerCaseString EnvoyUpstreamCanary{"x-envoy-upstream-canary"};
   const LowerCaseString EnvoyUpstreamRequestTimeoutAltResponse{
@@ -92,6 +93,12 @@ class HeaderValues {
     const std::string Retriable4xx{"retriable-4xx"};
   } EnvoyRetryOnValues;
 
+  struct {
+    const std::string Cancelled{"cancelled"};
+    const std::string DeadlineExceeded{"deadline-exceeded"};
+    const std::string ResourceExhausted{"resource-exhausted"};
+  } EnvoyRetryOnGrpcValues;
+
   struct {
     const std::string _100Continue{"100-continue"};
   } ExpectValues;

source/common/router/BUILD

@@ -79,6 +79,7 @@ envoy_cc_library(
         "//include/envoy/upstream:upstream_interface",
         "//source/common/common:assert_lib",
         "//source/common/common:utility_lib",
+        "//source/common/grpc:common_lib",
         "//source/common/http:codes_lib",
         "//source/common/http:headers_lib",
         "//source/common/http:utility_lib",

source/common/router/retry_state_impl.cc

@@ -7,6 +7,7 @@
 
 #include "common/common/assert.h"
 #include "common/common/utility.h"
+#include "common/grpc/common.h"
 #include "common/http/codes.h"
 #include "common/http/headers.h"
 #include "common/http/utility.h"
@@ -19,6 +20,11 @@ namespace Router {
 const uint32_t RetryPolicy::RETRY_ON_5XX;
 const uint32_t RetryPolicy::RETRY_ON_CONNECT_FAILURE;
 const uint32_t RetryPolicy::RETRY_ON_RETRIABLE_4XX;
+const uint32_t RetryPolicy::RETRY_ON_GRPC_CANCELLED;
+const uint32_t RetryPolicy::RETRY_ON_GRPC_DEADLINE_EXCEEDED;
+const uint32_t RetryPolicy::RETRY_ON_GRPC_RESOURCE_EXHAUSTED;
+
+
 
 RetryStatePtr RetryStateImpl::create(const RetryPolicy& route_policy,
                                      Http::HeaderMap& request_headers,
@@ -29,7 +35,7 @@ RetryStatePtr RetryStateImpl::create(const RetryPolicy& route_policy,
   RetryStatePtr ret;
 
   // We short circuit here and do not both with an allocation if there is no chance we will retry.
-  if (request_headers.EnvoyRetryOn() || route_policy.retryOn()) {
+  if (request_headers.EnvoyRetryOn() || request_headers.EnvoyRetryGrpcOn() || route_policy.retryOn()) {
     ret.reset(new RetryStateImpl(route_policy, request_headers, cluster, runtime, random,
                                  dispatcher, priority));
   }
@@ -48,12 +54,15 @@ RetryStateImpl::RetryStateImpl(const RetryPolicy& route_policy, Http::HeaderMap&
 
   if (request_headers.EnvoyRetryOn()) {
     retry_on_ = parseRetryOn(request_headers.EnvoyRetryOn()->value().c_str());
-    if (retry_on_ != 0 && request_headers.EnvoyMaxRetries()) {
-      const char* max_retries = request_headers.EnvoyMaxRetries()->value().c_str();
-      uint64_t temp;
-      if (StringUtil::atoul(max_retries, temp)) {
-        retries_remaining_ = temp;
-      }
+  }
+  if (request_headers.EnvoyRetryGrpcOn()) {
+    retry_on_ |= parseRetryGrpcOn(request_headers.EnvoyRetryGrpcOn()->value().c_str());
+  }
+  if (retry_on_ != 0 && request_headers.EnvoyMaxRetries()) {
+    const char* max_retries = request_headers.EnvoyMaxRetries()->value().c_str();
+    uint64_t temp;
+    if (StringUtil::atoul(max_retries, temp)) {
+      retries_remaining_ = temp;
     }
   }
 
@@ -96,6 +105,22 @@ uint32_t RetryStateImpl::parseRetryOn(const std::string& config) {
   return ret;
 }
 
+uint32_t RetryStateImpl::parseRetryGrpcOn(const std::string& retry_grpc_on_header) {
+  uint32_t ret = 0;
+  std::vector<std::string> retry_on_list = StringUtil::split(retry_grpc_on_header, ',');
+  for (const std::string& retry_on : retry_on_list) {
+    if (retry_on == Http::Headers::get().EnvoyRetryOnGrpcValues.Cancelled) {
+      ret |= RetryPolicy::RETRY_ON_GRPC_CANCELLED;
+    } else if (retry_on == Http::Headers::get().EnvoyRetryOnGrpcValues.DeadlineExceeded) {
+      ret |= RetryPolicy::RETRY_ON_GRPC_DEADLINE_EXCEEDED;
+    } else if (retry_on == Http::Headers::get().EnvoyRetryOnGrpcValues.ResourceExhausted) {
+      ret |= RetryPolicy::RETRY_ON_GRPC_RESOURCE_EXHAUSTED;
+    }
+  }
+
+  return ret;
+}
+
 void RetryStateImpl::resetRetry() {
   if (callback_) {
     cluster_.resourceManager(priority_).retries().dec();
@@ -169,6 +194,17 @@ bool RetryStateImpl::wouldRetry(const Http::HeaderMap* response_headers,
     }
   }
 
+  if (retry_on_ & (RetryPolicy::RETRY_ON_GRPC_CANCELLED |
+                   RetryPolicy::RETRY_ON_GRPC_DEADLINE_EXCEEDED |
+                   RetryPolicy::RETRY_ON_GRPC_RESOURCE_EXHAUSTED) && response_headers) {
+    Grpc::Common::GrpcStatus status = Grpc::Common::getGrpcStatus(*response_headers);
+    if ((status == Grpc::Common::Canceled && (retry_on_ & RetryPolicy::RETRY_ON_GRPC_CANCELLED)) ||
+        (status == Grpc::Common::DeadlineExceeded && (retry_on_ & RetryPolicy::RETRY_ON_GRPC_DEADLINE_EXCEEDED)) ||
+        (status == Grpc::Common::ResourceExhausted && (retry_on_ & RetryPolicy::RETRY_ON_GRPC_RESOURCE_EXHAUSTED))) {
+      return true;
+    }
+  }
+
   return false;
 }
 

source/common/router/retry_state_impl.h

@@ -27,6 +27,9 @@ class RetryStateImpl : public RetryState {
 
   static uint32_t parseRetryOn(const std::string& config);
 
+  // Returns the RetryPolicy extracted from the x-envoy-retry-grpc-on header.
+  static uint32_t parseRetryGrpcOn(const std::string& retry_grpc_on_header);
+
   // Router::RetryState
   bool enabled() override { return retry_on_ != 0; }
   bool shouldRetry(const Http::HeaderMap* response_headers,

test/common/grpc/BUILD

@@ -26,6 +26,7 @@ envoy_cc_test(
         "//source/common/http:headers_lib",
         "//test/mocks/upstream:upstream_mocks",
         "//test/proto:helloworld_proto",
+        "//test/test_common:utility_lib",
     ],
 )
 

test/common/grpc/common_test.cc

@@ -3,12 +3,27 @@
 
 #include "test/mocks/upstream/mocks.h"
 #include "test/proto/helloworld.pb.h"
+#include "test/test_common/utility.h"
 
 #include "gtest/gtest.h"
 
 namespace Envoy {
 namespace Grpc {
 
+TEST(GrpcCommonTest, getGrpcStatus) {
+  Http::TestHeaderMapImpl ok_trailers{{"grpc-status", "0"}};
+  EXPECT_EQ(Common::Ok, Common::getGrpcStatus(ok_trailers));
+
+  Http::TestHeaderMapImpl no_status_trailers{{"foo", "bar"}};
+  EXPECT_EQ(Common::InvalidCode, Common::getGrpcStatus(no_status_trailers));
+
+  Http::TestHeaderMapImpl aborted_trailers{{"grpc-status", "10"}};
+  EXPECT_EQ(Common::Aborted, Common::getGrpcStatus(aborted_trailers));
+
+  Http::TestHeaderMapImpl invalid_trailers{{"grpc-status", "-1"}};
+  EXPECT_EQ(Common::InvalidCode, Common::getGrpcStatus(invalid_trailers));
+}
+
 TEST(GrpcCommonTest, chargeStats) {
   NiceMock<Upstream::MockClusterInfo> cluster;
   Common::chargeStat(cluster, "service", "method", true);

test/common/router/retry_state_impl_test.cc

@@ -101,6 +101,48 @@ TEST_F(RouterRetryStateImplTest, Policy5xxRemote503) {
   EXPECT_FALSE(state_->shouldRetry(&response_headers, no_reset_, callback_));
 }
 
+TEST_F(RouterRetryStateImplTest, PolicyGrpcCancelled) {
+  Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-grpc-on", "cancelled"}};
+  setup(request_headers);
+  EXPECT_TRUE(state_->enabled());
+
+  Http::TestHeaderMapImpl response_headers{{":status", "200"}, {"grpc-status", "1"}};
+  expectTimerCreateAndEnable();
+  EXPECT_TRUE(state_->shouldRetry(&response_headers, no_reset_, callback_));
+  EXPECT_CALL(callback_ready_, ready());
+  retry_timer_->callback_();
+
+  EXPECT_FALSE(state_->shouldRetry(&response_headers, no_reset_, callback_));
+}
+
+TEST_F(RouterRetryStateImplTest, PolicyGrpcDeadlineExceeded) {
+  Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-grpc-on", "deadline-exceeded"}};
+  setup(request_headers);
+  EXPECT_TRUE(state_->enabled());
+
+  Http::TestHeaderMapImpl response_headers{{":status", "200"}, {"grpc-status", "4"}};
+  expectTimerCreateAndEnable();
+  EXPECT_TRUE(state_->shouldRetry(&response_headers, no_reset_, callback_));
+  EXPECT_CALL(callback_ready_, ready());
+  retry_timer_->callback_();
+
+  EXPECT_FALSE(state_->shouldRetry(&response_headers, no_reset_, callback_));
+}
+
+TEST_F(RouterRetryStateImplTest, PolicyGrpcResourceExhausted) {
+  Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-grpc-on", "resource-exhausted"}};
+  setup(request_headers);
+  EXPECT_TRUE(state_->enabled());
+
+  Http::TestHeaderMapImpl response_headers{{":status", "200"}, {"grpc-status", "8"}};
+  expectTimerCreateAndEnable();
+  EXPECT_TRUE(state_->shouldRetry(&response_headers, no_reset_, callback_));
+  EXPECT_CALL(callback_ready_, ready());
+  retry_timer_->callback_();
+
+  EXPECT_FALSE(state_->shouldRetry(&response_headers, no_reset_, callback_));
+}
+
 TEST_F(RouterRetryStateImplTest, Policy5xxRemote200RemoteReset) {
   // Don't retry after reply start.
   Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-on", "5xx"}};

test/common/router/router_test.cc

@@ -722,6 +722,48 @@ TEST_F(RouterTest, RetryUpstream5xxNotComplete) {
                     .value());
 }
 
+
+TEST_F(RouterTest, RetryUpstreamGrpcCancelled) {
+  NiceMock<Http::MockStreamEncoder> encoder1;
+  Http::StreamDecoder* response_decoder = nullptr;
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke([&](Http::StreamDecoder& decoder, Http::ConnectionPool::Callbacks& callbacks)
+                           -> Http::ConnectionPool::Cancellable* {
+                             response_decoder = &decoder;
+                             callbacks.onPoolReady(encoder1, cm_.conn_pool_.host_);
+                             return nullptr;
+                           }));
+  expectResponseTimerCreate();
+
+  Http::TestHeaderMapImpl headers{{"x-envoy-grpc-retry-on", "cancelled"}, {"x-envoy-internal", "true"}};
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, true);
+
+  // 5xx response.
+  router_.retry_state_->expectRetry();
+  Http::HeaderMapPtr response_headers1(new Http::TestHeaderMapImpl{{":status", "200"}, {":grpc-status", "1"}});
+  EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(200));
+  response_decoder->decodeHeaders(std::move(response_headers1), true);
+
+  // We expect the grpc-status to result in a retried request.
+  EXPECT_CALL(encoder1.stream_, resetStream(_)).Times(0);
+  NiceMock<Http::MockStreamEncoder> encoder2;
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke([&](Http::StreamDecoder& decoder, Http::ConnectionPool::Callbacks& callbacks)
+                           -> Http::ConnectionPool::Cancellable* {
+                             response_decoder = &decoder;
+                             callbacks.onPoolReady(encoder2, cm_.conn_pool_.host_);
+                             return nullptr;
+                           }));
+  router_.retry_state_->callback_();
+
+  // Normal response.
+  EXPECT_CALL(*router_.retry_state_, shouldRetry(_, _, _)).WillOnce(Return(false));
+  Http::HeaderMapPtr response_headers(new Http::TestHeaderMapImpl{{":status", "200"}});
+  EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(200));
+  response_decoder->decodeHeaders(std::move(response_headers), true);
+}
+
 TEST_F(RouterTest, Shadow) {
   callbacks_.route_->route_entry_.shadow_policy_.cluster_ = "foo";
   callbacks_.route_->route_entry_.shadow_policy_.runtime_key_ = "bar";

test/integration/http2_integration_test.cc

@@ -91,6 +91,8 @@ TEST_P(Http2IntegrationTest, TwoRequests) { testTwoRequests(Http::CodecClient::T
 
 TEST_P(Http2IntegrationTest, Retry) { testRetry(Http::CodecClient::Type::HTTP2); }
 
+TEST_P(Http2IntegrationTest, GrpcRetry) { testGrpcRetry(); }
+
 TEST_P(Http2IntegrationTest, MaxHeadersInCodec) {
   Http::TestHeaderMapImpl big_headers{
       {":method", "GET"}, {":path", "/test/long/url"}, {":scheme", "http"}, {":authority", "host"}};

test/integration/http2_upstream_integration_test.cc

@@ -93,6 +93,8 @@ TEST_P(Http2UpstreamIntegrationTest, TwoRequests) {
 
 TEST_P(Http2UpstreamIntegrationTest, Retry) { testRetry(Http::CodecClient::Type::HTTP2); }
 
+TEST_P(Http2UpstreamIntegrationTest, GrpcRetry) { testGrpcRetry(); }
+
 TEST_P(Http2UpstreamIntegrationTest, DownstreamResetBeforeResponseComplete) {
   testDownstreamResetBeforeResponseComplete();
 }