Adding default retry configuration changes for sync cluster client

Author: petyaslavova

redis/cluster.py

@@ -9,7 +9,7 @@
 
 from redis._parsers import CommandsParser, Encoder
 from redis._parsers.helpers import parse_scan
-from redis.backoff import default_backoff
+from redis.backoff import ExponentialWithJitterBackoff, NoBackoff
 from redis.cache import CacheConfig, CacheFactory, CacheFactoryInterface, CacheInterface
 from redis.client import CaseInsensitiveDict, PubSub, Redis
 from redis.commands import READ_COMMANDS, RedisClusterCommands
@@ -179,7 +179,7 @@ def parse_cluster_myshardid(resp, **options):
     "cache",
     "cache_config",
 )
-KWARGS_DISABLED_KEYS = ("host", "port")
+KWARGS_DISABLED_KEYS = ("host", "port", "retry")
 
 
 def cleanup_kwargs(**kwargs):
@@ -431,7 +431,7 @@ def replace_default_node(self, target_node: "ClusterNode" = None) -> None:
                 # Choose a primary if the cluster contains different primaries
                 self.nodes_manager.default_node = random.choice(primaries)
             else:
-                # Otherwise, hoose a primary if the cluster contains different primaries
+                # Otherwise, choose a primary if the cluster contains different primaries
                 replicas = [node for node in self.get_replicas() if node != curr_node]
                 if replicas:
                     self.nodes_manager.default_node = random.choice(replicas)
@@ -487,6 +487,13 @@ class initializer. In the case of conflicting arguments, querystring
         reason="Please configure the 'load_balancing_strategy' instead",
         version="5.0.3",
     )
