Added dynamic_startup_nodes configuration to RedisCluster. (#2244)

* Added dynamic_startup_nodes configuration to RedisCluster. By default, uses only the initial passed startup nodes to refresh the cluster topology. If set to true, sets the startup nodes to all of the discovered nodes.

* Added RedisCluster specific options to the README file and updated CHANGES

* Fixed timeout_error_topology_refresh test

Co-authored-by: dvora-h <[email protected]>

Author: barshaul

CHANGES

@@ -11,9 +11,9 @@
     * Fix broken connection writer lock-up for asyncio (#2065)
     * Fix auth bug when provided with no username (#2086)
     * Fix missing ClusterPipeline._lock (#2189)
+    * Added dynaminc_startup_nodes configuration to RedisCluster
     * Fix reusing the old nodes' connections when cluster topology refresh is being done
     * Fix RedisCluster to immediately raise AuthenticationError without a retry
-
 * 4.1.3 (Feb 8, 2022)
     * Fix flushdb and flushall (#1926)
     * Add redis5 and redis4 dockers (#1871)

README.md

@@ -1006,6 +1006,7 @@ a slots cache which maps each of the 16384 slots to the node/s handling them,
 a nodes cache that contains ClusterNode objects (name, host, port, redis connection)
 for all of the cluster's nodes, and a commands cache contains all the server
 supported commands that were retrieved using the Redis 'COMMAND' output.
+See *RedisCluster specific options* below for more.
 
 RedisCluster instance can be directly used to execute Redis commands. When a
 command is being executed through the cluster instance, the target node(s) will
@@ -1245,6 +1246,55 @@ The following commands are not supported:
 
 Using scripting within pipelines in cluster mode is **not supported**.
 
+
+**RedisCluster specific options**
+
+ require_full_coverage: (default=False)
+ 
+    When set to False (default value): the client will not require a
+    full coverage of the slots. However, if not all slots are covered,
+    and at least one node has 'cluster-require-full-coverage' set to
+    'yes,' the server will throw a ClusterDownError for some key-based
+    commands. See -
+    https://redis.io/topics/cluster-tutorial#redis-cluster-configuration-parameters
+    When set to True: all slots must be covered to construct the
+    cluster client. If not all slots are covered, RedisClusterException
+    will be thrown.
+    
+ read_from_replicas: (default=False)
+
+     Enable read from replicas in READONLY mode. You can read possibly
+     stale data.
+     When set to true, read commands will be assigned between the
+     primary and its replications in a Round-Robin manner.
+     
+ dynamic_startup_nodes: (default=False)
+
+     Set the RedisCluster's startup nodes to all of the discovered nodes.
+     If true, the cluster's discovered nodes will be used to determine the
+     cluster nodes-slots mapping in the next topology refresh.
+     It will remove the initial passed startup nodes if their endpoints aren't
+     listed in the CLUSTER SLOTS output.
+     If you use dynamic DNS endpoints for startup nodes but CLUSTER SLOTS lists
+     specific IP addresses, keep it at false.
+     
+ cluster_error_retry_attempts: (default=3)
+
+     Retry command execution attempts when encountering ClusterDownError
+     or ConnectionError
+     
+ reinitialize_steps: (default=10)
+
+    Specifies the number of MOVED errors that need to occur before
+    reinitializing the whole cluster topology. If a MOVED error occurs
+    and the cluster does not need to be reinitialized on this current
+    error handling, only the MOVED slot will be patched with the
+    redirected node.
+    To reinitialize the cluster on every MOVED error, set
+    reinitialize_steps to 1.
+    To avoid reinitializing the cluster on moved errors, set
+    reinitialize_steps to 0.
+
 ### Author
 
 redis-py is developed and maintained by [Redis Inc](https://redis.com). It can be found [here](

redis/cluster.py

@@ -482,6 +482,7 @@ def __init__(
         require_full_coverage=False,
         reinitialize_steps=10,
         read_from_replicas=False,
+        dynamic_startup_nodes=False,
         url=None,
         **kwargs,
     ):
@@ -509,6 +510,14 @@ def __init__(
              stale data.
              When set to true, read commands will be assigned between the
              primary and its replications in a Round-Robin manner.
+         :dynamic_startup_nodes: 'bool'
+             Set the RedisCluster's startup nodes to all of the discovered nodes.
+             If true, the cluster's discovered nodes will be used to determine the
+             cluster nodes-slots mapping in the next topology refresh.
+             It will remove the initial passed startup nodes if their endpoints aren't
+             listed in the CLUSTER SLOTS output.
+             If you use dynamic DNS endpoints for startup nodes but CLUSTER SLOTS lists
+             specific IP addresses, keep it at false.
         :cluster_error_retry_attempts: 'int'
              Retry command execution attempts when encountering ClusterDownError
              or ConnectionError
@@ -598,6 +607,7 @@ def __init__(
             startup_nodes=startup_nodes,
             from_url=from_url,
             require_full_coverage=require_full_coverage,
+            dynamic_startup_nodes=dynamic_startup_nodes,
             **kwargs,
         )
 
@@ -1283,6 +1293,7 @@ def __init__(
         from_url=False,
         require_full_coverage=False,
         lock=None,
+        dynamic_startup_nodes=False,
         **kwargs,
     ):
         self.nodes_cache = {}
@@ -1292,6 +1303,7 @@ def __init__(
         self.populate_startup_nodes(startup_nodes)
         self.from_url = from_url
         self._require_full_coverage = require_full_coverage
+        self._dynamic_startup_nodes = dynamic_startup_nodes
         self._moved_exception = None
         self.connection_kwargs = kwargs
         self.read_load_balancer = LoadBalancer()
@@ -1612,8 +1624,9 @@ def initialize(self):
         self.slots_cache = tmp_slots
         # Set the default node
         self.default_node = self.get_nodes_by_server_type(PRIMARY)[0]
-        # Populate the startup nodes with all discovered nodes
-        self.startup_nodes = tmp_nodes_cache
+        if self._dynamic_startup_nodes:
+            # Populate the startup nodes with all discovered nodes
+            self.startup_nodes = tmp_nodes_cache
         # If initialize was called after a MovedError, clear it
         self._moved_exception = None
 

tests/test_cluster.py

@@ -673,7 +673,7 @@ def __init__(self, val=0):
 
             def moved_redirect_effect(connection, *args, **options):
                 # raise a timeout for 5 times so we'll need to reinitilize the topology
-                if count.val >= 5:
+                if count.val == 4:
                     parse_response.side_effect = real_func
                 count.val += 1
                 raise TimeoutError()
@@ -2285,6 +2285,27 @@ def cmd_init_mock(self, r):
                 assert rc.get_node(host=default_host, port=7001) is not None
                 assert rc.get_node(host=default_host, port=7002) is not None
 
+    @pytest.mark.parametrize("dynamic_startup_nodes", [True, False])
+    def test_init_slots_dynamic_startup_nodes(self, dynamic_startup_nodes):
+        rc = get_mocked_redis_client(
+            host="[email protected]",
+            port=7000,
+            cluster_slots=default_cluster_slots,
+            dynamic_startup_nodes=dynamic_startup_nodes,
+        )
+        # Nodes are taken from default_cluster_slots
+        discovered_nodes = [
+            "127.0.0.1:7000",
+            "127.0.0.1:7001",
+            "127.0.0.1:7002",
+            "127.0.0.1:7003",
+        ]
+        startup_nodes = list(rc.nodes_manager.startup_nodes.keys())
+        if dynamic_startup_nodes is True:
+            assert startup_nodes.sort() == discovered_nodes.sort()
+        else:
+            assert startup_nodes == ["[email protected]:7000"]
+
 
 @pytest.mark.onlycluster
 class TestClusterPubSubObject: