Ensure Redis Proxy can follow redirection when doing request mirroring

docs/root/intro/version_history.rst

@@ -16,6 +16,7 @@ Version history
 * lb_subset_config: new fallback policy for selectors: :ref:`KEYS_SUBSET<envoy_api_enum_value_Cluster.LbSubsetConfig.LbSubsetSelector.LbSubsetSelectorFallbackPolicy.KEYS_SUBSET>`
 * logger: added :ref:`--log-format-escaped <operations_cli>` command line option to escape newline characters in application logs.
 * redis: performance improvement for larger split commands by avoiding string copies.
+* redis: correctly follow MOVE/ASK redirection for mirrored clusters.
 * router: added support for REQ(header-name) :ref:`header formatter <config_http_conn_man_headers_custom_request_headers>`.
 * router: skip the Location header when the response code is not a 201 or a 3xx.
 * server: fixed a bug in config validation for configs with runtime layers

source/extensions/clusters/redis/redis_cluster.h

@@ -194,7 +194,7 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
 
   struct RedisDiscoverySession
       : public Extensions::NetworkFilters::Common::Redis::Client::Config,
-        public Extensions::NetworkFilters::Common::Redis::Client::PoolCallbacks {
+        public Extensions::NetworkFilters::Common::Redis::Client::ClientCallbacks {
     RedisDiscoverySession(RedisCluster& parent,
                           NetworkFilters::Common::Redis::Client::ClientFactory& client_factory);
 
@@ -225,11 +225,14 @@ class RedisCluster : public Upstream::BaseDynamicClusterImpl {
       return Extensions::NetworkFilters::Common::Redis::Client::ReadPolicy::Master;
     }
 
-    // Extensions::NetworkFilters::Common::Redis::Client::PoolCallbacks
+    // Extensions::NetworkFilters::Common::Redis::Client::ClientCallbacks
     void onResponse(NetworkFilters::Common::Redis::RespValuePtr&& value) override;
     void onFailure() override;
     // Note: Below callback isn't used in topology updates
-    bool onRedirection(const NetworkFilters::Common::Redis::RespValue&) override { return true; }
+    bool onRedirection(NetworkFilters::Common::Redis::RespValuePtr&&, const std::string&,
+                       bool) override {
+      return true;
+    }
     void onUnexpectedResponse(const NetworkFilters::Common::Redis::RespValuePtr&);
 
     Network::Address::InstanceConstSharedPtr

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

@@ -30,9 +30,9 @@ class PoolRequest {
 /**
  * Outbound request callbacks.
  */
-class PoolCallbacks {
+class ClientCallbacks {
 public:
-  virtual ~PoolCallbacks() = default;
+  virtual ~ClientCallbacks() = default;
 
   /**
    * Called when a pipelined response is received.
@@ -48,21 +48,26 @@ class PoolCallbacks {
   /**
    * Called when a MOVED or ASK redirection error is received, and the request must be retried.
    * @param value supplies the MOVED error response
+   * @param host_address supplies the redirection host address and port
+   * @param ask_redirection indicates if this is a ASK redirection
    * @return bool true if the request is successfully redirected, false otherwise
    */
-  virtual bool onRedirection(const Common::Redis::RespValue& value) PURE;
+  virtual bool onRedirection(RespValuePtr&& value, const std::string& host_address,
+                             bool ask_redirection) PURE;
 };
 
 /**
  * DoNothingPoolCallbacks is used for internally generated commands whose response is
  * transparently filtered, and redirection never occurs (e.g., "asking", "auth", etc.).
  */
-class DoNothingPoolCallbacks : public PoolCallbacks {
+class DoNothingPoolCallbacks : public ClientCallbacks {
 public:
-  // PoolCallbacks
+  // ClientCallbacks
   void onResponse(Common::Redis::RespValuePtr&&) override {}
   void onFailure() override {}
-  bool onRedirection(const Common::Redis::RespValue&) override { return false; }
+  bool onRedirection(Common::Redis::RespValuePtr&&, const std::string&, bool) override {
+    return false;
+  }
 };
 
 /**
@@ -95,7 +100,7 @@ class Client : public Event::DeferredDeletable {
    * @return PoolRequest* a handle to the active request or nullptr if the request could not be made
    *         for some reason.
    */
-  virtual PoolRequest* makeRequest(const RespValue& request, PoolCallbacks& callbacks) PURE;
+  virtual PoolRequest* makeRequest(const RespValue& request, ClientCallbacks& callbacks) PURE;
 
   /**
    * Initialize the connection. Issue the auth command and readonly command as needed.

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

@@ -7,6 +7,8 @@ namespace Common {
 namespace Redis {
 namespace Client {
 namespace {
+// null_pool_callbacks is used for requests that must be filtered and not redirected such as
+// "asking".
 Common::Redis::Client::DoNothingPoolCallbacks null_pool_callbacks;
 } // namespace
 
@@ -97,7 +99,7 @@ void ClientImpl::flushBufferAndResetTimer() {
   connection_->write(encoder_buffer_, false);
 }
 
-PoolRequest* ClientImpl::makeRequest(const RespValue& request, PoolCallbacks& callbacks) {
+PoolRequest* ClientImpl::makeRequest(const RespValue& request, ClientCallbacks& callbacks) {
   ASSERT(connection_->state() == Network::Connection::State::Open);
 
   const bool empty_buffer = encoder_buffer_.length() == 0;
@@ -212,7 +214,7 @@ void ClientImpl::onRespValue(RespValuePtr&& value) {
   }
   request.aggregate_request_timer_->complete();
 
-  PoolCallbacks& callbacks = request.callbacks_;
+  ClientCallbacks& callbacks = request.callbacks_;
 
   // We need to ensure the request is popped before calling the callback, since the callback might
   // result in closing the connection.
@@ -223,9 +225,13 @@ void ClientImpl::onRespValue(RespValuePtr&& value) {
     std::vector<absl::string_view> err = StringUtil::splitToken(value->asString(), " ", false);
     bool redirected = false;
     if (err.size() == 3) {
+      // MOVED and ASK redirection errors have the following substrings: MOVED or ASK (err[0]), hash
+      // key slot (err[1]), and IP address and TCP port separated by a colon (err[2])
       if (err[0] == RedirectionResponse::get().MOVED || err[0] == RedirectionResponse::get().ASK) {
-        redirected = callbacks.onRedirection(*value);
-        if (redirected) {
+        redirected = true;
+        bool redirect_succeeded = callbacks.onRedirection(std::move(value), std::string(err[2]),
+                                                          err[0] == RedirectionResponse::get().ASK);
+        if (redirect_succeeded) {
           host_->cluster().stats().upstream_internal_redirect_succeeded_total_.inc();
         } else {
           host_->cluster().stats().upstream_internal_redirect_failed_total_.inc();
@@ -251,7 +257,7 @@ void ClientImpl::onRespValue(RespValuePtr&& value) {
   putOutlierEvent(Upstream::Outlier::Result::ExtOriginRequestSuccess);
 }
 
-ClientImpl::PendingRequest::PendingRequest(ClientImpl& parent, PoolCallbacks& callbacks,
+ClientImpl::PendingRequest::PendingRequest(ClientImpl& parent, ClientCallbacks& callbacks,
                                            Stats::StatName command)
     : parent_(parent), callbacks_(callbacks), command_{command},
       aggregate_request_timer_(parent_.redis_command_stats_->createAggregateTimer(

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

@@ -83,7 +83,7 @@ class ClientImpl : public Client, public DecoderCallbacks, public Network::Conne
     connection_->addConnectionCallbacks(callbacks);
   }
   void close() override;
-  PoolRequest* makeRequest(const RespValue& request, PoolCallbacks& callbacks) override;
+  PoolRequest* makeRequest(const RespValue& request, ClientCallbacks& callbacks) override;
   bool active() override { return !pending_requests_.empty(); }
   void flushBufferAndResetTimer();
   void initialize(const std::string& auth_password) override;
@@ -104,14 +104,14 @@ class ClientImpl : public Client, public DecoderCallbacks, public Network::Conne
   };
 
   struct PendingRequest : public PoolRequest {
-    PendingRequest(ClientImpl& parent, PoolCallbacks& callbacks, Stats::StatName stat_name);
+    PendingRequest(ClientImpl& parent, ClientCallbacks& callbacks, Stats::StatName stat_name);
     ~PendingRequest() override;
 
     // PoolRequest
     void cancel() override;
 
     ClientImpl& parent_;
-    PoolCallbacks& callbacks_;
+    ClientCallbacks& callbacks_;
     Stats::StatName command_;
     bool canceled_{};
     Stats::TimespanPtr aggregate_request_timer_;

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

@@ -19,6 +19,13 @@ AuthRequest::AuthRequest(const std::string& password) {
   asArray().swap(values);
 }
 
+RespValuePtr makeError(const std::string& error) {
+  Common::Redis::RespValuePtr response(new RespValue());
+  response->type(Common::Redis::RespType::Error);
+  response->asString() = error;
+  return response;
+}
+
 ReadOnlyRequest::ReadOnlyRequest() {
   std::vector<RespValue> values(1);
   values[0].type(RespType::BulkString);

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

@@ -16,6 +16,8 @@ class AuthRequest : public Redis::RespValue {
   AuthRequest(const std::string& password);
 };
 
+RespValuePtr makeError(const std::string& error);
+
 class ReadOnlyRequest : public Redis::RespValue {
 public:
   ReadOnlyRequest();