+    @deprecated_args(
+        args_to_warn=[
+            "cluster_error_retry_attempts",
+        ],
+        reason="Please configure the 'retry' object instead",
+        version="6.0.0",
+    )
     def __init__(
         self,
         host: Optional[str] = None,
@@ -544,9 +551,19 @@ def __init__(
              If you use dynamic DNS endpoints for startup nodes but CLUSTER SLOTS lists
              specific IP addresses, it is best to set it to false.
         :param cluster_error_retry_attempts:
+             @deprecated - Please configure the 'retry' object instead
+             In case 'retry' object is set - this argument is ignored!
+
              Number of times to retry before raising an error when
-             :class:`~.TimeoutError` or :class:`~.ConnectionError` or
+             :class:`~.TimeoutError` or :class:`~.ConnectionError`, :class:`~.SlotNotCoveredError` or
              :class:`~.ClusterDownError` are encountered
+        :param retry:
+            A retry object that defines the retry strategy and the number of
+            retries for the cluster client.
+            In current implementation for the cluster client (starting form redis-py version 6.0.0)
+            the retry object is not yet fully utilized, instead it is used just to determine
+            the number of retries for the cluster client.
+            In the future releases the retry object will be used to handle the cluster client retries!
         :param reinitialize_steps:
             Specifies the number of MOVED errors that need to occur before
             reinitializing the whole cluster topology. If a MOVED error occurs
@@ -566,7 +583,8 @@ def __init__(
 
          :**kwargs:
              Extra arguments that will be sent into Redis instance when created
-             (See Official redis-py doc for supported kwargs
+             (See Official redis-py doc for supported kwargs - the only limitation
+              is that you can't provide 'retry' object as part of kwargs.
          [https://github.com/andymccurdy/redis-py/blob/master/redis/client.py])
              Some kwargs are not supported and will raise a
              RedisClusterException:
@@ -581,6 +599,15 @@ def __init__(
                 "Argument 'db' is not possible to use in cluster mode"
             )
 
+        if "retry" in kwargs:
+            # Argument 'retry' is not possible to be used in kwargs when in cluster mode
+            # the kwargs are set to the lower level connections to the cluster nodes
+            # and there we provide retry configuration without retries allowed.
+            # The retries should be handled on cluster client level.
+            raise RedisClusterException(
+                "Argument 'retry' is not possible to be used in kwargs when in cluster mode"
+            )
+
         # Get the startup node/s
         from_url = False
         if url is not None:
@@ -623,9 +650,11 @@ def __init__(
         kwargs = cleanup_kwargs(**kwargs)
         if retry:
             self.retry = retry
-            kwargs.update({"retry": self.retry})
         else:
-            kwargs.update({"retry": Retry(default_backoff(), 0)})
+            self.retry = Retry(
+                backoff=ExponentialWithJitterBackoff(base=1, cap=10),
+                retries=cluster_error_retry_attempts,
+            )
 
         self.encoder = Encoder(
             kwargs.get("encoding", "utf-8"),
@@ -767,13 +796,13 @@ def set_default_node(self, node):
         self.nodes_manager.default_node = node
         return True
 
-    def get_retry(self) -> Optional["Retry"]:
+    def get_retry(self) -> Retry:
         return self.retry
 
-    def set_retry(self, retry: "Retry") -> None:
+    def set_retry(self, retry: Retry) -> None:
+        if not isinstance(retry, Retry):
+            raise TypeError("retry must be a valid instance of redis.retry.Retry")
         self.retry = retry
-        for node in self.get_nodes():
-            node.redis_connection.set_retry(retry)
 
     def monitor(self, target_node=None):
         """
@@ -824,6 +853,7 @@ def pipeline(self, transaction=None, shard_hint=None):
             read_from_replicas=self.read_from_replicas,
             load_balancing_strategy=self.load_balancing_strategy,
             reinitialize_steps=self.reinitialize_steps,
+            retry=self.retry,
             lock=self._lock,
         )
 
@@ -1112,9 +1142,7 @@ def _internal_execute_command(self, *args, **kwargs):
         # execution since the nodes may not be valid anymore after the tables
         # were reinitialized. So in case of passed target nodes,
         # retry_attempts will be set to 0.
-        retry_attempts = (
-            0 if target_nodes_specified else self.cluster_error_retry_attempts
-        )
+        retry_attempts = 0 if target_nodes_specified else self.retry.get_retries()
         # Add one for the first execution
         execute_attempts = 1 + retry_attempts
         for _ in range(execute_attempts):
@@ -1322,8 +1350,12 @@ def __eq__(self, obj):
         return isinstance(obj, ClusterNode) and obj.name == self.name
 
     def __del__(self):
-        if self.redis_connection is not None:
-            self.redis_connection.close()
+        try:
+            if self.redis_connection is not None:
+                self.redis_connection.close()
+        except Exception:
+            # Ignore errors when closing the connection
+            pass
 
 
 class LoadBalancingStrategy(Enum):
@@ -1574,17 +1606,27 @@ def create_redis_connections(self, nodes):
         )
 
     def create_redis_node(self, host, port, **kwargs):
+        # We are configuring the connection pool to not retry
+        # connections on lower level clients to avoid retrying
+        # connections to nodes that are not reachable
+        # and to avoid blocking the connection pool.
+        # The retries will be handled on cluster client level
+        # where we will have proper handling of the cluster topology
+        node_retry_config = Retry(backoff=NoBackoff(), retries=0)
+
         if self.from_url:
             # Create a redis node with a costumed connection pool
             kwargs.update({"host": host})
             kwargs.update({"port": port})
             kwargs.update({"cache": self._cache})
+            kwargs.update({"retry": node_retry_config})
             r = Redis(connection_pool=self.connection_pool_class(**kwargs))
         else:
             r = Redis(
                 host=host,
                 port=port,
                 cache=self._cache,
+                retry=node_retry_config,
                 **kwargs,
             )
         return r
@@ -2028,6 +2070,13 @@ class ClusterPipeline(RedisCluster):
         TryAgainError,
     )
 
+    @deprecated_args(
+        args_to_warn=[
+            "cluster_error_retry_attempts",
+        ],
+        reason="Please configure the 'retry' object instead",
+        version="6.0.0",
+    )
     def __init__(
         self,
         nodes_manager: "NodesManager",
@@ -2039,6 +2088,7 @@ def __init__(
         load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
         cluster_error_retry_attempts: int = 3,
         reinitialize_steps: int = 5,
+        retry: Optional[Retry] = None,
         lock=None,
         **kwargs,
     ):
@@ -2058,6 +2108,14 @@ def __init__(
         self.cluster_error_retry_attempts = cluster_error_retry_attempts
         self.reinitialize_counter = 0
         self.reinitialize_steps = reinitialize_steps
+        if retry is not None:
+            self.retry = retry
+        else:
+            self.retry = Retry(
+                backoff=ExponentialWithJitterBackoff(base=1, cap=10),
+                retries=self.cluster_error_retry_attempts,
+            )
+
         self.encoder = Encoder(
             kwargs.get("encoding", "utf-8"),
             kwargs.get("encoding_errors", "strict"),
@@ -2191,7 +2249,7 @@ def send_cluster_commands(
         """
         if not stack:
             return []
-        retry_attempts = self.cluster_error_retry_attempts
+        retry_attempts = self.retry.get_retries()
         while True:
             try:
                 return self._send_cluster_commands(

redis/retry.py

@@ -44,6 +44,20 @@ def update_supported_errors(
             set(self._supported_errors + tuple(specified_errors))
         )
 
+    def get_retries(self) -> int:
+        """
+        Get the number of retries.
+        """
+        return self._retries
+
+    def update_retries(self, value: int) -> None:
+        """
+        Set the number of retries.
+        """
+        if not isinstance(value, int):
+            raise ValueError("Retries count must be an integer.")
+        self._retries = value
+
     def call_with_retry(
         self,
         do: Callable[[], T],

tests/test_cluster.py

@@ -14,7 +14,11 @@
 import redis
 from redis import Redis
 from redis._parsers import CommandsParser
-from redis.backoff import ExponentialBackoff, NoBackoff, default_backoff
+from redis.backoff import (
+    ExponentialBackoff,
+    ExponentialWithJitterBackoff,
+    NoBackoff,
+)
 from redis.cluster import (
     PRIMARY,
     REDIS_CLUSTER_HASH_SLOTS,
@@ -884,46 +888,48 @@ def moved_redirect_effect(connection, *args, **options):
     def test_cluster_get_set_retry_object(self, request):
         retry = Retry(NoBackoff(), 2)
         r = _get_client(RedisCluster, request, retry=retry)
-        assert r.get_retry()._retries == retry._retries
+        assert r.get_retry().get_retries() == retry.get_retries()
         assert isinstance(r.get_retry()._backoff, NoBackoff)
         for node in r.get_nodes():
-            assert node.redis_connection.get_retry()._retries == retry._retries
+            assert node.redis_connection.get_retry().get_retries() == 0
             assert isinstance(node.redis_connection.get_retry()._backoff, NoBackoff)
         rand_node = r.get_random_node()
         existing_conn = rand_node.redis_connection.connection_pool.get_connection()
         # Change retry policy
         new_retry = Retry(ExponentialBackoff(), 3)
         r.set_retry(new_retry)
-        assert r.get_retry()._retries == new_retry._retries
+        assert r.get_retry().get_retries() == new_retry.get_retries()
         assert isinstance(r.get_retry()._backoff, ExponentialBackoff)
         for node in r.get_nodes():
-            assert node.redis_connection.get_retry()._retries == new_retry._retries
-            assert isinstance(
-                node.redis_connection.get_retry()._backoff, ExponentialBackoff
-            )
-        assert existing_conn.retry._retries == new_retry._retries
+            assert node.redis_connection.get_retry()._retries == 0
+            assert isinstance(node.redis_connection.get_retry()._backoff, NoBackoff)
+        assert existing_conn.retry._retries == 0
         new_conn = rand_node.redis_connection.connection_pool.get_connection()
-        assert new_conn.retry._retries == new_retry._retries
+        assert new_conn.retry._retries == 0
 
     def test_cluster_retry_object(self, r) -> None:
         # Test default retry
         # FIXME: Workaround for https://github.com/redis/redis-py/issues/3030
         host = r.get_default_node().host
 
-        retry = r.get_connection_kwargs().get("retry")
+        # test default retry config
+        retry = r.get_retry()
         assert isinstance(retry, Retry)
-        assert retry._retries == 0
-        assert isinstance(retry._backoff, type(default_backoff()))
+        assert retry.get_retries() == 3
+        assert isinstance(retry._backoff, type(ExponentialWithJitterBackoff()))
         node1 = r.get_node(host, 16379).redis_connection
         node2 = r.get_node(host, 16380).redis_connection
-        assert node1.get_retry()._retries == node2.get_retry()._retries
+        assert node1.get_retry()._retries == 0
+        assert node2.get_retry()._retries == 0
 
-        # Test custom retry
+        # Test custom retry is not applied to nodes
         retry = Retry(ExponentialBackoff(10, 5), 5)
         rc_custom_retry = RedisCluster(host, 16379, retry=retry)
         assert (
-            rc_custom_retry.get_node(host, 16379).redis_connection.get_retry()._retries
-            == retry._retries
+            rc_custom_retry.get_node(host, 16379)
+            .redis_connection.get_retry()
+            .get_retries()
+            == 0
         )
 
     def test_replace_cluster_node(self, r) -> None: