envoyproxy · mattklein123 · Jun 6, 2019 · Feb 18, 2019 · Feb 20, 2019 · Feb 20, 2019
diff --git a/docs/root/intro/version_history.rst b/docs/root/intro/version_history.rst
@@ -17,6 +17,7 @@ Version history
 * http: mitigated a race condition with the :ref:`delayed_close_timeout<envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.delayed_close_timeout>` where it could trigger while actively flushing a pending write buffer for a downstream connection.
 * jwt_authn: make filter's parsing of JWT more flexible, allowing syntax like ``jwt=eyJhbGciOiJS...ZFnFIw,extra=7,realm=123``
 * 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

diff --git a/include/envoy/upstream/upstream.h b/include/envoy/upstream/upstream.h
@@ -731,6 +731,12 @@ 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 envoy::api::v2::Cluster::CustomClusterType& clusterType() const PURE;
+
   /**
    * @return configuration for least request load balancing, only used if LB type is least request.
    */

diff --git a/source/common/common/hash.cc b/source/common/common/hash.cc
@@ -34,4 +34,40 @@ uint64_t MurmurHash::murmurHash2_64(absl::string_view key, uint64_t seed) {
   return hash;
 }
 
+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 Envoy
diff --git a/source/common/common/hash.h b/source/common/common/hash.h
@@ -77,6 +77,17 @@ class MurmurHash {
   static inline uint64_t shift_mix(uint64_t v) { return v ^ (v >> 47); }
 };
 
+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);
+};
+
 struct ConstCharStarHash {
   size_t operator()(const char* a) const { return HashUtil::xxHash64(a); }
 };

diff --git a/source/common/upstream/upstream_impl.cc b/source/common/upstream/upstream_impl.cc
@@ -624,6 +624,30 @@ ClusterInfoImpl::ClusterInfoImpl(const envoy::api::v2::Cluster& config,
     eds_service_name_ = config.eds_cluster_config().service_name();
   }
 
+  if (!config.has_cluster_type()) {
+    switch (config.type()) {
+    case envoy::api::v2::Cluster::STATIC:
+      cluster_type_.set_name(Extensions::Clusters::ClusterTypes::get().Static);
+      break;
+    case envoy::api::v2::Cluster::STRICT_DNS:
+      cluster_type_.set_name(Extensions::Clusters::ClusterTypes::get().StrictDns);
+      break;
+    case envoy::api::v2::Cluster::LOGICAL_DNS:
+      cluster_type_.set_name(Extensions::Clusters::ClusterTypes::get().LogicalDns);
+      break;
+    case envoy::api::v2::Cluster::ORIGINAL_DST:
+      cluster_type_.set_name(Extensions::Clusters::ClusterTypes::get().OriginalDst);
+      break;
+    case envoy::api::v2::Cluster::EDS:
+      cluster_type_.set_name(Extensions::Clusters::ClusterTypes::get().Eds);
+      break;
+    default:
+      NOT_REACHED_GCOVR_EXCL_LINE;
+    }
+  } else {
+    cluster_type_.CopyFrom(config.cluster_type());
+  }
+
   // TODO(htuch): Remove this temporary workaround when we have
   // https://github.com/lyft/protoc-gen-validate/issues/97 resolved. This just provides early
   // validation of sanity of fields that we should catch at config ingestion.

