load balancer: added maglev/ring hash load balancer extension

CODEOWNERS

@@ -289,6 +289,8 @@ extensions/filters/http/oauth2 @derekargueta @snowp
 /*/extensions/load_balancing_policies/least_request @wbpcode @UNOWNED
 /*/extensions/load_balancing_policies/random @wbpcode @UNOWNED
 /*/extensions/load_balancing_policies/round_robin @wbpcode @UNOWNED
+/*/extensions/load_balancing_policies/ring_hash @wbpcode @UNOWNED
+/*/extensions/load_balancing_policies/maglev @wbpcode @UNOWNED
 # Early header mutation
 /*/extensions/http/early_header_mutation/header_mutation @wbpcode @UNOWNED
 

api/envoy/extensions/load_balancing_policies/maglev/v3/maglev.proto

@@ -16,7 +16,7 @@ option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/loa
 option (udpa.annotations.file_status).package_version_status = ACTIVE;
 
 // [#protodoc-title: Maglev Load Balancing Policy]
-// [#not-implemented-hide:]
+// [#extension: envoy.load_balancing_policies.maglev]
 
 // This configuration allows the built-in Maglev LB policy to be configured via the LB policy
 // extension point. See the :ref:`load balancing architecture overview
@@ -30,4 +30,7 @@ message Maglev {
 
   // Common configuration for hashing-based load balancing policies.
   common.v3.ConsistentHashingLbConfig consistent_hashing_lb_config = 2;
+
+  // Enable locality weighted load balancing for maglev lb explicitly.
+  common.v3.LocalityLbConfig.LocalityWeightedLbConfig locality_weighted_lb_config = 3;
 }

api/envoy/extensions/load_balancing_policies/ring_hash/v3/ring_hash.proto

@@ -17,12 +17,12 @@ option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/loa
 option (udpa.annotations.file_status).package_version_status = ACTIVE;
 
 // [#protodoc-title: Ring Hash Load Balancing Policy]
-// [#not-implemented-hide:]
+// [#extension: envoy.load_balancing_policies.ring_hash]
 
 // This configuration allows the built-in RING_HASH LB policy to be configured via the LB policy
 // extension point. See the :ref:`load balancing architecture overview
 // <arch_overview_load_balancing_types>` for more information.
