upstream/extension: pass degraded load to plugins (envoyproxy#5628)

Changes the interface of the retry priority plugin and updates the
previous_priorities plugin to account for degraded hosts.

The plugin will ignore both healthy and degraded hosts for an attempted
priority, to retain the idea that we're excluding the entire priority
after it's been attempted.

Signed-off-by: Snow Pettersen <snowp@squareup.com>
Signed-off-by: Fred Douglas <fredlas@google.com>

docs/root/intro/arch_overview/http_connection_management.rst

@@ -67,7 +67,7 @@ this behavior retry plugins can be used, which fall into two categories:
   Envoy supports the following built-in priority predicates
 
   * *envoy.retry_priority.previous_priorities*: This will keep track of previously attempted priorities, 
-    and adjust the priority load such that other priorites will be targeted in subsequent retry attempts.
+    and adjust the priority load such that other priorities will be targeted in subsequent retry attempts.
 
 Host selection will continue until either the configured predicates accept the host or a configurable
 :ref:`max attempts <envoy_api_field_route.RetryPolicy.host_selection_retry_max_attempts>` has been reached. 

include/envoy/router/router.h

@@ -260,12 +260,12 @@ class RetryState {
   /**
    * Returns a reference to the PriorityLoad that should be used for the next retry.
    * @param priority_set current priority set.
-   * @param healthy_priority_load original priority load.
-   * @return PriorityLoad that should be used to select a priority for the next retry.
+   * @param original_priority_load original priority load.
+   * @return HealthyAndDegradedLoad that should be used to select a priority for the next retry.
    */
-  virtual const Upstream::HealthyLoad&
+  virtual const Upstream::HealthyAndDegradedLoad&
   priorityLoadForRetry(const Upstream::PrioritySet& priority_set,
-                       const Upstream::HealthyLoad& healthy_priority_load) PURE;
+                       const Upstream::HealthyAndDegradedLoad& original_priority_load) PURE;
   /**
    * return how many times host selection should be reattempted during host selection.
    */

include/envoy/upstream/load_balancer.h

@@ -55,8 +55,9 @@ class LoadBalancerContext {
    * @return a reference to the priority load data that should be used to select a priority.
    *
    */
-  virtual const HealthyLoad& determinePriorityLoad(const PrioritySet& priority_set,
-                                                   const HealthyLoad& original_priority_load) PURE;
+  virtual const HealthyAndDegradedLoad&
+  determinePriorityLoad(const PrioritySet& priority_set,
+                        const HealthyAndDegradedLoad& original_priority_load) PURE;
 
   /**
    * Called to determine whether we should reperform host selection. The load balancer

include/envoy/upstream/retry.h

@@ -21,12 +21,14 @@ class RetryPriority {
    * Determines what PriorityLoad to use.
    *
    * @param priority_set current priority set of cluster.
-   * @param original_priority the unmodified PriorityLoad.
-   * @return a reference to the PriorityLoad to use. Return original_priority if no changes should
-   * be made.
+   * @param original_priority_load the unmodified HealthAndDegradedLoad.
+   * @return HealthAndDegradedLoad load that should be used for the next retry. Return
+   * original_priority_load if the original load should be used. a pointer to riginal_priority,
+   * original_degraded_priority if no changes should be made.
    */
-  virtual const HealthyLoad& determinePriorityLoad(const PrioritySet& priority_set,
-                                                   const HealthyLoad& original_priority) PURE;
+  virtual const HealthyAndDegradedLoad&
+  determinePriorityLoad(const PrioritySet& priority_set,
+                        const HealthyAndDegradedLoad& original_priority_load) PURE;
 
   /**
    * Called after a host has been attempted but before host selection for the next attempt has

include/envoy/upstream/types.h

@@ -29,6 +29,11 @@ struct HealthyLoad : PriorityLoad {
   using PriorityLoad::PriorityLoad;
 };
 
+struct HealthyAndDegradedLoad {
+  HealthyLoad healthy_priority_load_;
+  DegradedLoad degraded_priority_load_;
+};
+
 // Phantom type indicating that the type is related to host availability.
 struct Availability {};
 

source/common/router/retry_state_impl.h

@@ -54,13 +54,13 @@ class RetryStateImpl : public RetryState {
         [&host](auto predicate) { return predicate->shouldSelectAnotherHost(host); });
   }
 
-  const Upstream::HealthyLoad&
+  const Upstream::HealthyAndDegradedLoad&
   priorityLoadForRetry(const Upstream::PrioritySet& priority_set,
-                       const Upstream::HealthyLoad& healthy_priority_load) override {
+                       const Upstream::HealthyAndDegradedLoad& original_priority_load) override {
     if (!retry_priority_) {
-      return healthy_priority_load;
+      return original_priority_load;
     }
-    return retry_priority_->determinePriorityLoad(priority_set, healthy_priority_load);
+    return retry_priority_->determinePriorityLoad(priority_set, original_priority_load);
   }
 
   uint32_t hostSelectionMaxAttempts() const override { return host_selection_max_attempts_; }

source/common/router/router.h

@@ -218,9 +218,9 @@ class Filter : Logger::Loggable<Logger::Id::router>,
     return retry_state_->shouldSelectAnotherHost(host);
   }
 
-  const Upstream::HealthyLoad&
+  const Upstream::HealthyAndDegradedLoad&
   determinePriorityLoad(const Upstream::PrioritySet& priority_set,
-                        const Upstream::HealthyLoad& original_priority_load) override {
+                        const Upstream::HealthyAndDegradedLoad& original_priority_load) override {
     // We only modify the priority load on retries.
     if (!is_retry_) {
       return original_priority_load;

source/common/upstream/load_balancer_impl.cc

@@ -90,18 +90,16 @@ 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(), priority_set_, healthy_per_priority_load_,
-                                degraded_per_priority_load_, per_priority_health_,
-                                per_priority_degraded_);
+    recalculatePerPriorityState(host_set->priority(), priority_set_, per_priority_load_,
+                                per_priority_health_, per_priority_degraded_);
   }
   // Reclaculate panic mode for all levels.
   recalculatePerPriorityPanic();
 
   priority_set_.addPriorityUpdateCb(
       [this](uint32_t priority, const HostVector&, const HostVector&) -> void {
-        recalculatePerPriorityState(priority, priority_set_, healthy_per_priority_load_,
-                                    degraded_per_priority_load_, per_priority_health_,
-                                    per_priority_degraded_);
+        recalculatePerPriorityState(priority, priority_set_, per_priority_load_,
+                                    per_priority_health_, per_priority_degraded_);
       });
 
   priority_set_.addPriorityUpdateCb(
@@ -125,12 +123,11 @@ LoadBalancerBase::LoadBalancerBase(const PrioritySet& priority_set, ClusterStats
 
 void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority,
                                                    const PrioritySet& priority_set,
-                                                   HealthyLoad& healthy_per_priority_load,
-                                                   DegradedLoad& degraded_per_priority_load,
+                                                   HealthyAndDegradedLoad& per_priority_load,
                                                    HealthyAvailability& per_priority_health,
                                                    DegradedAvailability& per_priority_degraded) {
-  healthy_per_priority_load.get().resize(priority_set.hostSetsPerPriority().size());
-  degraded_per_priority_load.get().resize(priority_set.hostSetsPerPriority().size());
+  per_priority_load.healthy_priority_load_.get().resize(priority_set.hostSetsPerPriority().size());
+  per_priority_load.degraded_priority_load_.get().resize(priority_set.hostSetsPerPriority().size());
   per_priority_health.get().resize(priority_set.hostSetsPerPriority().size());
   per_priority_degraded.get().resize(priority_set.hostSetsPerPriority().size());
 
@@ -171,22 +168,23 @@ void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority,
   if (normalized_total_availability == 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.
-    healthy_per_priority_load.get()[0] = 100;
+    per_priority_load.healthy_priority_load_.get()[0] = 100;
     return;
   }
 
   // We start of with a total load of 100 and distribute it between priorities based on
   // availability. We first attempt to distribute this load to healthy priorities based on healthy
   // availability.
-  const auto first_healthy_and_remaining = distributeLoad(
-      healthy_per_priority_load, per_priority_health, 100, normalized_total_availability);
+  const auto first_healthy_and_remaining =
+      distributeLoad(per_priority_load.healthy_priority_load_, per_priority_health, 100,
+                     normalized_total_availability);
 
   // Using the remaining load after allocating load to healthy priorities, distribute it based on
   // degraded availability.
   const auto remaining_load_for_degraded = first_healthy_and_remaining.second;
   const auto first_degraded_and_remaining =
-      distributeLoad(degraded_per_priority_load, per_priority_degraded, remaining_load_for_degraded,
-                     normalized_total_availability);
+      distributeLoad(per_priority_load.degraded_priority_load_, per_priority_degraded,
+                     remaining_load_for_degraded, normalized_total_availability);
 
   // Anything that remains should just be rounding errors, so allocate that to the first available
   // priority, either as healthy or degraded.
@@ -198,20 +196,20 @@ void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority,
 
     // Attempt to allocate the remainder to the first healthy priority first. If no such priority
     // exist, allocate to the first degraded priority.
-    ASSERT(remaining_load <
-           healthy_per_priority_load.get().size() + per_priority_degraded.get().size());
+    ASSERT(remaining_load < per_priority_load.healthy_priority_load_.get().size() +
+                                per_priority_load.degraded_priority_load_.get().size());
     if (first_healthy != -1) {
-      healthy_per_priority_load.get()[first_healthy] += remaining_load;
+      per_priority_load.healthy_priority_load_.get()[first_healthy] += remaining_load;
     } else {
-      degraded_per_priority_load.get()[first_degraded] += remaining_load;
+      per_priority_load.degraded_priority_load_.get()[first_degraded] += remaining_load;
     }
   }
 
   // The allocated load between healthy and degraded should be exactly 100.
-  ASSERT(100 == std::accumulate(healthy_per_priority_load.get().begin(),
-                                healthy_per_priority_load.get().end(), 0) +
-                    std::accumulate(degraded_per_priority_load.get().begin(),
-                                    degraded_per_priority_load.get().end(), 0));
+  ASSERT(100 == std::accumulate(per_priority_load.healthy_priority_load_.get().begin(),
+                                per_priority_load.healthy_priority_load_.get().end(), 0) +
+                    std::accumulate(per_priority_load.degraded_priority_load_.get().begin(),
+                                    per_priority_load.degraded_priority_load_.get().end(), 0));
 }
 
 // Method iterates through priority levels and turns on/off panic mode.
@@ -227,7 +225,7 @@ void LoadBalancerBase::recalculatePerPriorityPanic() {
 
   if (normalized_total_availability == 0) {
     // Everything is terrible. All load should be to P=0. Turn on panic mode.
-    ASSERT(healthy_per_priority_load_.get()[0] == 100);
+    ASSERT(per_priority_load_.healthy_priority_load_.get()[0] == 100);
     per_priority_panic_[0] = true;
     return;
   }
@@ -245,18 +243,18 @@ void LoadBalancerBase::recalculatePerPriorityPanic() {
 std::pair<HostSet&, LoadBalancerBase::HostAvailability>
 LoadBalancerBase::chooseHostSet(LoadBalancerContext* context) {
   if (context) {
-    const auto& healthy_per_priority_load =
-        context->determinePriorityLoad(priority_set_, healthy_per_priority_load_);
+    const auto priority_loads = context->determinePriorityLoad(priority_set_, per_priority_load_);
 
-    // TODO(snowp): pass degraded priority load to plugin.
     const auto priority_and_source =
-        choosePriority(random_.random(), healthy_per_priority_load, degraded_per_priority_load_);
+        choosePriority(random_.random(), priority_loads.healthy_priority_load_,
+                       priority_loads.degraded_priority_load_);
     return {*priority_set_.hostSetsPerPriority()[priority_and_source.first],
             priority_and_source.second};
   }
 
   const auto priority_and_source =
-      choosePriority(random_.random(), healthy_per_priority_load_, degraded_per_priority_load_);
+      choosePriority(random_.random(), per_priority_load_.healthy_priority_load_,
+                     per_priority_load_.degraded_priority_load_);
   return {*priority_set_.hostSetsPerPriority()[priority_and_source.first],
           priority_and_source.second};
 }

source/common/upstream/load_balancer_impl.h

@@ -73,10 +73,10 @@ class LoadBalancerBase : public LoadBalancer {
   std::pair<HostSet&, HostAvailability> chooseHostSet(LoadBalancerContext* context);
 
   uint32_t percentageLoad(uint32_t priority) const {
-    return healthy_per_priority_load_.get()[priority];
+    return per_priority_load_.healthy_priority_load_.get()[priority];
   }
   uint32_t percentageDegradedLoad(uint32_t priority) const {
-    return degraded_per_priority_load_.get()[priority];
+    return per_priority_load_.degraded_priority_load_.get()[priority];
   }
   bool isInPanic(uint32_t priority) const { return per_priority_panic_[priority]; }
 
@@ -92,8 +92,7 @@ class LoadBalancerBase : public LoadBalancer {
   // per_priority_health for that priority level, and may update per_priority_load for all
   // priority levels.
   void static recalculatePerPriorityState(uint32_t priority, const PrioritySet& priority_set,
-                                          HealthyLoad& healthy_priority_load,
-                                          DegradedLoad& degraded_priority_load,
+                                          HealthyAndDegradedLoad& priority_load,
                                           HealthyAvailability& per_priority_health,
                                           DegradedAvailability& per_priority_degraded);
   void recalculatePerPriorityPanic();
@@ -119,10 +118,9 @@ class LoadBalancerBase : public LoadBalancer {
 
     return std::min<uint32_t>(health + degraded, 100);
   }
-  // The percentage load (0-100) for each priority level when targeting healthy hosts.
-  HealthyLoad healthy_per_priority_load_;
-  // The percentage load (0-100) for each priority level when targeting degraded hosts.
-  DegradedLoad degraded_per_priority_load_;
+  // The percentage load (0-100) for each priority level when targeting healthy hosts and
+  // the percentage load (0-100) for each priority level when targeting degraded hosts.
+  HealthyAndDegradedLoad per_priority_load_;
   // The health percentage (0-100) for each priority level.
   HealthyAvailability per_priority_health_;
   // The degraded percentage (0-100) for each priority level.
@@ -141,8 +139,9 @@ class LoadBalancerContextBase : public LoadBalancerContext {
 
   const Http::HeaderMap* downstreamHeaders() const override { return nullptr; }
 
-  const HealthyLoad& determinePriorityLoad(const PrioritySet&,
-                                           const HealthyLoad& original_priority_load) override {
+  const HealthyAndDegradedLoad&
+  determinePriorityLoad(const PrioritySet&,
+                        const HealthyAndDegradedLoad& original_priority_load) override {
     return original_priority_load;
   }
 

source/common/upstream/thread_aware_lb_impl.cc

@@ -21,8 +21,10 @@ void ThreadAwareLoadBalancerBase::initialize() {
 void ThreadAwareLoadBalancerBase::refresh() {
   auto per_priority_state_vector = std::make_shared<std::vector<PerPriorityStatePtr>>(
       priority_set_.hostSetsPerPriority().size());
-  auto healthy_per_priority_load = std::make_shared<HealthyLoad>(healthy_per_priority_load_);
-  auto degraded_per_priority_load = std::make_shared<DegradedLoad>(degraded_per_priority_load_);
+  auto healthy_per_priority_load =
+      std::make_shared<HealthyLoad>(per_priority_load_.healthy_priority_load_);
+  auto degraded_per_priority_load =
+      std::make_shared<DegradedLoad>(per_priority_load_.degraded_priority_load_);
 
   for (const auto& host_set : priority_set_.hostSetsPerPriority()) {
     const uint32_t priority = host_set->priority();