upstream: Implement retry concurrency budgets

api/envoy/api/v2/route/route.proto

@@ -893,6 +893,20 @@ message RetryPolicy {
     google.protobuf.Duration max_interval = 2 [(validate.rules).duration = {gt {}}];
   }
 
+  message RetryBudget {
+    // Specifies the limit on concurrent retries as a percentage of outstanding requests. This
+    // parameter is optional.
+    //
+    // Defaults to 20%.
+    type.Percent percent_budget = 1;
+
+    // Specifies the minimum retry concurrency allowed for the retry budget. This parameter is
+    // optional.
+    //
+    // Defaults to 10.
+    google.protobuf.UInt32Value min_concurrency = 2;
+  }
+
   // Specifies the conditions under which retry takes place. These are the same
   // conditions documented for :ref:`config_http_filters_router_x-envoy-retry-on` and
   // :ref:`config_http_filters_router_x-envoy-retry-grpc-on`.
@@ -949,6 +963,10 @@ message RetryPolicy {
 
   // HTTP headers which must be present in the request for retries to be attempted.
   repeated HeaderMatcher retriable_request_headers = 10;
+
+  // Specifies a limit on concurrent retries in relation to the number of active
+  // requests/connections. This parameter is optional.
+  RetryBudget retry_budget = 11;
 }
 
 // HTTP request hedging :ref:`architecture overview <arch_overview_http_routing_hedging>`.

api/envoy/api/v3alpha/route/route.proto

@@ -837,6 +837,20 @@ message RetryPolicy {
     google.protobuf.Duration max_interval = 2 [(validate.rules).duration = {gt {}}];
   }
 
+  message RetryBudget {
+    // Specifies the limit on concurrent retries as a percentage of outstanding requests. This
+    // parameter is optional.
+    //
+    // Defaults to 20%.
+    type.Percent percent_budget = 1;
+
+    // Specifies the minimum retry concurrency allowed for the retry budget. This parameter is
+    // optional.
+    //
+    // Defaults to 10.
+    google.protobuf.UInt32Value min_concurrency = 2;
+  }
+
   // Specifies the conditions under which retry takes place. These are the same
   // conditions documented for :ref:`config_http_filters_router_x-envoy-retry-on` and
   // :ref:`config_http_filters_router_x-envoy-retry-grpc-on`.
@@ -893,6 +907,10 @@ message RetryPolicy {
 
   // HTTP headers which must be present in the request for retries to be attempted.
   repeated HeaderMatcher retriable_request_headers = 10;
+
+  // Specifies a limit on concurrent retries in relation to the number of active
+  // requests/connections. This parameter is optional.
+  RetryBudget retry_budget = 11;
 }
 
 // HTTP request hedging :ref:`architecture overview <arch_overview_http_routing_hedging>`.

docs/root/configuration/upstream/cluster_manager/cluster_stats.rst

@@ -73,6 +73,7 @@ Every cluster has a statistics tree rooted at *cluster.<name>.* with the followi
   upstream_rq_rx_reset, Counter, Total requests that were reset remotely
   upstream_rq_tx_reset, Counter, Total requests that were reset locally
   upstream_rq_retry, Counter, Total request retries
+  upstream_rq_retry_budget_exceeded, Counter, Total requests not retried due to retry budgets
   upstream_rq_retry_success, Counter, Total request retry successes
   upstream_rq_retry_overflow, Counter, Total requests not retried due to circuit breaking
   upstream_flow_control_paused_reading_total, Counter, Total number of times flow control paused reading from upstream

include/envoy/router/router.h