-// [#next-free-field: 7]
+// [#next-free-field: 8]
 message RingHash {
   // The hash function used to hash hosts onto the ketama ring.
   enum HashFunction {
@@ -90,4 +90,7 @@ message RingHash {
 
   // Common configuration for hashing-based load balancing policies.
   common.v3.ConsistentHashingLbConfig consistent_hashing_lb_config = 6;
+
+  // Enable locality weighted load balancing for ring hash lb explicitly.
+  common.v3.LocalityLbConfig.LocalityWeightedLbConfig locality_weighted_lb_config = 7;
 }

changelogs/current.yaml

@@ -17,5 +17,11 @@ removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`
 
 new_features:
+- area: upstream
+  change: |
+    added :ref:`ring hash extension <envoy_v3_api_msg_extensions.load_balancing_policies.ring_hash.v3.RingHash>` to suppport the :ref:`load balancer policy <envoy_v3_api_field_config.cluster.v3.Cluster.load_balancing_policy>`.
+- area: upstream
+  change: |
+    added :ref:`maglev extension <envoy_v3_api_msg_extensions.load_balancing_policies.maglev.v3.Maglev>` to suppport the :ref:`load balancer policy <envoy_v3_api_field_config.cluster.v3.Cluster.load_balancing_policy>`.
 
 deprecated:

source/common/upstream/BUILD

@@ -409,6 +409,7 @@ envoy_cc_library(
         ":thread_aware_lb_lib",
         ":upstream_lib",
         "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/load_balancing_policies/maglev/v3:pkg_cc_proto",
     ],
 )
 
@@ -423,6 +424,7 @@ envoy_cc_library(
         ":thread_aware_lb_lib",
         "//source/common/common:minimal_logger_lib",
         "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/load_balancing_policies/ring_hash/v3:pkg_cc_proto",
     ],
 )
 

source/common/upstream/load_balancer_factory_base.h

@@ -13,11 +13,13 @@ namespace Upstream {
  * TODO: provide a ThreadLocalLoadBalancer construct to abstract away thread-awareness from load
  * balancing extensions that don't require it.
  */
-class TypedLoadBalancerFactoryBase : public TypedLoadBalancerFactory {
+template <class Proto> class TypedLoadBalancerFactoryBase : public TypedLoadBalancerFactory {
 public:
   // Upstream::TypedLoadBalancerFactory
   std::string name() const override { return name_; }
 
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override { return std::make_unique<Proto>(); }
+
 protected:
   TypedLoadBalancerFactoryBase(const std::string& name) : name_(name) {}
 

source/common/upstream/maglev_lb.cc

@@ -97,7 +97,10 @@ MaglevLoadBalancer::MaglevLoadBalancer(
     Runtime::Loader& runtime, Random::RandomGenerator& random,
     OptRef<const envoy::config::cluster::v3::Cluster::MaglevLbConfig> config,
     const envoy::config::cluster::v3::Cluster::CommonLbConfig& common_config)
-    : ThreadAwareLoadBalancerBase(priority_set, stats, runtime, random, common_config),
+    : ThreadAwareLoadBalancerBase(priority_set, stats, runtime, random,
+                                  PROTOBUF_PERCENT_TO_ROUNDED_INTEGER_OR_DEFAULT(
+                                      common_config, healthy_panic_threshold, 100, 50),
+                                  common_config.has_locality_weighted_lb_config()),
       scope_(scope.createScope("maglev_lb.")), stats_(generateStats(*scope_)),
       table_size_(config ? PROTOBUF_GET_WRAPPED_OR_DEFAULT(config.ref(), table_size,
                                                            MaglevTable::DefaultTableSize)
@@ -115,6 +118,28 @@ MaglevLoadBalancer::MaglevLoadBalancer(
   }
 }
 
+MaglevLoadBalancer::MaglevLoadBalancer(
+    const PrioritySet& priority_set, ClusterLbStats& stats, Stats::Scope& scope,
+    Runtime::Loader& runtime, Random::RandomGenerator& random, uint32_t healthy_panic_threshold,
+    const envoy::extensions::load_balancing_policies::maglev::v3::Maglev& config)
+    : ThreadAwareLoadBalancerBase(priority_set, stats, runtime, random, healthy_panic_threshold,
+                                  config.has_locality_weighted_lb_config()),
+      scope_(scope.createScope("maglev_lb.")), stats_(generateStats(*scope_)),
+      table_size_(
+          PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, table_size, MaglevTable::DefaultTableSize)),
+      use_hostname_for_hashing_(
+          config.has_consistent_hashing_lb_config()
+              ? config.consistent_hashing_lb_config().use_hostname_for_hashing()
+              : false),
+      hash_balance_factor_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(config.consistent_hashing_lb_config(),
+                                                           hash_balance_factor, 0)) {
+  ENVOY_LOG(debug, "maglev table size: {}", table_size_);
+  // The table size must be prime number.
+  if (!Primes::isPrime(table_size_)) {
+    throw EnvoyException("The table size of maglev must be prime number");
+  }
+}
+
 MaglevLoadBalancerStats MaglevLoadBalancer::generateStats(Stats::Scope& scope) {
   return {ALL_MAGLEV_LOAD_BALANCER_STATS(POOL_GAUGE(scope))};
 }

source/common/upstream/maglev_lb.h

@@ -2,6 +2,8 @@
 
 #include "envoy/common/random_generator.h"
 #include "envoy/config/cluster/v3/cluster.pb.h"
+#include "envoy/extensions/load_balancing_policies/maglev/v3/maglev.pb.h"
+#include "envoy/extensions/load_balancing_policies/maglev/v3/maglev.pb.validate.h"
 #include "envoy/stats/scope.h"
 #include "envoy/stats/stats_macros.h"
 
@@ -76,6 +78,11 @@ class MaglevLoadBalancer : public ThreadAwareLoadBalancerBase,
                      OptRef<const envoy::config::cluster::v3::Cluster::MaglevLbConfig> config,
                      const envoy::config::cluster::v3::Cluster::CommonLbConfig& common_config);
 
+  MaglevLoadBalancer(const PrioritySet& priority_set, ClusterLbStats& stats, Stats::Scope& scope,
+                     Runtime::Loader& runtime, Random::RandomGenerator& random,
+                     uint32_t healthy_panic_threshold,
+                     const envoy::extensions::load_balancing_policies::maglev::v3::Maglev& config);
+
   const MaglevLoadBalancerStats& stats() const { return stats_; }
   uint64_t tableSize() const { return table_size_; }
 

source/common/upstream/ring_hash_lb.cc

@@ -21,7 +21,10 @@ RingHashLoadBalancer::RingHashLoadBalancer(
     Runtime::Loader& runtime, Random::RandomGenerator& random,
     OptRef<const envoy::config::cluster::v3::Cluster::RingHashLbConfig> config,
     const envoy::config::cluster::v3::Cluster::CommonLbConfig& common_config)
-    : ThreadAwareLoadBalancerBase(priority_set, stats, runtime, random, common_config),
+    : ThreadAwareLoadBalancerBase(priority_set, stats, runtime, random,
+                                  PROTOBUF_PERCENT_TO_ROUNDED_INTEGER_OR_DEFAULT(
+                                      common_config, healthy_panic_threshold, 100, 50),
+                                  common_config.has_locality_weighted_lb_config()),
       scope_(scope.createScope("ring_hash_lb.")), stats_(generateStats(*scope_)),
       min_ring_size_(config.has_value() ? PROTOBUF_GET_WRAPPED_OR_DEFAULT(
                                               config.ref(), minimum_ring_size, DefaultMinRingSize)
@@ -46,6 +49,34 @@ RingHashLoadBalancer::RingHashLoadBalancer(
   }
 }
 
+RingHashLoadBalancer::RingHashLoadBalancer(
+    const PrioritySet& priority_set, ClusterLbStats& stats, Stats::Scope& scope,
+    Runtime::Loader& runtime, Random::RandomGenerator& random, uint32_t healthy_panic_threshold,
+    const envoy::extensions::load_balancing_policies::ring_hash::v3::RingHash& config)
+    : ThreadAwareLoadBalancerBase(priority_set, stats, runtime, random, healthy_panic_threshold,
+                                  config.has_locality_weighted_lb_config()),
+      scope_(scope.createScope("ring_hash_lb.")), stats_(generateStats(*scope_)),
+      min_ring_size_(
+          PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, minimum_ring_size, DefaultMinRingSize)),
+      max_ring_size_(
+          PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, maximum_ring_size, DefaultMaxRingSize)),
+      hash_function_(static_cast<HashFunction>(config.hash_function())),
+      use_hostname_for_hashing_(
+          config.has_consistent_hashing_lb_config()
+              ? config.consistent_hashing_lb_config().use_hostname_for_hashing()
+              : config.use_hostname_for_hashing()),
+      hash_balance_factor_(config.has_consistent_hashing_lb_config()
+                               ? PROTOBUF_GET_WRAPPED_OR_DEFAULT(
+                                     config.consistent_hashing_lb_config(), hash_balance_factor, 0)
+                               : PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, hash_balance_factor, 0)) {
+  // It's important to do any config validation here, rather than deferring to Ring's ctor,
+  // because any exceptions thrown here will be caught and handled properly.
+  if (min_ring_size_ > max_ring_size_) {
+    throw EnvoyException(fmt::format("ring hash: minimum_ring_size ({}) > maximum_ring_size ({})",
+                                     min_ring_size_, max_ring_size_));
+  }
+}
+
 RingHashLoadBalancerStats RingHashLoadBalancer::generateStats(Stats::Scope& scope) {
   return {ALL_RING_HASH_LOAD_BALANCER_STATS(POOL_GAUGE(scope))};
 }

source/common/upstream/ring_hash_lb.h

@@ -3,6 +3,8 @@
 #include <vector>
 
 #include "envoy/config/cluster/v3/cluster.pb.h"
+#include "envoy/extensions/load_balancing_policies/ring_hash/v3/ring_hash.pb.h"
+#include "envoy/extensions/load_balancing_policies/ring_hash/v3/ring_hash.pb.validate.h"
 #include "envoy/runtime/runtime.h"
 #include "envoy/stats/scope.h"
 #include "envoy/stats/stats_macros.h"
@@ -45,6 +47,11 @@ class RingHashLoadBalancer : public ThreadAwareLoadBalancerBase,
                        OptRef<const envoy::config::cluster::v3::Cluster::RingHashLbConfig> config,
                        const envoy::config::cluster::v3::Cluster::CommonLbConfig& common_config);
 
+  RingHashLoadBalancer(
+      const PrioritySet& priority_set, ClusterLbStats& stats, Stats::Scope& scope,
+      Runtime::Loader& runtime, Random::RandomGenerator& random, uint32_t healthy_panic_threshold,
+      const envoy::extensions::load_balancing_policies::ring_hash::v3::RingHash& config);
+
   const RingHashLoadBalancerStats& stats() const { return stats_; }
 
 private:

source/common/upstream/thread_aware_lb_impl.cc

@@ -70,8 +70,10 @@ void normalizeLocalityWeights(const HostsPerLocality& hosts_per_locality,
 
 void normalizeWeights(const HostSet& host_set, bool in_panic,
                       NormalizedHostWeightVector& normalized_host_weights,
-                      double& min_normalized_weight, double& max_normalized_weight) {
-  if (host_set.localityWeights() == nullptr || host_set.localityWeights()->empty()) {
+                      double& min_normalized_weight, double& max_normalized_weight,
+                      bool locality_weighted_balancing) {
+  if (!locality_weighted_balancing || host_set.localityWeights() == nullptr ||
+      host_set.localityWeights()->empty()) {
     // If we're not dealing with locality weights, just normalize weights for the flat set of hosts.
     const auto& hosts = in_panic ? host_set.hosts() : host_set.healthyHosts();
     normalizeHostWeights(hosts, 1.0, normalized_host_weights, min_normalized_weight,
@@ -121,7 +123,7 @@ void ThreadAwareLoadBalancerBase::refresh() {
     double min_normalized_weight = 1.0;
     double max_normalized_weight = 0.0;
     normalizeWeights(*host_set, per_priority_state->global_panic_, normalized_host_weights,
-                     min_normalized_weight, max_normalized_weight);
+                     min_normalized_weight, max_normalized_weight, locality_weighted_balancing_);
     per_priority_state->current_lb_ = createLoadBalancer(
         std::move(normalized_host_weights), min_normalized_weight, max_normalized_weight);
   }

source/common/upstream/thread_aware_lb_impl.h

@@ -106,14 +106,12 @@ class ThreadAwareLoadBalancerBase : public LoadBalancerBase, public ThreadAwareL
   }
 
 protected:
-  ThreadAwareLoadBalancerBase(
-      const PrioritySet& priority_set, ClusterLbStats& stats, Runtime::Loader& runtime,
-      Random::RandomGenerator& random,
-      const envoy::config::cluster::v3::Cluster::CommonLbConfig& common_config)
-      : LoadBalancerBase(priority_set, stats, runtime, random,
-                         PROTOBUF_PERCENT_TO_ROUNDED_INTEGER_OR_DEFAULT(
-                             common_config, healthy_panic_threshold, 100, 50)),
-        factory_(new LoadBalancerFactoryImpl(stats, random)) {}
+  ThreadAwareLoadBalancerBase(const PrioritySet& priority_set, ClusterLbStats& stats,
+                              Runtime::Loader& runtime, Random::RandomGenerator& random,
+                              uint32_t healthy_panic_threshold, bool locality_weighted_balancing)
+      : LoadBalancerBase(priority_set, stats, runtime, random, healthy_panic_threshold),
+        factory_(new LoadBalancerFactoryImpl(stats, random)),
+        locality_weighted_balancing_(locality_weighted_balancing) {}
 
 private:
   struct PerPriorityState {
@@ -171,6 +169,7 @@ class ThreadAwareLoadBalancerBase : public LoadBalancerBase, public ThreadAwareL
   void refresh();
 
   std::shared_ptr<LoadBalancerFactoryImpl> factory_;
+  const bool locality_weighted_balancing_{};
   Common::CallbackHandlePtr priority_update_cb_;
 };
 

source/extensions/extensions_build_config.bzl

@@ -398,7 +398,8 @@ EXTENSIONS = {
     "envoy.load_balancing_policies.least_request":     "//source/extensions/load_balancing_policies/least_request:config",
     "envoy.load_balancing_policies.random":            "//source/extensions/load_balancing_policies/random:config",
     "envoy.load_balancing_policies.round_robin":       "//source/extensions/load_balancing_policies/round_robin:config",
-
+    "envoy.load_balancing_policies.maglev":            "//source/extensions/load_balancing_policies/maglev:config",
+    "envoy.load_balancing_policies.ring_hash":       "//source/extensions/load_balancing_policies/ring_hash:config",
 
     # HTTP Early Header Mutation
     #

source/extensions/extensions_metadata.yaml

@@ -1369,6 +1369,20 @@ envoy.load_balancing_policies.round_robin:
   status: stable
   type_urls:
   - envoy.extensions.load_balancing_policies.round_robin.v3.RoundRobin
+envoy.load_balancing_policies.ring_hash:
+  categories:
+  - envoy.load_balancing_policies
+  security_posture: robust_to_untrusted_downstream_and_upstream
+  status: stable
+  type_urls:
+  - envoy.extensions.load_balancing_policies.ring_hash.v3.RingHash
+envoy.load_balancing_policies.maglev:
+  categories:
+  - envoy.load_balancing_policies
+  security_posture: robust_to_untrusted_downstream_and_upstream
+  status: stable
+  type_urls:
+  - envoy.extensions.load_balancing_policies.maglev.v3.Maglev
 envoy.http.early_header_mutation.header_mutation:
   categories:
   - envoy.http.early_header_mutation

source/extensions/load_balancing_policies/maglev/BUILD

@@ -0,0 +1,22 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/upstream:load_balancer_factory_base_lib",
+        "//source/common/upstream:load_balancer_lib",
+        "//source/common/upstream:maglev_lb_lib",
+        "@envoy_api//envoy/extensions/load_balancing_policies/maglev/v3:pkg_cc_proto",
+    ],
+)

source/extensions/load_balancing_policies/maglev/config.cc

@@ -0,0 +1,41 @@
+#include "source/extensions/load_balancing_policies/maglev/config.h"
+
+#include "envoy/extensions/load_balancing_policies/maglev/v3/maglev.pb.h"
+
+#include "source/common/upstream/maglev_lb.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace LoadBalancingPolices {
+namespace Maglev {
+
+Upstream::ThreadAwareLoadBalancerPtr Factory::create(const Upstream::ClusterInfo& cluster_info,
+                                                     const Upstream::PrioritySet& priority_set,
+                                                     Runtime::Loader& runtime,
+                                                     Random::RandomGenerator& random, TimeSource&) {
+
+  const auto* typed_config =
+      dynamic_cast<const envoy::extensions::load_balancing_policies::maglev::v3::Maglev*>(
+          cluster_info.loadBalancingPolicy().get());
+
+  // The load balancing policy configuration will be loaded and validated in the main thread when we
+  // load the cluster configuration. So we can assume the configuration is valid here.
+  ASSERT(typed_config != nullptr,
+         "Invalid load balancing policy configuration for maglev load balancer");
+
+  return std::make_unique<Upstream::MaglevLoadBalancer>(
+      priority_set, cluster_info.lbStats(), cluster_info.statsScope(), runtime, random,
+      static_cast<uint32_t>(PROTOBUF_PERCENT_TO_ROUNDED_INTEGER_OR_DEFAULT(
+          cluster_info.lbConfig(), healthy_panic_threshold, 100, 50)),
+      *typed_config);
+}
+
+/**
+ * Static registration for the Factory. @see RegisterFactory.
+ */
+REGISTER_FACTORY(Factory, Upstream::TypedLoadBalancerFactory);
+
+} // namespace Maglev
+} // namespace LoadBalancingPolices
+} // namespace Extensions
+} // namespace Envoy