Redis health check is not sending the auth command on it's connection

docs/root/intro/version_history.rst

@@ -39,6 +39,7 @@ Version history
 * performance: stats symbol table implementation (disabled by default; to test it, add "--use-fake-symbol-table 0" to the command-line arguments when starting Envoy).
 * rbac: added support for DNS SAN as :ref:`principal_name <envoy_api_field_config.rbac.v2.Principal.Authenticated.principal_name>`.
 * 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.
 * 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

include/envoy/server/health_checker_config.h

@@ -43,6 +43,11 @@ class HealthCheckerFactoryContext {
    *         messages.
    */
   virtual ProtobufMessage::ValidationVisitor& messageValidationVisitor() PURE;
+
+  /**
+   * @return Api::Api& the API used by the server.
+   */
+  virtual Api::Api& api() PURE;
 };
 
 /**

source/common/upstream/cluster_factory_impl.cc

@@ -109,7 +109,8 @@ ClusterFactoryImplBase::create(const envoy::api::v2::Cluster& cluster,
     } else {
       new_cluster_pair.first->setHealthChecker(HealthCheckerFactory::create(
           cluster.health_checks()[0], *new_cluster_pair.first, context.runtime(), context.random(),
-          context.dispatcher(), context.logManager(), context.messageValidationVisitor()));
+          context.dispatcher(), context.logManager(), context.messageValidationVisitor(),
+          context.api()));
     }
   }
 

source/common/upstream/health_checker_impl.cc

@@ -29,9 +29,11 @@ class HealthCheckerFactoryContextImpl : public Server::Configuration::HealthChec
                                   Envoy::Runtime::RandomGenerator& random,
                                   Event::Dispatcher& dispatcher,
                                   HealthCheckEventLoggerPtr&& event_logger,
-                                  ProtobufMessage::ValidationVisitor& validation_visitor)
+                                  ProtobufMessage::ValidationVisitor& validation_visitor,
+                                  Api::Api& api)
       : cluster_(cluster), runtime_(runtime), random_(random), dispatcher_(dispatcher),
-        event_logger_(std::move(event_logger)), validation_visitor_(validation_visitor) {}
+        event_logger_(std::move(event_logger)), validation_visitor_(validation_visitor), api_(api) {
+  }
   Upstream::Cluster& cluster() override { return cluster_; }
   Envoy::Runtime::Loader& runtime() override { return runtime_; }
   Envoy::Runtime::RandomGenerator& random() override { return random_; }
@@ -40,6 +42,7 @@ class HealthCheckerFactoryContextImpl : public Server::Configuration::HealthChec
   ProtobufMessage::ValidationVisitor& messageValidationVisitor() override {
     return validation_visitor_;
   }
+  Api::Api& api() override { return api_; }
 
 private:
   Upstream::Cluster& cluster_;
@@ -48,14 +51,14 @@ class HealthCheckerFactoryContextImpl : public Server::Configuration::HealthChec
   Event::Dispatcher& dispatcher_;
   HealthCheckEventLoggerPtr event_logger_;
   ProtobufMessage::ValidationVisitor& validation_visitor_;
+  Api::Api& api_;
 };
 
-HealthCheckerSharedPtr
-HealthCheckerFactory::create(const envoy::api::v2::core::HealthCheck& health_check_config,
-                             Upstream::Cluster& cluster, Runtime::Loader& runtime,
-                             Runtime::RandomGenerator& random, Event::Dispatcher& dispatcher,
-                             AccessLog::AccessLogManager& log_manager,
-                             ProtobufMessage::ValidationVisitor& validation_visitor) {
+HealthCheckerSharedPtr HealthCheckerFactory::create(
+    const envoy::api::v2::core::HealthCheck& health_check_config, Upstream::Cluster& cluster,
+    Runtime::Loader& runtime, Runtime::RandomGenerator& random, Event::Dispatcher& dispatcher,
+    AccessLog::AccessLogManager& log_manager,
+    ProtobufMessage::ValidationVisitor& validation_visitor, Api::Api& api) {
   HealthCheckEventLoggerPtr event_logger;
   if (!health_check_config.event_log_path().empty()) {
     event_logger = std::make_unique<HealthCheckEventLoggerImpl>(
@@ -81,7 +84,7 @@ HealthCheckerFactory::create(const envoy::api::v2::core::HealthCheck& health_che
             health_check_config.custom_health_check().name());
     std::unique_ptr<Server::Configuration::HealthCheckerFactoryContext> context(
         new HealthCheckerFactoryContextImpl(cluster, runtime, random, dispatcher,
-                                            std::move(event_logger), validation_visitor));
+                                            std::move(event_logger), validation_visitor, api));
     return factory.createCustomHealthChecker(health_check_config, *context);
   }
   default:

source/common/upstream/health_checker_impl.h

@@ -11,6 +11,7 @@
 #include "common/stream_info/stream_info_impl.h"
 #include "common/upstream/health_checker_base_impl.h"
 
+#include "include/envoy/api/_virtual_includes/api_interface/envoy/api/api.h"
 #include "src/proto/grpc/health/v1/health.pb.h"
 
 namespace Envoy {
@@ -32,12 +33,11 @@ class HealthCheckerFactory : public Logger::Loggable<Logger::Id::health_checker>
    * @param validation_visitor message validation visitor instance.
    * @return a health checker.
    */
