diff --git a/bazel/repository_locations.bzl b/bazel/repository_locations.bzl
index f8e276a1ebc62..ecc90b6b42e12 100644
--- a/bazel/repository_locations.bzl
+++ b/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 = "ec157d3d8b359a1bd65c22116ba4387a459cc53a",
+        commit = "92ffa417b5d2e8ab3912ff4da4daa6aa93f8df0f",
         remote = "https://github.com/envoyproxy/data-plane-api",
     ),
     grpc_httpjson_transcoding = dict(
diff --git a/include/envoy/upstream/upstream.h b/include/envoy/upstream/upstream.h
index c3a9eb3182b4c..e6de663cbc121 100644
--- a/include/envoy/upstream/upstream.h
+++ b/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,6 +205,16 @@ 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.
    *
@@ -207,12 +222,14 @@ class HostSet {
    * @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;
 
   /**
diff --git a/source/common/upstream/BUILD b/source/common/upstream/BUILD
index a92a8119d6c7d..ed792fe212986 100644
--- a/source/common/upstream/BUILD
+++ b/source/common/upstream/BUILD
@@ -170,6 +170,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"],
@@ -260,6 +269,7 @@ envoy_cc_library(
     srcs = ["eds.cc"],
     hdrs = ["eds.h"],
     deps = [
+        ":locality_lib",
         ":sds_subscription_lib",
         ":upstream_includes",
         "//include/envoy/config:grpc_mux_interface",
@@ -374,6 +384,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",
     ],
 )
diff --git a/source/common/upstream/cluster_manager_impl.cc b/source/common/upstream/cluster_manager_impl.cc
index 4f56991246869..e5f35e8533890 100644
--- a/source/common/upstream/cluster_manager_impl.cc
+++ b/source/common/upstream/cluster_manager_impl.cc
@@ -547,6 +547,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();
@@ -555,14 +556,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) {
@@ -706,7 +706,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>();
@@ -716,7 +717,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) {
diff --git a/source/common/upstream/cluster_manager_impl.h b/source/common/upstream/cluster_manager_impl.h
index 9fc85bcce2ddb..acdfaec3548d1 100644
--- a/source/common/upstream/cluster_manager_impl.h
+++ b/source/common/upstream/cluster_manager_impl.h
@@ -243,6 +243,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);
diff --git a/source/common/upstream/edf_scheduler.h b/source/common/upstream/edf_scheduler.h
index f4e10eb7fe857..857a16aef4f7c 100644
--- a/source/common/upstream/edf_scheduler.h
+++ b/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 {}.",
diff --git a/source/common/upstream/eds.cc b/source/common/upstream/eds.cc
index 469227ee16aa9..4c0da18e290c3 100644
--- a/source/common/upstream/eds.cc
+++ b/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);
   }
 }
 
diff --git a/source/common/upstream/eds.h b/source/common/upstream/eds.h
index a24c13516c957..e7127c170832e 100644
--- a/source/common/upstream/eds.h
+++ b/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
diff --git a/source/common/upstream/load_balancer_impl.cc b/source/common/upstream/load_balancer_impl.cc
index 2cfb22a8d1980..603fc2adfea69 100644
--- a/source/common/upstream/load_balancer_impl.cc
+++ b/source/common/upstream/load_balancer_impl.cc
@@ -14,9 +14,11 @@
 namespace Envoy {
 namespace Upstream {
 
+namespace {
 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 +59,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 +97,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];
 }
@@ -340,7 +342,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();
 
@@ -351,6 +353,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_ ==
diff --git a/source/common/upstream/load_balancer_impl.h b/source/common/upstream/load_balancer_impl.h
index 69e56b0e73018..9d16a80d9b130 100644
--- a/source/common/upstream/load_balancer_impl.h
+++ b/source/common/upstream/load_balancer_impl.h
@@ -15,6 +15,11 @@
 namespace Envoy {
 namespace Upstream {
 
+// Priority levels and localities are considered overprovisioned with this factor. This means that
+// we don't consider a priority level or locality unhealthy until the percentage of healthy hosts
+// multiplied by kOverProvisioningFactor drops below 100.
+static constexpr uint32_t kOverProvisioningFactor = 140;
+
 /**
  * Base class for all LB implementations.
  */
@@ -39,7 +44,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]; }
 
diff --git a/source/common/upstream/locality.h b/source/common/upstream/locality.h
new file mode 100644
index 0000000000000..f4ff0948cfbd7
--- /dev/null
+++ b/source/common/upstream/locality.h
@@ -0,0 +1,61 @@
+#pragma once
+
+#include "envoy/api/v2/core/base.pb.h"
+
+#include "common/protobuf/utility.h"
+
+namespace Envoy {
+namespace Upstream {
+
+// TODO(htuch): should these be templated in protobuf/utility.h?
+struct LocalityHash {
+  size_t operator()(const envoy::api::v2::core::Locality& locality) const {
+    return MessageUtil::hash(locality);
+  }
+};
+
+struct LocalityEqualTo {
+  bool operator()(const envoy::api::v2::core::Locality& lhs,
+                  const envoy::api::v2::core::Locality& rhs) const {
+    return Protobuf::util::MessageDifferencer::Equivalent(lhs, rhs);
+  }
+};
+
+struct LocalityLess {
+  bool operator()(const envoy::api::v2::core::Locality& lhs,
+                  const envoy::api::v2::core::Locality& rhs) const {
+    using LocalityTuple = std::tuple<const std::string&, const std::string&, const std::string&>;
+    const LocalityTuple lhs_tuple = LocalityTuple(lhs.region(), lhs.zone(), lhs.sub_zone());
+    const LocalityTuple rhs_tuple = LocalityTuple(rhs.region(), rhs.zone(), rhs.sub_zone());
+    return lhs_tuple < rhs_tuple;
+  }
+};
+
+// For tests etc. where this is convenient.
+static inline envoy::api::v2::core::Locality
+Locality(const std::string& region, const std::string& zone, const std::string sub_zone) {
+  envoy::api::v2::core::Locality locality;
+  locality.set_region(region);
+  locality.set_zone(zone);
+  locality.set_sub_zone(sub_zone);
+  return locality;
+}
+
+} // namespace Upstream
+} // namespace Envoy
+
+// Something heinous this way comes. Required to allow == for LocalityWeightsMap.h in eds.h.
+namespace envoy {
+namespace api {
+namespace v2 {
+namespace core {
+
+inline bool operator==(const envoy::api::v2::core::Locality& x,
+                       const envoy::api::v2::core::Locality& y) {
+  return Envoy::Upstream::LocalityEqualTo()(x, y);
+}
+
+} // namespace core
+} // namespace v2
+} // namespace api
+} // namespace envoy
diff --git a/source/common/upstream/logical_dns_cluster.cc b/source/common/upstream/logical_dns_cluster.cc
index 97fe972d15c16..91ea2eb8e134e 100644
--- a/source/common/upstream/logical_dns_cluster.cc
+++ b/source/common/upstream/logical_dns_cluster.cc
@@ -113,7 +113,7 @@ void LogicalDnsCluster::startResolve() {
             auto& first_host_set = priority_set_.getOrCreateHostSet(0);
             first_host_set.updateHosts(new_hosts, createHealthyHostList(*new_hosts),
                                        HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(),
-                                       *new_hosts, {});
+                                       {}, *new_hosts, {});
           }
         }
 
diff --git a/source/common/upstream/original_dst_cluster.cc b/source/common/upstream/original_dst_cluster.cc
index cdeb7e2129674..5f24154eb7939 100644
--- a/source/common/upstream/original_dst_cluster.cc
+++ b/source/common/upstream/original_dst_cluster.cc
@@ -110,7 +110,7 @@ void OriginalDstCluster::addHost(HostSharedPtr& host) {
   HostVectorSharedPtr new_hosts(new HostVector(first_host_set.hosts()));
   new_hosts->emplace_back(host);
   first_host_set.updateHosts(new_hosts, createHealthyHostList(*new_hosts),
-                             HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(),
+                             HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(), {},
                              {std::move(host)}, {});
 }
 
