Add ClientSideWeightedRoundRobin LB Policy. (#35905)

Commit Message: - Add `ClientSideWeightedRoundRobinLoadBalancer`. - Attach `ClientSideHostLbPolicyData` to each `Host`. - Add `OrcaLoadReportCallbacks` callback to each `LoadBalancerContext`. - Calculate weights in `Host::ClientSideHostLbPolicyData` based on OrcaLoadReport. - Periodically update `Host::weight` for load balancing using calculated weights. - Add `host` reference to `OrcaLoadReportCallbacks::onOrcaLoadReport` callback. Risk Level: low Testing: `bazel test //test/extensions/load_balancing_policies/client_side_weighted_round_robin/...` Docs Changes: n/a Release Notes: n/a Platform Specific Features: n/a #34777 --------- Signed-off-by: Misha Efimov <[email protected]>
envoyproxy · Sep 25, 2024 · 89d485d · 89d485d
1 parent 49a4bcc
commit 89d485d
Show file tree

Hide file tree

Showing 28 changed files with 1,552 additions and 46 deletions.
diff --git a/CODEOWNERS b/CODEOWNERS
@@ -329,6 +329,7 @@ extensions/filters/http/oauth2 @derekargueta @mattklein123
 /*/extensions/load_balancing_policies/maglev @wbpcode @nezdolik
 /*/extensions/load_balancing_policies/subset @wbpcode @zuercher @nezdolik
 /*/extensions/load_balancing_policies/cluster_provided @wbpcode @zuercher
+/*/extensions/load_balancing_policies/client_side_weighted_round_robin @wbpcode @adisuissa @efimki
 # Early header mutation
 /*/extensions/http/early_header_mutation/header_mutation @wbpcode @tyxia
 # Network matching extensions

diff --git a/api/BUILD b/api/BUILD
@@ -276,6 +276,7 @@ proto_library(
         "//envoy/extensions/internal_redirect/previous_routes/v3:pkg",
         "//envoy/extensions/internal_redirect/safe_cross_scheme/v3:pkg",
         "//envoy/extensions/key_value/file_based/v3:pkg",
+        "//envoy/extensions/load_balancing_policies/client_side_weighted_round_robin/v3:pkg",
         "//envoy/extensions/load_balancing_policies/cluster_provided/v3:pkg",
         "//envoy/extensions/load_balancing_policies/common/v3:pkg",
         "//envoy/extensions/load_balancing_policies/least_request/v3:pkg",

diff --git a/...ncing_policies/client_side_weighted_round_robin/v3/client_side_weighted_round_robin.proto b/...ncing_policies/client_side_weighted_round_robin/v3/client_side_weighted_round_robin.proto
@@ -15,7 +15,7 @@ option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/loa
 option (udpa.annotations.file_status).package_version_status = ACTIVE;
 
 // [#protodoc-title: Client-Side Weighted Round Robin Load Balancing Policy]
-// [#not-implemented-hide:]
+// [#extension: envoy.load_balancing_policies.client_side_weighted_round_robin]
 
 // Configuration for the client_side_weighted_round_robin LB policy.
 //

diff --git a/changelogs/current.yaml b/changelogs/current.yaml
@@ -429,5 +429,10 @@ new_features:
     Added allow list
     :ref:`allowed_override_modes <envoy_v3_api_field_extensions.filters.http.ext_proc.v3.ExternalProcessor.allowed_override_modes>`
     for :ref:`mode_override <envoy_v3_api_field_service.ext_proc.v3.ProcessingResponse.mode_override>`.
+- area: load_balancing
+  change: |
+    WIP: Added implementation of :ref:`client_side_weighted_round_robin
+    <envoy_v3_api_msg_extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin>`
+    load balancing policy that uses ``OrcaLoadReport`` provided by the upstream host to calculate host load balancing weight.
 
 deprecated:
diff --git a/envoy/upstream/load_balancer.h b/envoy/upstream/load_balancer.h
@@ -110,16 +110,24 @@ class LoadBalancerContext {
   class OrcaLoadReportCallbacks {
   public:
     virtual ~OrcaLoadReportCallbacks() = default;
-    // Invoked when a new orca report is received for this LB context.
+    /**
+     * Invoked when a new orca report is received for this LB context.
+     * @param orca_load_report supplies the ORCA load report.
+     * @param host supplies the upstream host, which provided the load report.
+     * @return absl::Status the result of ORCA load report processing by the load balancer.
+     */
     virtual absl::Status
-    onOrcaLoadReport(const xds::data::orca::v3::OrcaLoadReport& orca_load_report) PURE;
+    onOrcaLoadReport(const xds::data::orca::v3::OrcaLoadReport& orca_load_report,
+                     const HostDescription& host) PURE;
   };
 
   /**
    * Install a callback to be invoked when ORCA Load report is received for this
    * LB context.
+   * Note: LB Context keeps a weak pointer to `callbacks` and doesn't invoke the callback
+   * if it is `expired()`.
    */
