Setup NullLogging, code lint, Fixed ConnectionError retry code logic

Author: Grokzen

docs/logging.rst

@@ -0,0 +1,17 @@
+Setup client logging
+####################
+
+To setup logging for debugging inside the client during development you can add this as an example to your own code to enable `DEBUG` logging when using the library.
+
+.. code-block:: python
+
+	import logging
+
+	from rediscluster import RedisCluster
+
+	logging.basicConfig()
+	logger = logging.getLogger('rediscluster')
+	logger.setLevel(logging.DEBUG)
+	logger.propergate = True
+
+Note that this logging is not reccommended to be used inside production as it can cause a performance drain and a slowdown of your client.

rediscluster/__init__.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 
 # python std lib
+import logging
 import sys
 
 # rediscluster imports
@@ -51,3 +52,6 @@ def int_or_str(value):
     RedisClusterException,
     TryAgainError,
 ]
+
+# Set default logging handler to avoid "No handler found" warnings.
+logging.getLogger(__name__).addHandler(logging.NullHandler())

rediscluster/client.py

@@ -3,6 +3,8 @@
 
 # python std lib
 import datetime
+import json
+import logging
 import random
 import string
 import time
@@ -54,6 +56,9 @@
 )
 
 
+log = logging.getLogger(__name__)
+
+
 class CaseInsensitiveDict(dict):
     "Case insensitive dict implementation. Assumes string keys only."
 
