Add redis cluster load balancing

CODEOWNERS

@@ -32,3 +32,5 @@
 /*/extensions/quic_listeners/ @alyssawilk @danzh2010 @mattklein123 @mpwarres @wu-bin
 # zookeeper_proxy extension
 /*/extensions/filters/network/zookeeper_proxy @rgs1 @snowp
+# redis cluster extension
+/*/extensions/clusters/redis @msukalski @henryyyang @mattklein123

docs/root/intro/version_history.rst

@@ -32,6 +32,7 @@ Version history
 * original_src filter: added the :ref:`filter<config_http_filters_original_src>`.
 * rbac: migrated from v2alpha to v2.
 * redis: add support for Redis cluster custom cluster type.
+* redis: automatically route commands using cluster slots for Redis cluster.
 * redis: added :ref:`prefix routing <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.prefix_routes>` to enable routing commands based on their key's prefix to different upstream.
 * redis: add support for zpopmax and zpopmin commands.
 * redis: added
@@ -66,6 +67,8 @@ Version history
   that allows ignoring new hosts for the purpose of load balancing calculations until they have
   been health checked for the first time.
 * upstream: added runtime error checking to prevent setting dns type to STRICT_DNS or LOGICAL_DNS when custom resolver name is specified.
+* grpc-json: added support for :ref:`auto mapping
+  <envoy_api_field_config.filter.http.transcoder.v2.GrpcJsonTranscoder.auto_mapping>`.
 
 1.10.0 (Apr 5, 2019)
 ====================

include/envoy/upstream/upstream.h

@@ -724,6 +724,13 @@ class ClusterInfo {
    */
   virtual envoy::api::v2::Cluster::DiscoveryType type() const PURE;
 
+  /**
+   * @return the type of cluster, this includes custom discovery type in addition to the build-in
+   * discovery types.
+   */
+  virtual const absl::optional<envoy::api::v2::Cluster::CustomClusterType>&
+  clusterType() const PURE;
+
   /**
    * @return configuration for least request load balancing, only used if LB type is least request.
    */

source/common/upstream/upstream_impl.cc

