retry extensions: implement "other priority" extension

api/envoy/config/retry/other_priority/BUILD

@@ -0,0 +1,11 @@
+licenses(["notice"])  # Apache 2
+
+load("//bazel:api_build_system.bzl", "api_proto_library_internal")
+
+api_proto_library_internal(
+    name = "other_priority",
+    srcs = ["other_priority_config.proto"],
+    deps = [
+        "//envoy/api/v2/core:base",
+    ],
+)

api/envoy/config/retry/other_priority/other_priority_config.proto

@@ -0,0 +1,11 @@
+syntax = "proto3";
+
+package envoy.config.retry.other_priority;
+
+// A retry host selector that attempts to spread retries between priorities, even if certain
+// priorities would not normally be attempted due to higher priorities being available.
+message OtherPriorityConfig {
+  // How often the priority load should be updated based on previously attempted priorities. Useful
+  // to allow each priorities to receive more than one request before being excluded.
+  int32 update_frequency = 1;
+}

include/envoy/upstream/retry.h

@@ -109,6 +109,8 @@ class RetryPriorityFactory {
                                    const Protobuf::Message& config) PURE;
 
   virtual std::string name() const PURE;
+
+  virtual ProtobufTypes::MessagePtr createEmptyConfigProto() PURE;
 };
 
 /**
@@ -125,6 +127,8 @@ class RetryHostPredicateFactory {
    * @return name name of this factory.
    */
   virtual std::string name() PURE;
+
+  virtual ProtobufTypes::MessagePtr createEmptyConfigProto() PURE;
 };
 
 } // namespace Upstream

source/common/router/config_impl.cc

@@ -50,15 +50,21 @@ RetryPolicyImpl::RetryPolicyImpl(const envoy::api::v2::route::RouteAction& confi
   retry_on_ |= RetryStateImpl::parseRetryGrpcOn(config.retry_policy().retry_on());
 
   for (const auto& host_predicate : config.retry_policy().retry_host_predicate()) {
-    Registry::FactoryRegistry<Upstream::RetryHostPredicateFactory>::getFactory(
-        host_predicate.name())
-        ->createHostPredicate(*this, host_predicate.config());
+    auto& factory =
+        ::Envoy::Config::Utility::getAndCheckFactory<Upstream::RetryHostPredicateFactory>(
+            host_predicate.name());
+
+    auto config = ::Envoy::Config::Utility::translateToFactoryConfig(host_predicate, factory);
+    factory.createHostPredicate(*this, *config);
   }
 
   const auto retry_priority = config.retry_policy().retry_priority();
   if (!retry_priority.name().empty()) {
-    Registry::FactoryRegistry<Upstream::RetryPriorityFactory>::getFactory(retry_priority.name())
-        ->createRetryPriority(*this, retry_priority.config());
+    auto& factory = ::Envoy::Config::Utility::getAndCheckFactory<Upstream::RetryPriorityFactory>(
+        retry_priority.name());
+
+    auto config = ::Envoy::Config::Utility::translateToFactoryConfig(retry_priority, factory);
+    factory.createRetryPriority(*this, *config);
   }
 
   auto host_selection_attempts = config.retry_policy().host_selection_retry_max_attempts();

source/common/upstream/load_balancer_impl.cc

@@ -44,25 +44,29 @@ LoadBalancerBase::LoadBalancerBase(const PrioritySet& priority_set, ClusterStats
           common_config, healthy_panic_threshold, 100, 50)),
       priority_set_(priority_set) {
   for (auto& host_set : priority_set_.hostSetsPerPriority()) {
-    recalculatePerPriorityState(host_set->priority());
+    recalculatePerPriorityState(host_set->priority(), priority_set_, per_priority_load_,
+                                per_priority_health_);
   }
-  priority_set_.addMemberUpdateCb(
-      [this](uint32_t priority, const HostVector&, const HostVector&) -> void {
-        recalculatePerPriorityState(priority);
-      });
+  priority_set_.addMemberUpdateCb([this](uint32_t priority, const HostVector&,
+                                         const HostVector&) -> void {
+    recalculatePerPriorityState(priority, priority_set_, per_priority_load_, per_priority_health_);
+  });
 }
 