@@ -135,7 +135,7 @@ void OriginalDstCluster::cleanup() {
 
   if (to_be_removed.size() > 0) {
     host_set.updateHosts(new_hosts, createHealthyHostList(*new_hosts),
-                         HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(), {},
+                         HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(), {}, {},
                          to_be_removed);
   }
 
diff --git a/source/common/upstream/subset_lb.cc b/source/common/upstream/subset_lb.cc
index 65de541056bc8..9d32d440665f9 100644
--- a/source/common/upstream/subset_lb.cc
+++ b/source/common/upstream/subset_lb.cc
@@ -463,7 +463,18 @@ void SubsetLoadBalancer::HostSubsetImpl::update(const HostVector& hosts_added,
       original_host_set_.hostsPerLocality().filter(
           [&predicate](const Host& host) { return predicate(host) && host.healthy(); });
 
-  HostSetImpl::updateHosts(hosts, healthy_hosts, hosts_per_locality, healthy_hosts_per_locality,
+  // We pass in an empty list of locality weights here. This effectively disables locality balancing
+  // for subset LB.
+  // TODO(htuch): We should consider adding locality awareness here, but we need to do some design
+  // work first, and this might not even be a desirable thing to do. Consider for example a
+  // situation in which you have 50/50 split across two localities X/Y which have 100 hosts each
+  // without subsetting. If the subset LB results in X having only 1 host selected but Y having 100,
+  // then a lot more load is being dumped on the single host in X than originally anticipated in the
+  // load balancing assignment delivered via EDS. It might seem you want to further weight by subset
+  // size in order for this to make sense. However, while the original X/Y weightings can be
+  // respected this way, those weightings were made by a management server that was not taking into
+  // consideration subsets (e.g. LRS only reports at locality level).
+  HostSetImpl::updateHosts(hosts, healthy_hosts, hosts_per_locality, healthy_hosts_per_locality, {},
                            filtered_added, filtered_removed);
 }
 
diff --git a/source/common/upstream/upstream_impl.cc b/source/common/upstream/upstream_impl.cc
index 2e948854626ad..455d51f362936 100644
--- a/source/common/upstream/upstream_impl.cc
+++ b/source/common/upstream/upstream_impl.cc
@@ -128,6 +128,69 @@ HostsPerLocalityImpl::filter(std::function<bool(const Host&)> predicate) const {
   return shared_filtered_clone;
 }
 
+void HostSetImpl::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) {
+  hosts_ = std::move(hosts);
+  healthy_hosts_ = std::move(healthy_hosts);
+  hosts_per_locality_ = std::move(hosts_per_locality);
+  healthy_hosts_per_locality_ = std::move(healthy_hosts_per_locality);
+  locality_weights_ = std::move(locality_weights);
+  // Rebuild the locality scheduler.
+  // TODO(htuch): if the underlying locality index ->
+  // envoy::api::v2::core::Locality hasn't changed in hosts_/healthy_hosts_, we
+  // could just update locality_weight_ without rebuilding. Similar to how host
+  // level WRR works, we would age out the existing entries via picks and lazily
+  // apply the new weights.
+  if (hosts_per_locality_ != nullptr && locality_weights_ != nullptr &&
+      !locality_weights_->empty()) {
+    locality_scheduler_ = std::make_unique<EdfScheduler<LocalityEntry>>();
+    locality_entries_.clear();
+    for (uint32_t i = 0; i < hosts_per_locality_->get().size(); ++i) {
+      const double effective_weight = effectiveLocalityWeight(i);
+      if (effective_weight > 0) {
+        locality_entries_.emplace_back(std::make_shared<LocalityEntry>(i, effective_weight));
+        locality_scheduler_->add(effective_weight, locality_entries_.back());
+      }
+    }
+  } else {
+    locality_scheduler_ = nullptr;
+  }
+  runUpdateCallbacks(hosts_added, hosts_removed);
+}
+
+absl::optional<uint32_t> HostSetImpl::chooseLocality() {
+  if (locality_scheduler_ == nullptr) {
+    return {};
+  }
+  const std::shared_ptr<LocalityEntry> locality = locality_scheduler_->pick();
+  // We don't build a schedule if there are no weighted localities, so we should always succeed.
+  ASSERT(locality != nullptr);
+  // If we picked it before, its weight must have been positive.
+  ASSERT(locality->effective_weight_ > 0);
+  locality_scheduler_->add(locality->effective_weight_, locality);
+  return locality->index_;
+}
+
+double HostSetImpl::effectiveLocalityWeight(uint32_t index) const {
+  ASSERT(locality_weights_ != nullptr);
+  ASSERT(hosts_per_locality_ != nullptr);
+  const auto& locality_hosts = hosts_per_locality_->get()[index];
+  const auto& locality_healthy_hosts = healthy_hosts_per_locality_->get()[index];
+  ASSERT(!locality_hosts.empty());
+  const double locality_healthy_ratio = 1.0 * locality_healthy_hosts.size() / locality_hosts.size();
+  const uint32_t weight = (*locality_weights_)[index];
+  // Health ranges from 0-1.0, and is the ratio of healthy hosts to total hosts, modified by the
+  // somewhat arbitrary overprovision factor of kOverProvisioningFactor.
+  // Eventually the overprovision factor will likely be made configurable.
+  const double effective_locality_health_ratio =
+      std::min(1.0, (kOverProvisioningFactor / 100.0) * locality_healthy_ratio);
+  return weight * effective_locality_health_ratio;
+}
+
 HostSet& PrioritySetImpl::getOrCreateHostSet(uint32_t priority) {
   if (host_sets_.size() < priority + 1) {
     for (size_t i = host_sets_.size(); i <= priority; ++i) {
@@ -450,11 +513,12 @@ void ClusterImplBase::reloadHealthyHosts() {
   }
 
   for (auto& host_set : prioritySet().hostSetsPerPriority()) {
+    // TODO(htuch): Can we skip these copies by exporting out const shared_ptr from HostSet?
     HostVectorConstSharedPtr hosts_copy(new HostVector(host_set->hosts()));
     HostsPerLocalityConstSharedPtr hosts_per_locality_copy = host_set->hostsPerLocality().clone();
-    host_set->updateHosts(hosts_copy, createHealthyHostList(host_set->hosts()),
-                          hosts_per_locality_copy,
-                          createHealthyHostLists(host_set->hostsPerLocality()), {}, {});
+    host_set->updateHosts(
+        hosts_copy, createHealthyHostList(host_set->hosts()), hosts_per_locality_copy,
+        createHealthyHostLists(host_set->hostsPerLocality()), host_set->localityWeights(), {}, {});
   }
 }
 
@@ -550,7 +614,7 @@ void StaticClusterImpl::startPreInit() {
   ASSERT(priority_set_.hostSetsPerPriority().size() == 1);
   auto& first_host_set = priority_set_.getOrCreateHostSet(0);
   first_host_set.updateHosts(initial_hosts_, createHealthyHostList(*initial_hosts_),
-                             HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(),
+                             HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(), {},
                              *initial_hosts_, {});
   initial_hosts_ = nullptr;
 
@@ -693,7 +757,7 @@ void StrictDnsClusterImpl::updateAllHosts(const HostVector& hosts_added,
   ASSERT(priority_set_.hostSetsPerPriority().size() == 1);
   auto& first_host_set = priority_set_.getOrCreateHostSet(0);
   first_host_set.updateHosts(new_hosts, createHealthyHostList(*new_hosts),
-                             HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(),
+                             HostsPerLocalityImpl::empty(), HostsPerLocalityImpl::empty(), {},
                              hosts_added, hosts_removed);
 }
 
diff --git a/source/common/upstream/upstream_impl.h b/source/common/upstream/upstream_impl.h
index f92f141fa5c4d..3602532ce26e7 100644
--- a/source/common/upstream/upstream_impl.h
+++ b/source/common/upstream/upstream_impl.h
@@ -31,27 +31,13 @@
 #include "common/config/well_known_names.h"
 #include "common/stats/stats_impl.h"
 #include "common/upstream/load_balancer_impl.h"
+#include "common/upstream/locality.h"
 #include "common/upstream/outlier_detection_impl.h"
 #include "common/upstream/resource_manager_impl.h"
 
 namespace Envoy {
 namespace Upstream {
 
-// Wrapper around envoy::api::v2::core::Locality to make it easier to compare for ordering in
-// std::map and in tests to construct literals.
-// TODO(htuch): Consider making this reference based when we have a single string implementation.
-class Locality : public std::tuple<std::string, std::string, std::string> {
-public:
-  Locality(const std::string& region, const std::string& zone, const std::string& sub_zone)
-      : std::tuple<std::string, std::string, std::string>(region, zone, sub_zone) {}
-  Locality(const envoy::api::v2::core::Locality& locality)
-      : std::tuple<std::string, std::string, std::string>(locality.region(), locality.zone(),
-                                                          locality.sub_zone()) {}
-  bool empty() const {
-    return std::get<0>(*this).empty() && std::get<1>(*this).empty() && std::get<2>(*this).empty();
-  }
-};
-
 /**
  * Null implementation of HealthCheckHostMonitor.
  */
@@ -203,13 +189,8 @@ class HostSetImpl : public HostSet {
   void updateHosts(HostVectorConstSharedPtr hosts, HostVectorConstSharedPtr healthy_hosts,
                    HostsPerLocalityConstSharedPtr hosts_per_locality,
                    HostsPerLocalityConstSharedPtr healthy_hosts_per_locality,
-                   const HostVector& hosts_added, const HostVector& hosts_removed) override {
-    hosts_ = std::move(hosts);
-    healthy_hosts_ = std::move(healthy_hosts);
-    hosts_per_locality_ = std::move(hosts_per_locality);
-    healthy_hosts_per_locality_ = std::move(healthy_hosts_per_locality);
-    runUpdateCallbacks(hosts_added, hosts_removed);
-  }
+                   LocalityWeightsConstSharedPtr locality_weights, const HostVector& hosts_added,
+                   const HostVector& hosts_removed) override;
 
   /**
    * Install a callback that will be invoked when the host set membership changes.
@@ -230,6 +211,8 @@ class HostSetImpl : public HostSet {
   const HostsPerLocality& healthyHostsPerLocality() const override {
     return *healthy_hosts_per_locality_;
   }
+  LocalityWeightsConstSharedPtr localityWeights() const override { return locality_weights_; }
+  absl::optional<uint32_t> chooseLocality() override;
   uint32_t priority() const override { return priority_; }
 
 protected:
@@ -238,6 +221,9 @@ class HostSetImpl : public HostSet {
   }
 
 private:
+  // Weight for a locality taking into account health status.
+  double effectiveLocalityWeight(uint32_t index) const;
+
   uint32_t priority_;
   HostVectorConstSharedPtr hosts_;
   HostVectorConstSharedPtr healthy_hosts_;
@@ -246,6 +232,17 @@ class HostSetImpl : public HostSet {
   // TODO(mattklein123): Remove mutable.
   mutable Common::CallbackManager<uint32_t, const HostVector&, const HostVector&>
       member_update_cb_helper_;
+  // Locality weights (used to build WRR locality_scheduler_);
+  LocalityWeightsConstSharedPtr locality_weights_;
+  // WRR locality scheduler state.
+  struct LocalityEntry {
+    LocalityEntry(uint32_t index, double effective_weight)
+        : index_(index), effective_weight_(effective_weight) {}
+    const uint32_t index_;
+    const double effective_weight_;
+  };
+  std::vector<std::shared_ptr<LocalityEntry>> locality_entries_;
+  std::unique_ptr<EdfScheduler<LocalityEntry>> locality_scheduler_;
 };
 
 typedef std::unique_ptr<HostSetImpl> HostSetImplPtr;
diff --git a/test/common/upstream/eds_test.cc b/test/common/upstream/eds_test.cc
index 848b2be43b640..69aac9701a657 100644
--- a/test/common/upstream/eds_test.cc
+++ b/test/common/upstream/eds_test.cc
@@ -25,22 +25,23 @@ class EdsTest : public testing::Test {
 
   void resetCluster() {
     resetCluster(R"EOF(
-    {
-      "name": "name",
-      "connect_timeout_ms": 250,
-      "type": "sds",
-      "lb_type": "round_robin",
-      "service_name": "fare"
-    }
+      name: name
+      connect_timeout: 0.25s
+      type: EDS
+      lb_policy: ROUND_ROBIN
+      eds_cluster_config:
+        service_name: fare
+        eds_config:
+          api_config_source:
+            cluster_names:
+            - eds
+            refresh_delay: 1s
     )EOF");
   }
 
-  void resetCluster(const std::string& json_config) {
-    envoy::api::v2::core::ConfigSource eds_config;
-    eds_config.mutable_api_config_source()->add_cluster_names("eds");
-    eds_config.mutable_api_config_source()->mutable_refresh_delay()->set_seconds(1);
+  void resetCluster(const std::string& yaml_config) {
     local_info_.node_.mutable_locality()->set_zone("us-east-1a");
-    eds_cluster_ = parseSdsClusterFromJson(json_config, eds_config);
+    eds_cluster_ = parseClusterFromV2Yaml(yaml_config);
     Upstream::ClusterManager::ClusterInfoMap cluster_map;
     Upstream::MockCluster cluster;
     cluster_map.emplace("eds", cluster);
@@ -119,12 +120,16 @@ TEST_F(EdsTest, OnConfigUpdateSuccess) {
 // Validate that onConfigUpdate() with no service name accepts config.
 TEST_F(EdsTest, NoServiceNameOnSuccessConfigUpdate) {
   resetCluster(R"EOF(
-    {
-      "name": "name",
-      "connect_timeout_ms": 250,
-      "type": "sds",
-      "lb_type": "round_robin"
-    }
+      name: name
+      connect_timeout: 0.25s
+      type: EDS
+      lb_policy: ROUND_ROBIN
+      eds_cluster_config:
+        eds_config:
+          api_config_source:
+            cluster_names:
+            - eds
+            refresh_delay: 1s
     )EOF");
   Protobuf::RepeatedPtrField<envoy::api::v2::ClusterLoadAssignment> resources;
   auto* cluster_load_assignment = resources.Add();
@@ -230,6 +235,120 @@ TEST_F(EdsTest, EndpointLocality) {
     EXPECT_EQ("hello", locality.zone());
     EXPECT_EQ("world", locality.sub_zone());
   }
+  EXPECT_EQ(nullptr, cluster_->prioritySet().hostSetsPerPriority()[0]->localityWeights());
+}
+
+// Validate that onConfigUpdate() propagatees locality weights to the host set when locality
+// weighted balancing isn't configured.
+TEST_F(EdsTest, EndpointLocalityWeightsIgnored) {
+  Protobuf::RepeatedPtrField<envoy::api::v2::ClusterLoadAssignment> resources;
+  auto* cluster_load_assignment = resources.Add();
+  cluster_load_assignment->set_cluster_name("fare");
+
+  {
+    auto* endpoints = cluster_load_assignment->add_endpoints();
+    auto* locality = endpoints->mutable_locality();
+    locality->set_region("oceania");
+    locality->set_zone("hello");
+    locality->set_sub_zone("world");
+    endpoints->mutable_load_balancing_weight()->set_value(42);
+
+    auto* endpoint_address = endpoints->add_lb_endpoints()
+                                 ->mutable_endpoint()
+                                 ->mutable_address()
+                                 ->mutable_socket_address();
+    endpoint_address->set_address("1.2.3.4");
+    endpoint_address->set_port_value(80);
+  }
+
+  bool initialized = false;
+  cluster_->initialize([&initialized] { initialized = true; });
+  VERBOSE_EXPECT_NO_THROW(cluster_->onConfigUpdate(resources));
+  EXPECT_TRUE(initialized);
+
+  EXPECT_EQ(nullptr, cluster_->prioritySet().hostSetsPerPriority()[0]->localityWeights());
+}
+
+// Validate that onConfigUpdate() propagates locality weights to the host set when locality
+// weighted balancing is configured.
+TEST_F(EdsTest, EndpointLocalityWeights) {
+  resetCluster(R"EOF(
+      name: name
+      connect_timeout: 0.25s
+      type: EDS
+      lb_policy: ROUND_ROBIN
+      common_lb_config:
+        locality_weighted_lb_config: {}
+      eds_cluster_config:
+        service_name: fare
+        eds_config:
+          api_config_source:
+            cluster_names:
+            - eds
+            refresh_delay: 1s
+    )EOF");
+  Protobuf::RepeatedPtrField<envoy::api::v2::ClusterLoadAssignment> resources;
+  auto* cluster_load_assignment = resources.Add();
+  cluster_load_assignment->set_cluster_name("fare");
+
+  {
+    auto* endpoints = cluster_load_assignment->add_endpoints();
+    auto* locality = endpoints->mutable_locality();
+    locality->set_region("oceania");
+    locality->set_zone("hello");
+    locality->set_sub_zone("world");
+    endpoints->mutable_load_balancing_weight()->set_value(42);
+
+    auto* endpoint_address = endpoints->add_lb_endpoints()
+                                 ->mutable_endpoint()
+                                 ->mutable_address()
+                                 ->mutable_socket_address();
+    endpoint_address->set_address("1.2.3.4");
+    endpoint_address->set_port_value(80);
+  }
+
+  {
+    auto* endpoints = cluster_load_assignment->add_endpoints();
+    auto* locality = endpoints->mutable_locality();
+    locality->set_region("space");
+    locality->set_zone("station");
+    locality->set_sub_zone("international");
+
+    auto* endpoint_address = endpoints->add_lb_endpoints()
+                                 ->mutable_endpoint()
+                                 ->mutable_address()
+                                 ->mutable_socket_address();
+    endpoint_address->set_address("1.2.3.5");
+    endpoint_address->set_port_value(80);
+  }
+
+  {
+    auto* endpoints = cluster_load_assignment->add_endpoints();
+    auto* locality = endpoints->mutable_locality();
+    locality->set_region("sugar");
+    locality->set_zone("candy");
+    locality->set_sub_zone("mountain");
+    endpoints->mutable_load_balancing_weight()->set_value(37);
+
+    auto* endpoint_address = endpoints->add_lb_endpoints()
+                                 ->mutable_endpoint()
+                                 ->mutable_address()
+                                 ->mutable_socket_address();
+    endpoint_address->set_address("1.2.3.6");
+    endpoint_address->set_port_value(80);
+  }
+
+  bool initialized = false;
+  cluster_->initialize([&initialized] { initialized = true; });
+  VERBOSE_EXPECT_NO_THROW(cluster_->onConfigUpdate(resources));
+  EXPECT_TRUE(initialized);
+
+  const auto& locality_weights =
+      *cluster_->prioritySet().hostSetsPerPriority()[0]->localityWeights();
+  EXPECT_EQ(3, locality_weights.size());
+  EXPECT_EQ(42, locality_weights[0]);
+  EXPECT_EQ(0, locality_weights[1]);
+  EXPECT_EQ(37, locality_weights[2]);
 }
 
 // Validate that onConfigUpdate() updates bins hosts per locality as expected.
@@ -269,12 +388,13 @@ TEST_F(EdsTest, EndpointHostsPerLocality) {
     auto& hosts_per_locality = cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality();
     EXPECT_EQ(2, hosts_per_locality.get().size());
     EXPECT_EQ(1, hosts_per_locality.get()[0].size());
-    EXPECT_EQ(Locality("", "us-east-1a", ""), Locality(hosts_per_locality.get()[0][0]->locality()));
+    EXPECT_THAT(Locality("", "us-east-1a", ""),
+                ProtoEq(hosts_per_locality.get()[0][0]->locality()));
     EXPECT_EQ(2, hosts_per_locality.get()[1].size());
-    EXPECT_EQ(Locality("oceania", "koala", "ingsoc"),
-              Locality(hosts_per_locality.get()[1][0]->locality()));
-    EXPECT_EQ(Locality("oceania", "koala", "ingsoc"),
-              Locality(hosts_per_locality.get()[1][1]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "ingsoc"),
+                ProtoEq(hosts_per_locality.get()[1][0]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "ingsoc"),
+                ProtoEq(hosts_per_locality.get()[1][1]->locality()));
   }
 
   add_hosts_to_locality("oceania", "koala", "eucalyptus", 3);
@@ -286,16 +406,17 @@ TEST_F(EdsTest, EndpointHostsPerLocality) {
     auto& hosts_per_locality = cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality();
     EXPECT_EQ(4, hosts_per_locality.get().size());
     EXPECT_EQ(1, hosts_per_locality.get()[0].size());
-    EXPECT_EQ(Locality("", "us-east-1a", ""), Locality(hosts_per_locality.get()[0][0]->locality()));
+    EXPECT_THAT(Locality("", "us-east-1a", ""),
+                ProtoEq(hosts_per_locality.get()[0][0]->locality()));
     EXPECT_EQ(5, hosts_per_locality.get()[1].size());
-    EXPECT_EQ(Locality("general", "koala", "ingsoc"),
-              Locality(hosts_per_locality.get()[1][0]->locality()));
+    EXPECT_THAT(Locality("general", "koala", "ingsoc"),
+                ProtoEq(hosts_per_locality.get()[1][0]->locality()));
     EXPECT_EQ(3, hosts_per_locality.get()[2].size());
-    EXPECT_EQ(Locality("oceania", "koala", "eucalyptus"),
-              Locality(hosts_per_locality.get()[2][0]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "eucalyptus"),
+                ProtoEq(hosts_per_locality.get()[2][0]->locality()));
     EXPECT_EQ(2, hosts_per_locality.get()[3].size());
-    EXPECT_EQ(Locality("oceania", "koala", "ingsoc"),
-              Locality(hosts_per_locality.get()[3][0]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "ingsoc"),
+                ProtoEq(hosts_per_locality.get()[3][0]->locality()));
   }
 }
 
@@ -437,13 +558,13 @@ TEST_F(EdsTest, PriorityAndLocality) {
         cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality();
     EXPECT_EQ(2, first_hosts_per_locality.get().size());
     EXPECT_EQ(1, first_hosts_per_locality.get()[0].size());
-    EXPECT_EQ(Locality("", "us-east-1a", ""),
-              Locality(first_hosts_per_locality.get()[0][0]->locality()));
+    EXPECT_THAT(Locality("", "us-east-1a", ""),
+                ProtoEq(first_hosts_per_locality.get()[0][0]->locality()));
     EXPECT_EQ(2, first_hosts_per_locality.get()[1].size());
-    EXPECT_EQ(Locality("oceania", "koala", "ingsoc"),
-              Locality(first_hosts_per_locality.get()[1][0]->locality()));
-    EXPECT_EQ(Locality("oceania", "koala", "ingsoc"),
-              Locality(first_hosts_per_locality.get()[1][1]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "ingsoc"),
+                ProtoEq(first_hosts_per_locality.get()[1][0]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "ingsoc"),
+                ProtoEq(first_hosts_per_locality.get()[1][1]->locality()));
 
     auto& second_hosts_per_locality =
         cluster_->prioritySet().hostSetsPerPriority()[1]->hostsPerLocality();
@@ -463,27 +584,27 @@ TEST_F(EdsTest, PriorityAndLocality) {
         cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality();
     EXPECT_EQ(3, first_hosts_per_locality.get().size());
     EXPECT_EQ(1, first_hosts_per_locality.get()[0].size());
-    EXPECT_EQ(Locality("", "us-east-1a", ""),
-              Locality(first_hosts_per_locality.get()[0][0]->locality()));
+    EXPECT_THAT(Locality("", "us-east-1a", ""),
+                ProtoEq(first_hosts_per_locality.get()[0][0]->locality()));
     EXPECT_EQ(3, first_hosts_per_locality.get()[1].size());
-    EXPECT_EQ(Locality("oceania", "koala", "eucalyptus"),
-              Locality(first_hosts_per_locality.get()[1][0]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "eucalyptus"),
+                ProtoEq(first_hosts_per_locality.get()[1][0]->locality()));
     EXPECT_EQ(2, first_hosts_per_locality.get()[2].size());
-    EXPECT_EQ(Locality("oceania", "koala", "ingsoc"),
-              Locality(first_hosts_per_locality.get()[2][0]->locality()));
+    EXPECT_THAT(Locality("oceania", "koala", "ingsoc"),
+                ProtoEq(first_hosts_per_locality.get()[2][0]->locality()));
 
     auto& second_hosts_per_locality =
         cluster_->prioritySet().hostSetsPerPriority()[1]->hostsPerLocality();
     EXPECT_EQ(3, second_hosts_per_locality.get().size());
     EXPECT_EQ(8, second_hosts_per_locality.get()[0].size());
-    EXPECT_EQ(Locality("", "us-east-1a", ""),
-              Locality(second_hosts_per_locality.get()[0][0]->locality()));
+    EXPECT_THAT(Locality("", "us-east-1a", ""),
+                ProtoEq(second_hosts_per_locality.get()[0][0]->locality()));
     EXPECT_EQ(2, second_hosts_per_locality.get()[1].size());
-    EXPECT_EQ(Locality("foo", "bar", "eep"),
-              Locality(second_hosts_per_locality.get()[1][0]->locality()));
+    EXPECT_THAT(Locality("foo", "bar", "eep"),
+                ProtoEq(second_hosts_per_locality.get()[1][0]->locality()));
     EXPECT_EQ(5, second_hosts_per_locality.get()[2].size());
-    EXPECT_EQ(Locality("general", "koala", "ingsoc"),
-              Locality(second_hosts_per_locality.get()[2][0]->locality()));
+    EXPECT_THAT(Locality("general", "koala", "ingsoc"),
+                ProtoEq(second_hosts_per_locality.get()[2][0]->locality()));
   }
 }
 
diff --git a/test/common/upstream/load_balancer_benchmark.cc b/test/common/upstream/load_balancer_benchmark.cc
index 69e51fc0d948f..c1c78333776d6 100644
--- a/test/common/upstream/load_balancer_benchmark.cc
+++ b/test/common/upstream/load_balancer_benchmark.cc
@@ -23,7 +23,7 @@ class BaseTester {
       hosts.push_back(makeTestHost(info_, fmt::format("tcp://10.0.{}.{}:6379", i / 256, i % 256)));
     }
     HostVectorConstSharedPtr updated_hosts{new HostVector(hosts)};
-    host_set.updateHosts(updated_hosts, updated_hosts, nullptr, nullptr, hosts, {});
+    host_set.updateHosts(updated_hosts, updated_hosts, nullptr, nullptr, {}, hosts, {});
   }
 
   PrioritySetImpl priority_set_;
diff --git a/test/common/upstream/load_balancer_impl_test.cc b/test/common/upstream/load_balancer_impl_test.cc
index 79949e3f98add..6718a80e4ce3d 100644
--- a/test/common/upstream/load_balancer_impl_test.cc
+++ b/test/common/upstream/load_balancer_impl_test.cc
@@ -304,7 +304,7 @@ TEST_P(FailoverTest, ExtendPrioritiesWithLocalPrioritySet) {
   // test, but it should at least do no harm.
   HostVectorSharedPtr hosts(new HostVector({makeTestHost(info_, "tcp://127.0.0.1:82")}));
   local_priority_set_->getOrCreateHostSet(0).updateHosts(
-      hosts, hosts, empty_locality_, empty_locality_, empty_host_vector_, empty_host_vector_);
+      hosts, hosts, empty_locality_, empty_locality_, {}, empty_host_vector_, empty_host_vector_);
   EXPECT_EQ(tertiary_host_set_.hosts_[0], lb_->chooseHost(nullptr));
 }
 
@@ -333,6 +333,36 @@ TEST_P(RoundRobinLoadBalancerTest, Normal) {
   EXPECT_EQ(hostSet().healthy_hosts_[1], lb_->chooseHost(nullptr));
 }
 
+TEST_P(RoundRobinLoadBalancerTest, Locality) {
+  HostVectorSharedPtr hosts(new HostVector({makeTestHost(info_, "tcp://127.0.0.1:80"),
+                                            makeTestHost(info_, "tcp://127.0.0.1:81"),
+                                            makeTestHost(info_, "tcp://127.0.0.1:82")}));
+  HostsPerLocalitySharedPtr hosts_per_locality =
+      makeHostsPerLocality({{(*hosts)[1]}, {(*hosts)[0]}, {(*hosts)[2]}});
+  hostSet().hosts_ = *hosts;
+  hostSet().healthy_hosts_ = *hosts;
+  hostSet().healthy_hosts_per_locality_ = hosts_per_locality;
+  init(false);
+  // chooseLocality() return value determines which locality we use.
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(0));
+  EXPECT_EQ(hostSet().healthy_hosts_[1], lb_->chooseHost(nullptr));
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(1));
+  EXPECT_EQ(hostSet().healthy_hosts_[0], lb_->chooseHost(nullptr));
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(0));
+  EXPECT_EQ(hostSet().healthy_hosts_[1], lb_->chooseHost(nullptr));
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(1));
+  EXPECT_EQ(hostSet().healthy_hosts_[0], lb_->chooseHost(nullptr));
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(0));
+  EXPECT_EQ(hostSet().healthy_hosts_[1], lb_->chooseHost(nullptr));
+  // When there is no locality, we RR over all available hosts.
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(absl::optional<uint32_t>()));
+  EXPECT_EQ(hostSet().healthy_hosts_[0], lb_->chooseHost(nullptr));
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(absl::optional<uint32_t>()));
+  EXPECT_EQ(hostSet().healthy_hosts_[1], lb_->chooseHost(nullptr));
+  EXPECT_CALL(hostSet(), chooseLocality()).WillOnce(Return(absl::optional<uint32_t>()));
+  EXPECT_EQ(hostSet().healthy_hosts_[2], lb_->chooseHost(nullptr));
+}
+
 TEST_P(RoundRobinLoadBalancerTest, Weighted) {
   hostSet().healthy_hosts_ = {makeTestHost(info_, "tcp://127.0.0.1:80", 1),
                               makeTestHost(info_, "tcp://127.0.0.1:81", 2)};
@@ -433,7 +463,7 @@ TEST_P(RoundRobinLoadBalancerTest, ZoneAwareSmallCluster) {
   hostSet().healthy_hosts_per_locality_ = hosts_per_locality;
   common_config_.mutable_healthy_panic_threshold()->set_value(0);
   init(true);
-  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality,
+  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality, {},
                                empty_host_vector_, empty_host_vector_);
 
   EXPECT_CALL(runtime_.snapshot_, getInteger("upstream.healthy_panic_threshold", 0))
@@ -457,7 +487,7 @@ TEST_P(RoundRobinLoadBalancerTest, ZoneAwareSmallCluster) {
   EXPECT_CALL(runtime_.snapshot_, getInteger("upstream.zone_routing.min_cluster_size", 6))
       .WillRepeatedly(Return(1));
   // Trigger reload.
-  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality,
+  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality, {},
                                empty_host_vector_, empty_host_vector_);
   EXPECT_EQ(hostSet().healthy_hosts_per_locality_->get()[0][0], lb_->chooseHost(nullptr));
 }
@@ -481,7 +511,7 @@ TEST_P(RoundRobinLoadBalancerTest, NoZoneAwareDifferentZoneSize) {
   hostSet().healthy_hosts_per_locality_ = upstream_hosts_per_locality;
   common_config_.mutable_healthy_panic_threshold()->set_value(100);
   init(true);
-  local_host_set_->updateHosts(hosts, hosts, local_hosts_per_locality, local_hosts_per_locality,
+  local_host_set_->updateHosts(hosts, hosts, local_hosts_per_locality, local_hosts_per_locality, {},
                                empty_host_vector_, empty_host_vector_);
 
   EXPECT_CALL(runtime_.snapshot_, getInteger("upstream.healthy_panic_threshold", 100))
@@ -516,7 +546,7 @@ TEST_P(RoundRobinLoadBalancerTest, ZoneAwareRoutingLargeZoneSwitchOnOff) {
   hostSet().hosts_ = *hosts;
   hostSet().healthy_hosts_per_locality_ = hosts_per_locality;
   init(true);
-  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality,
+  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality, {},
                                empty_host_vector_, empty_host_vector_);
 
   // There is only one host in the given zone for zone aware routing.
@@ -565,7 +595,8 @@ TEST_P(RoundRobinLoadBalancerTest, ZoneAwareRoutingSmallZone) {
   hostSet().healthy_hosts_per_locality_ = upstream_hosts_per_locality;
   init(true);
   local_host_set_->updateHosts(local_hosts, local_hosts, local_hosts_per_locality,
-                               local_hosts_per_locality, empty_host_vector_, empty_host_vector_);
+                               local_hosts_per_locality, {}, empty_host_vector_,
+                               empty_host_vector_);
 
   // There is only one host in the given zone for zone aware routing.
   EXPECT_CALL(random_, random()).WillOnce(Return(0)).WillOnce(Return(100));
@@ -635,7 +666,7 @@ TEST_P(RoundRobinLoadBalancerTest, LowPrecisionForDistribution) {
   // To trigger update callback.
   auto local_hosts_per_locality_shared = makeHostsPerLocality(std::move(local_hosts_per_locality));
   local_host_set_->updateHosts(local_hosts, local_hosts, local_hosts_per_locality_shared,
-                               local_hosts_per_locality_shared, empty_host_vector_,
+                               local_hosts_per_locality_shared, {}, empty_host_vector_,
                                empty_host_vector_);
 
   // Force request out of small zone and to randomly select zone.
@@ -656,7 +687,7 @@ TEST_P(RoundRobinLoadBalancerTest, NoZoneAwareRoutingOneZone) {
   hostSet().hosts_ = *hosts;
   hostSet().healthy_hosts_per_locality_ = hosts_per_locality;
   init(true);
-  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality,
+  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality, {},
                                empty_host_vector_, empty_host_vector_);
   EXPECT_EQ(hostSet().healthy_hosts_[0], lb_->chooseHost(nullptr));
 }
@@ -671,7 +702,7 @@ TEST_P(RoundRobinLoadBalancerTest, NoZoneAwareRoutingNotHealthy) {
   hostSet().hosts_ = *hosts;
   hostSet().healthy_hosts_per_locality_ = hosts_per_locality;
   init(true);
-  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality,
+  local_host_set_->updateHosts(hosts, hosts, hosts_per_locality, hosts_per_locality, {},
                                empty_host_vector_, empty_host_vector_);
 
   // local zone has no healthy hosts, take from the all healthy hosts.
@@ -704,7 +735,8 @@ TEST_P(RoundRobinLoadBalancerTest, NoZoneAwareRoutingLocalEmpty) {
   hostSet().healthy_hosts_per_locality_ = upstream_hosts_per_locality;
   init(true);
   local_host_set_->updateHosts(local_hosts, local_hosts, local_hosts_per_locality,
-                               local_hosts_per_locality, empty_host_vector_, empty_host_vector_);
+                               local_hosts_per_locality, {}, empty_host_vector_,
+                               empty_host_vector_);
 
   // Local cluster is not OK, we'll do regular routing.
   EXPECT_EQ(hostSet().healthy_hosts_[0], lb_->chooseHost(nullptr));
@@ -732,7 +764,8 @@ TEST_P(RoundRobinLoadBalancerTest, NoZoneAwareRoutingNoLocalLocality) {
   hostSet().healthy_hosts_per_locality_ = upstream_hosts_per_locality;
   init(true);
   local_host_set_->updateHosts(local_hosts, local_hosts, local_hosts_per_locality,
-                               local_hosts_per_locality, empty_host_vector_, empty_host_vector_);
+                               local_hosts_per_locality, {}, empty_host_vector_,
+                               empty_host_vector_);
 
   // Local cluster is not OK, we'll do regular routing.
   EXPECT_EQ(hostSet().healthy_hosts_[0], lb_->chooseHost(nullptr));
diff --git a/test/common/upstream/load_balancer_simulation_test.cc b/test/common/upstream/load_balancer_simulation_test.cc
index 10d033c4d8438..8e92d7f884a72 100644
--- a/test/common/upstream/load_balancer_simulation_test.cc
+++ b/test/common/upstream/load_balancer_simulation_test.cc
@@ -101,7 +101,7 @@ class DISABLED_SimulationTest : public testing::Test {
       }
       auto per_zone_local_shared = makeHostsPerLocality(std::move(per_zone_local));
       local_priority_set_->getOrCreateHostSet(0).updateHosts(
-          originating_hosts, originating_hosts, per_zone_local_shared, per_zone_local_shared,
+          originating_hosts, originating_hosts, per_zone_local_shared, per_zone_local_shared, {},
           empty_vector_, empty_vector_);
 
       HostConstSharedPtr selected = lb.chooseHost(nullptr);
diff --git a/test/common/upstream/original_dst_cluster_test.cc b/test/common/upstream/original_dst_cluster_test.cc
index 6ed7cf3db2780..f975d4a00ce10 100644
--- a/test/common/upstream/original_dst_cluster_test.cc
+++ b/test/common/upstream/original_dst_cluster_test.cc
@@ -429,7 +429,7 @@ TEST_F(OriginalDstClusterTest, MultipleClusters) {
         const HostsPerLocalityConstSharedPtr empty_hosts_per_locality{new HostsPerLocalityImpl()};
 
         second.getOrCreateHostSet(0).updateHosts(new_hosts, healthy_hosts, empty_hosts_per_locality,
-                                                 empty_hosts_per_locality, added, removed);
+                                                 empty_hosts_per_locality, {}, added, removed);
       });
 
   EXPECT_CALL(membership_updated_, ready());
diff --git a/test/common/upstream/subset_lb_test.cc b/test/common/upstream/subset_lb_test.cc
index ef74c5d6697c0..e54d424f248cf 100644
--- a/test/common/upstream/subset_lb_test.cc
+++ b/test/common/upstream/subset_lb_test.cc
@@ -181,8 +181,9 @@ class SubsetLoadBalancerTest : public testing::TestWithParam<UpdateOrder> {
     }
     local_hosts_per_locality_ = makeHostsPerLocality(std::move(local_hosts_per_locality_vector));
 
-    local_priority_set_.getOrCreateHostSet(0).updateHosts(
-        local_hosts_, local_hosts_, local_hosts_per_locality_, local_hosts_per_locality_, {}, {});
+    local_priority_set_.getOrCreateHostSet(0).updateHosts(local_hosts_, local_hosts_,
+                                                          local_hosts_per_locality_,
+                                                          local_hosts_per_locality_, {}, {}, {});
 
     lb_.reset(new SubsetLoadBalancer(lb_type_, priority_set_, &local_priority_set_, stats_,
                                      runtime_, random_, subset_info_, ring_hash_lb_config_,
@@ -276,9 +277,9 @@ class SubsetLoadBalancerTest : public testing::TestWithParam<UpdateOrder> {
     }
 
     if (GetParam() == REMOVES_FIRST && !remove.empty()) {
-      local_priority_set_.getOrCreateHostSet(0).updateHosts(local_hosts_, local_hosts_,
-                                                            local_hosts_per_locality_,
-                                                            local_hosts_per_locality_, {}, remove);
+      local_priority_set_.getOrCreateHostSet(0).updateHosts(
+          local_hosts_, local_hosts_, local_hosts_per_locality_, local_hosts_per_locality_, {}, {},
+          remove);
     }
 
     for (const auto& host : add) {
@@ -290,14 +291,14 @@ class SubsetLoadBalancerTest : public testing::TestWithParam<UpdateOrder> {
 
     if (GetParam() == REMOVES_FIRST) {
       if (!add.empty()) {
-        local_priority_set_.getOrCreateHostSet(0).updateHosts(local_hosts_, local_hosts_,
-                                                              local_hosts_per_locality_,
-                                                              local_hosts_per_locality_, add, {});
+        local_priority_set_.getOrCreateHostSet(0).updateHosts(
+            local_hosts_, local_hosts_, local_hosts_per_locality_, local_hosts_per_locality_, {},
+            add, {});
       }
     } else if (!add.empty() || !remove.empty()) {
-      local_priority_set_.getOrCreateHostSet(0).updateHosts(local_hosts_, local_hosts_,
-                                                            local_hosts_per_locality_,
-                                                            local_hosts_per_locality_, add, remove);
+      local_priority_set_.getOrCreateHostSet(0).updateHosts(
+          local_hosts_, local_hosts_, local_hosts_per_locality_, local_hosts_per_locality_, {}, add,
+          remove);
     }
   }
 
diff --git a/test/common/upstream/upstream_impl_test.cc b/test/common/upstream/upstream_impl_test.cc
index 5e70e443624e0..59795d185f900 100644
--- a/test/common/upstream/upstream_impl_test.cc
+++ b/test/common/upstream/upstream_impl_test.cc
@@ -735,7 +735,7 @@ TEST(PrioritySet, Extend) {
   HostVector hosts_removed{};
 
   priority_set.hostSetsPerPriority()[1]->updateHosts(
-      hosts, hosts, hosts_per_locality, hosts_per_locality, hosts_added, hosts_removed);
+      hosts, hosts, hosts_per_locality, hosts_per_locality, {}, hosts_added, hosts_removed);
   EXPECT_EQ(1, changes);
   EXPECT_EQ(last_priority, 1);
   EXPECT_EQ(1, priority_set.hostSetsPerPriority()[1]->hosts().size());
@@ -841,6 +841,106 @@ TEST(HostsPerLocalityImpl, Filter) {
   }
 }
 
+class HostSetImplLocalityTest : public ::testing::Test {
+public:
+  LocalityWeightsConstSharedPtr locality_weights_;
+  HostSetImpl host_set_{0};
+  std::shared_ptr<MockClusterInfo> info_{new NiceMock<MockClusterInfo>()};
+  HostVector hosts_{
+      makeTestHost(info_, "tcp://127.0.0.1:80"), makeTestHost(info_, "tcp://127.0.0.1:81"),
+      makeTestHost(info_, "tcp://127.0.0.1:82"), makeTestHost(info_, "tcp://127.0.0.1:83"),
+      makeTestHost(info_, "tcp://127.0.0.1:84"), makeTestHost(info_, "tcp://127.0.0.1:85")};
+};
+
+// When no locality weights belong to the host set, there's an empty pick.
+TEST_F(HostSetImplLocalityTest, Empty) {
+  EXPECT_EQ(nullptr, host_set_.localityWeights());
+  EXPECT_FALSE(host_set_.chooseLocality().has_value());
+}
+
+// When all locality weights are the same we have unweighted RR behavior.
+TEST_F(HostSetImplLocalityTest, Unweighted) {
+  HostsPerLocalitySharedPtr hosts_per_locality =
+      makeHostsPerLocality({{hosts_[0]}, {hosts_[1]}, {hosts_[2]}});
+  LocalityWeightsConstSharedPtr locality_weights{new LocalityWeights{1, 1, 1}};
+  host_set_.updateHosts({}, {}, hosts_per_locality, hosts_per_locality, locality_weights, {}, {});
+  EXPECT_EQ(0, host_set_.chooseLocality().value());
+  EXPECT_EQ(1, host_set_.chooseLocality().value());
+  EXPECT_EQ(2, host_set_.chooseLocality().value());
+  EXPECT_EQ(0, host_set_.chooseLocality().value());
+  EXPECT_EQ(1, host_set_.chooseLocality().value());
+  EXPECT_EQ(2, host_set_.chooseLocality().value());
+}
+
+// When locality weights differ, we have weighted RR behavior.
+TEST_F(HostSetImplLocalityTest, Weighted) {
+  HostsPerLocalitySharedPtr hosts_per_locality = makeHostsPerLocality({{hosts_[0]}, {hosts_[1]}});
+  LocalityWeightsConstSharedPtr locality_weights{new LocalityWeights{1, 2}};
+  host_set_.updateHosts({}, {}, hosts_per_locality, hosts_per_locality, locality_weights, {}, {});
+  EXPECT_EQ(1, host_set_.chooseLocality().value());
+  EXPECT_EQ(0, host_set_.chooseLocality().value());
+  EXPECT_EQ(1, host_set_.chooseLocality().value());
+  EXPECT_EQ(1, host_set_.chooseLocality().value());
+  EXPECT_EQ(0, host_set_.chooseLocality().value());
+  EXPECT_EQ(1, host_set_.chooseLocality().value());
+}
+
+// Localities with no weight assignment are never picked.
+TEST_F(HostSetImplLocalityTest, MissingWeight) {
+  HostsPerLocalitySharedPtr hosts_per_locality =
+      makeHostsPerLocality({{hosts_[0]}, {hosts_[1]}, {hosts_[2]}});
+  LocalityWeightsConstSharedPtr locality_weights{new LocalityWeights{1, 0, 1}};
+  host_set_.updateHosts({}, {}, hosts_per_locality, hosts_per_locality, locality_weights, {}, {});
+  EXPECT_EQ(0, host_set_.chooseLocality().value());
+  EXPECT_EQ(2, host_set_.chooseLocality().value());
+  EXPECT_EQ(0, host_set_.chooseLocality().value());
+  EXPECT_EQ(2, host_set_.chooseLocality().value());
+  EXPECT_EQ(0, host_set_.chooseLocality().value());
+  EXPECT_EQ(2, host_set_.chooseLocality().value());
+}
+
+// Gentle failover between localities as health diminishes.
+TEST_F(HostSetImplLocalityTest, UnhealthyFailover) {
+  const auto setHealthyHostCount = [this](uint32_t host_count) {
+    LocalityWeightsConstSharedPtr locality_weights{new LocalityWeights{1, 2}};
+    HostsPerLocalitySharedPtr hosts_per_locality = makeHostsPerLocality(
+        {{hosts_[0], hosts_[1], hosts_[2], hosts_[3], hosts_[4]}, {hosts_[5]}});
+    HostVector healthy_hosts;
+    for (uint32_t i = 0; i < host_count; ++i) {
+      healthy_hosts.emplace_back(hosts_[i]);
+    }
+    HostsPerLocalitySharedPtr healthy_hosts_per_locality =
+        makeHostsPerLocality({healthy_hosts, {hosts_[5]}});
+    host_set_.updateHosts({}, {}, hosts_per_locality, healthy_hosts_per_locality, locality_weights,
+                          {}, {});
+  };
+
+  const auto expectPicks = [this](uint32_t locality_0_picks, uint32_t locality_1_picks) {
+    uint32_t count[2] = {0, 0};
+    for (uint32_t i = 0; i < 100; ++i) {
+      const uint32_t locality_index = host_set_.chooseLocality().value();
+      ASSERT_LT(locality_index, 2);
+      ++count[locality_index];
+    }
+    ENVOY_LOG_MISC(debug, "Locality picks {} {}", count[0], count[1]);
+    EXPECT_EQ(locality_0_picks, count[0]);
+    EXPECT_EQ(locality_1_picks, count[1]);
+  };
+
+  setHealthyHostCount(5);
+  expectPicks(33, 67);
+  setHealthyHostCount(4);
+  expectPicks(33, 67);
+  setHealthyHostCount(3);
+  expectPicks(29, 71);
+  setHealthyHostCount(2);
+  expectPicks(22, 78);
+  setHealthyHostCount(1);
+  expectPicks(12, 88);
+  setHealthyHostCount(0);
+  expectPicks(0, 100);
+}
+
 } // namespace
 } // namespace Upstream
 } // namespace Envoy
diff --git a/test/integration/load_stats_integration_test.cc b/test/integration/load_stats_integration_test.cc
index 66263f5c2d1e6..5ebacc84b261e 100644
--- a/test/integration/load_stats_integration_test.cc
+++ b/test/integration/load_stats_integration_test.cc
@@ -29,12 +29,25 @@ class LoadStatsIntegrationTest : public HttpIntegrationTest,
     ++num_endpoints;
   }
 
+  // Used as args to updateClusterLocalityAssignment().
+  struct LocalityAssignment {
+    LocalityAssignment() : LocalityAssignment({}, 0) {}
+    LocalityAssignment(const std::vector<uint32_t>& endpoints) : LocalityAssignment(endpoints, 0) {}
+    LocalityAssignment(const std::vector<uint32_t>& endpoints, uint32_t weight)
+        : endpoints_(endpoints), weight_(weight) {}
+
+    // service_upstream_ indices for endpoints in the cluster.
+    const std::vector<uint32_t> endpoints_;
+    // If non-zero, locality level weighting.
+    const uint32_t weight_{};
+  };
+
   // We need to supply the endpoints via EDS to provide locality information for
   // load reporting. Use a filesystem delivery to simplify test mechanics.
-  void updateClusterLoadAssignment(const std::vector<uint32_t>& winter_upstreams,
-                                   const std::vector<uint32_t>& dragon_upstreeams,
-                                   const std::vector<uint32_t>& p1_winter_upstreams,
-                                   const std::vector<uint32_t>& p1_dragon_upstreams) {
+  void updateClusterLoadAssignment(const LocalityAssignment& winter_upstreams,
+                                   const LocalityAssignment& dragon_upstreams,
+                                   const LocalityAssignment& p1_winter_upstreams,
+                                   const LocalityAssignment& p1_dragon_upstreams) {
     uint32_t num_endpoints = 0;
     envoy::api::v2::ClusterLoadAssignment cluster_load_assignment;
     cluster_load_assignment.set_cluster_name("cluster_0");
@@ -43,7 +56,10 @@ class LoadStatsIntegrationTest : public HttpIntegrationTest,
     winter->mutable_locality()->set_region("some_region");
     winter->mutable_locality()->set_zone("zone_name");
     winter->mutable_locality()->set_sub_zone("winter");
-    for (uint32_t index : winter_upstreams) {
+    if (winter_upstreams.weight_ > 0) {
+      winter->mutable_load_balancing_weight()->set_value(winter_upstreams.weight_);
+    }
+    for (uint32_t index : winter_upstreams.endpoints_) {
       addEndpoint(*winter, index, num_endpoints);
     }
 
@@ -51,7 +67,10 @@ class LoadStatsIntegrationTest : public HttpIntegrationTest,
     dragon->mutable_locality()->set_region("some_region");
     dragon->mutable_locality()->set_zone("zone_name");
     dragon->mutable_locality()->set_sub_zone("dragon");
-    for (uint32_t index : dragon_upstreeams) {
+    if (dragon_upstreams.weight_ > 0) {
+      dragon->mutable_load_balancing_weight()->set_value(dragon_upstreams.weight_);
+    }
+    for (uint32_t index : dragon_upstreams.endpoints_) {
       addEndpoint(*dragon, index, num_endpoints);
     }
 
@@ -60,7 +79,7 @@ class LoadStatsIntegrationTest : public HttpIntegrationTest,
     winter_p1->mutable_locality()->set_region("some_region");
     winter_p1->mutable_locality()->set_zone("zone_name");
     winter_p1->mutable_locality()->set_sub_zone("winter");
-    for (uint32_t index : p1_winter_upstreams) {
+    for (uint32_t index : p1_winter_upstreams.endpoints_) {
       addEndpoint(*winter_p1, index, num_endpoints);
     }
 
@@ -69,7 +88,7 @@ class LoadStatsIntegrationTest : public HttpIntegrationTest,
     dragon_p1->mutable_locality()->set_region("some_region");
     dragon_p1->mutable_locality()->set_zone("zone_name");
     dragon_p1->mutable_locality()->set_sub_zone("dragon");
-    for (uint32_t index : p1_dragon_upstreams) {
+    for (uint32_t index : p1_dragon_upstreams.endpoints_) {
       addEndpoint(*dragon_p1, index, num_endpoints);
     }
 
@@ -128,6 +147,9 @@ class LoadStatsIntegrationTest : public HttpIntegrationTest,
       cluster_0->set_type(envoy::api::v2::Cluster::EDS);
       auto* eds_cluster_config = cluster_0->mutable_eds_cluster_config();
       eds_cluster_config->mutable_eds_config()->set_path(eds_path_);
+      if (locality_weighted_lb_) {
+        cluster_0->mutable_common_lb_config()->mutable_locality_weighted_lb_config();
+      }
     });
     HttpIntegrationTest::initialize();
     for (uint32_t i = 0; i < upstream_endpoints_; ++i) {
@@ -308,6 +330,7 @@ class LoadStatsIntegrationTest : public HttpIntegrationTest,
   std::string eds_path_;
   uint32_t eds_version_{};
   uint32_t load_requests_{};
+  bool locality_weighted_lb_{};
 
   const uint64_t request_size_ = 1024;
   const uint64_t response_size_ = 512;
@@ -330,7 +353,7 @@ TEST_P(LoadStatsIntegrationTest, Success) {
   // unknown cluster to exercise the handling of this case.
   requestLoadStatsResponse({"cluster_0", "cluster_1"});
 
-  updateClusterLoadAssignment({0}, {1}, {3}, {});
+  updateClusterLoadAssignment({{0}}, {{1}}, {{3}}, {});
 
   for (uint32_t i = 0; i < 4; ++i) {
     sendAndReceiveUpstream(i % 2);
@@ -345,7 +368,7 @@ TEST_P(LoadStatsIntegrationTest, Success) {
   EXPECT_EQ(0, test_server_->counter("load_reporter.errors")->value());
 
   // 33%/67% split between dragon/winter primary localities.
-  updateClusterLoadAssignment({0}, {1, 2}, {}, {4});
+  updateClusterLoadAssignment({{0}}, {{1, 2}}, {}, {{4}});
   requestLoadStatsResponse({"cluster_0"});
 
   for (uint32_t i = 0; i < 6; ++i) {
@@ -361,7 +384,7 @@ TEST_P(LoadStatsIntegrationTest, Success) {
   EXPECT_EQ(0, test_server_->counter("load_reporter.errors")->value());
 
   // Change to 50/50 for the failover clusters.
-  updateClusterLoadAssignment({}, {}, {3}, {4});
+  updateClusterLoadAssignment({}, {}, {{3}}, {{4}});
   requestLoadStatsResponse({"cluster_0"});
   test_server_->waitForGaugeEq("cluster.cluster_0.membership_total", 2);
 
@@ -377,7 +400,7 @@ TEST_P(LoadStatsIntegrationTest, Success) {
 
   // 100% winter locality.
   updateClusterLoadAssignment({}, {}, {}, {});
-  updateClusterLoadAssignment({1}, {}, {}, {});
+  updateClusterLoadAssignment({{1}}, {}, {}, {});
   requestLoadStatsResponse({"cluster_0"});
 
   for (uint32_t i = 0; i < 1; ++i) {
@@ -405,6 +428,42 @@ TEST_P(LoadStatsIntegrationTest, Success) {
   cleanupLoadStatsConnection();
 }
 
+// Validate the load reports for successful requests when using locality
+// weighted LB. This serves as a de facto integration test for locality weighted
+// LB.
+TEST_P(LoadStatsIntegrationTest, LocalityWeighted) {
+  locality_weighted_lb_ = true;
+  initialize();
+
+  waitForLoadStatsStream();
+  waitForLoadStatsRequest({});
+  loadstats_stream_->startGrpcStream();
+
+  requestLoadStatsResponse({"cluster_0"});
+
+  // Simple 33%/67% split between dragon/winter localities.
+  // Even though there are more endpoints in the dragon locality, the winter locality gets the
+  // expected weighting in the WRR locality schedule.
+  updateClusterLoadAssignment({{0}, 2}, {{1, 2}, 1}, {}, {});
+
+  sendAndReceiveUpstream(0);
+  sendAndReceiveUpstream(1);
+  sendAndReceiveUpstream(0);
+  sendAndReceiveUpstream(0);
+  sendAndReceiveUpstream(2);
+  sendAndReceiveUpstream(0);
+
+  // Verify we get the expect request distribution.
+  waitForLoadStatsRequest({localityStats("winter", 4, 0, 0), localityStats("dragon", 2, 0, 0)});
+
+  EXPECT_EQ(1, test_server_->counter("load_reporter.requests")->value());
+  // On slow machines, more than one load stats response may be pushed while we are simulating load.
+  EXPECT_LE(2, test_server_->counter("load_reporter.responses")->value());
+  EXPECT_EQ(0, test_server_->counter("load_reporter.errors")->value());
+
+  cleanupLoadStatsConnection();
+}
+
 // Validate the load reports for requests when all endpoints are non-local.
 TEST_P(LoadStatsIntegrationTest, NoLocalLocality) {
   sub_zone_ = "summer";
@@ -418,7 +477,7 @@ TEST_P(LoadStatsIntegrationTest, NoLocalLocality) {
   // unknown cluster to exercise the handling of this case.
   requestLoadStatsResponse({"cluster_0", "cluster_1"});
 
-  updateClusterLoadAssignment({0}, {1}, {3}, {});
+  updateClusterLoadAssignment({{0}}, {{1}}, {{3}}, {});
 
   for (uint32_t i = 0; i < 4; ++i) {
     sendAndReceiveUpstream(i % 2);
@@ -447,7 +506,7 @@ TEST_P(LoadStatsIntegrationTest, Error) {
   loadstats_stream_->startGrpcStream();
 
   requestLoadStatsResponse({"cluster_0"});
-  updateClusterLoadAssignment({0}, {}, {}, {});
+  updateClusterLoadAssignment({{0}}, {}, {}, {});
 
   // This should count as an error since 5xx.
   sendAndReceiveUpstream(0, 503);
@@ -471,7 +530,7 @@ TEST_P(LoadStatsIntegrationTest, InProgress) {
   waitForLoadStatsStream();
   waitForLoadStatsRequest({});
   loadstats_stream_->startGrpcStream();
-  updateClusterLoadAssignment({0}, {}, {}, {});
+  updateClusterLoadAssignment({{0}}, {}, {}, {});
 
   requestLoadStatsResponse({"cluster_0"});
   initiateClientConnection();
@@ -500,7 +559,7 @@ TEST_P(LoadStatsIntegrationTest, Dropped) {
   waitForLoadStatsRequest({});
   loadstats_stream_->startGrpcStream();
 
-  updateClusterLoadAssignment({0}, {}, {}, {});
+  updateClusterLoadAssignment({{0}}, {}, {}, {});
   requestLoadStatsResponse({"cluster_0"});
   // This should count as dropped, since we trigger circuit breaking.
   initiateClientConnection();
diff --git a/test/mocks/upstream/mocks.cc b/test/mocks/upstream/mocks.cc
index c86092d6a2909..874ef72f4483c 100644
--- a/test/mocks/upstream/mocks.cc
+++ b/test/mocks/upstream/mocks.cc
@@ -28,6 +28,9 @@ MockHostSet::MockHostSet(uint32_t priority) : priority_(priority) {
   ON_CALL(*this, healthyHostsPerLocality())
       .WillByDefault(
           Invoke([this]() -> const HostsPerLocality& { return *healthy_hosts_per_locality_; }));
+  ON_CALL(*this, localityWeights()).WillByDefault(Invoke([this]() -> LocalityWeightsConstSharedPtr {
+    return locality_weights_;
+  }));
 }
 
 MockPrioritySet::MockPrioritySet() {
diff --git a/test/mocks/upstream/mocks.h b/test/mocks/upstream/mocks.h
index ae1f60f027e21..884c3b074b0de 100644
--- a/test/mocks/upstream/mocks.h
+++ b/test/mocks/upstream/mocks.h
@@ -46,10 +46,13 @@ class MockHostSet : public HostSet {
   MOCK_CONST_METHOD0(healthyHosts, const HostVector&());
   MOCK_CONST_METHOD0(hostsPerLocality, const HostsPerLocality&());
   MOCK_CONST_METHOD0(healthyHostsPerLocality, const HostsPerLocality&());
-  MOCK_METHOD6(updateHosts, void(std::shared_ptr<const HostVector> hosts,
+  MOCK_CONST_METHOD0(localityWeights, LocalityWeightsConstSharedPtr());
+  MOCK_METHOD0(chooseLocality, absl::optional<uint32_t>());
+  MOCK_METHOD7(updateHosts, void(std::shared_ptr<const HostVector> hosts,
                                  std::shared_ptr<const HostVector> healthy_hosts,
                                  HostsPerLocalityConstSharedPtr hosts_per_locality,
                                  HostsPerLocalityConstSharedPtr healthy_hosts_per_locality,
+                                 LocalityWeightsConstSharedPtr locality_weights,
                                  const HostVector& hosts_added, const HostVector& hosts_removed));
   MOCK_CONST_METHOD0(priority, uint32_t());
 
@@ -57,6 +60,7 @@ class MockHostSet : public HostSet {
   HostVector healthy_hosts_;
   HostsPerLocalitySharedPtr hosts_per_locality_{new HostsPerLocalityImpl()};
   HostsPerLocalitySharedPtr healthy_hosts_per_locality_{new HostsPerLocalityImpl()};
+  LocalityWeightsConstSharedPtr locality_weights_{{}};
   Common::CallbackManager<uint32_t, const HostVector&, const HostVector&> member_update_cb_helper_;
   uint32_t priority_{};
 };