Added MultiDbClient support with OSS Cluster API (#3734)

* Added Database, Healthcheck, CircuitBreaker, FailureDetector

* Added DatabaseSelector, exceptions, refactored existing entities

* Added MultiDbConfig

* Added DatabaseConfig

* Added DatabaseConfig test coverage

* Renamed DatabaseSelector into FailoverStrategy

* Added CommandExecutor

* Updated healthcheck to close circuit on success

* Added thread-safeness

* Added missing thread-safeness

* Added missing thread-safenes for dispatcher

* Refactored client to keep databases in WeightedList

* Added database CRUD operations

* Added on-fly configuration

* Added background health checks

* Added background healthcheck + half-open event

* Refactored background scheduling

* Added support for Active-Active pipeline

* Refactored healthchecks

* Added Pipeline testing

* Added support for transactions

* Removed code repetitions, fixed weight assignment, added loops enhancement, fixed data structure

* Added missing doc blocks

* Added support for Pub/Sub in MultiDBClient

* Refactored configuration

* Refactored failure detector

* Refactored retry logic

* Added scenario tests

* Added pybreaker optional dependency

* Added pybreaker to dev dependencies

* Rename tests directory

* Added scenario tests for Pipeline and Transaction

* Added handling of ConnectionRefusedError, added timeouts so cluster could recover

* Increased timeouts

* Refactored integration tests

* Added scenario tests for Pub/Sub

* Updated healthcheck retry

* Increased timeout to avoid unprepared state before tests

* Added backoff retry and changed timeouts

* Added retry for healthchecks to avoid fluctuations

* Changed retry configuration for healthchecks

* Fixed property name

* Added check for thread results

* Added MultiDbClient support with OSS Cluster API

* Removed database statuses

* Increased test timeouts

* Increased retry timeout

* Increased timeout retries

* Updated base threshold for retries

* Fixed flacky tests

* Added missing docblocks

Author: vladvildanov

redis/client.py

@@ -1217,7 +1217,8 @@ def run_in_thread(
         sleep_time: float = 0.0,
         daemon: bool = False,
         exception_handler: Optional[Callable] = None,
-        pubsub = None
+        pubsub = None,
+        sharded_pubsub: bool = False,
     ) -> "PubSubWorkerThread":
         for channel, handler in self.channels.items():
             if handler is None:
@@ -1233,7 +1234,7 @@ def run_in_thread(
 
         pubsub = self if pubsub is None else pubsub
         thread = PubSubWorkerThread(
-            pubsub, sleep_time, daemon=daemon, exception_handler=exception_handler
+            pubsub, sleep_time, daemon=daemon, exception_handler=exception_handler, sharded_pubsub=sharded_pubsub
         )
         thread.start()
         return thread
@@ -1248,12 +1249,14 @@ def __init__(
         exception_handler: Union[
             Callable[[Exception, "PubSub", "PubSubWorkerThread"], None], None
         ] = None,
+        sharded_pubsub: bool = False,
     ):
         super().__init__()
         self.daemon = daemon
         self.pubsub = pubsub
         self.sleep_time = sleep_time
         self.exception_handler = exception_handler
+        self.sharded_pubsub = sharded_pubsub
         self._running = threading.Event()
 
     def run(self) -> None:
@@ -1264,7 +1267,10 @@ def run(self) -> None:
         sleep_time = self.sleep_time
         while self._running.is_set():
             try:
-                pubsub.get_message(ignore_subscribe_messages=True, timeout=sleep_time)
+                if not self.sharded_pubsub:
+                    pubsub.get_message(ignore_subscribe_messages=True, timeout=sleep_time)
+                else:
+                    pubsub.get_sharded_message(ignore_subscribe_messages=True, timeout=sleep_time)
             except BaseException as e:
                 if self.exception_handler is None:
                     raise

redis/cluster.py

@@ -3154,7 +3154,8 @@ def _reinitialize_on_error(self, error):
                 self._nodes_manager.initialize()
                 self.reinitialize_counter = 0
             else:
-                self._nodes_manager.update_moved_exception(error)
+                if type(error) == MovedError:
+                    self._nodes_manager.update_moved_exception(error)
 
         self._executing = False
 

redis/multidb/client.py

@@ -9,7 +9,7 @@
 from redis.multidb.command_executor import DefaultCommandExecutor
 from redis.multidb.config import MultiDbConfig, DEFAULT_GRACE_PERIOD
 from redis.multidb.circuit import State as CBState, CircuitBreaker
-from redis.multidb.database import State as DBState, Database, AbstractDatabase, Databases
+from redis.multidb.database import Database, AbstractDatabase, Databases
 from redis.multidb.exception import NoValidDatabaseException
 from redis.multidb.failure_detector import FailureDetector
 from redis.multidb.healthcheck import HealthCheck
@@ -78,13 +78,8 @@ def raise_exception_on_failed_hc(error):
 
             # Set states according to a weights and circuit state
             if database.circuit.state == CBState.CLOSED and not is_active_db_found:
-                database.state = DBState.ACTIVE
                 self.command_executor.active_database = database
                 is_active_db_found = True
-            elif database.circuit.state == CBState.CLOSED and is_active_db_found:
-                database.state = DBState.PASSIVE
-            else:
-                database.state = DBState.DISCONNECTED
 
         if not is_active_db_found:
             raise NoValidDatabaseException('Initial connection failed - no active database found')
@@ -115,8 +110,6 @@ def set_active_database(self, database: AbstractDatabase) -> None:
 
         if database.circuit.state == CBState.CLOSED:
             highest_weighted_db, _ = self._databases.get_top_n(1)[0]
-            highest_weighted_db.state = DBState.PASSIVE
-            database.state = DBState.ACTIVE
             self.command_executor.active_database = database
             return
 
@@ -138,9 +131,7 @@ def add_database(self, database: AbstractDatabase):
 
     def _change_active_database(self, new_database: AbstractDatabase, highest_weight_database: AbstractDatabase):
         if new_database.weight > highest_weight_database.weight and new_database.circuit.state == CBState.CLOSED:
-            new_database.state = DBState.ACTIVE
             self.command_executor.active_database = new_database
-            highest_weight_database.state = DBState.PASSIVE
 
     def remove_database(self, database: Database):
         """
@@ -150,7 +141,6 @@ def remove_database(self, database: Database):
         highest_weighted_db, highest_weight = self._databases.get_top_n(1)[0]
 
         if highest_weight <= weight and highest_weighted_db.circuit.state == CBState.CLOSED:
-            highest_weighted_db.state = DBState.ACTIVE
             self.command_executor.active_database = highest_weighted_db
 
     def update_database_weight(self, database: AbstractDatabase, weight: float):
@@ -240,7 +230,7 @@ def _check_db_health(self, database: AbstractDatabase, on_error: Callable[[Excep
                         database.circuit.state = CBState.OPEN
                     elif is_healthy and database.circuit.state != CBState.CLOSED:
                         database.circuit.state = CBState.CLOSED
-                except (ConnectionError, TimeoutError, socket.timeout, ConnectionRefusedError) as e:
+                except (ConnectionError, TimeoutError, socket.timeout, ConnectionRefusedError, ValueError) as e:
                     if database.circuit.state != CBState.OPEN:
                         database.circuit.state = CBState.OPEN
                     is_healthy = False
@@ -334,7 +324,7 @@ def execute(self) -> List[Any]:
 
 class PubSub:
     """
-    PubSub object for multi-database client.
+    PubSub object for multi database client.
     """
     def __init__(self, client: MultiDBClient, **kwargs):
         """Initialize the PubSub object for a multi-database client.
@@ -438,18 +428,33 @@ def get_message(
             ignore_subscribe_messages=ignore_subscribe_messages, timeout=timeout
         )
 
-    get_sharded_message = get_message
+    def get_sharded_message(
+        self, ignore_subscribe_messages: bool = False, timeout: float = 0.0
+    ):
+        """
+        Get the next message if one is available in a sharded channel, otherwise None.
+
+        If timeout is specified, the system will wait for `timeout` seconds
+        before returning. Timeout should be specified as a floating point
+        number, or None, to wait indefinitely.
+        """
+        return self._client.command_executor.execute_pubsub_method(
+            'get_sharded_message',
+            ignore_subscribe_messages=ignore_subscribe_messages, timeout=timeout
+        )
 
     def run_in_thread(
         self,
         sleep_time: float = 0.0,
         daemon: bool = False,
         exception_handler: Optional[Callable] = None,
+        sharded_pubsub: bool = False,
     ) -> "PubSubWorkerThread":
         return self._client.command_executor.execute_pubsub_run_in_thread(
             sleep_time=sleep_time,
             daemon=daemon,
             exception_handler=exception_handler,
-            pubsub=self
+            pubsub=self,
+            sharded_pubsub=sharded_pubsub,
         )
 

redis/multidb/command_executor.py

@@ -235,10 +235,15 @@ def execute_pubsub_run_in_thread(
         sleep_time: float = 0.0,
         daemon: bool = False,
         exception_handler: Optional[Callable] = None,
+        sharded_pubsub: bool = False,
     ) -> "PubSubWorkerThread":
         def callback():
             return self._active_pubsub.run_in_thread(
-                sleep_time, daemon=daemon, exception_handler=exception_handler, pubsub=pubsub
+                sleep_time,
+                daemon=daemon,
+                exception_handler=exception_handler,
+                pubsub=pubsub,
+                sharded_pubsub=sharded_pubsub
             )
 
         return self._execute_with_failure_detection(callback)

redis/multidb/database.py

@@ -8,12 +8,6 @@
 from redis.multidb.circuit import CircuitBreaker
 from redis.typing import Number
 
-
-class State(Enum):
-    ACTIVE = 0
-    PASSIVE = 1
-    DISCONNECTED = 2
-
 class AbstractDatabase(ABC):
     @property
     @abstractmethod
@@ -39,18 +33,6 @@ def weight(self, weight: float):
         """Set the weight of this database in compare to others."""
         pass
 
-    @property
-    @abstractmethod
-    def state(self) -> State:
-        """The state of the current database."""
-        pass
-
-    @state.setter
-    @abstractmethod
-    def state(self, state: State):
-        """Set the state of the current database."""
-        pass
-
     @property
     @abstractmethod
     def circuit(self) -> CircuitBreaker:
@@ -70,8 +52,7 @@ def __init__(
             self,
             client: Union[redis.Redis, RedisCluster],
             circuit: CircuitBreaker,
-            weight: float,
-            state: State = State.DISCONNECTED,
+            weight: float
     ):
         """
         Initialize a new Database instance.
@@ -86,7 +67,6 @@ def __init__(
         self._cb = circuit
         self._cb.database = self
         self._weight = weight
-        self._state = state
 
     @property
     def client(self) -> Union[redis.Redis, RedisCluster]:
@@ -104,14 +84,6 @@ def weight(self) -> float:
     def weight(self, weight: float):
         self._weight = weight
 
-    @property
-    def state(self) -> State:
-        return self._state
-
-    @state.setter
-    def state(self, state: State):
-        self._state = state
-
     @property
     def circuit(self) -> CircuitBreaker:
         return self._cb

redis/multidb/event.py

@@ -39,7 +39,7 @@ def kwargs(self):
 
 class ResubscribeOnActiveDatabaseChanged(EventListenerInterface):
     """
-    Re-subscribe the currently active pub / sub to a new active database.
+    Re-subscribe currently active pub/sub to a new active database.
     """
     def listen(self, event: ActiveDatabaseChanged):
         old_pubsub = event.command_executor.active_pubsub

redis/multidb/healthcheck.py

@@ -1,4 +1,7 @@
 from abc import abstractmethod, ABC
+
+import redis
+from redis import Redis
 from redis.retry import Retry
 
 
@@ -51,8 +54,20 @@ def check_health(self, database) -> bool:
 
     def _returns_echoed_message(self, database) -> bool:
         expected_message = ["healthcheck", b"healthcheck"]
-        actual_message = database.client.execute_command('ECHO', "healthcheck")
-        return actual_message in expected_message
+
+        if isinstance(database.client, Redis):
+            actual_message = database.client.execute_command("ECHO" ,"healthcheck")
+            return actual_message in expected_message
+        else:
+            # For a cluster checks if all nodes are healthy.
+            all_nodes = database.client.get_nodes()
+            for node in all_nodes:
+                actual_message = node.redis_connection.execute_command("ECHO" ,"healthcheck")
+
+                if actual_message not in expected_message:
+                    return False
+
+            return True
 
     def _dummy_fail(self):
         pass

tests/test_multidb/conftest.py

@@ -7,7 +7,7 @@
 from redis.multidb.circuit import CircuitBreaker, State as CBState
 from redis.multidb.config import MultiDbConfig, DatabaseConfig, DEFAULT_HEALTH_CHECK_INTERVAL, \
      DEFAULT_AUTO_FALLBACK_INTERVAL
-from redis.multidb.database import Database, State, Databases
+from redis.multidb.database import Database, Databases
 from redis.multidb.failover import FailoverStrategy
 from redis.multidb.failure_detector import FailureDetector
 from redis.multidb.healthcheck import HealthCheck
@@ -38,7 +38,6 @@ def mock_hc() -> HealthCheck:
 def mock_db(request) -> Database:
      db = Mock(spec=Database)
      db.weight = request.param.get("weight", 1.0)
-     db.state = request.param.get("state", State.ACTIVE)
      db.client = Mock(spec=Redis)
 
      cb = request.param.get("circuit", {})
@@ -53,7 +52,6 @@ def mock_db(request) -> Database:
 def mock_db1(request) -> Database:
      db = Mock(spec=Database)
      db.weight = request.param.get("weight", 1.0)
-     db.state = request.param.get("state", State.ACTIVE)
      db.client = Mock(spec=Redis)
 
      cb = request.param.get("circuit", {})
@@ -68,7 +66,6 @@ def mock_db1(request) -> Database:
 def mock_db2(request) -> Database:
      db = Mock(spec=Database)
      db.weight = request.param.get("weight", 1.0)
-     db.state = request.param.get("state", State.ACTIVE)
      db.client = Mock(spec=Redis)
 
      cb = request.param.get("circuit", {})