-void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority) {
-  per_priority_load_.resize(priority_set_.hostSetsPerPriority().size());
-  per_priority_health_.resize(priority_set_.hostSetsPerPriority().size());
+void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority,
+                                                   const PrioritySet& priority_set,
+                                                   PriorityLoad& per_priority_load,
+                                                   std::vector<uint32_t>& per_priority_health) {
+  per_priority_load.resize(priority_set.hostSetsPerPriority().size());
+  per_priority_health.resize(priority_set.hostSetsPerPriority().size());
 
   // Determine the health of the newly modified priority level.
   // Health ranges from 0-100, and is the ratio of healthy hosts to total hosts, modified by the
   // overprovisioning factor.
-  HostSet& host_set = *priority_set_.hostSetsPerPriority()[priority];
-  per_priority_health_[priority] = 0;
+  HostSet& host_set = *priority_set.hostSetsPerPriority()[priority];
+  per_priority_health[priority] = 0;
   if (host_set.hosts().size() > 0) {
-    per_priority_health_[priority] =
+    per_priority_health[priority] =
         std::min<uint32_t>(100, (host_set.overprovisioning_factor() *
                                  host_set.healthyHosts().size() / host_set.hosts().size()));
   }
@@ -74,25 +78,25 @@ void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority) {
   // 3 host sets with 20% / 20% / 10% health they will get 40% / 40% / 20% load to ensure total load
   // adds up to 100.
   const uint32_t total_health = std::min<uint32_t>(
-      std::accumulate(per_priority_health_.begin(), per_priority_health_.end(), 0), 100);
+      std::accumulate(per_priority_health.begin(), per_priority_health.end(), 0), 100);
   if (total_health == 0) {
     // Everything is terrible. Send all load to P=0.
     // In this one case sumEntries(per_priority_load_) != 100 since we sinkhole all traffic in P=0.
-    per_priority_load_[0] = 100;
+    per_priority_load[0] = 100;
     return;
   }
   size_t total_load = 100;
-  for (size_t i = 0; i < per_priority_health_.size(); ++i) {
+  for (size_t i = 0; i < per_priority_health.size(); ++i) {
     // Now assign as much load as possible to the high priority levels and cease assigning load when
     // total_load runs out.
-    per_priority_load_[i] =
-        std::min<uint32_t>(total_load, per_priority_health_[i] * 100 / total_health);
-    total_load -= per_priority_load_[i];
+    per_priority_load[i] =
+        std::min<uint32_t>(total_load, per_priority_health[i] * 100 / total_health);
+    total_load -= per_priority_load[i];
   }
   if (total_load != 0) {
     // Account for rounding errors.
-    ASSERT(total_load < per_priority_load_.size());
-    per_priority_load_[0] += total_load;
+    ASSERT(total_load < per_priority_load.size());
+    per_priority_load[0] += total_load;
   }
 }
 

source/common/upstream/load_balancer_impl.h

@@ -70,10 +70,13 @@ class LoadBalancerBase : public LoadBalancer {
   // The priority-ordered set of hosts to use for load balancing.
   const PrioritySet& priority_set_;
 
+public:
   // Called when a host set at the given priority level is updated. This updates
   // per_priority_health_ for that priority level, and may update per_priority_load_ for all
   // priority levels.
-  void recalculatePerPriorityState(uint32_t priority);
+  void static recalculatePerPriorityState(uint32_t priority, const PrioritySet& priority_set,
+                                          PriorityLoad& priority_load,
+                                          std::vector<uint32_t>& per_priority_health_);
 
   // The percentage load (0-100) for each priority level
   std::vector<uint32_t> per_priority_load_;

source/extensions/extensions_build_config.bzl

@@ -111,6 +111,9 @@ EXTENSIONS = {
 
     # Retry host predicates
     "envoy.retry_host_predicates.other_hosts":          "//source/extensions/retry/host/other_hosts:config",
+
+    # Retry priorities
+    "envoy.retry_priorities.previous_priorities":       "//source/extensions/retry/priority/other_priority:config",
 }
 
 WINDOWS_EXTENSIONS = {

source/extensions/retry/host/other_hosts/config.h

@@ -18,6 +18,10 @@ class OtherHostsRetryPredicateFactory : public Upstream::RetryHostPredicateFacto
   }
 
   std::string name() override { return RetryHostPredicateValues::get().PreviousHostsPredicate; }
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return ProtobufTypes::MessagePtr{new Envoy::ProtobufWkt::Empty()};
+  }
 };
 
 } // namespace Host

source/extensions/retry/priority/BUILD

@@ -0,0 +1,17 @@
+licenses(["notice"])  # Apache 2
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "well_known_names",
+    hdrs = ["well_known_names.h"],
+    deps = [
+        "//source/common/singleton:const_singleton",
+    ],
+)

source/extensions/retry/priority/other_priority/BUILD

@@ -0,0 +1,33 @@
+licenses(["notice"])  # Apache 2
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "other_priority_lib",
+    srcs = ["other_priority.cc"],
+    hdrs = ["other_priority.h"],
+    deps = [
+        "//include/envoy/upstream:retry_interface",
+        "//source/common/upstream:load_balancer_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":other_priority_lib",
+        "//include/envoy/registry",
+        "//include/envoy/upstream:retry_interface",
+        "//source/common/protobuf",
+        "//source/extensions/retry/priority:well_known_names",
+        "@envoy_api//envoy/config/retry/other_priority:other_priority_cc",
+    ],
+)

source/extensions/retry/priority/other_priority/config.cc

@@ -0,0 +1,26 @@
+#include "extensions/retry/priority/other_priority/config.h"
+
+#include "envoy/config/retry/other_priority/other_priority_config.pb.validate.h"
+#include "envoy/registry/registry.h"
+#include "envoy/upstream/retry.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Retry {
+namespace Priority {
+
+void OtherPriorityRetryPriorityFactory::createRetryPriority(
+    Upstream::RetryPriorityFactoryCallbacks& callbacks, const Protobuf::Message& config) {
+  callbacks.addRetryPriority(std::make_shared<OtherPriorityRetryPriority>(
+      MessageUtil::downcastAndValidate<
+          const envoy::config::retry::other_priority::OtherPriorityConfig&>(config)
+          .update_frequency()));
+}
+
+static Registry::RegisterFactory<OtherPriorityRetryPriorityFactory, Upstream::RetryPriorityFactory>
+    register_;
+
+} // namespace Priority
+} // namespace Retry
+} // namespace Extensions
+} // namespace Envoy

source/extensions/retry/priority/other_priority/config.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include "envoy/upstream/retry.h"
+
+#include "common/protobuf/protobuf.h"
+
+#include "extensions/retry/priority/other_priority/other_priority.h"
+#include "extensions/retry/priority/well_known_names.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Retry {
+namespace Priority {
+
+class OtherPriorityRetryPriorityFactory : public Upstream::RetryPriorityFactory {
+public:
+  void createRetryPriority(Upstream::RetryPriorityFactoryCallbacks& callbacks,
+                           const Protobuf::Message& config) override;
+
+  std::string name() const override {
+    return RetryPriorityValues::get().PreviousPrioritiesRetryPriority;
+  }
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return ProtobufTypes::MessagePtr(new ::Envoy::ProtobufWkt::Empty());
+  }
+};
+
+} // namespace Priority
+} // namespace Retry
+} // namespace Extensions
+} // namespace Envoy

source/extensions/retry/priority/other_priority/other_priority.cc

@@ -0,0 +1,98 @@
+#include "extensions/retry/priority/other_priority/other_priority.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Retry {
+namespace Priority {
+const Upstream::PriorityLoad& OtherPriorityRetryPriority::determinePriorityLoad(
+    const Upstream::PrioritySet& priority_set,
+    const Upstream::PriorityLoad& original_priority_load) {
+  if (!priority_set_) {
+    priority_set_ = &priority_set;
+
+    excluded_priorities_.resize(original_priority_load.size());
+    // Initialize our local priority_load_ and priority_health_,
+    // keeping them in sync with the member update cb.
+    for (auto& host_set : priority_set_->hostSetsPerPriority()) {
+      recalculatePerPriorityState(host_set->priority());
+    }
+    callback_ = priority_set.addMemberUpdateCb([&, this](int priority, auto, auto) {
+      this->recalculatePerPriorityState(priority);
+      this->adjustForAttemptedPriorities();
+    });
+  }
+
+  // If we've not seen enough retries to modify the priority load, just
+  // return the original.
+  // If this retry should trigger an update, recalculate the priority load by excluding attempted
+  // priorities.
+  if (attempted_priorites_.size() < update_frequency_) {
+    return original_priority_load;
+  } else if (attempted_priorites_.size() % update_frequency_ == 0) {
+    for (auto priority : attempted_priorites_) {
+      excluded_priorities_[priority] = true;
+    }
+
+    adjustForAttemptedPriorities();
+  }
+
+  return per_priority_load_;
+}
+
+void OtherPriorityRetryPriority::adjustForAttemptedPriorities() {
+  // If all priorities are unhealthy to begin with, there's nothing to do.
+  if (!std::accumulate(per_priority_load_.begin(), per_priority_load_.end(), 0)) {
+    return;
+  }
+
+  // First we create an adjusted health view of the priorities, where attempted priorities are
+  // given a zero weight.
+  uint32_t total_health = 0;
+  std::vector<uint32_t> adjusted_per_priority_health;
+  adjusted_per_priority_health.resize(per_priority_health_.size());
+  for (size_t i = 0; i < per_priority_health_.size(); ++i) {
+    if (!excluded_priorities_[i]) {
+      adjusted_per_priority_health[i] = per_priority_health_[i];
+      total_health += per_priority_health_[i];
+    }
+  }
+
+  total_health = std::min(total_health, 100u);
+
+  // If there are no healthy priorities left, we reset the attempted priorities and recurse.
+  // This allows us to fall back to the unmodified priority load when we run out of priorites
+  // instead of failing to route requests.
+  if (total_health == 0) {
+    for (size_t i = 0; i < excluded_priorities_.size(); ++i) {
+      excluded_priorities_[i] = false;
+    }
+    attempted_priorites_.clear();
+
+    adjustForAttemptedPriorities();
+    return;
+  }
+
+  // We then adjust the load by rebalancing priorities with the adjusted health values.
+  Upstream::PriorityLoad per_priority_load;
+  per_priority_load.resize(per_priority_load_.size());
+
+  size_t total_load = 100;
+  // The outer loop is used to eliminate rounding errors: any remaining load will be assigned to the
+  // first healthy priority.
+  while (total_load != 0) {
+    for (size_t i = 0; i < adjusted_per_priority_health.size(); ++i) {
+      // Now assign as much load as possible to the high priority levels and cease assigning load
+      // when total_load runs out.
+      auto delta =
+          std::min<uint32_t>(total_load, adjusted_per_priority_health[i] * 100 / total_health);
+      per_priority_load[i] += delta;
+      total_load -= delta;
+    }
+  }
+
+  per_priority_load_ = per_priority_load;
+}
+} // namespace Priority
+} // namespace Retry
+} // namespace Extensions
+} // namespace Envoy