router: add weighted priority cluster specifier plugin (POC)

api/envoy/extensions/router/cluster_specifiers/weighted_priority/v3/BUILD

@@ -0,0 +1,9 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = ["@xds//udpa/annotations:pkg"],
+)

api/envoy/extensions/router/cluster_specifiers/weighted_priority/v3/weighted_priority.proto

@@ -0,0 +1,73 @@
+syntax = "proto3";
+
+package envoy.extensions.router.cluster_specifiers.weighted_priority.v3;
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.router.cluster_specifiers.weighted_priority.v3";
+option java_outer_classname = "WeightedPriorityProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/router/cluster_specifiers/weighted_priority/v3;weighted_priorityv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Weighted priority cluster specifier]
+
+// Selects a cluster by walking ordered priority groups and performing a weighted choice among the
+// clusters of the first group that has any healthy cluster.
+//
+// This differs from :ref:`weighted_clusters
+// <envoy_v3_api_field_config.route.v3.RouteAction.weighted_clusters>` in one respect: a group whose
+// clusters all lack healthy hosts is skipped, so a degraded tier does not absorb its share of the
+// traffic. Weights therefore describe how to split load *within* a tier, while groups describe the
+// order in which tiers are tried.
+//
+// Example configuration:
+//
+// .. code-block:: yaml
+//
+//     priority_groups:
+//     - clusters:
+//       - name: primary_a
+//         weight: 70
+//       - name: primary_b
+//         weight: 30
+//     - clusters:
+//       - name: fallback
+//         weight: 1
+//
+// While either primary cluster has healthy hosts, requests are split 70/30 between them, and the
+// split renormalizes over whichever of the two remains healthy. Once both are unhealthy, requests
+// go to ``fallback``.
+//
+// The cluster is chosen once per request, when the route is resolved, so retries of a request stay
+// within the chosen cluster. Combine this with a cluster type such as :ref:`the composite cluster
+// <arch_overview_composite_cluster>` when retries need to move between clusters.
+//
+// [#extension: envoy.router.cluster_specifier_plugin.weighted_priority]
+message WeightedPriorityClusterSpecifier {
+  // A cluster and its share of the traffic within a priority group.
+  message ClusterWeight {
+    // Name of the cluster. This cluster must be defined elsewhere in the configuration.
+    string name = 1 [(validate.rules).string = {min_len: 1}];
+
+    // Share of the group's traffic that this cluster receives, relative to the other clusters of
+    // the same group. Weights are renormalized over whichever clusters of the group are healthy,
+    // so removing a cluster from consideration redistributes its share proportionally.
+    uint32 weight = 2 [(validate.rules).uint32 = {gt: 0}];
+  }
+
+  // A set of clusters that are considered together, and among which traffic is split by weight.
+  message PriorityGroup {
+    // Clusters belonging to this group. Must contain at least one cluster.
+    repeated ClusterWeight clusters = 1 [(validate.rules).repeated = {min_items: 1}];
+  }
+
+  // Groups in descending order of preference. The first group containing a cluster that both
+  // exists and has healthy hosts is used; earlier groups that have none are skipped.
+  //
+  // If no group has a healthy cluster, the weighted choice is made over the last group while
+  // ignoring health, so the request fails against a real cluster and is attributed to it in stats
+  // rather than being dropped as an unroutable request.
+  repeated PriorityGroup priority_groups = 1 [(validate.rules).repeated = {min_items: 1}];
+}

changelogs/current/new_features/router__weighted-priority-cluster-specifier.rst

@@ -0,0 +1,7 @@
+Added a :ref:`weighted priority cluster specifier
+<envoy_v3_api_msg_extensions.router.cluster_specifiers.weighted_priority.v3.WeightedPriorityClusterSpecifier>`,
+which selects a cluster by walking ordered priority groups and making a weighted choice among the
+clusters of the first group that has a healthy cluster. Unlike :ref:`weighted_clusters
+<envoy_v3_api_field_config.route.v3.RouteAction.weighted_clusters>`, a group whose clusters all lack
+healthy hosts is skipped rather than absorbing its share of the traffic, and weights are
+renormalized over whichever clusters remain healthy.

source/extensions/extensions_build_config.bzl

@@ -644,6 +644,7 @@ EXTENSIONS = {
     #
     "envoy.router.cluster_specifier_plugin.lua":     "//source/extensions/router/cluster_specifiers/lua:config",
     "envoy.router.cluster_specifier_plugin.matcher": "//source/extensions/router/cluster_specifiers/matcher:config",
+    "envoy.router.cluster_specifier_plugin.weighted_priority": "//source/extensions/router/cluster_specifiers/weighted_priority:config",
 
     #
     # Extensions for generic proxy

source/extensions/extensions_metadata.yaml

@@ -1528,6 +1528,13 @@ envoy.router.cluster_specifier_plugin.matcher:
   status: alpha
   type_urls:
   - envoy.extensions.router.cluster_specifiers.matcher.v3.MatcherClusterSpecifier
+envoy.router.cluster_specifier_plugin.weighted_priority:
+  categories:
+  - envoy.router.cluster_specifier_plugin
+  security_posture: robust_to_untrusted_downstream_and_upstream
+  status: alpha
+  type_urls:
+  - envoy.extensions.router.cluster_specifiers.weighted_priority.v3.WeightedPriorityClusterSpecifier
 envoy.stat_sinks.dog_statsd:
   categories:
   - envoy.stats_sinks

source/extensions/router/cluster_specifiers/weighted_priority/BUILD

@@ -0,0 +1,40 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+# Weighted priority cluster specifier plugin.
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "weighted_priority_cluster_specifier_lib",
+    srcs = [
+        "weighted_priority_cluster_specifier.cc",
+    ],
+    hdrs = [
+        "weighted_priority_cluster_specifier.h",
+    ],
+    deps = [
+        "//envoy/router:cluster_specifier_plugin_interface",
+        "//envoy/upstream:cluster_manager_interface",
+        "//source/common/common:logger_lib",
+        "//source/common/router:delegating_route_lib",
+        "@envoy_api//envoy/extensions/router/cluster_specifiers/weighted_priority/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":weighted_priority_cluster_specifier_lib",
+        "//envoy/registry",
+        "@envoy_api//envoy/extensions/router/cluster_specifiers/weighted_priority/v3:pkg_cc_proto",
+    ],
+)

source/extensions/router/cluster_specifiers/weighted_priority/config.cc

@@ -0,0 +1,27 @@
+#include "source/extensions/router/cluster_specifiers/weighted_priority/config.h"
+
+#include "envoy/extensions/router/cluster_specifiers/weighted_priority/v3/weighted_priority.pb.validate.h"
+#include "envoy/registry/registry.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Router {
+namespace WeightedPriority {
+
+Envoy::Router::ClusterSpecifierPluginSharedPtr
+WeightedPriorityClusterSpecifierPluginFactoryConfig::createClusterSpecifierPlugin(
+    const Protobuf::Message& config, Server::Configuration::ServerFactoryContext& context) {
+  const auto& typed_config =
+      MessageUtil::downcastAndValidate<const WeightedPriorityClusterSpecifierConfigProto&>(
+          config, context.messageValidationVisitor());
+  return std::make_shared<WeightedPriorityClusterSpecifierPlugin>(typed_config,
+                                                                  context.clusterManager());
+}
+
+REGISTER_FACTORY(WeightedPriorityClusterSpecifierPluginFactoryConfig,
+                 Envoy::Router::ClusterSpecifierPluginFactoryConfig);
+
+} // namespace WeightedPriority
+} // namespace Router
+} // namespace Extensions
+} // namespace Envoy

source/extensions/router/cluster_specifiers/weighted_priority/config.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include "source/extensions/router/cluster_specifiers/weighted_priority/weighted_priority_cluster_specifier.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Router {
+namespace WeightedPriority {
+
+class WeightedPriorityClusterSpecifierPluginFactoryConfig
+    : public Envoy::Router::ClusterSpecifierPluginFactoryConfig {
+public:
+  WeightedPriorityClusterSpecifierPluginFactoryConfig() = default;
+
+  Envoy::Router::ClusterSpecifierPluginSharedPtr
+  createClusterSpecifierPlugin(const Protobuf::Message& config,
+                               Server::Configuration::ServerFactoryContext& context) override;
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return std::make_unique<WeightedPriorityClusterSpecifierConfigProto>();
+  }
+
+  std::string name() const override {
+    return "envoy.router.cluster_specifier_plugin.weighted_priority";
+  }
+};
+
+} // namespace WeightedPriority
+} // namespace Router
+} // namespace Extensions
+} // namespace Envoy

source/extensions/router/cluster_specifiers/weighted_priority/weighted_priority_cluster_specifier.cc

@@ -0,0 +1,113 @@
+#include "source/extensions/router/cluster_specifiers/weighted_priority/weighted_priority_cluster_specifier.h"
+
+#include "source/common/router/delegating_route_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Router {
+namespace WeightedPriority {
+namespace {
+
+bool hasHealthyHosts(Upstream::ThreadLocalCluster& cluster) {
+  for (const auto& host_set : cluster.prioritySet().hostSetsPerPriority()) {
+    if (!host_set->healthyHosts().empty()) {
+      return true;
+    }
+  }
+  return false;
+}
+
+// Serves the parent route with the cluster replaced by the selected one.
+class WeightedPriorityRouteEntry : public Envoy::Router::DelegatingRouteEntry {
+public:
+  WeightedPriorityRouteEntry(Envoy::Router::RouteEntryAndRouteConstSharedPtr parent,
+                             absl::string_view cluster_name)
+      : DelegatingRouteEntry(std::move(parent)), cluster_name_(cluster_name) {}
+
+  const std::string& clusterName() const override { return cluster_name_; }
+
+private:
+  const std::string cluster_name_;
+};
+
+} // namespace
+
+WeightedPriorityClusterSpecifierPlugin::WeightedPriorityClusterSpecifierPlugin(
+    const WeightedPriorityClusterSpecifierConfigProto& config,
+    Upstream::ClusterManager& cluster_manager)
+    : cluster_manager_(cluster_manager) {
+  priority_groups_.reserve(config.priority_groups_size());
+  for (const auto& group : config.priority_groups()) {
+    PriorityGroup entries;
+    entries.reserve(group.clusters_size());
+    for (const auto& cluster : group.clusters()) {
+      entries.push_back({cluster.name(), cluster.weight()});
+    }
+    priority_groups_.push_back(std::move(entries));
+  }
+}
+
+absl::string_view
+WeightedPriorityClusterSpecifierPlugin::selectFromGroup(const PriorityGroup& group, uint64_t random,
+                                                        bool require_healthy) const {
+  // Sum the weights of the eligible clusters first, so the choice is renormalized over them rather
+  // than over the configured total. Otherwise an excluded cluster's share would fall through to
+  // whichever cluster happens to be last.
+  uint64_t total_weight = 0;
+  for (const auto& cluster : group) {
+    if (require_healthy) {
+      auto* thread_local_cluster = cluster_manager_.getThreadLocalCluster(cluster.name_);
+      if (thread_local_cluster == nullptr || !hasHealthyHosts(*thread_local_cluster)) {
+        continue;
+      }
+    }
+    total_weight += cluster.weight_;
+  }
+
+  if (total_weight == 0) {
+    return {};
+  }
+
+  uint64_t selector = random % total_weight;
+  for (const auto& cluster : group) {
+    if (require_healthy) {
+      auto* thread_local_cluster = cluster_manager_.getThreadLocalCluster(cluster.name_);
+      if (thread_local_cluster == nullptr || !hasHealthyHosts(*thread_local_cluster)) {
+        continue;
+      }
+    }
+    if (selector < cluster.weight_) {
+      return cluster.name_;
+    }
+    selector -= cluster.weight_;
+  }
+
+  // Unreachable: the loops above visit the same clusters and `selector` is below their total.
+  return {};
+}
+
+Envoy::Router::RouteConstSharedPtr WeightedPriorityClusterSpecifierPlugin::route(
+    Envoy::Router::RouteEntryAndRouteConstSharedPtr parent, const Http::RequestHeaderMap&,
+    const StreamInfo::StreamInfo&, uint64_t random) const {
+  for (size_t i = 0; i < priority_groups_.size(); ++i) {
+    const absl::string_view selected =
+        selectFromGroup(priority_groups_[i], random, /*require_healthy=*/true);
+    if (!selected.empty()) {
+      ENVOY_LOG(debug, "weighted priority: selected cluster '{}' from group {}", selected, i);
+      return std::make_shared<WeightedPriorityRouteEntry>(std::move(parent), selected);
+    }
+  }
+
+  // No group had a healthy cluster. Choose from the last group regardless of health so the request
+  // fails against a real cluster, and is counted against it, rather than being dropped as
+  // unroutable.
+  const absl::string_view fallback =
+      selectFromGroup(priority_groups_.back(), random, /*require_healthy=*/false);
+  ENVOY_LOG(debug, "weighted priority: no healthy cluster, falling back to '{}'", fallback);
+  return std::make_shared<WeightedPriorityRouteEntry>(std::move(parent), fallback);
+}
+
+} // namespace WeightedPriority
+} // namespace Router
+} // namespace Extensions
+} // namespace Envoy

source/extensions/router/cluster_specifiers/weighted_priority/weighted_priority_cluster_specifier.h

@@ -0,0 +1,54 @@
+#pragma once
+
+#include <string>
+#include <vector>
+
+#include "envoy/extensions/router/cluster_specifiers/weighted_priority/v3/weighted_priority.pb.h"
+#include "envoy/router/cluster_specifier_plugin.h"
+#include "envoy/upstream/cluster_manager.h"
+
+#include "source/common/common/logger.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Router {
+namespace WeightedPriority {
+
+using WeightedPriorityClusterSpecifierConfigProto = envoy::extensions::router::cluster_specifiers::
+    weighted_priority::v3::WeightedPriorityClusterSpecifier;
+
+// Selects a cluster by walking priority groups in order and making a weighted choice among the
+// clusters of the first group that has a healthy cluster.
+class WeightedPriorityClusterSpecifierPlugin : public Envoy::Router::ClusterSpecifierPlugin,
+                                               Logger::Loggable<Logger::Id::router> {
+public:
+  WeightedPriorityClusterSpecifierPlugin(const WeightedPriorityClusterSpecifierConfigProto& config,
+                                         Upstream::ClusterManager& cluster_manager);
+
+  // Envoy::Router::ClusterSpecifierPlugin
+  Envoy::Router::RouteConstSharedPtr route(Envoy::Router::RouteEntryAndRouteConstSharedPtr parent,
+                                           const Http::RequestHeaderMap& headers,
+                                           const StreamInfo::StreamInfo& stream_info,
+                                           uint64_t random) const override;
+
+private:
+  struct WeightedCluster {
+    std::string name_;
+    uint32_t weight_{};
+  };
+  using PriorityGroup = std::vector<WeightedCluster>;
+
+  // Weighted choice among the clusters of `group`. When `require_healthy` is set, clusters that
+  // are unknown to the cluster manager or have no healthy hosts are excluded and the weights are
+  // renormalized over the remainder. Returns an empty view when nothing is eligible.
+  absl::string_view selectFromGroup(const PriorityGroup& group, uint64_t random,
+                                    bool require_healthy) const;
+
+  std::vector<PriorityGroup> priority_groups_;
+  Upstream::ClusterManager& cluster_manager_;
+};
+
+} // namespace WeightedPriority
+} // namespace Router
+} // namespace Extensions
+} // namespace Envoy