Add RedisClusterTrackerStore and RedisClusterLockStore

rasa/core/lock_store.py

@@ -266,6 +266,38 @@ def save_lock(self, lock: TicketLock) -> None:
         self.red.set(self.key_prefix + lock.conversation_id, lock.dumps())
 
 
+class RedisClusterLockStore(RedisLockStore):
+    """Redis store for ticket locks."""
+
+    def __init__(
+        self,
+        host: Text = "localhost",
+        port: int = 6379,
+        db: int = 1,
+        password: Optional[Text] = None,
+        use_ssl: bool = False,
+        key_prefix: Optional[Text] = None,
+        socket_timeout: float = DEFAULT_SOCKET_TIMEOUT_IN_SECONDS,
+    ) -> None:
+        """Create a lock store which uses Redis Cluster for persistence.
+        """
+        import redis
+
+        self.red = redis.cluster.RedisCluster(
+            host=host,
+            port=int(port),
+            password=password,
+            ssl=use_ssl,
+            socket_timeout=socket_timeout,
+        )
+
+        self.key_prefix = DEFAULT_REDIS_LOCK_STORE_KEY_PREFIX
+        if key_prefix:
+            logger.debug(f"Setting non-default redis key prefix: '{key_prefix}'.")
+            self._set_key_prefix(key_prefix)
+
+        super(RedisLockStore, self).__init__()
+
 class InMemoryLockStore(LockStore):
     """In-memory store for ticket locks."""
 
@@ -304,6 +336,8 @@ def _create_from_endpoint_config(
         lock_store: LockStore = InMemoryLockStore()
     elif endpoint_config.type == "redis":
         lock_store = RedisLockStore(host=endpoint_config.url, **endpoint_config.kwargs)
+    elif endpoint_config.type == "redis_cluster":
+        lock_store = RedisClusterLockStore(host=endpoint_config.url, **endpoint_config.kwargs)
     else:
         lock_store = _load_from_module_name_in_endpoint_config(endpoint_config)
 

rasa/core/tracker_store.py

@@ -615,6 +615,47 @@ def _merge_trackers(
         return merged
 
 
+class RedisClusterTrackerStore(RedisTrackerStore):
+    """Stores conversation history in Redis."""
+
+    def __init__(
+        self,
+        domain: Domain,
+        host: Text = "localhost",
+        port: int = 6379,
+        db: int = 0,
+        password: Optional[Text] = None,
+        event_broker: Optional[EventBroker] = None,
+        record_exp: Optional[float] = None,
+        key_prefix: Optional[Text] = None,
+        use_ssl: bool = False,
+        ssl_keyfile: Optional[Text] = None,
+        ssl_certfile: Optional[Text] = None,
+        ssl_ca_certs: Optional[Text] = None,
+        **kwargs: Dict[Text, Any],
+    ) -> None:
+        """Initializes the tracker store."""
+        import redis
+
+        self.red = redis.cluster.RedisCluster(
+            host=host,
+            port=port,
+            password=password,
+            ssl=use_ssl,
+            ssl_keyfile=ssl_keyfile,
+            ssl_certfile=ssl_certfile,
+            ssl_ca_certs=ssl_ca_certs,
+            decode_responses=True,
+        )
+        self.record_exp = record_exp
+
+        self.key_prefix = DEFAULT_REDIS_TRACKER_STORE_KEY_PREFIX
+        if key_prefix:
+            logger.debug(f"Setting non-default redis key prefix: '{key_prefix}'.")
+            self._set_key_prefix(key_prefix)
+
+        super(RedisTrackerStore, self).__init__(domain, event_broker, **kwargs)
+
 class DynamoTrackerStore(TrackerStore, SerializedTrackerAsDict):
     """Stores conversation history in DynamoDB."""
 
@@ -1504,6 +1545,13 @@ def _create_from_endpoint_config(
             event_broker=event_broker,
             **endpoint_config.kwargs,
         )
+    elif endpoint_config.type.lower() == "redis_cluster":
+        tracker_store = RedisClusterTrackerStore(
+            domain=domain,
+            host=endpoint_config.url,
+            event_broker=event_broker,
+            **endpoint_config.kwargs,
+        )
     elif endpoint_config.type.lower() == "mongod":
         tracker_store = MongoTrackerStore(
             domain=domain,

version

@@ -1 +1 @@
-3.6.5-e8-0.2.1
+3.6.5-e8-0.3.0