@@ -320,13 +325,19 @@ def __init__(self, host=None, port=None, startup_nodes=None, max_connections=Non
             - db (Redis do not support database SELECT in cluster mode)
         """
         # Tweaks to Redis client arguments when running in cluster mode
+        log.info("Created new instance of RedisCluster client instance")
+        log.debug("startup_nodes : " + json.dumps(startup_nodes, indent=2))
+
         if "db" in kwargs:
             raise RedisClusterException("Argument 'db' is not possible to use in cluster mode")
 
-        if kwargs.pop('ssl', False):  # Needs to be removed to avoid exception in redis Connection init
+        # Needs to be removed to avoid exception in redis Connection init
+        if kwargs.pop('ssl', False):
+            log.info("Patching connection_class to SSLClusterConnection")
             connection_class = SSLClusterConnection
 
         if "connection_pool" in kwargs:
+            log.info("Using custom created connection pool")
             pool = kwargs.pop('connection_pool')
         else:
             startup_nodes = [] if startup_nodes is None else startup_nodes
@@ -337,10 +348,15 @@ def __init__(self, host=None, port=None, startup_nodes=None, max_connections=Non
 
             if readonly_mode:
                 connection_pool_cls = ClusterReadOnlyConnectionPool
+                log.info("Using ClusterReadOnlyConnectionPool")
             elif read_from_replicas:
                 connection_pool_cls = ClusterWithReadReplicasConnectionPool
+                log.info("Using ClusterWithReadReplicasConnectionPool")
             else:
                 connection_pool_cls = ClusterConnectionPool
+                log.info("Using ClusterConnectionPool")
+
+            log.debug("Connection pool class " + str(connection_pool_cls))
 
             pool = connection_pool_cls(
                 startup_nodes=startup_nodes,
@@ -545,6 +561,7 @@ def _execute_command(self, *args, **kwargs):
             raise RedisClusterException("Unable to determine command to use")
 
         command = args[0]
+        log.debug("Command to execute : " + str(command) + " : " + str(args) + " : " + str(kwargs))
 
         # If set externally we must update it before calling any commands
         if self.refresh_table_asap:
@@ -562,28 +579,34 @@ def _execute_command(self, *args, **kwargs):
         try_random_node = False
         slot = self._determine_slot(*args)
         ttl = int(self.RedisClusterRequestTTL)
+        connection_error_retry_counter = 0
 
         while ttl > 0:
             ttl -= 1
 
-            if asking:
-                node = self.connection_pool.nodes.nodes[redirect_addr]
-                connection = self.connection_pool.get_connection_by_node(node)
-            elif try_random_node:
-                connection = self.connection_pool.get_random_connection()
-                try_random_node = False
-            else:
-                if self.refresh_table_asap:
-                    # MOVED
-                    node = self.connection_pool.get_master_node_by_slot(slot)
-                    # Reset the flag when it has been consumed to avoid it being
-                    self.refresh_table_asap = False
+            try:
+                if asking:
+                    node = self.connection_pool.nodes.nodes[redirect_addr]
+                    connection = self.connection_pool.get_connection_by_node(node)
+                elif try_random_node:
+                    connection = self.connection_pool.get_random_connection()
+                    try_random_node = False
                 else:
-                    node = self.connection_pool.get_node_by_slot(slot, self.read_from_replicas and (command in self.READ_COMMANDS))
-                    is_read_replica = node['server_type'] == 'slave'
-                connection = self.connection_pool.get_connection_by_node(node)
+                    if self.refresh_table_asap:
+                        # MOVED
+                        node = self.connection_pool.get_master_node_by_slot(slot)
+                        self.refresh_table_asap = False
+                    else:
+                        node = self.connection_pool.get_node_by_slot(
+                            slot,
+                            self.read_from_replicas and (command in self.READ_COMMANDS)
+                        )
+                        is_read_replica = node['server_type'] == 'slave'
+
+                    connection = self.connection_pool.get_connection_by_node(node)
+
+                log.debug("Determined node to execute : " + str(node))
 
-            try:
                 if asking:
                     connection.send_command('ASKING')
                     self.parse_response(connection, "ASKING", **kwargs)
@@ -598,25 +621,53 @@ def _execute_command(self, *args, **kwargs):
                 connection.send_command(*args)
                 return self.parse_response(connection, command, **kwargs)
             except SlotNotCoveredError as e:
+                log.exception("SlotNotCoveredError")
+
                 # In some cases during failover to a replica is happening
                 # a slot sometimes is not covered by the cluster layout and
                 # we need to attempt to refresh the cluster layout and try again
                 self.refresh_table_asap = True
-                time.sleep(0.05)
+                time.sleep(0.1)
 
                 # This is the last attempt before we run out of TTL, raise the exception
                 if ttl == 1:
                     raise e
-            except (RedisClusterException, BusyLoadingError):
+            except (RedisClusterException, BusyLoadingError) as e:
+                log.exception("RedisClusterException || BusyLoadingError")
                 raise
-            except ConnectionError:
+            except ConnectionError as e:
+                log.exception("ConnectionError")
+
                 connection.disconnect()
-            except TimeoutError:
+                connection_error_retry_counter += 1
+
+                # Give the node 0.1 seconds to get back up and retry again with same
+                # node and configuration. After 5 attempts then try to reinitialize
+                # the cluster and see if the nodes configuration has changed or not
+                if connection_error_retry_counter < 5:
+                    time.sleep(0.25)
+                else:
+                    # Reset the counter back to 0 as it should have 5 new attempts
+                    # after the client tries to reinitailize the cluster setup to the
+                    # new configuration.
+                    connection_error_retry_counter = 0
+                    self.refresh_table_asap = True
+
+                    # Hard force of reinitialize of the node/slots setup
+                    self.connection_pool.nodes.increment_reinitialize_counter(
+                        count=self.connection_pool.nodes.reinitialize_steps,
+                    )
+
+            except TimeoutError as e:
+                log.exception("TimeoutError")
+
                 if ttl < self.RedisClusterRequestTTL / 2:
                     time.sleep(0.05)
                 else:
                     try_random_node = True
             except ClusterDownError as e:
+                log.exception("ClusterDownError")
+
                 self.connection_pool.disconnect()
                 self.connection_pool.reset()
                 self.refresh_table_asap = True
@@ -627,19 +678,27 @@ def _execute_command(self, *args, **kwargs):
                 # This counter will increase faster when the same client object
                 # is shared between multiple threads. To reduce the frequency you
                 # can set the variable 'reinitialize_steps' in the constructor.
+                log.exception("MovedError")
+
                 self.refresh_table_asap = True
                 self.connection_pool.nodes.increment_reinitialize_counter()
 
                 node = self.connection_pool.nodes.set_node(e.host, e.port, server_type='master')
                 self.connection_pool.nodes.slots[e.slot_id][0] = node
             except TryAgainError as e:
+                log.exception("TryAgainError")
+
                 if ttl < self.RedisClusterRequestTTL / 2:
                     time.sleep(0.05)
             except AskError as e:
+                log.exception("AskError")
+
                 redirect_addr, asking = "{0}:{1}".format(e.host, e.port), True
             finally:
                 self.connection_pool.release(connection)
 
+            log.debug("TTL loop : " + str(ttl))
+
         raise ClusterError('TTL exhausted.')
 
     def _execute_command_on_nodes(self, nodes, *args, **kwargs):

rediscluster/connection.py

@@ -2,6 +2,7 @@
 
 # python std lib
 from __future__ import unicode_literals
+import logging
 import os
 import random
 import threading
@@ -23,6 +24,8 @@
 from redis.connection import ConnectionPool, Connection, DefaultParser, SSLConnection, UnixDomainSocketConnection
 from redis.exceptions import ConnectionError
 
+log = logging.getLogger(__name__)
+
 
 class ClusterParser(DefaultParser):
     """
@@ -42,6 +45,9 @@ class ClusterConnection(Connection):
     "Manages TCP communication to and from a Redis server"
 
     def __init__(self, *args, **kwargs):
+        log.info("Createing new ClusterConnection instance")
+        log.debug(str(args) + " : " + str(kwargs))
+
         self.readonly = kwargs.pop('readonly', False)
         kwargs['parser_class'] = ClusterParser
         super(ClusterConnection, self).__init__(*args, **kwargs)
@@ -54,6 +60,9 @@ def on_connect(self):
         super(ClusterConnection, self).on_connect()
 
         if self.readonly:
+            log.debug("Sending READONLY command to server to configure connection as readonly")
+            log.debug(str(self))
+
             self.send_command('READONLY')
 
             if nativestr(self.read_response()) != 'OK':
@@ -68,7 +77,10 @@ class SSLClusterConnection(SSLConnection):
         client = RedisCluster(connection_pool=pool)
     """
 
-    def __init__(self, **kwargs):
+    def __init__(self, *args, **kwargs):
+        log.info("Createing new SSLClusterConnection instance")
+        log.debug(str(args) + " : " + str(kwargs))
+
         self.readonly = kwargs.pop('readonly', False)
         # need to pop this off as the redis/connection.py SSLConnection init doesn't work with ssl passed in
         if 'ssl' in kwargs:
@@ -84,6 +96,8 @@ def on_connect(self):
         super(SSLClusterConnection, self).on_connect()
 
         if self.readonly:
+            log.debug("Sending READONLY command to server to configure connection as readonly")
+
             self.send_command('READONLY')
 
             if nativestr(self.read_response()) != 'OK':
@@ -109,8 +123,11 @@ def __init__(self, startup_nodes=None, init_slot_cache=True, connection_class=No
             it was operating on. This will allow the client to drift along side the cluster
             if the cluster nodes move around a lot.
         """
+        log.info("Creating new ClusterConnectionPool instance")
+
         if connection_class is None:
             connection_class = ClusterConnection
+
         super(ClusterConnectionPool, self).__init__(connection_class=connection_class, max_connections=max_connections)
 
         # Special case to make from_url method compliant with cluster setting.
@@ -153,7 +170,10 @@ def __repr__(self):
         """
         Return a string with all unique ip:port combinations that this pool is connected to.
         """
-        nodes = [{'host': i['host'], 'port': i['port']} for i in self.nodes.startup_nodes]
+        nodes = [
+            {'host': i['host'], 'port': i['port']}
+            for i in self.nodes.startup_nodes
+        ]
 
         return "{0}<{1}>".format(
             type(self).__name__,
@@ -164,6 +184,8 @@ def reset(self):
         """
         Resets the connection pool back to a clean state.
         """
+        log.debug("Resetting ConnectionPool")
+
         self.pid = os.getpid()
         self._created_connections = 0
         self._created_connections_per_node = {}  # Dict(Node, Int)