Add a request time metric to redis upstream

api/envoy/config/filter/network/redis_proxy/v2/redis_proxy.proto

@@ -91,6 +91,10 @@ message RedisProxy {
     // downstream unchanged. This limit defaults to 100.
     google.protobuf.UInt32Value max_upstream_unknown_connections = 6;
 
+    // Enable per-command statistics per upstream cluster, in addition to the filter level aggregate
+    // count.
+    bool enable_command_stats = 8;
+
     // ReadPolicy controls how Envoy routes read commands to Redis nodes. This is currently
     // supported for Redis Cluster. All ReadPolicy settings except MASTER may return stale data
     // because replication is asynchronous and requires some delay. You need to ensure that your

docs/root/intro/arch_overview/other_protocols/redis.rst

@@ -90,6 +90,11 @@ Every Redis cluster has its own extra statistics tree rooted at *cluster.<name>.
 
   max_upstream_unknown_connections_reached, Counter, Total number of times that an upstream connection to an unknown host is not created after redirection having reached the connection pool's max_upstream_unknown_connections limit
   upstream_cx_drained, Counter, Total number of upstream connections drained of active requests before being closed
+  upstream_commands.upstream_rq_time, Histogram, Histogram of upstream request times for all types of requests
+  upstream_commands.[command].success, Counter, Total number of successful requests for a specific Redis command
+  upstream_commands.[command].error, Counter, Total number of failed or cancelled requests for a specific Redis command
+  upstream_commands.[command].total, Counter, Total number of requests for a specific Redis command (sum of success and error)
+  upstream_commands.[command].latency, Histogram, Latency of requests for a specific Redis command
 
 Supported commands
 ------------------

docs/root/intro/version_history.rst

@@ -28,6 +28,7 @@ Version history
 * listeners: added :ref:`continue_on_listener_filters_timeout <envoy_api_field_Listener.continue_on_listener_filters_timeout>` to configure whether a listener will still create a connection when listener filters time out.
 * listeners: added :ref:`HTTP inspector listener filter <config_listener_filters_http_inspector>`.
 * 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: added :ref:`enable_command_stats <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.ConnPoolSettings.enable_command_stats>` to enable per command statistics for upstream clusters.
 * rbac: added support for DNS SAN as :ref:`principal_name <envoy_api_field_config.rbac.v2.Principal.Authenticated.principal_name>`.
 * lua: extended `httpCall()` and `respond()` APIs to accept headers with entry values that can be a string or table of strings.
 * performance: new buffer implementation enabled by default (to disable add "--use-libevent-buffers 1" to the command-line arguments when starting Envoy).

source/extensions/clusters/redis/redis_cluster.h

@@ -214,6 +214,7 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
     uint32_t maxBufferSizeBeforeFlush() const override { return 0; }
     std::chrono::milliseconds bufferFlushTimeoutInMs() const override { return buffer_timeout_; }
     uint32_t maxUpstreamUnknownConnections() const override { return 0; }
+    bool enableCommandStats() const override { return false; }
     // This is effectively not in used for making the "Cluster Slots" calls.
     // since we call cluster slots on both the master and slaves, ANY is more appropriate here.
     Extensions::NetworkFilters::Common::Redis::Client::ReadPolicy readPolicy() const override {

source/extensions/filters/network/common/redis/BUILD

@@ -57,7 +57,9 @@ envoy_cc_library(
     deps = [
         ":client_interface",
         ":codec_lib",
+        ":redis_command_stats_lib",
         "//include/envoy/router:router_interface",
+        "//include/envoy/stats:timespan",
         "//include/envoy/thread_local:thread_local_interface",
         "//include/envoy/upstream:cluster_manager_interface",
         "//source/common/buffer:buffer_lib",
@@ -78,3 +80,17 @@ envoy_cc_library(
         ":codec_lib",
     ],
 )
+
+envoy_cc_library(
+    name = "redis_command_stats_lib",
+    srcs = ["redis_command_stats.cc"],
+    hdrs = ["redis_command_stats.h"],
+    deps = [
+        ":codec_interface",
+        ":supported_commands_lib",
+        "//include/envoy/stats:stats_interface",
+        "//include/envoy/stats:timespan",
+        "//source/common/common:utility_lib",
+        "//source/common/stats:symbol_table_lib",
+    ],
+)

source/extensions/filters/network/common/redis/client.h

@@ -163,6 +163,11 @@ class Config {
    */
   virtual uint32_t maxUpstreamUnknownConnections() const PURE;
 
+  /**
+   * @return when enabled, upstream cluster per-command statistics will be recorded.
+   */
+  virtual bool enableCommandStats() const PURE;
+
   /**
    * @return the read policy the proxy should use.
    */

source/extensions/filters/network/common/redis/client_impl.cc

@@ -19,7 +19,8 @@ ConfigImpl::ConfigImpl(
           3)), // Default timeout is 3ms. If max_buffer_size_before_flush is zero, this is not used
                // as the buffer is flushed on each request immediately.
       max_upstream_unknown_connections_(
-          PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, max_upstream_unknown_connections, 100)) {
+          PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, max_upstream_unknown_connections, 100)),
+      enable_command_stats_(config.enable_command_stats()) {
   switch (config.read_policy()) {
   case envoy::config::filter::network::redis_proxy::v2::
       RedisProxy_ConnPoolSettings_ReadPolicy_MASTER:
@@ -49,9 +50,11 @@ ConfigImpl::ConfigImpl(
 ClientPtr ClientImpl::create(Upstream::HostConstSharedPtr host, Event::Dispatcher& dispatcher,
                              EncoderPtr&& encoder, DecoderFactory& decoder_factory,
                              const Config& config) {
-
-  std::unique_ptr<ClientImpl> client(
-      new ClientImpl(host, dispatcher, std::move(encoder), decoder_factory, config));
+  auto redis_command_stats = std::make_shared<RedisCommandStats>(
+      host->cluster().statsScope(), "upstream_commands", config.enableCommandStats());
+  std::unique_ptr<ClientImpl> client(new ClientImpl(host, dispatcher, std::move(encoder),
+                                                    decoder_factory, config,
+                                                    std::move(redis_command_stats)));
   client->connection_ = host->createConnection(dispatcher, nullptr, nullptr).connection_;
   client->connection_->addConnectionCallbacks(*client);
   client->connection_->addReadFilter(Network::ReadFilterSharedPtr{new UpstreamReadFilter(*client)});
@@ -61,11 +64,13 @@ ClientPtr ClientImpl::create(Upstream::HostConstSharedPtr host, Event::Dispatche
 }
 
 ClientImpl::ClientImpl(Upstream::HostConstSharedPtr host, Event::Dispatcher& dispatcher,
-                       EncoderPtr&& encoder, DecoderFactory& decoder_factory, const Config& config)
+                       EncoderPtr&& encoder, DecoderFactory& decoder_factory, const Config& config,
+                       RedisCommandStatsPtr&& redis_command_stats)
     : host_(host), encoder_(std::move(encoder)), decoder_(decoder_factory.create(*this)),
       config_(config),
       connect_or_op_timer_(dispatcher.createTimer([this]() -> void { onConnectOrOpTimeout(); })),
-      flush_timer_(dispatcher.createTimer([this]() -> void { flushBufferAndResetTimer(); })) {
+      flush_timer_(dispatcher.createTimer([this]() -> void { flushBufferAndResetTimer(); })),
+      time_source_(dispatcher.timeSource()), redis_command_stats_(redis_command_stats) {
   host->cluster().stats().upstream_cx_total_.inc();
   host->stats().cx_total_.inc();
   host->cluster().stats().upstream_cx_active_.inc();
@@ -94,9 +99,15 @@ PoolRequest* ClientImpl::makeRequest(const RespValue& request, PoolCallbacks& ca
 
   const bool empty_buffer = encoder_buffer_.length() == 0;
 
-  pending_requests_.emplace_back(*this, callbacks);
+  std::string to_lower_command(redis_command_stats_->getCommandFromRequest(request));
+  to_lower_table_.toLowerCase(to_lower_command);
+  pending_requests_.emplace_back(*this, callbacks, to_lower_command);
   encoder_->encode(request, encoder_buffer_);
 
+  if (config_.enableCommandStats()) {
+    redis_command_stats_->updateStatsTotal(to_lower_command);
+  }
+
   // If buffer is full, flush. If the buffer was empty before the request, start the timer.
   if (encoder_buffer_.length() >= config_.maxBufferSizeBeforeFlush()) {
     flushBufferAndResetTimer();
@@ -186,6 +197,14 @@ void ClientImpl::onRespValue(RespValuePtr&& value) {
   ASSERT(!pending_requests_.empty());
   PendingRequest& request = pending_requests_.front();
   const bool canceled = request.canceled_;
+
+  if (redis_command_stats_->enabled()) {
+    bool success = !canceled && (value->type() != Common::Redis::RespType::Error);
+    redis_command_stats_->updateStats(success, request.command_);
+    request.command_request_timer_->complete();
+  }
+  request.aggregate_request_timer_->complete();
+
   PoolCallbacks& callbacks = request.callbacks_;
 
   // We need to ensure the request is popped before calling the callback, since the callback might
@@ -225,8 +244,13 @@ void ClientImpl::onRespValue(RespValuePtr&& value) {
   putOutlierEvent(Upstream::Outlier::Result::EXT_ORIGIN_REQUEST_SUCCESS);
 }
 
-ClientImpl::PendingRequest::PendingRequest(ClientImpl& parent, PoolCallbacks& callbacks)
-    : parent_(parent), callbacks_(callbacks) {
+ClientImpl::PendingRequest::PendingRequest(ClientImpl& parent, PoolCallbacks& callbacks,
+                                           std::string command)
+    : parent_(parent), callbacks_(callbacks), command_{command},
+      aggregate_request_timer_(
+          parent_.redis_command_stats_->createAggregateTimer(parent_.time_source_)),
+      command_request_timer_(
+          parent_.redis_command_stats_->createCommandTimer(command_, parent_.time_source_)) {
   parent.host_->cluster().stats().upstream_rq_total_.inc();
   parent.host_->stats().rq_total_.inc();
   parent.host_->cluster().stats().upstream_rq_active_.inc();

source/extensions/filters/network/common/redis/client_impl.h

@@ -3,18 +3,21 @@
 #include <chrono>
 
 #include "envoy/config/filter/network/redis_proxy/v2/redis_proxy.pb.h"
+#include "envoy/stats/timespan.h"
 #include "envoy/thread_local/thread_local.h"
 #include "envoy/upstream/cluster_manager.h"
 
 #include "common/buffer/buffer_impl.h"
 #include "common/common/hash.h"
+#include "common/common/to_lower_table.h"
 #include "common/network/filter_impl.h"
 #include "common/protobuf/utility.h"
 #include "common/singleton/const_singleton.h"
 #include "common/upstream/load_balancer_impl.h"
 #include "common/upstream/upstream_impl.h"
 
 #include "extensions/filters/network/common/redis/client.h"
+#include "extensions/filters/network/common/redis/redis_command_stats.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -49,6 +52,7 @@ class ConfigImpl : public Config {
   uint32_t maxUpstreamUnknownConnections() const override {
     return max_upstream_unknown_connections_;
   }
+  bool enableCommandStats() const override { return enable_command_stats_; }
   ReadPolicy readPolicy() const override { return read_policy_; }
 
 private:
@@ -58,6 +62,7 @@ class ConfigImpl : public Config {
   const uint32_t max_buffer_size_before_flush_;
   const std::chrono::milliseconds buffer_flush_timeout_;
   const uint32_t max_upstream_unknown_connections_;
+  const bool enable_command_stats_;
   ReadPolicy read_policy_;
 };
 
@@ -94,19 +99,23 @@ class ClientImpl : public Client, public DecoderCallbacks, public Network::Conne
   };
 
   struct PendingRequest : public PoolRequest {
-    PendingRequest(ClientImpl& parent, PoolCallbacks& callbacks);
+    PendingRequest(ClientImpl& parent, PoolCallbacks& callbacks, std::string command);
     ~PendingRequest() override;
 
     // PoolRequest
     void cancel() override;
 
     ClientImpl& parent_;
     PoolCallbacks& callbacks_;
+    std::string command_;
     bool canceled_{};
+    Stats::CompletableTimespanPtr aggregate_request_timer_;
+    Stats::CompletableTimespanPtr command_request_timer_;
   };
 
   ClientImpl(Upstream::HostConstSharedPtr host, Event::Dispatcher& dispatcher, EncoderPtr&& encoder,
-             DecoderFactory& decoder_factory, const Config& config);
+             DecoderFactory& decoder_factory, const Config& config,
+             RedisCommandStatsPtr&& redis_command_stats);
   void onConnectOrOpTimeout();
   void onData(Buffer::Instance& data);
   void putOutlierEvent(Upstream::Outlier::Result result);
@@ -129,6 +138,9 @@ class ClientImpl : public Client, public DecoderCallbacks, public Network::Conne
   Event::TimerPtr connect_or_op_timer_;
   bool connected_{};
   Event::TimerPtr flush_timer_;
+  Envoy::TimeSource& time_source_;
+  const RedisCommandStatsPtr redis_command_stats_;
+  const ToLowerTable to_lower_table_;
 };
 
 class ClientFactoryImpl : public ClientFactory {

source/extensions/filters/network/common/redis/redis_command_stats.cc

@@ -0,0 +1,106 @@
+#include "extensions/filters/network/common/redis/redis_command_stats.h"
+
+#include "extensions/filters/network/common/redis/supported_commands.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Common {
+namespace Redis {
+
+RedisCommandStats::RedisCommandStats(Stats::Scope& scope, const std::string& prefix, bool enabled)
+    : scope_(scope), stat_name_set_(scope.symbolTable()), prefix_(stat_name_set_.add(prefix)),
+      enabled_(enabled), upstream_rq_time_(stat_name_set_.add("upstream_rq_time")) {
+  // Note: Even if this is disabled, we track the upstream_rq_time.
+  if (enabled_) {
+    // Create StatName for each Redis command. Note that we don't include Auth or Ping.
+    for (const std::string& command :
+         Extensions::NetworkFilters::Common::Redis::SupportedCommands::simpleCommands()) {
+      createStats(command);
+    }
+    for (const std::string& command :
+         Extensions::NetworkFilters::Common::Redis::SupportedCommands::evalCommands()) {
+      createStats(command);
+    }
+    for (const std::string& command : Extensions::NetworkFilters::Common::Redis::SupportedCommands::
+             hashMultipleSumResultCommands()) {
+      createStats(command);
+    }
+    createStats(Extensions::NetworkFilters::Common::Redis::SupportedCommands::mget());
+    createStats(Extensions::NetworkFilters::Common::Redis::SupportedCommands::mset());
+  }
+}
+
+void RedisCommandStats::createStats(std::string name) {
+  stat_name_set_.add(name + ".total");
+  stat_name_set_.add(name + ".success");
+  stat_name_set_.add(name + ".error");
+  stat_name_set_.add(name + ".latency");
+}
+
+Stats::SymbolTable::StoragePtr RedisCommandStats::addPrefix(const Stats::StatName name) {
+  Stats::StatNameVec names_with_prefix;
+  names_with_prefix.reserve(2);
+  names_with_prefix.push_back(prefix_);
+  names_with_prefix.insert(names_with_prefix.end(), name);
+  return scope_.symbolTable().join(names_with_prefix);
+}
+
+Stats::Counter& RedisCommandStats::counter(std::string name) {
+  Stats::StatName stat_name = stat_name_set_.getStatName(name);
+  const Stats::SymbolTable::StoragePtr stat_name_storage = addPrefix(stat_name);
+  return scope_.counterFromStatName(Stats::StatName(stat_name_storage.get()));
+}
+
+Stats::Histogram& RedisCommandStats::histogram(std::string name) {
+  Stats::StatName stat_name = stat_name_set_.getStatName(name);
+  return histogram(stat_name);
+}
+
+Stats::Histogram& RedisCommandStats::histogram(Stats::StatName stat_name) {
+  const Stats::SymbolTable::StoragePtr stat_name_storage = addPrefix(stat_name);
+  return scope_.histogramFromStatName(Stats::StatName(stat_name_storage.get()));
+}
+
+Stats::CompletableTimespanPtr
+RedisCommandStats::createCommandTimer(std::string name, Envoy::TimeSource& time_source) {
+  Stats::StatName stat_name = stat_name_set_.getStatName(name + latency_suffix_);
+  return std::make_unique<Stats::TimespanWithUnit<std::chrono::microseconds>>(histogram(stat_name),
+                                                                              time_source);
+}
+
+Stats::CompletableTimespanPtr
+RedisCommandStats::createAggregateTimer(Envoy::TimeSource& time_source) {
+  return std::make_unique<Stats::TimespanWithUnit<std::chrono::microseconds>>(
+      histogram(upstream_rq_time_), time_source);
+}
+
+std::string RedisCommandStats::getCommandFromRequest(const RespValue& request) {
+  // Get command from RespValue
+  switch (request.type()) {
+  case RespType::Array:
+    return getCommandFromRequest(request.asArray().front());
+  case RespType::Integer:
+    return std::to_string(request.asInteger());
+  case RespType::Null:
+    return "null";
+  default:
+    return request.asString();
+  }
+}
+
+void RedisCommandStats::updateStatsTotal(std::string command) { counter(command + ".total").inc(); }
+
+void RedisCommandStats::updateStats(const bool success, std::string command) {
+  if (success) {
+    counter(command + ".success").inc();
+  } else {
+    counter(command + ".error").inc();
+  }
+}
+
+} // namespace Redis
+} // namespace Common
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy