Added support for Lag-Aware Healthcheck and OSS Cluster API

redis/asyncio/cluster.py

@@ -404,6 +404,7 @@ def __init__(
         else:
             self._event_dispatcher = event_dispatcher
 
+        self.startup_nodes = startup_nodes
         self.nodes_manager = NodesManager(
             startup_nodes,
             require_full_coverage,
@@ -2199,7 +2200,8 @@ async def _reinitialize_on_error(self, error):
                 await self._pipe.cluster_client.nodes_manager.initialize()
                 self.reinitialize_counter = 0
             else:
-                self._pipe.cluster_client.nodes_manager.update_moved_exception(error)
+                if type(error) == MovedError:
+                    self._pipe.cluster_client.nodes_manager.update_moved_exception(error)
 
         self._executing = False
 

redis/asyncio/http/http_client.py

@@ -0,0 +1,216 @@
+import asyncio
+from abc import ABC, abstractmethod
+from concurrent.futures import ThreadPoolExecutor
+from typing import Optional, Mapping, Union, Any
+from redis.http.http_client import HttpResponse, HttpClient
+
+DEFAULT_USER_AGENT = "HttpClient/1.0 (+https://example.invalid)"
+DEFAULT_TIMEOUT = 30.0
+RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
+
+class AsyncHTTPClient(ABC):
+    @abstractmethod
+    async def get(
+        self,
+        path: str,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP GET request."""
+        pass
+
+    @abstractmethod
+    async def delete(
+        self,
+        path: str,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP DELETE request."""
+        pass
+
+    @abstractmethod
+    async def post(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP POST request."""
+        pass
+
+    @abstractmethod
+    async def put(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP PUT request."""
+        pass
+
+    @abstractmethod
+    async def patch(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP PATCH request."""
+        pass
+
+    @abstractmethod
+    async def request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        body: Optional[Union[bytes, str]] = None,
+        timeout: Optional[float] = None,
+    ) -> HttpResponse:
+        """
+        Invoke HTTP request with given method."""
+        pass
+
+class AsyncHTTPClientWrapper(AsyncHTTPClient):
+    """
+    An async wrapper around sync HTTP client with thread pool execution.
+    """
+    def __init__(
+        self,
+        client: HttpClient,
+        max_workers: int = 10
+    ) -> None:
+        """
+        Initialize a new HTTP client instance.
+
+        Args:
+            client: Sync HTTP client instance.
+            max_workers: Maximum number of concurrent requests.
+
+        The client supports both regular HTTPS with server verification and mutual TLS
+        authentication. For server verification, provide CA certificate information via
+        ca_file, ca_path or ca_data. For mutual TLS, additionally provide a client
+        certificate and key via client_cert_file and client_key_file.
+        """
+        self.client = client
+        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+
+    async def get(
+        self,
+        path: str,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.get,
+            path, params, headers, timeout, expect_json
+        )
+
+    async def delete(
+        self,
+        path: str,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.delete,
+            path, params, headers, timeout, expect_json
+        )
+
+    async def post(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.post,
+            path, json_body, data, params, headers, timeout, expect_json
+        )
+
+    async def put(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.put,
+            path, json_body, data, params, headers, timeout, expect_json
+        )
+
+    async def patch(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.patch,
+            path, json_body, data, params, headers, timeout, expect_json
+        )
+
+    async def request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[Mapping[str, Union[None, str, int, float, bool, list, tuple]]] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        body: Optional[Union[bytes, str]] = None,
+        timeout: Optional[float] = None,
+    ) -> HttpResponse:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.request,
+            method, path, params, headers, body, timeout
+        )

redis/asyncio/multidb/client.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 from typing import Callable, Optional, Coroutine, Any, List, Union, Awaitable
 
 from redis.asyncio.client import PubSubHandler
@@ -13,6 +14,7 @@
 from redis.multidb.exception import NoValidDatabaseException
 from redis.typing import KeyT, EncodableT, ChannelT
 
+logger = logging.getLogger(__name__)
 
 class MultiDBClient(AsyncRedisModuleCommands, AsyncCoreCommands):
     """
@@ -274,6 +276,8 @@ async def _check_db_health(
                     database.circuit.state = CBState.OPEN
                 is_healthy = False
 
+                logger.exception('Health check failed, due to exception', exc_info=e)
+
                 if on_error:
                     await on_error(e)
 

redis/asyncio/multidb/command_executor.py

@@ -3,6 +3,7 @@
 from datetime import datetime
 from typing import List, Optional, Callable, Any, Union, Awaitable
 
+from redis.asyncio import RedisCluster
 from redis.asyncio.client import PubSub, Pipeline
 from redis.asyncio.multidb.database import Databases, AsyncDatabase, Database
 from redis.asyncio.multidb.event import AsyncActiveDatabaseChanged, RegisterCommandFailure, \
@@ -181,6 +182,9 @@ def command_retry(self) -> Retry:
 
     def pubsub(self, **kwargs):
         if self._active_pubsub is None:
+            if isinstance(self._active_database.client, RedisCluster):
+                raise ValueError("PubSub is not supported for RedisCluster")
+
             self._active_pubsub = self._active_database.client.pubsub(**kwargs)
             self._active_pubsub_kwargs = kwargs