@@ -164,6 +164,18 @@ class RetryPolicy {
   static const uint32_t RETRY_ON_RETRIABLE_HEADERS       = 0x1000;
   // clang-format on
 
+  /**
+   * Limitations placed on concurrent retries as a percentage of the number of active requests.
+   */
+  struct RetryBudget {
+    // The percentage of active requests that are allowed to be retries.
+    double budget_pct;
+
+    // The minimum number of active requests before enforcing the retry budget.
+    uint32_t min_concurrency;
+  };
+  using RetryBudget = struct RetryBudget;
+
   virtual ~RetryPolicy() = default;
 
   /**
@@ -227,6 +239,12 @@ class RetryPolicy {
    * @return absl::optional<std::chrono::milliseconds> maximum retry interval
    */
   virtual absl::optional<std::chrono::milliseconds> maxInterval() const PURE;
+
+  /**
+   * @return absl::optional<RetryBudget> limit on allowed concurrent retries in relation to current
+   * outstanding requests.
+   */
+  virtual absl::optional<RetryBudget> retryBudget() const PURE;
 };
 
 /**

include/envoy/upstream/resource_manager.h

@@ -47,6 +47,11 @@ class Resource {
    * @return the current maximum allowed number of this resource.
    */
   virtual uint64_t max() PURE;
+
+  /**
+   * @return the current resource count.
+   */
+  virtual uint64_t count() PURE;
 };
 
 /**

include/envoy/upstream/upstream.h

@@ -568,6 +568,7 @@ class PrioritySet {
   COUNTER(upstream_rq_pending_total)                                                               \
   COUNTER(upstream_rq_per_try_timeout)                                                             \
   COUNTER(upstream_rq_retry)                                                                       \
+  COUNTER(upstream_rq_retry_budget_exceeded)                                                       \
   COUNTER(upstream_rq_retry_overflow)                                                              \
   COUNTER(upstream_rq_retry_success)                                                               \
   COUNTER(upstream_rq_rx_reset)                                                                    \

source/common/http/async_client_impl.h

@@ -143,6 +143,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
       return absl::nullopt;
     }
     absl::optional<std::chrono::milliseconds> maxInterval() const override { return absl::nullopt; }
+    absl::optional<RetryBudget> retryBudget() const override { return absl::nullopt; }
 
     const std::vector<uint32_t> retriable_status_codes_;
     const std::vector<Http::HeaderMatcherSharedPtr> retriable_headers_;

source/common/router/config_impl.cc

@@ -121,6 +121,14 @@ RetryPolicyImpl::RetryPolicyImpl(const envoy::api::v2::route::RetryPolicy& retry
             "retry_policy.max_interval must greater than or equal to the base_interval");
       }
     }
+
+    if (retry_policy.has_retry_budget()) {
+      retry_budget_ = {};
+      retry_budget_->budget_pct =
+          PROTOBUF_PERCENT_TO_DOUBLE_OR_DEFAULT(retry_policy.retry_budget(), percent_budget, 20.0);
+      retry_budget_->min_concurrency =
+          PROTOBUF_GET_WRAPPED_OR_DEFAULT(retry_policy.retry_budget(), min_concurrency, 10);
+    }
   }
 }
 

source/common/router/config_impl.h

@@ -255,6 +255,7 @@ class RetryPolicyImpl : public RetryPolicy {
   }
   absl::optional<std::chrono::milliseconds> baseInterval() const override { return base_interval_; }
   absl::optional<std::chrono::milliseconds> maxInterval() const override { return max_interval_; }
+  absl::optional<RetryBudget> retryBudget() const override { return retry_budget_; }
 
 private:
   std::chrono::milliseconds per_try_timeout_{0};
@@ -274,6 +275,7 @@ class RetryPolicyImpl : public RetryPolicy {
   absl::optional<std::chrono::milliseconds> base_interval_;
   absl::optional<std::chrono::milliseconds> max_interval_;
   ProtobufMessage::ValidationVisitor* validation_visitor_{};
+  absl::optional<RetryBudget> retry_budget_;
 };
 
 /**

source/common/router/retry_state_impl.cc

@@ -59,7 +59,8 @@ RetryStateImpl::RetryStateImpl(const RetryPolicy& route_policy, Http::HeaderMap&
       retry_priority_(route_policy.retryPriority()),
       retriable_status_codes_(route_policy.retriableStatusCodes()),
       retriable_headers_(route_policy.retriableHeaders()),
-      retriable_request_headers_(route_policy.retriableRequestHeaders()) {
+      retriable_request_headers_(route_policy.retriableRequestHeaders()),
+      retry_budget_(route_policy.retryBudget()) {
 
   retry_on_ = route_policy.retryOn();
   retries_remaining_ = std::max(retries_remaining_, route_policy.numRetries());
@@ -232,6 +233,11 @@ RetryStatus RetryStateImpl::shouldRetry(bool would_retry, DoRetryCallback callba
     return RetryStatus::No;
   }
 
+  if (retryBudgetExceeded()) {
+    cluster_.stats().upstream_rq_retry_budget_exceeded_.inc();
+    return RetryStatus::NoRetryLimitExceeded;
+  }
+
   ASSERT(!callback_);
   callback_ = callback;
   cluster_.resourceManager(priority_).retries().inc();
@@ -240,6 +246,25 @@ RetryStatus RetryStateImpl::shouldRetry(bool would_retry, DoRetryCallback callba
   return RetryStatus::Yes;
 }
 
+bool RetryStateImpl::retryBudgetExceeded() {
+  if (!retry_budget_) {
+    return false;
+  }
+
+  // If a retry budget was configured, we cannot exceed the configured percentage of total
+  // outstanding requests/connections.
+  const uint64_t current_active = cluster_.resourceManager(priority_).connections().count() +
+                                  cluster_.resourceManager(priority_).requests().count() +
+                                  cluster_.resourceManager(priority_).pendingRequests().count();
+
+  if (current_active < retry_budget_->min_concurrency) {
+    return false;
+  }
+
+  const double normalized_pct = retry_budget_->budget_pct / 100.0;
+  return cluster_.resourceManager(priority_).retries().count() >= normalized_pct * current_active;
+}
+
 RetryStatus RetryStateImpl::shouldRetryHeaders(const Http::HeaderMap& response_headers,
                                                DoRetryCallback callback) {
   return shouldRetry(wouldRetryFromHeaders(response_headers), callback);

source/common/router/retry_state_impl.h

@@ -93,6 +93,7 @@ class RetryStateImpl : public RetryState {
   void enableBackoffTimer();
   void resetRetry();
   bool wouldRetryFromReset(const Http::StreamResetReason reset_reason);
+  bool retryBudgetExceeded();
   RetryStatus shouldRetry(bool would_retry, DoRetryCallback callback);
 
   const Upstream::ClusterInfo& cluster_;
@@ -111,6 +112,7 @@ class RetryStateImpl : public RetryState {
   std::vector<uint32_t> retriable_status_codes_;
   std::vector<Http::HeaderMatcherSharedPtr> retriable_headers_;
   std::vector<Http::HeaderMatcherSharedPtr> retriable_request_headers_;
+  absl::optional<RetryPolicy::RetryBudget> retry_budget_;
 };
 
 } // namespace Router

source/common/upstream/resource_manager_impl.h

@@ -72,6 +72,7 @@ class ResourceManagerImpl : public ResourceManager {
       open_gauge_.set(canCreate() ? 0 : 1);
     }
     uint64_t max() override { return runtime_.snapshot().getInteger(runtime_key_, max_); }
+    uint64_t count() override { return current_.load(); }
 
     /**
      * We set the gauge instead of incrementing and decrementing because,

test/common/router/retry_state_impl_test.cc

@@ -26,6 +26,8 @@ namespace {
 
 class RouterRetryStateImplTest : public testing::Test {
 public:
+  enum TestResourceType { Connection, Request, PendingRequest, Retry };
+
   RouterRetryStateImplTest() : callback_([this]() -> void { callback_ready_.ready(); }) {
     ON_CALL(runtime_.snapshot_, featureEnabled("upstream.use_retry", 100))
         .WillByDefault(Return(true));
@@ -46,6 +48,46 @@ class RouterRetryStateImplTest : public testing::Test {
     EXPECT_CALL(*retry_timer_, enableTimer(_, _));
   }
 
+  void incrOutstandingResource(TestResourceType resource, uint32_t num) {
+    for (uint32_t i = 0; i < num; ++i) {
+      switch (resource) {
+      case TestResourceType::Retry:
+        cluster_.resourceManager(Upstream::ResourcePriority::Default).retries().inc();
+        resource_manager_cleanup_tasks_.emplace_back([this]() {
+          cluster_.resourceManager(Upstream::ResourcePriority::Default).retries().dec();
+        });
+        break;
+      case TestResourceType::Connection:
+        cluster_.resourceManager(Upstream::ResourcePriority::Default).connections().inc();
+        resource_manager_cleanup_tasks_.emplace_back([this]() {
+          cluster_.resourceManager(Upstream::ResourcePriority::Default).connections().dec();
+        });
+        break;
+      case TestResourceType::Request:
+        cluster_.resourceManager(Upstream::ResourcePriority::Default).requests().inc();
+        resource_manager_cleanup_tasks_.emplace_back([this]() {
+          cluster_.resourceManager(Upstream::ResourcePriority::Default).requests().dec();
+        });
+        break;
+      case TestResourceType::PendingRequest:
+        cluster_.resourceManager(Upstream::ResourcePriority::Default).pendingRequests().inc();
+        resource_manager_cleanup_tasks_.emplace_back([this]() {
+          cluster_.resourceManager(Upstream::ResourcePriority::Default).pendingRequests().dec();
+        });
+        break;
+      }
+    }
+  }
+
+  void cleanupOutstandingResources() {
+    for (auto& task : resource_manager_cleanup_tasks_) {
+      task();
+    }
+    resource_manager_cleanup_tasks_.clear();
+  }
+
+  void TearDown() override { cleanupOutstandingResources(); }
+
   NiceMock<TestRetryPolicy> policy_;
   NiceMock<Upstream::MockClusterInfo> cluster_;
   NiceMock<Runtime::MockLoader> runtime_;
@@ -55,6 +97,7 @@ class RouterRetryStateImplTest : public testing::Test {
   RetryStatePtr state_;
   ReadyWatcher callback_ready_;
   RetryState::DoRetryCallback callback_;
+  std::vector<std::function<void()>> resource_manager_cleanup_tasks_;
 
   const Http::StreamResetReason remote_reset_{Http::StreamResetReason::RemoteReset};
   const Http::StreamResetReason remote_refused_stream_reset_{
@@ -892,6 +935,83 @@ TEST_F(RouterRetryStateImplTest, NoPreferredOverLimitExceeded) {
   EXPECT_EQ(RetryStatus::No, state_->shouldRetryHeaders(good_response_headers, callback_));
 }
 
+// Test percentage-based retry budgets configured to allow retries.
+TEST_F(RouterRetryStateImplTest, RetryBudgetEnforceCxAllow) {
+  policy_.retry_budget_ = RetryPolicy::RetryBudget();
+  policy_.retry_budget_->budget_pct = 20.0;
+  policy_.retry_budget_->min_concurrency = 5;
+  cluster_.resetResourceManager(10 /* cx */, 10 /* rq_pending */, 10 /* rq */, 10 /* rq_retry */,
+                                10 /* conn_pool */);
+
+  Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-on", "5xx"},
+                                          {"x-envoy-max-retries", "5"}};
+
+  // Mix of requests/connections that meet the minimum concurrency.
+  incrOutstandingResource(TestResourceType::Connection, 1);
+  incrOutstandingResource(TestResourceType::Request, 2);
+  incrOutstandingResource(TestResourceType::PendingRequest, 2);
+
+  setup(request_headers);
+  EXPECT_TRUE(state_->enabled());
+
+  expectTimerCreateAndEnable();
+  Http::TestHeaderMapImpl bad_response_headers{{":status", "503"}};
+  EXPECT_EQ(RetryStatus::Yes, state_->shouldRetryHeaders(bad_response_headers, callback_));
+  EXPECT_EQ(0UL, cluster_.stats().upstream_rq_retry_budget_exceeded_.value());
+}
+
+// Test percentage-based retry budgets that haven't met the minimum concurrency. It is expected that
+// the budget will not be enforced.
+TEST_F(RouterRetryStateImplTest, RetryBudgetMinimumConcurrency) {
+  policy_.retry_budget_ = RetryPolicy::RetryBudget();
+  policy_.retry_budget_->budget_pct = 20.0;
+  policy_.retry_budget_->min_concurrency = 5;
+  cluster_.resetResourceManager(10 /* cx */, 10 /* rq_pending */, 10 /* rq */, 11 /* rq_retry */,
+                                10 /* conn_pool */);
+
+  incrOutstandingResource(TestResourceType::Retry, 1);
+  incrOutstandingResource(TestResourceType::Request, 1);
+
+  Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-on", "5xx"},
+                                          {"x-envoy-max-retries", "5"}};
+
+  setup(request_headers);
+  EXPECT_TRUE(state_->enabled());
+
+  // Even though there's a 1:1 ratio between retries and outstanding requests with a 20% budget
+  // configured, the minimum concurrency requirement hasn't been met. Therefore, we expect the retry
+  // to be allowed since max retries are 5.
+  expectTimerCreateAndEnable();
+  Http::TestHeaderMapImpl bad_response_headers{{":status", "503"}};
+  EXPECT_EQ(RetryStatus::Yes, state_->shouldRetryHeaders(bad_response_headers, callback_));
+  EXPECT_EQ(0UL, cluster_.stats().upstream_rq_retry_budget_exceeded_.value());
+}
+
+// Test percentage-based retry budgets configured to disallow retries.
+TEST_F(RouterRetryStateImplTest, RetryBudgetEnforceDisallow) {
+  policy_.retry_budget_ = RetryPolicy::RetryBudget();
+  policy_.retry_budget_->budget_pct = 20.0;
+  policy_.retry_budget_->min_concurrency = 5;
+  cluster_.resetResourceManager(50 /* cx */, 10 /* rq_pending */, 10 /* rq */, 11 /* rq_retry */,
+                                10 /* conn_pool */);
+
+  Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-on", "5xx"},
+                                          {"x-envoy-max-retries", "5"}};
+
+  setup(request_headers);
+  EXPECT_TRUE(state_->enabled());
+
+  // Mix of requests/connections that meet the minimum concurrency.
+  incrOutstandingResource(TestResourceType::Retry, 1);
+  incrOutstandingResource(TestResourceType::Connection, 3);
+  incrOutstandingResource(TestResourceType::Request, 2);
+
+  Http::TestHeaderMapImpl bad_response_headers{{":status", "503"}};
+  EXPECT_EQ(RetryStatus::NoRetryLimitExceeded,
+            state_->shouldRetryHeaders(bad_response_headers, callback_));
+  EXPECT_EQ(1UL, cluster_.stats().upstream_rq_retry_budget_exceeded_.value());
+}
+
 } // namespace
 } // namespace Router
 } // namespace Envoy

test/mocks/router/mocks.h

@@ -110,6 +110,7 @@ class TestRetryPolicy : public RetryPolicy {
 
   absl::optional<std::chrono::milliseconds> baseInterval() const override { return base_interval_; }
   absl::optional<std::chrono::milliseconds> maxInterval() const override { return max_interval_; }
+  absl::optional<RetryBudget> retryBudget() const override { return retry_budget_; }
 
   std::chrono::milliseconds per_try_timeout_{0};
   uint32_t num_retries_{};
@@ -120,6 +121,7 @@ class TestRetryPolicy : public RetryPolicy {
   std::vector<Http::HeaderMatcherSharedPtr> retriable_request_headers_;
   absl::optional<std::chrono::milliseconds> base_interval_{};
   absl::optional<std::chrono::milliseconds> max_interval_{};
+  absl::optional<RetryBudget> retry_budget_{};
 };
 
 class MockRetryState : public RetryState {