upstream: locality weighted load balancing.

bazel/repository_locations.bzl

@@ -76,7 +76,7 @@ REPOSITORY_LOCATIONS = dict(
         urls = ["https://github.com/google/protobuf/archive/v3.5.0.tar.gz"],
     ),
     envoy_api = dict(
-        commit = "e73b8264a26c909f75d38ec2c73d9f47ad1b3b57",
+        commit = "771aeeaeece5ab035724871ff9cbbb337f8c98c8",
         remote = "https://github.com/envoyproxy/data-plane-api",
     ),
     grpc_httpjson_transcoding = dict(

include/envoy/upstream/upstream.h

@@ -169,6 +169,11 @@ class HostsPerLocality {
 typedef std::shared_ptr<HostsPerLocality> HostsPerLocalitySharedPtr;
 typedef std::shared_ptr<const HostsPerLocality> HostsPerLocalityConstSharedPtr;
 
+// Weight for each locality index in HostsPerLocality.
+typedef std::vector<uint32_t> LocalityWeights;
+typedef std::shared_ptr<LocalityWeights> LocalityWeightsSharedPtr;
+typedef std::shared_ptr<const LocalityWeights> LocalityWeightsConstSharedPtr;
+
 /**
  * Base host set interface. This contains all of the endpoints for a given LocalityLbEndpoints
  * priority level.
@@ -200,19 +205,31 @@ class HostSet {
    */
   virtual const HostsPerLocality& healthyHostsPerLocality() const PURE;
 
+  /**
+   * @return weights for each locality in the host set.
+   */
+  virtual LocalityWeightsConstSharedPtr localityWeights() const PURE;
+
+  /**
+   * @return next locality index to route to if performing locality weighted balancing.
+   */
+  virtual absl::optional<uint32_t> chooseLocality() PURE;
+
   /**
    * Updates the hosts in a given host set.
    *
    * @param hosts supplies the (usually new) list of hosts in the host set.
    * @param healthy hosts supplies the subset of hosts which are healthy.
    * @param hosts_per_locality supplies the hosts subdivided by locality.
    * @param hosts_per_locality supplies the healthy hosts subdivided by locality.
+   * @param locality_weights supplies a map from locality to associated weight.
    * @param hosts_added supplies the hosts added since the last update.
    * @param hosts_removed supplies the hosts removed since the last update.
    */
   virtual void updateHosts(HostVectorConstSharedPtr hosts, HostVectorConstSharedPtr healthy_hosts,
                            HostsPerLocalityConstSharedPtr hosts_per_locality,
                            HostsPerLocalityConstSharedPtr healthy_hosts_per_locality,
+                           LocalityWeightsConstSharedPtr locality_weights,
                            const HostVector& hosts_added, const HostVector& hosts_removed) PURE;
 
   /**

source/common/upstream/BUILD

@@ -168,6 +168,15 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "locality_lib",
+    hdrs = ["locality.h"],
+    deps = [
+        "//source/common/protobuf:utility_lib",
+        "@envoy_api//envoy/api/v2/core:base_cc",
+    ],
+)
+
 envoy_cc_library(
     name = "logical_dns_cluster_lib",
     srcs = ["logical_dns_cluster.cc"],
@@ -258,6 +267,7 @@ envoy_cc_library(
     srcs = ["eds.cc"],
     hdrs = ["eds.h"],
     deps = [
+        ":locality_lib",
         ":sds_subscription_lib",
         ":upstream_includes",
         "//include/envoy/config:grpc_mux_interface",
@@ -371,6 +381,7 @@ envoy_cc_library(
         "//source/common/config:metadata_lib",
         "//source/common/config:well_known_names",
         "//source/common/stats:stats_lib",
+        "//source/common/upstream:locality_lib",
         "@envoy_api//envoy/api/v2/core:base_cc",
     ],
 )

source/common/upstream/cluster_manager_impl.cc

@@ -551,6 +551,7 @@ void ClusterManagerImpl::postThreadLocalClusterUpdate(const Cluster& cluster, ui
                                                       const HostVector& hosts_removed) {
   const auto& host_set = cluster.prioritySet().hostSetsPerPriority()[priority];
 
+  // TODO(htuch): Can we skip these copies by exporting out const shared_ptr from HostSet?
   HostVectorConstSharedPtr hosts_copy(new HostVector(host_set->hosts()));
   HostVectorConstSharedPtr healthy_hosts_copy(new HostVector(host_set->healthyHosts()));
   HostsPerLocalityConstSharedPtr hosts_per_locality_copy = host_set->hostsPerLocality().clone();
@@ -559,14 +560,13 @@ void ClusterManagerImpl::postThreadLocalClusterUpdate(const Cluster& cluster, ui
 
   tls_->runOnAllThreads([
     this, name = cluster.info()->name(), priority, hosts_copy, healthy_hosts_copy,
-    hosts_per_locality_copy, healthy_hosts_per_locality_copy, hosts_added, hosts_removed
-  ]()
-                            ->void {
-                              ThreadLocalClusterManagerImpl::updateClusterMembership(
-                                  name, priority, hosts_copy, healthy_hosts_copy,
-                                  hosts_per_locality_copy, healthy_hosts_per_locality_copy,
-                                  hosts_added, hosts_removed, *tls_);
-                            });
+    hosts_per_locality_copy, healthy_hosts_per_locality_copy,
+    locality_weights = host_set->localityWeights(), hosts_added, hosts_removed
+  ]() {
+    ThreadLocalClusterManagerImpl::updateClusterMembership(
+        name, priority, hosts_copy, healthy_hosts_copy, hosts_per_locality_copy,
+        healthy_hosts_per_locality_copy, locality_weights, hosts_added, hosts_removed, *tls_);
+  });
 }
 
 void ClusterManagerImpl::postThreadLocalHealthFailure(const HostSharedPtr& host) {
@@ -710,7 +710,8 @@ void ClusterManagerImpl::ThreadLocalClusterManagerImpl::drainConnPools(
 void ClusterManagerImpl::ThreadLocalClusterManagerImpl::updateClusterMembership(
     const std::string& name, uint32_t priority, HostVectorConstSharedPtr hosts,
     HostVectorConstSharedPtr healthy_hosts, HostsPerLocalityConstSharedPtr hosts_per_locality,
-    HostsPerLocalityConstSharedPtr healthy_hosts_per_locality, const HostVector& hosts_added,
+    HostsPerLocalityConstSharedPtr healthy_hosts_per_locality,
+    LocalityWeightsConstSharedPtr locality_weights, const HostVector& hosts_added,
     const HostVector& hosts_removed, ThreadLocal::Slot& tls) {
 
   ThreadLocalClusterManagerImpl& config = tls.getTyped<ThreadLocalClusterManagerImpl>();
@@ -720,7 +721,8 @@ void ClusterManagerImpl::ThreadLocalClusterManagerImpl::updateClusterMembership(
   ENVOY_LOG(debug, "membership update for TLS cluster {}", name);
   cluster_entry->priority_set_.getOrCreateHostSet(priority).updateHosts(
       std::move(hosts), std::move(healthy_hosts), std::move(hosts_per_locality),
-      std::move(healthy_hosts_per_locality), hosts_added, hosts_removed);
+      std::move(healthy_hosts_per_locality), std::move(locality_weights), hosts_added,
+      hosts_removed);
 
   // If an LB is thread aware, create a new worker local LB on membership changes.
   if (cluster_entry->lb_factory_ != nullptr) {

source/common/upstream/cluster_manager_impl.h

@@ -245,6 +245,7 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
                                         HostVectorConstSharedPtr healthy_hosts,
                                         HostsPerLocalityConstSharedPtr hosts_per_locality,
                                         HostsPerLocalityConstSharedPtr healthy_hosts_per_locality,
+                                        LocalityWeightsConstSharedPtr locality_weights,
                                         const HostVector& hosts_added,
                                         const HostVector& hosts_removed, ThreadLocal::Slot& tls);
     static void onHostHealthFailure(const HostSharedPtr& host, ThreadLocal::Slot& tls);

source/common/upstream/edf_scheduler.h

@@ -53,10 +53,10 @@ template <class C> class EdfScheduler {
 
   /**
    * Insert entry into queue with a given weight. The deadline will be current_time_ + 1 / weight.
-   * @param weight integer weight.
+   * @param weight floating point weight.
    * @param entry shared pointer to entry, only a weak reference will be retained.
    */
-  void add(uint64_t weight, std::shared_ptr<C> entry) {
+  void add(double weight, std::shared_ptr<C> entry) {
     ASSERT(weight > 0);
     const double deadline = current_time_ + 1.0 / weight;
     EDF_TRACE("Insertion {} in queue with deadline {} and weight {}.",

source/common/upstream/eds.cc

@@ -46,7 +46,7 @@ void EdsClusterImpl::startPreInit() { subscription_->start({cluster_name_}, *thi
 
 void EdsClusterImpl::onConfigUpdate(const ResourceVector& resources) {
   typedef std::unique_ptr<HostVector> HostListPtr;
-  std::vector<HostListPtr> new_hosts(1);
+  std::vector<std::pair<HostListPtr, LocalityWeightsMap>> priority_state(1);
   if (resources.empty()) {
     ENVOY_LOG(debug, "Missing ClusterLoadAssignment for {} in onConfigUpdate()", cluster_name_);
     info_->stats().update_empty_.inc();
@@ -69,22 +69,27 @@ void EdsClusterImpl::onConfigUpdate(const ResourceVector& resources) {
       throw EnvoyException(
           fmt::format("Unexpected non-zero priority for local cluster '{}'.", cluster_name_));
     }
-    if (new_hosts.size() <= priority) {
-      new_hosts.resize(priority + 1);
+    if (priority_state.size() <= priority) {
+      priority_state.resize(priority + 1);
     }
-    if (new_hosts[priority] == nullptr) {
-      new_hosts[priority] = HostListPtr{new HostVector};
+    if (priority_state[priority].first == nullptr) {
+      priority_state[priority].first.reset(new HostVector());
+    }
+    if (locality_lb_endpoint.has_locality() && locality_lb_endpoint.has_load_balancing_weight()) {
+      priority_state[priority].second[locality_lb_endpoint.locality()] =
+          locality_lb_endpoint.load_balancing_weight().value();
     }
     for (const auto& lb_endpoint : locality_lb_endpoint.lb_endpoints()) {
-      new_hosts[priority]->emplace_back(new HostImpl(
+      priority_state[priority].first->emplace_back(new HostImpl(
           info_, "", resolveProtoAddress(lb_endpoint.endpoint().address()), lb_endpoint.metadata(),
           lb_endpoint.load_balancing_weight().value(), locality_lb_endpoint.locality()));
     }
   }
 
-  for (size_t i = 0; i < new_hosts.size(); ++i) {
-    if (new_hosts[i] != nullptr) {
-      updateHostsPerLocality(priority_set_.getOrCreateHostSet(i), *new_hosts[i]);
+  for (size_t i = 0; i < priority_state.size(); ++i) {
+    if (priority_state[i].first != nullptr) {
+      updateHostsPerLocality(priority_set_.getOrCreateHostSet(i), *priority_state[i].first,
+                             priority_state[i].second);
     }
   }
 
@@ -93,45 +98,69 @@ void EdsClusterImpl::onConfigUpdate(const ResourceVector& resources) {
   onPreInitComplete();
 }
 
-void EdsClusterImpl::updateHostsPerLocality(HostSet& host_set, HostVector& new_hosts) {
+void EdsClusterImpl::updateHostsPerLocality(HostSet& host_set, const HostVector& new_hosts,
+                                            LocalityWeightsMap& locality_weights_map) {
   HostVectorSharedPtr current_hosts_copy(new HostVector(host_set.hosts()));
 
   HostVector hosts_added;
   HostVector hosts_removed;
+  // We need to trigger updateHosts with the new host vectors if they have changed. We also do this
+  // when the locality weight map changes.
+  // TODO(htuch): We eagerly update all the host sets here on weight changes, which isn't great,
+  // since this has the knock on effect that we rebuild the load balancers and locality scheduler.
+  // We could make this happen lazily, as we do for host-level weight updates, where as things age
+  // out of the locality scheduler, we discover their new weights. We don't currently have a shared
+  // object for locality weights that we can update here, we should add something like this to
+  // improve performance and scalability of locality weight updates.
   if (updateDynamicHostList(new_hosts, *current_hosts_copy, hosts_added, hosts_removed,
-                            health_checker_ != nullptr)) {
-    ENVOY_LOG(debug, "EDS hosts changed for cluster: {} ({}) priority {}", info_->name(),
-              host_set.hosts().size(), host_set.priority());
+                            health_checker_ != nullptr) ||
+      current_locality_weights_map_ != locality_weights_map) {
+    current_locality_weights_map_ = locality_weights_map;
+    LocalityWeightsSharedPtr locality_weights;
+    ENVOY_LOG(debug, "EDS hosts or locality weights changed for cluster: {} ({}) priority {}",
+              info_->name(), host_set.hosts().size(), host_set.priority());
     std::vector<HostVector> per_locality;
 
+    // If we are configured for locality weighted LB we populate the locality
+    // weights.
+    const bool locality_weighted_lb = info()->lbConfig().has_locality_weighted_lb_config();
+    if (locality_weighted_lb) {
+      locality_weights = std::make_shared<LocalityWeights>();
+    }
     // If local locality is not defined then skip populating per locality hosts.
-    const Locality local_locality(local_info_.node().locality());
+    const auto& local_locality = local_info_.node().locality();
     ENVOY_LOG(trace, "Local locality: {}", local_info_.node().locality().DebugString());
 
     // We use std::map to guarantee a stable ordering for zone aware routing.
-    std::map<Locality, HostVector> hosts_per_locality;
+    std::map<envoy::api::v2::core::Locality, HostVector, LocalityLess> hosts_per_locality;
 
     for (const HostSharedPtr& host : *current_hosts_copy) {
-      hosts_per_locality[Locality(host->locality())].push_back(host);
+      hosts_per_locality[host->locality()].push_back(host);
     }
 
     // Do we have hosts for the local locality?
     const bool non_empty_local_locality =
-        !local_locality.empty() &&
+        local_info_.node().has_locality() &&
         hosts_per_locality.find(local_locality) != hosts_per_locality.end();
 
     // As per HostsPerLocality::get(), the per_locality vector must have the
     // local locality hosts first if non_empty_local_locality.
     if (non_empty_local_locality) {
-      per_locality.push_back(hosts_per_locality[local_locality]);
+      per_locality.emplace_back(hosts_per_locality[local_locality]);
+      if (locality_weighted_lb) {
+        locality_weights->emplace_back(locality_weights_map[local_locality]);
+      }
     }
 
     // After the local locality hosts (if any), we place the remaining locality
     // host groups in lexicographic order. This provides a stable ordering for
     // zone aware routing.
     for (auto& entry : hosts_per_locality) {
-      if (!non_empty_local_locality || local_locality != entry.first) {
-        per_locality.push_back(entry.second);
+      if (!non_empty_local_locality || !LocalityEqualTo()(local_locality, entry.first)) {
+        per_locality.emplace_back(entry.second);
+        if (locality_weighted_lb) {
+          locality_weights->emplace_back(locality_weights_map[entry.first]);
+        }
       }
     }
 
@@ -140,7 +169,7 @@ void EdsClusterImpl::updateHostsPerLocality(HostSet& host_set, HostVector& new_h
 
     host_set.updateHosts(current_hosts_copy, createHealthyHostList(*current_hosts_copy),
                          per_locality_shared, createHealthyHostLists(*per_locality_shared),
-                         hosts_added, hosts_removed);
+                         std::move(locality_weights), hosts_added, hosts_removed);
   }
 }
 

source/common/upstream/eds.h

@@ -5,6 +5,7 @@
 #include "envoy/config/subscription.h"
 #include "envoy/local_info/local_info.h"
 
+#include "common/upstream/locality.h"
 #include "common/upstream/upstream_impl.h"
 
 namespace Envoy {
@@ -35,7 +36,10 @@ class EdsClusterImpl : public BaseDynamicClusterImpl,
   }
 
 private:
-  void updateHostsPerLocality(HostSet& host_set, HostVector& new_hosts);
+  using LocalityWeightsMap =
+      std::unordered_map<envoy::api::v2::core::Locality, uint32_t, LocalityHash, LocalityEqualTo>;
+  void updateHostsPerLocality(HostSet& host_set, const HostVector& new_hosts,
+                              LocalityWeightsMap& locality_weights_map);
 
   // ClusterImplBase
   void startPreInit() override;
@@ -44,6 +48,7 @@ class EdsClusterImpl : public BaseDynamicClusterImpl,
   std::unique_ptr<Config::Subscription<envoy::api::v2::ClusterLoadAssignment>> subscription_;
   const LocalInfo::LocalInfo& local_info_;
   const std::string cluster_name_;
+  LocalityWeightsMap current_locality_weights_map_;
 };
 
 } // namespace Upstream

source/common/upstream/load_balancer_impl.cc

@@ -14,9 +14,16 @@
 namespace Envoy {
 namespace Upstream {
 
+namespace {
+// Priority levels are considered overprovisioned with this factor. This means that we don't
+// consider a priority level unhealthy until the percentage of healthy hosts multiplied by
+// kOverProvisioningFactor drops below 100.
+static constexpr uint32_t kOverProvisioningFactor = 140;
+
 static const std::string RuntimeZoneEnabled = "upstream.zone_routing.enabled";
 static const std::string RuntimeMinClusterSize = "upstream.zone_routing.min_cluster_size";
 static const std::string RuntimePanicThreshold = "upstream.healthy_panic_threshold";
+} // namespace
 
 uint32_t LoadBalancerBase::choosePriority(uint64_t hash,
                                           const std::vector<uint32_t>& per_priority_load) {
@@ -57,13 +64,13 @@ void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority) {
 
   // 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
-  // somewhat arbitrary overprovision factor of 1.4.
+  // somewhat arbitrary overprovision factor of kOverProvisioningFactor.
   // Eventually the overprovision factor will likely be made configurable.
   HostSet& host_set = *priority_set_.hostSetsPerPriority()[priority];
   per_priority_health_[priority] = 0;
   if (host_set.hosts().size() > 0) {
-    per_priority_health_[priority] =
-        std::min<uint32_t>(100, 140 * host_set.healthyHosts().size() / host_set.hosts().size());
+    per_priority_health_[priority] = std::min<uint32_t>(
+        100, kOverProvisioningFactor * host_set.healthyHosts().size() / host_set.hosts().size());
   }
 
   // Now that we've updated health for the changed priority level, we need to caculate percentage
@@ -95,7 +102,7 @@ void LoadBalancerBase::recalculatePerPriorityState(uint32_t priority) {
   }
 }
 
-const HostSet& LoadBalancerBase::chooseHostSet() {
+HostSet& LoadBalancerBase::chooseHostSet() {
   const uint32_t priority = choosePriority(random_.random(), per_priority_load_);
   return *priority_set_.hostSetsPerPriority()[priority];
 }
@@ -341,7 +348,7 @@ uint32_t ZoneAwareLoadBalancerBase::tryChooseLocalLocalityHosts(const HostSet& h
 }
 
 ZoneAwareLoadBalancerBase::HostsSource ZoneAwareLoadBalancerBase::hostSourceToUse() {
-  const HostSet& host_set = chooseHostSet();
+  HostSet& host_set = chooseHostSet();
   HostsSource hosts_source;
   hosts_source.priority_ = host_set.priority();
 
@@ -352,6 +359,14 @@ ZoneAwareLoadBalancerBase::HostsSource ZoneAwareLoadBalancerBase::hostSourceToUs
     return hosts_source;
   }
 
+  // If we're doing locality weighted balancing, pick locality.
+  const absl::optional<uint32_t> locality = host_set.chooseLocality();
+  if (locality.has_value()) {
+    hosts_source.source_type_ = HostsSource::SourceType::LocalityHealthyHosts;
+    hosts_source.locality_index_ = locality.value();
+    return hosts_source;
+  }
+
   // If we've latched that we can't do priority-based routing, return healthy hosts for the selected
   // host set.
   if (per_priority_state_[host_set.priority()]->locality_routing_state_ ==

source/common/upstream/load_balancer_impl.h

@@ -39,7 +39,7 @@ class LoadBalancerBase {
                    const envoy::api::v2::Cluster::CommonLbConfig& common_config);
 
   // Choose host set randomly, based on the per_priority_load_;
-  const HostSet& chooseHostSet();
+  HostSet& chooseHostSet();
 
   uint32_t percentageLoad(uint32_t priority) const { return per_priority_load_[priority]; }