-  virtual void setOrcaLoadReportCallbacks(OrcaLoadReportCallbacks& callbacks) PURE;
+  virtual void setOrcaLoadReportCallbacks(std::weak_ptr<OrcaLoadReportCallbacks> callbacks) PURE;
 };
 
 /**

diff --git a/source/common/orca/orca_load_metrics.cc b/source/common/orca/orca_load_metrics.cc
@@ -21,54 +21,77 @@ static constexpr absl::string_view kRpsFractionalField = "rps_fractional";
 static constexpr absl::string_view kNamedMetricsFieldPrefix = "named_metrics.";
 static constexpr absl::string_view kRequestCostFieldPrefix = "request_cost.";
 static constexpr absl::string_view kUtilizationFieldPrefix = "utilization.";
-} // namespace
 
-void addOrcaNamedMetricToLoadMetricStats(const Protobuf::Map<std::string, double>& metrics_map,
-                                         const absl::string_view metric_name,
-                                         const absl::string_view metric_name_prefix,
-                                         Upstream::LoadMetricStats& stats) {
-  absl::string_view metric_name_without_prefix = absl::StripPrefix(metric_name, metric_name_prefix);
-  // If the metric name is "*", add all metrics from the map.
+using OnLoadReportMetricFn =
+    std::function<void(absl::string_view metric_name, double metric_value)>;
+
+void scanOrcaLoadReportMetricsMap(const Protobuf::Map<std::string, double>& metrics_map,
+                                  absl::string_view metric_name,
+                                  absl::string_view metric_name_prefix,
+                                  OnLoadReportMetricFn on_load_report_metric) {
+  absl::string_view metric_name_without_prefix = metric_name.substr(metric_name_prefix.size());
+  // If the metric name is "*", report all metrics from the map.
   if (metric_name_without_prefix == "*") {
     for (const auto& [key, value] : metrics_map) {
-      stats.add(absl::StrCat(metric_name_prefix, key), value);
+      on_load_report_metric(absl::StrCat(metric_name_prefix, key), value);
     }
   } else {
-    // Add the metric if it exists in the map.
+    // Report the metric if it exists in the map.
     const auto metric_it = metrics_map.find(metric_name_without_prefix);
     if (metric_it != metrics_map.end()) {
-      stats.add(metric_name, metric_it->second);
+      on_load_report_metric(metric_name, metric_it->second);
     }
   }
 }
 
-void addOrcaLoadReportToLoadMetricStats(const LrsReportMetricNames& metric_names,
-                                        const xds::data::orca::v3::OrcaLoadReport& report,
-                                        Upstream::LoadMetricStats& stats) {
+void scanOrcaLoadReport(const LrsReportMetricNames& metric_names,
+                        const xds::data::orca::v3::OrcaLoadReport& report,
+                        OnLoadReportMetricFn on_load_report_metric) {
   // TODO(efimki): Use InlineMap to speed up this loop.
   for (const std::string& metric_name : metric_names) {
     if (metric_name == kCpuUtilizationField) {
-      stats.add(metric_name, report.cpu_utilization());
+      on_load_report_metric(metric_name, report.cpu_utilization());
     } else if (metric_name == kMemUtilizationField) {
-      stats.add(metric_name, report.mem_utilization());
+      on_load_report_metric(metric_name, report.mem_utilization());
     } else if (metric_name == kApplicationUtilizationField) {
-      stats.add(metric_name, report.application_utilization());
+      on_load_report_metric(metric_name, report.application_utilization());
     } else if (metric_name == kEpsField) {
-      stats.add(metric_name, report.eps());
+      on_load_report_metric(metric_name, report.eps());
     } else if (metric_name == kRpsFractionalField) {
-      stats.add(metric_name, report.rps_fractional());
+      on_load_report_metric(metric_name, report.rps_fractional());
     } else if (absl::StartsWith(metric_name, kNamedMetricsFieldPrefix)) {
-      addOrcaNamedMetricToLoadMetricStats(report.named_metrics(), metric_name,
-                                          kNamedMetricsFieldPrefix, stats);
+      scanOrcaLoadReportMetricsMap(report.named_metrics(), metric_name, kNamedMetricsFieldPrefix,
+                                   on_load_report_metric);
     } else if (absl::StartsWith(metric_name, kUtilizationFieldPrefix)) {
-      addOrcaNamedMetricToLoadMetricStats(report.utilization(), metric_name,
-                                          kUtilizationFieldPrefix, stats);
+      scanOrcaLoadReportMetricsMap(report.utilization(), metric_name, kUtilizationFieldPrefix,
+                                   on_load_report_metric);
     } else if (absl::StartsWith(metric_name, kRequestCostFieldPrefix)) {
-      addOrcaNamedMetricToLoadMetricStats(report.request_cost(), metric_name,
-                                          kRequestCostFieldPrefix, stats);
+      scanOrcaLoadReportMetricsMap(report.request_cost(), metric_name, kRequestCostFieldPrefix,
+                                   on_load_report_metric);
     }
   }
 }
 
+} // namespace
+
+void addOrcaLoadReportToLoadMetricStats(const LrsReportMetricNames& metric_names,
+                                        const xds::data::orca::v3::OrcaLoadReport& report,
+                                        Upstream::LoadMetricStats& stats) {
+  scanOrcaLoadReport(metric_names, report,
+                     [&stats](absl::string_view metric_name, double metric_value) {
+                       stats.add(metric_name, metric_value);
+                     });
+}
+
+double getMaxUtilization(const LrsReportMetricNames& metric_names,
+                         const xds::data::orca::v3::OrcaLoadReport& report) {
+  double max_utilization = 0;
+  scanOrcaLoadReport(metric_names, report,
+                     [&max_utilization](absl::string_view, double metric_value) {
+                       max_utilization = std::max<double>(max_utilization, metric_value);
+                     });
+  return max_utilization;
+}
+
 } // namespace Orca
 } // namespace Envoy
diff --git a/source/common/orca/orca_load_metrics.h b/source/common/orca/orca_load_metrics.h
@@ -10,9 +10,14 @@ namespace Orca {
 // List of metric names to report to the LRS.
 typedef std::vector<std::string> LrsReportMetricNames;
 
+// Adds metrics with `metric_names` from the `report` to the `stats`.
 void addOrcaLoadReportToLoadMetricStats(const LrsReportMetricNames& metric_names,
                                         const xds::data::orca::v3::OrcaLoadReport& report,
                                         Upstream::LoadMetricStats& stats);
 
+// Returns the maximum value of metrics with `metric_names` in the `report`.
+double getMaxUtilization(const LrsReportMetricNames& metric_names,
+                         const xds::data::orca::v3::OrcaLoadReport& report);
+
 } // namespace Orca
 } // namespace Envoy
diff --git a/source/common/router/router.cc b/source/common/router/router.cc
@@ -2123,7 +2123,7 @@ void Filter::maybeProcessOrcaLoadReport(const Envoy::Http::HeaderMap& headers_or
   auto host = upstream_request.upstreamHost();
   const bool need_to_send_load_report =
       (host != nullptr) && cluster_->lrsReportMetricNames().has_value();
-  if (!need_to_send_load_report && !orca_load_report_callbacks_.has_value()) {
+  if (!need_to_send_load_report && orca_load_report_callbacks_.expired()) {
     return;
   }
 
@@ -2144,8 +2144,8 @@ void Filter::maybeProcessOrcaLoadReport(const Envoy::Http::HeaderMap& headers_or
                                                     orca_load_report.value(),
                                                     host->loadMetricStats());
   }
-  if (orca_load_report_callbacks_.has_value()) {
-    const absl::Status status = orca_load_report_callbacks_->onOrcaLoadReport(*orca_load_report);
+  if (auto callbacks = orca_load_report_callbacks_.lock(); callbacks != nullptr) {
+    const absl::Status status = callbacks->onOrcaLoadReport(*orca_load_report, *host);
     if (!status.ok()) {
       ENVOY_STREAM_LOG(error, "Failed to invoke OrcaLoadReportCallbacks: {}", *callbacks_,
                        status.message());

diff --git a/source/common/router/router.h b/source/common/router/router.h
@@ -424,7 +424,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
     return callbacks_->upstreamOverrideHost();
   }
 
-  void setOrcaLoadReportCallbacks(OrcaLoadReportCallbacks& callbacks) override {
+  void setOrcaLoadReportCallbacks(std::weak_ptr<OrcaLoadReportCallbacks> callbacks) override {
     orca_load_report_callbacks_ = callbacks;
   }
 
@@ -610,7 +610,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   Http::Code timeout_response_code_ = Http::Code::GatewayTimeout;
   FilterUtility::HedgingParams hedging_params_;
   Http::StreamFilterSidestreamWatermarkCallbacks watermark_callbacks_;
-  OptRef<OrcaLoadReportCallbacks> orca_load_report_callbacks_;
+  std::weak_ptr<OrcaLoadReportCallbacks> orca_load_report_callbacks_;
   bool grpc_request_ : 1;
   bool exclude_http_code_stats_ : 1;
   bool downstream_response_started_ : 1;

diff --git a/source/common/upstream/load_balancer_context_base.h b/source/common/upstream/load_balancer_context_base.h
@@ -35,7 +35,7 @@ class LoadBalancerContextBase : public LoadBalancerContext {
 
   absl::optional<OverrideHost> overrideHostToSelect() const override { return {}; }
 
-  void setOrcaLoadReportCallbacks(OrcaLoadReportCallbacks&) override {}
+  void setOrcaLoadReportCallbacks(std::weak_ptr<OrcaLoadReportCallbacks>) override {}
 };
 
 } // namespace Upstream

diff --git a/source/extensions/extensions_build_config.bzl b/source/extensions/extensions_build_config.bzl
@@ -496,6 +496,7 @@ EXTENSIONS = {
     "envoy.load_balancing_policies.ring_hash":         "//source/extensions/load_balancing_policies/ring_hash:config",
     "envoy.load_balancing_policies.subset":            "//source/extensions/load_balancing_policies/subset:config",
     "envoy.load_balancing_policies.cluster_provided":  "//source/extensions/load_balancing_policies/cluster_provided:config",
+    "envoy.load_balancing_policies.client_side_weighted_round_robin": "//source/extensions/load_balancing_policies/client_side_weighted_round_robin:config",
 
     #
     # HTTP Early Header Mutation

diff --git a/source/extensions/extensions_metadata.yaml b/source/extensions/extensions_metadata.yaml
@@ -1674,6 +1674,13 @@ envoy.load_balancing_policies.cluster_provided:
   status: stable
   type_urls:
   - envoy.extensions.load_balancing_policies.cluster_provided.v3.ClusterProvided
+envoy.load_balancing_policies.client_side_weighted_round_robin:
+  categories:
+  - envoy.load_balancing_policies
+  security_posture: unknown
+  status: wip
+  type_urls:
+  - envoy.extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin
 envoy.http.early_header_mutation.header_mutation:
   categories:
   - envoy.http.early_header_mutation

diff --git a/source/extensions/load_balancing_policies/client_side_weighted_round_robin/BUILD b/source/extensions/load_balancing_policies/client_side_weighted_round_robin/BUILD
@@ -0,0 +1,36 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":client_side_weighted_round_robin_lb_lib",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/upstream:load_balancer_context_base_lib",
+        "//source/extensions/load_balancing_policies/common:factory_base",
+        "@envoy_api//envoy/extensions/load_balancing_policies/client_side_weighted_round_robin/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_library(
+    name = "client_side_weighted_round_robin_lb_lib",
+    srcs = ["client_side_weighted_round_robin_lb.cc"],
+    hdrs = ["client_side_weighted_round_robin_lb.h"],
+    deps = [
+        "//source/common/orca:orca_load_metrics_lib",
+        "//source/extensions/load_balancing_policies/common:load_balancer_lib",
+        "//source/extensions/load_balancing_policies/round_robin:round_robin_lb_lib",
+        "@com_github_cncf_xds//xds/data/orca/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/load_balancing_policies/client_side_weighted_round_robin/v3:pkg_cc_proto",
+    ],
+)