Make SLOW_COMMAND_TIMEOUT a configuration option (#292)

Author: learoyklinginsmith

README.md

@@ -26,6 +26,7 @@ gem 'redis-cluster-client'
 | `:replica` | Boolean | `false` | `true` if client should use scale read feature |
 | `:replica_affinity` | Symbol or String | `:random` | scale reading strategy, `:random`, `random_with_primary` or `:latency` are valid |
 | `:fixed_hostname` | String | `nil` | required if client should connect to single endpoint with SSL |
+| `:slow_command_timeout` | Integer | `-1` | timeout used for "slow" queries that fetch metdata e.g. CLUSTER NODES, COMMAND |
 | `:concurrency` | Hash | `{ model: :on_demand, size: 5}` | concurrency settings, `:on_demand`, `:pooled` and `:none` are valid models, size is a max number of workers, `:none` model is no concurrency, Please choose the one suited your environment if needed. |
 
 Also, [the other generic options](https://github.com/redis-rb/redis-client#configuration) can be passed.
@@ -90,6 +91,11 @@ RedisClient.cluster(nodes: 'rediss://endpoint.example.com:6379').new_client
 RedisClient.cluster(nodes: 'rediss://endpoint.example.com:6379', fixed_hostname: 'endpoint.example.com').new_client
 ```
 
+```ruby
+# To specify a timeout for "slow" commands (CLUSTER NODES, COMMAND)
+RedisClient.cluster(slow_command_timeout: 4).new_client
+```
+
 ```ruby
 # To specify concurrency settings
 RedisClient.cluster(concurrency: { model: :on_demand, size: 6 }).new_client

lib/redis_client/cluster/command.rb

@@ -7,8 +7,6 @@
 class RedisClient
   class Cluster
     class Command
-      SLOW_COMMAND_TIMEOUT = Float(ENV.fetch('REDIS_CLIENT_SLOW_COMMAND_TIMEOUT', -1))
-
       EMPTY_STRING = ''
       LEFT_BRACKET = '{'
       RIGHT_BRACKET = '}'
@@ -23,12 +21,12 @@ class Command
       )
 
       class << self
-        def load(nodes)
+        def load(nodes, slow_command_timeout: -1)
           cmd = errors = nil
 
           nodes&.each do |node|
             regular_timeout = node.read_timeout
-            node.read_timeout = SLOW_COMMAND_TIMEOUT > 0.0 ? SLOW_COMMAND_TIMEOUT : regular_timeout
+            node.read_timeout = slow_command_timeout > 0.0 ? slow_command_timeout : regular_timeout
             reply = node.call('COMMAND')
             node.read_timeout = regular_timeout
             commands = parse_command_reply(reply)

lib/redis_client/cluster/node.rb

@@ -16,9 +16,6 @@ class Node
       # It affects to strike a balance between load and stability in initialization or changed states.
       MAX_STARTUP_SAMPLE = Integer(ENV.fetch('REDIS_CLIENT_MAX_STARTUP_SAMPLE', 3))
 
-      # It's used with slow queries of fetching meta data like CLUSTER NODES, COMMAND and so on.
-      SLOW_COMMAND_TIMEOUT = Float(ENV.fetch('REDIS_CLIENT_SLOW_COMMAND_TIMEOUT', -1))
-
       # less memory consumption, but slow
       USE_CHAR_ARRAY_SLOT = Integer(ENV.fetch('REDIS_CLIENT_USE_CHAR_ARRAY_SLOT', 1)) == 1
 
@@ -96,7 +93,7 @@ def build_connection_prelude
       end
 
       class << self
-        def load_info(options, concurrent_worker, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
+        def load_info(options, concurrent_worker, slow_command_timeout: -1, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
           raise ::RedisClient::Cluster::InitialSetupError, [] if options.nil? || options.empty?
 
           startup_size = options.size > MAX_STARTUP_SAMPLE ? MAX_STARTUP_SAMPLE : options.size
@@ -107,7 +104,7 @@ def load_info(options, concurrent_worker, **kwargs) # rubocop:disable Metrics/Ab
           startup_nodes.each_with_index do |raw_client, i|
             work_group.push(i, raw_client) do |client|
               regular_timeout = client.read_timeout
-              client.read_timeout = SLOW_COMMAND_TIMEOUT > 0.0 ? SLOW_COMMAND_TIMEOUT : regular_timeout
+              client.read_timeout = slow_command_timeout > 0.0 ? slow_command_timeout : regular_timeout
               reply = client.call('CLUSTER', 'NODES')
               client.read_timeout = regular_timeout
               parse_cluster_node_reply(reply)

lib/redis_client/cluster/router.rb

@@ -22,7 +22,7 @@ def initialize(config, concurrent_worker, pool: nil, **kwargs)
         @pool = pool
         @client_kwargs = kwargs
         @node = fetch_cluster_info(@config, @concurrent_worker, pool: @pool, **@client_kwargs)
-        @command = ::RedisClient::Cluster::Command.load(@node.replica_clients.shuffle)
+        @command = ::RedisClient::Cluster::Command.load(@node.replica_clients.shuffle, slow_command_timeout: config.slow_command_timeout)
         @mutex = Mutex.new
         @command_builder = @config.command_builder
       end
@@ -305,7 +305,7 @@ def send_pubsub_command(method, command, args, &block) # rubocop:disable Metrics
       end
 
       def fetch_cluster_info(config, concurrent_worker, pool: nil, **kwargs)
-        node_info_list = ::RedisClient::Cluster::Node.load_info(config.per_node_key, concurrent_worker, **kwargs)
+        node_info_list = ::RedisClient::Cluster::Node.load_info(config.per_node_key, concurrent_worker, slow_command_timeout: config.slow_command_timeout, **kwargs)
         node_addrs = node_info_list.map { |i| ::RedisClient::Cluster::NodeKey.hashify(i.node_key) }
         config.update_node(node_addrs)
         ::RedisClient::Cluster::Node.new(

lib/redis_client/cluster_config.rb

@@ -18,10 +18,12 @@ class ClusterConfig
     MERGE_CONFIG_KEYS = %i[ssl username password].freeze
     IGNORE_GENERIC_CONFIG_KEYS = %i[url host port path].freeze
     MAX_WORKERS = Integer(ENV.fetch('REDIS_CLIENT_MAX_THREADS', 5))
+    # It's used with slow queries of fetching meta data like CLUSTER NODES, COMMAND and so on.
+    SLOW_COMMAND_TIMEOUT = Float(ENV.fetch('REDIS_CLIENT_SLOW_COMMAND_TIMEOUT', -1))
 
     InvalidClientConfigError = Class.new(::RedisClient::Error)
 
-    attr_reader :command_builder, :client_config, :replica_affinity
+    attr_reader :command_builder, :client_config, :replica_affinity, :slow_command_timeout
 
     def initialize(
       nodes: DEFAULT_NODES,
@@ -30,6 +32,7 @@ def initialize(
       fixed_hostname: '',
       concurrency: nil,
       client_implementation: ::RedisClient::Cluster, # for redis gem
+      slow_command_timeout: SLOW_COMMAND_TIMEOUT,
       **client_config
     )
 
@@ -42,6 +45,7 @@ def initialize(
       @client_config = merge_generic_config(client_config, @node_configs)
       @concurrency = merge_concurrency_option(concurrency)
       @client_implementation = client_implementation
+      @slow_command_timeout = slow_command_timeout
       @mutex = Mutex.new
     end
 
@@ -53,6 +57,7 @@ def dup
         fixed_hostname: @fixed_hostname,
         concurrency: @concurrency,
         client_implementation: @client_implementation,
+        slow_command_timeout: @slow_command_timeout,
         **@client_config
       )
     end

test/redis_client/cluster/test_command.rb

@@ -31,6 +31,20 @@ def test_load
         end
       end
 
+      def test_load_slow_timeout
+        nodes = @raw_clients
+        assert_equal(TEST_TIMEOUT_SEC, nodes.first.read_timeout)
+        nodes.first.singleton_class.prepend(Module.new do
+          def call(...)
+            @slow_timeout = read_timeout
+            super
+          end
+        end)
+        ::RedisClient::Cluster::Command.load(nodes, slow_command_timeout: 9)
+        assert_equal(9, nodes.first.instance_variable_get(:@slow_timeout))
+        assert_equal(TEST_TIMEOUT_SEC, nodes.first.read_timeout)
+      end
+
       def test_parse_command_reply
         [
           {