Merge remote-tracking branch 'upstream/master' into ADS_incremental

Signed-off-by: Fred Douglas <fredlas@google.com>

docs/root/intro/version_history.rst

@@ -50,6 +50,7 @@ Version history
 * redis: added :ref:`enable_command_stats <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.ConnPoolSettings.enable_command_stats>` to enable :ref:`per command statistics <arch_overview_redis_cluster_command_stats>` for upstream clusters.
 * redis: added :ref:`read_policy <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.ConnPoolSettings.read_policy>` to allow reading from redis replicas for Redis Cluster deployments.
 * redis: fix a bug where the redis health checker ignored the upstream auth password.
+* redis: enable_hashtaging is always enabled when the upstream uses open source Redis cluster protocol.
 * regex: introduce new :ref:`RegexMatcher <envoy_api_msg_type.matcher.RegexMatcher>` type that
   provides a safe regex implementation for untrusted user input. This type is now used in all
   configuration that processes user provided input. See :ref:`deprecated configuration details

source/common/stats/BUILD

@@ -87,6 +87,15 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "recent_lookups_lib",
+    srcs = ["recent_lookups.cc"],
+    hdrs = ["recent_lookups.h"],
+    deps = [
+        "//source/common/common:assert_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "store_impl_lib",
     hdrs = ["store_impl.h"],

source/common/stats/recent_lookups.cc

@@ -0,0 +1,60 @@
+#include "common/stats/recent_lookups.h"
+
+#include "common/common/assert.h"
+
+namespace Envoy {
+namespace Stats {
+
+void RecentLookups::lookup(absl::string_view str) {
+  ++total_;
+  if (capacity_ == 0) {
+    return;
+  }
+  auto map_iter = map_.find(str);
+  if (map_iter != map_.end()) {
+    // The item is already in the list. Bump its reference-count and move it to
+    // the front of the list.
+    auto list_iter = map_iter->second;
+    ++list_iter->count_;
+    if (list_iter != list_.begin()) {
+      list_.splice(list_.begin(), list_, list_iter);
+    }
+  } else {
+    ASSERT(list_.size() <= capacity_);
+    // Evict oldest item if needed.
+    if (list_.size() >= capacity_) {
+      evictOne();
+    }
+
+    // The string storage is in the list entry.
+    list_.push_front(ItemCount{std::string(str), 1});
+    auto list_iter = list_.begin();
+    map_[list_iter->item_] = list_iter;
+  }
+  ASSERT(list_.size() == map_.size());
+}
+
+void RecentLookups::forEach(const IterFn& fn) const {
+  for (const ItemCount& item_count : list_) {
+    fn(item_count.item_, item_count.count_);
+  }
+}
+
+void RecentLookups::setCapacity(uint64_t capacity) {
+  capacity_ = capacity;
+  while (capacity_ < list_.size()) {
+    evictOne();
+  }
+}
+
+void RecentLookups::evictOne() {
+  ASSERT(!list_.empty());
+  ASSERT(!map_.empty());
+  const ItemCount& item_count = list_.back();
+  int erased = map_.erase(item_count.item_);
+  ASSERT(erased == 1);
+  list_.pop_back();
+}
+
+} // namespace Stats
+} // namespace Envoy

source/common/stats/recent_lookups.h

@@ -0,0 +1,77 @@
+#pragma once
+
+#include <functional>
+#include <list>
+#include <utility>
+
+#include "absl/container/flat_hash_map.h"
+#include "absl/strings/string_view.h"
+
+namespace Envoy {
+namespace Stats {
+
+// Remembers the last 'Capacity' items passed to lookup().
+class RecentLookups {
+public:
+  /**
+   * Records a lookup of a string. Only the last 'Capacity' lookups are remembered.
+   *
+   * @param str the item being looked up.
+   */
+  void lookup(absl::string_view str);
+
+  using IterFn = std::function<void(absl::string_view, uint64_t)>;
+
+  /**
+   * Calls fn(item, count) for each of the remembered lookups.
+   *
+   * @param fn The function to call for every recently looked up item.
+   */
+  void forEach(const IterFn& fn) const;
+
+  /**
+   * @return the total number of lookups since tracking began.
+   */
+  uint64_t total() const { return total_; }
+
+  /**
+   * Clears out all contents.
+   */
+  void clear() {
+    total_ = 0;
+    map_.clear();
+    list_.clear();
+  }
+
+  /**
+   * Controls the maximum number of recent lookups to remember. If set to 0,
+   * then only lookup counts is tracked.
+   * @param capacity The number of lookups to remember.
+   */
+  void setCapacity(uint64_t capacity);
+
+  /**
+   * @return The configured capacity.
+   */
+  uint64_t capacity() const { return capacity_; }
+
+private:
+  void evictOne();
+
+  struct ItemCount {
+    std::string item_;
+    uint64_t count_;
+  };
+  using List = std::list<ItemCount>;
+  List list_;
+
+  // TODO(jmarantz): we could make this more compact by making this a set of
+  // list-iterators with heterogeneous hash/compare functors.
+  using Map = absl::flat_hash_map<absl::string_view, List::iterator>;
+  Map map_;
+  uint64_t total_{0};
+  uint64_t capacity_{0};
+};
+
+} // namespace Stats
+} // namespace Envoy

source/extensions/clusters/redis/redis_cluster_lb.cc

@@ -176,11 +176,11 @@ bool isReadRequest(const NetworkFilters::Common::Redis::RespValue& request) {
 } // namespace
 
 RedisLoadBalancerContextImpl::RedisLoadBalancerContextImpl(
-    const std::string& key, bool enabled_hashtagging, bool use_crc16,
+    const std::string& key, bool enabled_hashtagging, bool is_redis_cluster,
     const NetworkFilters::Common::Redis::RespValue& request,
     NetworkFilters::Common::Redis::Client::ReadPolicy read_policy)
-    : hash_key_(use_crc16 ? Crc16::crc16(hashtag(key, enabled_hashtagging))
-                          : MurmurHash::murmurHash2_64(hashtag(key, enabled_hashtagging))),
+    : hash_key_(is_redis_cluster ? Crc16::crc16(hashtag(key, true))
+                                 : MurmurHash::murmurHash2_64(hashtag(key, enabled_hashtagging))),
       is_read_(isReadRequest(request)), read_policy_(read_policy) {}
 
 // Inspired by the redis-cluster hashtagging algorithm

source/extensions/clusters/redis/redis_cluster_lb.h

@@ -67,7 +67,19 @@ class RedisLoadBalancerContext {
 class RedisLoadBalancerContextImpl : public RedisLoadBalancerContext,
                                      public Upstream::LoadBalancerContextBase {
 public:
-  RedisLoadBalancerContextImpl(const std::string& key, bool enabled_hashtagging, bool use_crc16,
+  /**
+   * The load balancer context for Redis requests. Note that is_redis_cluster implies using Redis
+   * cluster which require us to always enable hashtagging.
+   * @param key specify the key for the Redis request.
+   * @param enabled_hashtagging specify whether to enable hashtagging, this will always be true if
+   * is_redis_cluster is true.
+   * @param is_redis_cluster specify whether this is a request for redis cluster, if true the key
+   * will be hashed using crc16.
+   * @param request specify the Redis request.
+   * @param read_policy specify the read policy.
+   */
+  RedisLoadBalancerContextImpl(const std::string& key, bool enabled_hashtagging,
+                               bool is_redis_cluster,
                                const NetworkFilters::Common::Redis::RespValue& request,
                                NetworkFilters::Common::Redis::Client::ReadPolicy read_policy =
                                    NetworkFilters::Common::Redis::Client::ReadPolicy::Master);

source/extensions/filters/network/redis_proxy/conn_pool_impl.cc

@@ -208,9 +208,9 @@ InstanceImpl::ThreadLocalPool::makeRequest(const std::string& key,
     return nullptr;
   }
 
-  const bool use_crc16 = is_redis_cluster_;
-  Clusters::Redis::RedisLoadBalancerContextImpl lb_context(
-      key, parent_.config_.enableHashtagging(), use_crc16, request, parent_.config_.readPolicy());
+  Clusters::Redis::RedisLoadBalancerContextImpl lb_context(key, parent_.config_.enableHashtagging(),
+                                                           is_redis_cluster_, request,
+                                                           parent_.config_.readPolicy());
   Upstream::HostConstSharedPtr host = cluster_->loadBalancer().chooseHost(&lb_context);
   if (!host) {
     return nullptr;

test/common/stats/BUILD

@@ -39,6 +39,30 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "recent_lookups_test",
+    srcs = ["recent_lookups_test.cc"],
+    deps = [
+        "//source/common/common:utility_lib",
+        "//source/common/stats:recent_lookups_lib",
+        "//test/test_common:logging_lib",
+        "//test/test_common:simulated_time_system_lib",
+    ],
+)
+
+envoy_cc_test_binary(
+    name = "recent_lookups_speed_test",
+    srcs = ["recent_lookups_speed_test.cc"],
+    external_deps = [
+        "benchmark",
+    ],
+    deps = [
+        "//source/common/common:utility_lib",
+        "//source/common/runtime:runtime_lib",
+        "//source/common/stats:recent_lookups_lib",
+    ],
+)
+
 envoy_cc_test(
     name = "stat_merger_test",
     srcs = ["stat_merger_test.cc"],

test/common/stats/recent_lookups_speed_test.cc

@@ -0,0 +1,82 @@
+// Note: this should be run with --compilation_mode=opt, and would benefit from a
+// quiescent system with disabled cstate power management.
+//
+// NOLINT(namespace-envoy)
+//
+// Running bazel-bin/test/common/stats/recent_lookups_speed_test
+// Run on (12 X 4500 MHz CPU s)
+// CPU Caches:
+//   L1 Data 32K (x6)
+//   L1 Instruction 32K (x6)
+//   L2 Unified 1024K (x6)
+//   L3 Unified 8448K (x1)
+// Load Average: 1.32, 7.40, 10.21
+// ***WARNING*** CPU scaling is enabled, the benchmark real time measurements may be noisy and will
+// incur extra overhead.
+// -----------------------------------------------------------------
+// Benchmark                       Time             CPU   Iterations
+// -----------------------------------------------------------------
+// BM_LookupsMixed             87068 ns        87068 ns         6955
+// BM_LookupsNoEvictions       45662 ns        45662 ns        15329
+// BM_LookupsAllEvictions      83015 ns        83015 ns         8435
+
+#include "common/runtime/runtime_impl.h"
+#include "common/stats/recent_lookups.h"
+
+#include "absl/strings/str_cat.h"
+#include "benchmark/benchmark.h"
+
+class RecentLookupsSpeedTest {
+public:
+  RecentLookupsSpeedTest(uint64_t lookup_variants, uint64_t capacity) {
+    recent_lookups_.setCapacity(capacity);
+    Envoy::Runtime::RandomGeneratorImpl random;
+    lookups_.reserve(lookup_variants);
+    for (size_t i = 0; i < lookup_variants; ++i) {
+      lookups_.push_back(absl::StrCat("lookup #", random.random()));
+    }
+  }
+
+  void test(benchmark::State& state) {
+    for (auto _ : state) {
+      Envoy::Runtime::RandomGeneratorImpl random;
+      for (uint64_t i = 0; i < lookups_.size(); ++i) {
+        recent_lookups_.lookup(lookups_[random.random() % lookups_.size()]);
+      }
+    }
+  }
+
+private:
+  std::vector<std::string> lookups_;
+  Envoy::Stats::RecentLookups recent_lookups_;
+};
+
+static void BM_LookupsMixed(benchmark::State& state) {
+  RecentLookupsSpeedTest speed_test(1000, 500);
+  speed_test.test(state);
+}
+BENCHMARK(BM_LookupsMixed);
+
+static void BM_LookupsNoEvictions(benchmark::State& state) {
+  RecentLookupsSpeedTest speed_test(1000, 1000);
+  speed_test.test(state);
+}
+BENCHMARK(BM_LookupsNoEvictions);
+
+static void BM_LookupsAllEvictions(benchmark::State& state) {
+  RecentLookupsSpeedTest speed_test(1000, 10);
+  speed_test.test(state);
+}
+BENCHMARK(BM_LookupsAllEvictions);
+
+int main(int argc, char** argv) {
+  Envoy::Thread::MutexBasicLockable lock;
+  Envoy::Logger::Context logger_context(spdlog::level::warn,
+                                        Envoy::Logger::Logger::DEFAULT_LOG_FORMAT, lock);
+  benchmark::Initialize(&argc, argv);
+
+  if (benchmark::ReportUnrecognizedArguments(argc, argv)) {
+    return 1;
+  }
+  benchmark::RunSpecifiedBenchmarks();
+}