-  static HealthCheckerSharedPtr create(const envoy::api::v2::core::HealthCheck& health_check_config,
-                                       Upstream::Cluster& cluster, Runtime::Loader& runtime,
-                                       Runtime::RandomGenerator& random,
-                                       Event::Dispatcher& dispatcher,
-                                       AccessLog::AccessLogManager& log_manager,
-                                       ProtobufMessage::ValidationVisitor& validation_visitor);
+  static HealthCheckerSharedPtr
+  create(const envoy::api::v2::core::HealthCheck& health_check_config, Upstream::Cluster& cluster,
+         Runtime::Loader& runtime, Runtime::RandomGenerator& random, Event::Dispatcher& dispatcher,
+         AccessLog::AccessLogManager& log_manager,
+         ProtobufMessage::ValidationVisitor& validation_visitor, Api::Api& api);
 };
 
 /**

source/common/upstream/health_discovery_service.cc

@@ -152,7 +152,8 @@ void HdsDelegate::processMessage(
                                               info_factory_, cm_, local_info_, dispatcher_, random_,
                                               singleton_manager_, tls_, validation_visitor_, api_));
 
-    hds_clusters_.back()->startHealthchecks(access_log_manager_, runtime_, random_, dispatcher_);
+    hds_clusters_.back()->startHealthchecks(access_log_manager_, runtime_, random_, dispatcher_,
+                                            api_);
   }
 }
 
@@ -243,10 +244,11 @@ ProdClusterInfoFactory::createClusterInfo(const CreateClusterInfoParams& params)
 
 void HdsCluster::startHealthchecks(AccessLog::AccessLogManager& access_log_manager,
                                    Runtime::Loader& runtime, Runtime::RandomGenerator& random,
-                                   Event::Dispatcher& dispatcher) {
+                                   Event::Dispatcher& dispatcher, Api::Api& api) {
   for (auto& health_check : cluster_.health_checks()) {
-    health_checkers_.push_back(Upstream::HealthCheckerFactory::create(
-        health_check, *this, runtime, random, dispatcher, access_log_manager, validation_visitor_));
+    health_checkers_.push_back(
+        Upstream::HealthCheckerFactory::create(health_check, *this, runtime, random, dispatcher,
+                                               access_log_manager, validation_visitor_, api));
     health_checkers_.back()->start();
   }
 }

source/common/upstream/health_discovery_service.h

@@ -59,7 +59,8 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
 
   // Creates and starts healthcheckers to its endpoints
   void startHealthchecks(AccessLog::AccessLogManager& access_log_manager, Runtime::Loader& runtime,
-                         Runtime::RandomGenerator& random, Event::Dispatcher& dispatcher);
+                         Runtime::RandomGenerator& random, Event::Dispatcher& dispatcher,
+                         Api::Api& api);
 
   std::vector<Upstream::HealthCheckerSharedPtr> healthCheckers() { return health_checkers_; };
 

source/extensions/clusters/redis/redis_cluster.cc

@@ -249,16 +249,10 @@ void RedisCluster::RedisDiscoverySession::startResolveRedis() {
   if (!client) {
     client = std::make_unique<RedisDiscoveryClient>(*this);
     client->host_ = current_host_address_;
-    client->client_ = client_factory_.create(host, dispatcher_, *this);
-    client->client_->addConnectionCallbacks(*client);
     std::string auth_password =
         Envoy::Config::DataSource::read(parent_.auth_password_datasource_, true, parent_.api_);
-    if (!auth_password.empty()) {
-      // Send an AUTH command to the upstream server.
-      client->client_->makeRequest(
-          Extensions::NetworkFilters::Common::Redis::Utility::makeAuthCommand(auth_password),
-          null_pool_callbacks);
-    }
+    client->client_ = client_factory_.create(host, dispatcher_, *this, auth_password);
+    client->client_->addConnectionCallbacks(*client);
   }
 
   current_request_ = client->client_->makeRequest(ClusterSlotsRequest::instance_, *this);

source/extensions/clusters/redis/redis_cluster.h

@@ -214,10 +214,10 @@ 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; }
-    // 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.
+    // This is effectively not in used for making the "Cluster Slots" calls, using Master to avoid
+    // the READONLY command when establishing connection.
     Extensions::NetworkFilters::Common::Redis::Client::ReadPolicy readPolicy() const override {
-      return Extensions::NetworkFilters::Common::Redis::Client::ReadPolicy::Any;
+      return Extensions::NetworkFilters::Common::Redis::Client::ReadPolicy::Master;
     }
 
     // Extensions::NetworkFilters::Common::Redis::Client::PoolCallbacks

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

@@ -57,6 +57,7 @@ envoy_cc_library(
     deps = [
         ":client_interface",
         ":codec_lib",
+        ":utility_lib",
         "//include/envoy/router:router_interface",
         "//include/envoy/thread_local:thread_local_interface",
         "//include/envoy/upstream:cluster_manager_interface",

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

@@ -95,6 +95,12 @@ class Client : public Event::DeferredDeletable {
    *         for some reason.
    */
   virtual PoolRequest* makeRequest(const RespValue& request, PoolCallbacks& callbacks) PURE;