diff --git a/source/common/upstream/upstream_impl.h b/source/common/upstream/upstream_impl.h
@@ -532,6 +532,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 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_;
@@ -614,6 +617,7 @@ class ClusterInfoImpl : public ClusterInfo {
   const bool drain_connections_on_host_removal_;
   const bool warm_hosts_;
   absl::optional<std::string> eds_service_name_;
+  envoy::api::v2::Cluster::CustomClusterType cluster_type_;
 };
 
 /**

diff --git a/source/extensions/clusters/redis/redis_cluster.cc b/source/extensions/clusters/redis/redis_cluster.cc
@@ -7,7 +7,7 @@ namespace Extensions {
 namespace Clusters {
 namespace Redis {
 
-RedisCluster::RedisCluster(
+RedisClusterImpl::RedisClusterImpl(
     const envoy::api::v2::Cluster& cluster,
     const envoy::config::cluster::redis::RedisClusterConfig& redisCluster,
     NetworkFilters::Common::Redis::Client::ClientFactory& redis_client_factory,
@@ -40,15 +40,15 @@ RedisCluster::RedisCluster(
   }
 };
 
-void RedisCluster::startPreInit() {
+void RedisClusterImpl::startPreInit() {
   for (const DnsDiscoveryResolveTargetPtr& target : dns_discovery_resolve_targets_) {
     target->startResolve();
   }
 }
 
-void RedisCluster::updateAllHosts(const Upstream::HostVector& hosts_added,
-                                  const Upstream::HostVector& hosts_removed,
-                                  uint32_t current_priority) {
+void RedisClusterImpl::updateAllHosts(const Upstream::HostVector& hosts_added,
+                                      const Upstream::HostVector& hosts_removed,
+                                      uint32_t current_priority) {
   Upstream::PriorityStateManager priority_state_manager(*this, local_info_, nullptr);
 
   auto locality_lb_endpoint = localityLbEndpoint();
@@ -64,7 +64,7 @@ void RedisCluster::updateAllHosts(const Upstream::HostVector& hosts_added,
       hosts_added, hosts_removed, absl::nullopt);
 }
 
-void RedisCluster::onClusterSlotUpdate(const std::vector<ClusterSlot>& slots) {
+void RedisClusterImpl::onClusterSlotUpdate(const std::vector<ClusterSlot>& slots) {
   Upstream::HostVector new_hosts;
   SlotArray slots_;
 
@@ -104,20 +104,20 @@ void RedisCluster::onClusterSlotUpdate(const std::vector<ClusterSlot>& slots) {
 }
 
 // DnsDiscoveryResolveTarget
-RedisCluster::DnsDiscoveryResolveTarget::DnsDiscoveryResolveTarget(
-    RedisCluster& parent, const std::string& dns_address, const uint32_t port,
+RedisClusterImpl::DnsDiscoveryResolveTarget::DnsDiscoveryResolveTarget(
+    RedisClusterImpl& parent, const std::string& dns_address, const uint32_t port,
     const envoy::api::v2::endpoint::LocalityLbEndpoints& locality_lb_endpoint,
     const envoy::api::v2::endpoint::LbEndpoint& lb_endpoint)
     : parent_(parent), dns_address_(dns_address), port_(port),
       locality_lb_endpoint_(locality_lb_endpoint), lb_endpoint_(lb_endpoint) {}
 
-RedisCluster::DnsDiscoveryResolveTarget::~DnsDiscoveryResolveTarget() {
+RedisClusterImpl::DnsDiscoveryResolveTarget::~DnsDiscoveryResolveTarget() {
   if (active_query_) {
     active_query_->cancel();
   }
 }
 
-void RedisCluster::DnsDiscoveryResolveTarget::startResolve() {
+void RedisClusterImpl::DnsDiscoveryResolveTarget::startResolve() {
   ENVOY_LOG(trace, "starting async DNS resolution for {}", dns_address_);
 
   active_query_ = parent_.dns_resolver_->resolve(
@@ -130,9 +130,9 @@ void RedisCluster::DnsDiscoveryResolveTarget::startResolve() {
       });
 }
 
-// RedisCluster
-RedisCluster::RedisDiscoverySession::RedisDiscoverySession(
-    Envoy::Extensions::Clusters::Redis::RedisCluster& parent,
+// RedisClusterImpl
+RedisClusterImpl::RedisDiscoverySession::RedisDiscoverySession(
+    Envoy::Extensions::Clusters::Redis::RedisClusterImpl& parent,
     NetworkFilters::Common::Redis::Client::ClientFactory& client_factory)
     : parent_(parent), dispatcher_(parent.dispatcher_),
       resolve_timer_(parent.dispatcher_.createTimer([this]() -> void { startResolve(); })),
@@ -162,7 +162,7 @@ ProcessCluster(const NetworkFilters::Common::Redis::RespValue& value) {
 }
 } // namespace
 
-RedisCluster::RedisDiscoverySession::~RedisDiscoverySession() {
+RedisClusterImpl::RedisDiscoverySession::~RedisDiscoverySession() {
   if (current_request_) {
     current_request_->cancel();
     current_request_ = nullptr;
@@ -173,7 +173,7 @@ RedisCluster::RedisDiscoverySession::~RedisDiscoverySession() {
   }
 }
 
-void RedisCluster::RedisDiscoveryClient::onEvent(Network::ConnectionEvent event) {
+void RedisClusterImpl::RedisDiscoveryClient::onEvent(Network::ConnectionEvent event) {
   if (event == Network::ConnectionEvent::RemoteClose ||
       event == Network::ConnectionEvent::LocalClose) {
     auto client_to_delete = parent_.client_map_.find(host_);
@@ -183,7 +183,7 @@ void RedisCluster::RedisDiscoveryClient::onEvent(Network::ConnectionEvent event)
   }
 }
 
-void RedisCluster::RedisDiscoverySession::registerDiscoveryAddress(
+void RedisClusterImpl::RedisDiscoverySession::registerDiscoveryAddress(
     const std::list<Envoy::Network::Address::InstanceConstSharedPtr>& address_list,
     const uint32_t port) {
   // Since the address from DNS does not have port, we need to make a new address that has port in
@@ -194,7 +194,7 @@ void RedisCluster::RedisDiscoverySession::registerDiscoveryAddress(
   }
 }
 
-void RedisCluster::RedisDiscoverySession::startResolve() {
+void RedisClusterImpl::RedisDiscoverySession::startResolve() {
   parent_.info_->stats().update_attempt_.inc();
   // If a resolution is currently in progress, skip it.
   if (current_request_) {
@@ -226,7 +226,7 @@ void RedisCluster::RedisDiscoverySession::startResolve() {
   current_request_ = client->client_->makeRequest(ClusterSlotsRequest::instance_, *this);
 }
 
-void RedisCluster::RedisDiscoverySession::onResponse(
+void RedisClusterImpl::RedisDiscoverySession::onResponse(
     NetworkFilters::Common::Redis::RespValuePtr&& value) {
   current_request_ = nullptr;
 
@@ -270,14 +270,14 @@ void RedisCluster::RedisDiscoverySession::onResponse(
   resolve_timer_->enableTimer(parent_.cluster_refresh_rate_);
 }
 
-void RedisCluster::RedisDiscoverySession::onUnexpectedResponse(
+void RedisClusterImpl::RedisDiscoverySession::onUnexpectedResponse(
     const NetworkFilters::Common::Redis::RespValuePtr& value) {
   ENVOY_LOG(warn, "Unexpected response to cluster slot command: {}", value->toString());
   this->parent_.info_->stats().update_failure_.inc();
   resolve_timer_->enableTimer(parent_.cluster_refresh_rate_);
 }
 
-void RedisCluster::RedisDiscoverySession::onFailure() {
+void RedisClusterImpl::RedisDiscoverySession::onFailure() {
   current_request_ = nullptr;
   if (!current_host_address_.empty()) {
     auto client_to_delete = client_map_.find(current_host_address_);
@@ -287,7 +287,7 @@ void RedisCluster::RedisDiscoverySession::onFailure() {
   resolve_timer_->enableTimer(parent_.cluster_refresh_rate_);
 }
 
-RedisCluster::ClusterSlotsRequest RedisCluster::ClusterSlotsRequest::instance_;
+RedisClusterImpl::ClusterSlotsRequest RedisClusterImpl::ClusterSlotsRequest::instance_;
 
 Upstream::ClusterImplBaseSharedPtr RedisClusterFactory::createClusterWithConfig(
     const envoy::api::v2::Cluster& cluster,
@@ -299,7 +299,7 @@ Upstream::ClusterImplBaseSharedPtr RedisClusterFactory::createClusterWithConfig(
       cluster.cluster_type().name() != Extensions::Clusters::ClusterTypes::get().Redis) {
     throw EnvoyException("Redis cluster can only created with redis cluster type");
   }
-  return std::make_shared<RedisCluster>(
+  return std::make_shared<RedisClusterImpl>(
       cluster, proto_config, NetworkFilters::Common::Redis::Client::ClientFactoryImpl::instance_,
       context.clusterManager(), context.runtime(), selectDnsResolver(cluster, context),
       socket_factory_context, std::move(stats_scope), context.addedViaApi());

diff --git a/source/extensions/clusters/redis/redis_cluster.h b/source/extensions/clusters/redis/redis_cluster.h
@@ -84,18 +84,25 @@ namespace Redis {
  * provide a fast lookup constant time lookup similar to Maglev. This will be used by the redis
  * proxy filter for load balancing purpose.
  */