@@ -574,7 +574,8 @@ ClusterInfoImpl::ClusterInfoImpl(const envoy::api::v2::Cluster& config,
       cluster_socket_options_(parseClusterSocketOptions(config, bind_config)),
       drain_connections_on_host_removal_(config.drain_connections_on_host_removal()),
       warm_hosts_(!config.health_checks().empty() &&
-                  common_lb_config_.ignore_new_hosts_until_first_hc()) {
+                  common_lb_config_.ignore_new_hosts_until_first_hc()),
+      cluster_type_(config.cluster_type()) {
   switch (config.lb_policy()) {
   case envoy::api::v2::Cluster::ROUND_ROBIN:
     lb_type_ = LoadBalancerType::RoundRobin;

source/common/upstream/upstream_impl.h

@@ -537,6 +537,9 @@ class ClusterInfoImpl : public ClusterInfo {
   extensionProtocolOptions(const std::string& name) const override;
   LoadBalancerType lbType() const override { return lb_type_; }
   envoy::api::v2::Cluster::DiscoveryType type() const override { return type_; }
+  const absl::optional<envoy::api::v2::Cluster::CustomClusterType>& clusterType() const override {
+    return cluster_type_;
+  }
   const absl::optional<envoy::api::v2::Cluster::LeastRequestLbConfig>&
   lbLeastRequestConfig() const override {
     return lb_least_request_config_;
@@ -619,6 +622,7 @@ class ClusterInfoImpl : public ClusterInfo {
   const bool drain_connections_on_host_removal_;
   const bool warm_hosts_;
   absl::optional<std::string> eds_service_name_;
+  const absl::optional<envoy::api::v2::Cluster::CustomClusterType> cluster_type_;
 };
 
 /**

source/extensions/clusters/redis/BUILD

@@ -8,13 +8,38 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "crc16_lib",
+    srcs = [
+        "crc16.cc",
+        "crc16.h",
+    ],
+)
+
+envoy_cc_library(
+    name = "redis_cluster_lb",
+    srcs = [
+        "redis_cluster_lb.cc",
+        "redis_cluster_lb.h",
+    ],
+    deps = [
+        ":crc16_lib",
+        "//include/envoy/upstream:thread_local_cluster_interface",
+        "//include/envoy/upstream:upstream_interface",
+        "//source/common/upstream:upstream_includes",
+        "//source/common/upstream:upstream_lib",
+        "//source/extensions/clusters:well_known_names",
+    ],
+)
+
 envoy_cc_library(
     name = "redis_cluster",
     srcs = [
         "redis_cluster.cc",
         "redis_cluster.h",
     ],
     deps = [
+        "redis_cluster_lb",
         "//include/envoy/api:api_interface",
         "//include/envoy/http:codec_interface",
         "//include/envoy/upstream:cluster_factory_interface",

source/extensions/clusters/redis/crc16.cc

@@ -0,0 +1,55 @@
+#include "crc16.h"
+
+#include "absl/strings/string_view.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Clusters {
+namespace Redis {
+
+/**
+ * XMODEM CRC16 implementation according to CITT standards.
+ * Based on (F).
+ * @param key The string to hash.
+ * @return The CRC16 hash code.
+ */
+static const uint16_t crc16tab[256] = {
+    0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7, 0x8108, 0x9129, 0xa14a, 0xb16b,
+    0xc18c, 0xd1ad, 0xe1ce, 0xf1ef, 0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6,
+    0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de, 0x2462, 0x3443, 0x0420, 0x1401,
+    0x64e6, 0x74c7, 0x44a4, 0x5485, 0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d,
+    0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4, 0xb75b, 0xa77a, 0x9719, 0x8738,
+    0xf7df, 0xe7fe, 0xd79d, 0xc7bc, 0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823,
+    0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b, 0x5af5, 0x4ad4, 0x7ab7, 0x6a96,
+    0x1a71, 0x0a50, 0x3a33, 0x2a12, 0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a,
+    0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41, 0xedae, 0xfd8f, 0xcdec, 0xddcd,
+    0xad2a, 0xbd0b, 0x8d68, 0x9d49, 0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70,
+    0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78, 0x9188, 0x81a9, 0xb1ca, 0xa1eb,
+    0xd10c, 0xc12d, 0xf14e, 0xe16f, 0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067,
+    0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e, 0x02b1, 0x1290, 0x22f3, 0x32d2,
+    0x4235, 0x5214, 0x6277, 0x7256, 0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d,
+    0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, 0xa7db, 0xb7fa, 0x8799, 0x97b8,
+    0xe75f, 0xf77e, 0xc71d, 0xd73c, 0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634,
+    0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab, 0x5844, 0x4865, 0x7806, 0x6827,
+    0x18c0, 0x08e1, 0x3882, 0x28a3, 0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a,
+    0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92, 0xfd2e, 0xed0f, 0xdd6c, 0xcd4d,
+    0xbdaa, 0xad8b, 0x9de8, 0x8dc9, 0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1,
+    0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8, 0x6e17, 0x7e36, 0x4e55, 0x5e74,
+    0x2e93, 0x3eb2, 0x0ed1, 0x1ef0};
+
+// (https://github.com/antirez/redis/blob/unstable/src/crc16.c)
+uint16_t Crc16::crc16(absl::string_view key) {
+  const char* buf = static_cast<const char*>(key.data());
+  uint64_t len = key.size();
+  uint64_t counter;
+  uint16_t crc = 0;
+  for (counter = 0; counter < len; counter++) {
+    crc = (crc << 8) ^ crc16tab[((crc >> 8) ^ *buf++) & 0x00FF];
+  }
+  return crc;
+}
+
+} // namespace Redis
+} // namespace Clusters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/clusters/redis/crc16.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <string>
+
+#include "absl/strings/ascii.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Clusters {
+namespace Redis {
+
+class Crc16 {
+public:
+  /**
+   * XMODEM CRC16 implementation according to CITT standards.
+   * Based on (https://github.com/antirez/redis/blob/unstable/src/crc16.c).
+   * @param key The string to hash.
+   * @return The CRC16 hash code.
+   */
+  static uint16_t crc16(absl::string_view key);
+};
+} // namespace Redis
+} // namespace Clusters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/clusters/redis/redis_cluster.cc

@@ -18,7 +18,8 @@ RedisCluster::RedisCluster(
     Upstream::ClusterManager& clusterManager, Runtime::Loader& runtime, Api::Api& api,
     Network::DnsResolverSharedPtr dns_resolver,
     Server::Configuration::TransportSocketFactoryContext& factory_context,
-    Stats::ScopePtr&& stats_scope, bool added_via_api)
+    Stats::ScopePtr&& stats_scope, bool added_via_api,
+    ClusterSlotUpdateCallBackSharedPtr lb_factory)
     : Upstream::BaseDynamicClusterImpl(cluster, runtime, factory_context, std::move(stats_scope),
                                        added_via_api),
       cluster_manager_(clusterManager),
@@ -32,7 +33,8 @@ RedisCluster::RedisCluster(
                            ? cluster.load_assignment()
                            : Config::Utility::translateClusterHosts(cluster.hosts())),
       local_info_(factory_context.localInfo()), random_(factory_context.random()),
-      redis_discovery_session_(*this, redis_client_factory), api_(api) {
+      redis_discovery_session_(*this, redis_client_factory), lb_factory_(std::move(lb_factory)),
+      api_(api) {
   const auto& locality_lb_endpoints = load_assignment_.endpoints();
   for (const auto& locality_lb_endpoint : locality_lb_endpoints) {
     for (const auto& lb_endpoint : locality_lb_endpoint.lb_endpoints()) {
@@ -77,17 +79,22 @@ void RedisCluster::updateAllHosts(const Upstream::HostVector& hosts_added,
 
 void RedisCluster::onClusterSlotUpdate(const std::vector<ClusterSlot>& slots) {
   Upstream::HostVector new_hosts;
-  SlotArray slots_;
 
   for (const ClusterSlot& slot : slots) {
-    new_hosts.emplace_back(new RedisHost(info(), "", slot.master_, *this, true));
+    new_hosts.emplace_back(new RedisHost(info(), "", slot.master(), *this, true));
   }
 
   std::unordered_map<std::string, Upstream::HostSharedPtr> updated_hosts;
   Upstream::HostVector hosts_added;
   Upstream::HostVector hosts_removed;
-  if (updateDynamicHostList(new_hosts, hosts_, hosts_added, hosts_removed, updated_hosts,
-                            all_hosts_)) {
+  const bool host_updated = updateDynamicHostList(new_hosts, hosts_, hosts_added, hosts_removed,
+                                                  updated_hosts, all_hosts_);
+  const bool slot_updated =
+      lb_factory_ ? lb_factory_->onClusterSlotUpdate(slots, updated_hosts) : false;
+
+  // If slot is updated, call updateAllHosts regardless of if there's new hosts to force
+  // update of the thread local load balancers.
+  if (host_updated || slot_updated) {
     ASSERT(std::all_of(hosts_.begin(), hosts_.end(), [&](const auto& host) {
       return host->priority() == localityLbEndpoint().priority();
     }));
@@ -96,16 +103,7 @@ void RedisCluster::onClusterSlotUpdate(const std::vector<ClusterSlot>& slots) {
     info_->stats().update_no_rebuild_.inc();
   }
 
-  for (const ClusterSlot& slot : slots) {
-    auto host = updated_hosts.find(slot.master_->asString());
-    ASSERT(host != updated_hosts.end(), "we expect all address to be found in the updated_hosts");
-    for (auto i = slot.start_; i <= slot.end_; ++i) {
-      slots_[i] = host->second;
-    }
-  }
-
   all_hosts_ = std::move(updated_hosts);
-  cluster_slots_map_.swap(slots_);
 
   // TODO(hyang): If there is an initialize callback, fire it now. Note that if the
   // cluster refers to multiple DNS names, this will return initialized after a single
@@ -317,17 +315,27 @@ RedisClusterFactory::createClusterWithConfig(
     Envoy::Stats::ScopePtr&& stats_scope) {
   if (!cluster.has_cluster_type() ||
       cluster.cluster_type().name() != Extensions::Clusters::ClusterTypes::get().Redis) {
-    throw EnvoyException("Redis cluster can only created with redis cluster type");
+    throw EnvoyException("Redis cluster can only created with redis cluster type.");
+  }
+  // TODO(hyang): This is needed to migrate existing cluster, disallow using other lb_policy
+  // in the future
+  if (cluster.lb_policy() != envoy::api::v2::Cluster::CLUSTER_PROVIDED) {
+    return std::make_pair(std::make_shared<RedisCluster>(
+                              cluster, proto_config,
+                              NetworkFilters::Common::Redis::Client::ClientFactoryImpl::instance_,
+                              context.clusterManager(), context.runtime(), context.api(),
+                              selectDnsResolver(cluster, context), socket_factory_context,
+                              std::move(stats_scope), context.addedViaApi(), nullptr),
+                          nullptr);
   }
-  // TODO(Henry): Implement a thread aware load balancer for Redis Cluster. This can come from
-  //              inside the created cluster.
+  auto lb_factory = std::make_shared<RedisClusterLoadBalancerFactory>();
   return std::make_pair(std::make_shared<RedisCluster>(
                             cluster, proto_config,
                             NetworkFilters::Common::Redis::Client::ClientFactoryImpl::instance_,
                             context.clusterManager(), context.runtime(), context.api(),
                             selectDnsResolver(cluster, context), socket_factory_context,
-                            std::move(stats_scope), context.addedViaApi()),
-                        nullptr);
+                            std::move(stats_scope), context.addedViaApi(), lb_factory),
+                        std::make_unique<RedisClusterThreadAwareLoadBalancer>(lb_factory));
 }
 
 REGISTER_FACTORY(RedisClusterFactory, Upstream::ClusterFactory);

source/extensions/clusters/redis/redis_cluster.h

@@ -51,6 +51,8 @@
 #include "common/upstream/resource_manager_impl.h"
 #include "common/upstream/upstream_impl.h"
 
+#include "source/extensions/clusters/redis/redis_cluster_lb.h"
+
 #include "server/transport_socket_config_impl.h"
 
 #include "extensions/clusters/well_known_names.h"
@@ -80,16 +82,11 @@ namespace Redis {
  * the `CLUSTER SLOTS command <https://redis.io/commands/cluster-slots>`_, and the responses and
  * failure cases.
  *
- * The topology is stored in cluster_slots_map_. According to the
- * `Redis Cluster Spec <https://redis.io/topics/cluster-spec#keys-distribution-model`_, the key
- * space is split into a fixed size 16384 slots. The current implementation uses a fixed size
- * std::array() of shared_ptr pointing to the master host. This has a fixed cpu and memory cost and
- * provide a fast lookup constant time lookup similar to Maglev. This will be used by the redis
- * proxy filter for load balancing purpose.
+ * Once the topology is fetched from Redis, the cluster will update the
+ * RedisClusterLoadBalancerFactory, which will be used by the redis proxy filter for load balancing
+ * purpose.
  */
 
-typedef std::array<Upstream::HostSharedPtr, 16384> SlotArray;
-
 class RedisCluster : public Upstream::BaseDynamicClusterImpl {
 public:
   RedisCluster(const envoy::api::v2::Cluster& cluster,
@@ -98,7 +95,8 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
                Upstream::ClusterManager& clusterManager, Runtime::Loader& runtime, Api::Api& api,
                Network::DnsResolverSharedPtr dns_resolver,
                Server::Configuration::TransportSocketFactoryContext& factory_context,
-               Stats::ScopePtr&& stats_scope, bool added_via_api);
+               Stats::ScopePtr&& stats_scope, bool added_via_api,
+               ClusterSlotUpdateCallBackSharedPtr factory);
 
   struct ClusterSlotsRequest : public Extensions::NetworkFilters::Common::Redis::RespValue {
   public:
@@ -124,15 +122,6 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
   void updateAllHosts(const Upstream::HostVector& hosts_added,
                       const Upstream::HostVector& hosts_removed, uint32_t priority);
 
-  struct ClusterSlot {
-    ClusterSlot(int64_t start, int64_t end, Network::Address::InstanceConstSharedPtr master)
-        : start_(start), end_(end), master_(std::move(master)) {}
-
-    int64_t start_;
-    int64_t end_;
-    Network::Address::InstanceConstSharedPtr master_;
-  };
-
   void onClusterSlotUpdate(const std::vector<ClusterSlot>&);
 
   const envoy::api::v2::endpoint::LocalityLbEndpoints& localityLbEndpoint() const {
@@ -259,8 +248,7 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
   const LocalInfo::LocalInfo& local_info_;
   Runtime::RandomGenerator& random_;
   RedisDiscoverySession redis_discovery_session_;
-  // The slot to master node map.
-  SlotArray cluster_slots_map_;
+  const ClusterSlotUpdateCallBackSharedPtr lb_factory_;
 
   Upstream::HostVector hosts_;
   Upstream::HostMap all_hosts_;