+
+  /**
+   * Initialize the connection. Issue the auth command and readonly command as needed.
+   * @param auth password for upstream host.
+   */
+  virtual void initialize(const std::string& auth_password) PURE;
 };
 
 using ClientPtr = std::unique_ptr<Client>;
@@ -181,10 +187,11 @@ class ClientFactory {
    * @param host supplies the upstream host.
    * @param dispatcher supplies the owning thread's dispatcher.
    * @param config supplies the connection pool configuration.
+   * @param auth password for upstream host.
    * @return ClientPtr a new connection pool client.
    */
   virtual ClientPtr create(Upstream::HostConstSharedPtr host, Event::Dispatcher& dispatcher,
-                           const Config& config) PURE;
+                           const Config& config, const std::string& auth_password) PURE;
 };
 
 } // namespace Client

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

@@ -6,6 +6,9 @@ namespace NetworkFilters {
 namespace Common {
 namespace Redis {
 namespace Client {
+namespace {
+Common::Redis::Client::DoNothingPoolCallbacks null_pool_callbacks;
+} // namespace
 
 ConfigImpl::ConfigImpl(
     const envoy::config::filter::network::redis_proxy::v2::RedisProxy::ConnPoolSettings& config)
@@ -49,7 +52,6 @@ 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));
   client->connection_ = host->createConnection(dispatcher, nullptr, nullptr).connection_;
@@ -245,12 +247,28 @@ void ClientImpl::PendingRequest::cancel() {
   canceled_ = true;
 }
 
+void ClientImpl::initialize(const std::string& auth_password) {
+  if (!auth_password.empty()) {
+    // Send an AUTH command to the upstream server.
+    makeRequest(Utility::makeAuthCommand(auth_password), null_pool_callbacks);
+  }
+  // Any connection to replica requires the READONLY command in order to perform read.
+  // Also the READONLY command is a no-opt for the master.
+  // We only need to send the READONLY command iff it's possible that the host is a replica.
+  if (config_.readPolicy() != Common::Redis::Client::ReadPolicy::Master) {
+    makeRequest(Utility::ReadOnlyRequest::instance(), null_pool_callbacks);
+  }
+}
+
 ClientFactoryImpl ClientFactoryImpl::instance_;
 
 ClientPtr ClientFactoryImpl::create(Upstream::HostConstSharedPtr host,
-                                    Event::Dispatcher& dispatcher, const Config& config) {
-  return ClientImpl::create(host, dispatcher, EncoderPtr{new EncoderImpl()}, decoder_factory_,
-                            config);
+                                    Event::Dispatcher& dispatcher, const Config& config,
+                                    const std::string& auth_password) {
+  ClientPtr client =
+      ClientImpl::create(host, dispatcher, EncoderPtr{new EncoderImpl()}, decoder_factory_, config);
+  client->initialize(auth_password);
+  return client;
 }
 
 } // namespace Client

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

@@ -15,6 +15,7 @@
 #include "common/upstream/upstream_impl.h"
 
 #include "extensions/filters/network/common/redis/client.h"
+#include "extensions/filters/network/common/redis/utility.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -77,6 +78,7 @@ class ClientImpl : public Client, public DecoderCallbacks, public Network::Conne
   PoolRequest* makeRequest(const RespValue& request, PoolCallbacks& callbacks) override;
   bool active() override { return !pending_requests_.empty(); }
   void flushBufferAndResetTimer();
+  void initialize(const std::string& auth_password) override;
 
 private:
   friend class RedisClientImplTest;
@@ -135,7 +137,7 @@ class ClientFactoryImpl : public ClientFactory {
 public:
   // RedisProxy::ConnPool::ClientFactoryImpl
   ClientPtr create(Upstream::HostConstSharedPtr host, Event::Dispatcher& dispatcher,
-                   const Config& config) override;
+                   const Config& config, const std::string& auth_password) override;
 
   static ClientFactoryImpl instance_;
 

source/extensions/filters/network/redis_proxy/BUILD

@@ -116,6 +116,7 @@ envoy_cc_library(
     hdrs = ["config.h"],
     deps = [
         "//include/envoy/registry",
+        "//include/envoy/upstream:upstream_interface",
         "//source/common/config:filter_json_lib",
         "//source/extensions/filters/network:well_known_names",
         "//source/extensions/filters/network/common:factory_base_lib",

source/extensions/filters/network/redis_proxy/config.h

@@ -2,9 +2,12 @@
 
 #include <string>
 
+#include "envoy/api/api.h"
 #include "envoy/config/filter/network/redis_proxy/v2/redis_proxy.pb.h"
 #include "envoy/config/filter/network/redis_proxy/v2/redis_proxy.pb.validate.h"
+#include "envoy/upstream/upstream.h"
 
+#include "common/common/empty_string.h"
 #include "common/config/datasource.h"
 
 #include "extensions/filters/network/common/factory_base.h"
@@ -29,6 +32,15 @@ class ProtocolOptionsConfigImpl : public Upstream::ProtocolOptionsConfig {
     return auth_password_;
   }
 
+  static const std::string auth_password(Upstream::ClusterInfoConstSharedPtr info, Api::Api& api) {
+    auto options = info->extensionProtocolOptionsTyped<ProtocolOptionsConfigImpl>(
+        NetworkFilterNames::get().RedisProxy);
+    if (options) {
+      return options->auth_password(api);
+    }
+    return EMPTY_STRING;
+  }
+
 private:
   envoy::api::v2::core::DataSource auth_password_;
 };

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

@@ -15,9 +15,6 @@ namespace Extensions {
 namespace NetworkFilters {
 namespace RedisProxy {
 namespace ConnPool {
-namespace {
-Common::Redis::Client::DoNothingPoolCallbacks null_pool_callbacks;
-} // namespace
 
 InstanceImpl::InstanceImpl(
     const std::string& cluster_name, Upstream::ClusterManager& cm,
@@ -54,11 +51,7 @@ InstanceImpl::ThreadLocalPool::ThreadLocalPool(InstanceImpl& parent, Event::Disp
   cluster_update_handle_ = parent_.cm_.addThreadLocalClusterUpdateCallbacks(*this);
   Upstream::ThreadLocalCluster* cluster = parent_.cm_.get(cluster_name_);
   if (cluster != nullptr) {
-    auto options = cluster->info()->extensionProtocolOptionsTyped<ProtocolOptionsConfigImpl>(
-        NetworkFilterNames::get().RedisProxy);
-    if (options) {
-      auth_password_ = options->auth_password(parent_.api_);
-    }
+    auth_password_ = ProtocolOptionsConfigImpl::auth_password(cluster->info(), parent_.api_);
     onClusterAddOrUpdateNonVirtual(*cluster);
   }
 }
@@ -195,21 +188,9 @@ InstanceImpl::ThreadLocalPool::threadLocalActiveClient(Upstream::HostConstShared
   if (!client) {
     client = std::make_unique<ThreadLocalActiveClient>(*this);
     client->host_ = host;
-    client->redis_client_ = parent_.client_factory_.create(host, dispatcher_, parent_.config_);
+    client->redis_client_ =
+        parent_.client_factory_.create(host, dispatcher_, parent_.config_, auth_password_);
     client->redis_client_->addConnectionCallbacks(*client);
-    // TODO(hyang): should the auth command and readonly command be moved to the factory method?
-    if (!auth_password_.empty()) {
-      // Send an AUTH command to the upstream server.
-      client->redis_client_->makeRequest(Common::Redis::Utility::makeAuthCommand(auth_password_),
-                                         null_pool_callbacks);
-    }
-    // Any connection to replica requires the READONLY command in order to perform read.
-    // Also the READONLY command is a no-opt for the master.
-    // We only need to send the READONLY command iff it's possible that the host is a replica.
-    if (parent_.config_.readPolicy() != Common::Redis::Client::ReadPolicy::Master) {
-      client->redis_client_->makeRequest(Common::Redis::Utility::ReadOnlyRequest::instance(),
-                                         null_pool_callbacks);
-    }
   }
   return client;
 }

source/extensions/health_checkers/redis/BUILD

@@ -17,6 +17,7 @@ envoy_cc_library(
     deps = [
         "//source/common/upstream:health_checker_base_lib",
         "//source/extensions/filters/network/common/redis:client_lib",
+        "//source/extensions/filters/network/redis_proxy:config",
         "//source/extensions/filters/network/redis_proxy:conn_pool_lib",
         "@envoy_api//envoy/api/v2/core:health_check_cc",
         "@envoy_api//envoy/config/health_checker/redis/v2:redis_cc",

source/extensions/health_checkers/redis/config.cc

@@ -17,7 +17,7 @@ Upstream::HealthCheckerSharedPtr RedisHealthCheckerFactory::createCustomHealthCh
   return std::make_shared<RedisHealthChecker>(
       context.cluster(), config,
       getRedisHealthCheckConfig(config, context.messageValidationVisitor()), context.dispatcher(),
-      context.runtime(), context.random(), context.eventLogger(),
+      context.runtime(), context.random(), context.eventLogger(), context.api(),
       NetworkFilters::Common::Redis::Client::ClientFactoryImpl::instance_);
 };