+static const int MAX_SLOT = 16384;
 
-typedef std::array<Upstream::HostSharedPtr, 16384> SlotArray;
+typedef std::array<Upstream::HostSharedPtr, MAX_SLOT> SlotArray;
 
-class RedisCluster : public Upstream::BaseDynamicClusterImpl {
+class RedisCluster {
 public:
-  RedisCluster(const envoy::api::v2::Cluster& cluster,
-               const envoy::config::cluster::redis::RedisClusterConfig& redisCluster,
-               NetworkFilters::Common::Redis::Client::ClientFactory& client_factory,
-               Upstream::ClusterManager& clusterManager, Runtime::Loader& runtime,
-               Network::DnsResolverSharedPtr dns_resolver,
-               Server::Configuration::TransportSocketFactoryContext& factory_context,
-               Stats::ScopePtr&& stats_scope, bool added_via_api);
+  virtual ~RedisCluster() = default;
+  virtual const SlotArray& slotArray() const PURE;
+};
+
+class RedisClusterImpl : public RedisCluster, public Upstream::BaseDynamicClusterImpl {
+public:
+  RedisClusterImpl(const envoy::api::v2::Cluster& cluster,
+                   const envoy::config::cluster::redis::RedisClusterConfig& redisCluster,
+                   NetworkFilters::Common::Redis::Client::ClientFactory& client_factory,
+                   Upstream::ClusterManager& clusterManager, Runtime::Loader& runtime,
+                   Network::DnsResolverSharedPtr dns_resolver,
+                   Server::Configuration::TransportSocketFactoryContext& factory_context,
+                   Stats::ScopePtr&& stats_scope, bool added_via_api);
 
   struct ClusterSlotsRequest : public Extensions::NetworkFilters::Common::Redis::RespValue {
   public:
@@ -113,6 +120,8 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
 
   InitializePhase initializePhase() const override { return InitializePhase::Primary; }
 
+  const SlotArray& slotArray() const override { return cluster_slots_map_; }
+
 private:
   friend class RedisClusterTest;
 
@@ -146,7 +155,8 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
   class RedisHost : public Upstream::HostImpl {
   public:
     RedisHost(Upstream::ClusterInfoConstSharedPtr cluster, const std::string& hostname,
-              Network::Address::InstanceConstSharedPtr address, RedisCluster& parent, bool master)
+              Network::Address::InstanceConstSharedPtr address, RedisClusterImpl& parent,
+              bool master)
         : Upstream::HostImpl(cluster, hostname, address, parent.lbEndpoint().metadata(),
                              parent.lbEndpoint().load_balancing_weight().value(),
                              parent.localityLbEndpoint().locality(),
@@ -164,15 +174,15 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
   // Resolves the discovery endpoint.
   struct DnsDiscoveryResolveTarget {
     DnsDiscoveryResolveTarget(
-        RedisCluster& parent, const std::string& dns_address, const uint32_t port,
+        RedisClusterImpl& parent, const std::string& dns_address, const uint32_t port,
         const envoy::api::v2::endpoint::LocalityLbEndpoints& locality_lb_endpoint,
         const envoy::api::v2::endpoint::LbEndpoint& lb_endpoint);
 
     ~DnsDiscoveryResolveTarget();
 
     void startResolve();
 
-    RedisCluster& parent_;
+    RedisClusterImpl& parent_;
     Network::ActiveDnsQuery* active_query_{};
     const std::string dns_address_;
     const uint32_t port_;
@@ -202,7 +212,7 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
   struct RedisDiscoverySession
       : public Extensions::NetworkFilters::Common::Redis::Client::Config,
         public Extensions::NetworkFilters::Common::Redis::Client::PoolCallbacks {
-    RedisDiscoverySession(RedisCluster& parent,
+    RedisDiscoverySession(RedisClusterImpl& parent,
                           NetworkFilters::Common::Redis::Client::ClientFactory& client_factory);
 
     ~RedisDiscoverySession();
@@ -232,7 +242,7 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
     bool onRedirection(const NetworkFilters::Common::Redis::RespValue&) override { return true; }
     void onUnexpectedResponse(const NetworkFilters::Common::Redis::RespValuePtr&);
 
-    RedisCluster& parent_;
+    RedisClusterImpl& parent_;
     Event::Dispatcher& dispatcher_;
     std::string current_host_address_;
     Extensions::NetworkFilters::Common::Redis::Client::PoolRequest* current_request_{};

diff --git a/source/extensions/filters/network/redis_proxy/BUILD b/source/extensions/filters/network/redis_proxy/BUILD
@@ -72,6 +72,7 @@ envoy_cc_library(
         "//source/common/protobuf:utility_lib",
         "//source/common/upstream:load_balancer_lib",
         "//source/common/upstream:upstream_lib",
+        "//source/extensions/clusters/redis:redis_cluster",
         "//source/extensions/filters/network/common/redis:client_lib",
         "@envoy_api//envoy/config/filter/network/redis_proxy/v2:redis_proxy_cc",
     ],