Merge branch 'master' into ps_fix_read_response_timeout_for_pubsub

Author: petyaslavova

redis/asyncio/cluster.py

@@ -1989,6 +1989,15 @@ def __init__(
             else TransactionStrategy(self)
         )
 
+    @property
+    def nodes_manager(self) -> "NodesManager":
+        """Get the nodes manager from the cluster client."""
+        return self.cluster_client.nodes_manager
+
+    def set_response_callback(self, command: str, callback: ResponseCallbackT) -> None:
+        """Set a custom response callback on the cluster client."""
+        self.cluster_client.set_response_callback(command, callback)
+
     async def initialize(self) -> "ClusterPipeline":
         await self._execution_strategy.initialize()
         return self

redis/client.py

@@ -1,4 +1,5 @@
 import copy
+import logging
 import re
 import threading
 import time
@@ -90,6 +91,20 @@
 NEVER_DECODE = "NEVER_DECODE"
 
 
+logger = logging.getLogger(__name__)
+
+
+def is_debug_log_enabled():
+    return logger.isEnabledFor(logging.DEBUG)
+
+
+def add_debug_log_for_operation_failure(connection: "AbstractConnection"):
+    logger.debug(
+        f"Operation failed, "
+        f"with connection: {connection}, details: {connection.extract_connection_details() if connection else 'no connection'}",
+    )
+
+
 class CaseInsensitiveDict(dict):
     "Case insensitive dict implementation. Assumes string keys only."
 
@@ -728,6 +743,8 @@ def _execute_command(self, *args, **options):
         actual_retry_attempts = [0]
 
         def failure_callback(error, failure_count):
+            if is_debug_log_enabled():
+                add_debug_log_for_operation_failure(conn)
             actual_retry_attempts[0] = failure_count
             self._close_connection(conn, error, failure_count, start_time, command_name)
 
@@ -1751,6 +1768,8 @@ def immediate_execute_command(self, *args, **options):
         actual_retry_attempts = [0]
 
         def failure_callback(error, failure_count):
+            if is_debug_log_enabled():
+                add_debug_log_for_operation_failure(conn)
             actual_retry_attempts[0] = failure_count
             self._disconnect_reset_raise_on_watching(
                 conn, error, failure_count, start_time, command_name
@@ -1988,6 +2007,8 @@ def execute(self, raise_on_error: bool = True) -> List[Any]:
         actual_retry_attempts = [0]
 
         def failure_callback(error, failure_count):
+            if is_debug_log_enabled():
+                add_debug_log_for_operation_failure(conn)
             actual_retry_attempts[0] = failure_count
             self._disconnect_raise_on_watching(
                 conn, error, failure_count, start_time, operation_name

redis/cluster.py

@@ -3955,6 +3955,12 @@ def _reinitialize_on_error(self, error, failure_count):
             or type(error) in self.CONNECTION_ERRORS
         ):
             if self._transaction_connection:
+                if is_debug_log_enabled():
+                    logger.debug(
+                        f"Operation failed, "
+                        f"with connection: {self._transaction_connection}, "
+                        f"details: {self._transaction_connection.extract_connection_details()}",
+                    )
                 # Disconnect and release back to pool
                 self._transaction_connection.disconnect()
                 node = self._nodes_manager.find_connection_owner(

redis/commands/core.py

@@ -2900,11 +2900,11 @@ class ListCommands(CommandsProtocol):
     """
 
     def blpop(
-        self, keys: List, timeout: Optional[Number] = 0
+        self, keys: KeysT, timeout: Optional[Number] = 0
     ) -> Union[Awaitable[list], list]:
         """
         LPOP a value off of the first non-empty list
-        named in the ``keys`` list.
+        named in ``keys``.
 
         If none of the lists in ``keys`` has a value to LPOP, then block
         for ``timeout`` seconds, or until a value gets pushed on to one
@@ -2921,11 +2921,11 @@ def blpop(
         return self.execute_command("BLPOP", *keys)
 
     def brpop(
-        self, keys: List, timeout: Optional[Number] = 0
+        self, keys: KeysT, timeout: Optional[Number] = 0
     ) -> Union[Awaitable[list], list]:
         """
         RPOP a value off of the first non-empty list
-        named in the ``keys`` list.
+        named in ``keys``.
 
         If none of the lists in ``keys`` has a value to RPOP, then block
         for ``timeout`` seconds, or until a value gets pushed on to one

redis/connection.py

@@ -287,6 +287,10 @@ def reset_should_reconnect(self):
         """
         pass
 
+    @abstractmethod
+    def extract_connection_details(self) -> str:
+        pass
+
 
 class MaintNotificationsAbstractConnection:
     """
@@ -1448,6 +1452,18 @@ def socket_connect_timeout(self) -> Optional[Union[float, int]]:
     def socket_connect_timeout(self, value: Optional[Union[float, int]]):
         self._socket_connect_timeout = value
 
+    def extract_connection_details(self) -> str:
+        socket_address = None
+        if self._sock is None:
+            return "not connected"
+        try:
+            socket_address = self._sock.getsockname() if self._sock else None
+            socket_address = socket_address[1] if socket_address else None
+        except (AttributeError, OSError):
+            pass
+
+        return f"connected to ip {self.get_resolved_ip()}, local socket port: {socket_address}"
+
 
 class Connection(AbstractConnection):
     "Manages TCP communication to and from a Redis server"
@@ -1911,6 +1927,9 @@ def _on_invalidation_callback(self, data: List[Union[str, Optional[List[bytes]]]
                         reason=CSCReason.INVALIDATION,
                     )
 
+    def extract_connection_details(self) -> str:
+        return self._conn.extract_connection_details()
+
 
 class SSLConnection(Connection):
     """Manages SSL connections to and from the Redis server(s).

redis/maint_notifications.py

@@ -1006,7 +1006,12 @@ def handle_maintenance_completed_notification(self, **kwargs):
             or not self.config.is_relaxed_timeouts_enabled()
         ):
             return
-        add_debug_log_for_notification(self.connection, "MAINTENANCE_COMPLETED")
+        notification = None
+        if kwargs.get("notification"):
+            notification = kwargs["notification"]
+        add_debug_log_for_notification(
+            self.connection, notification if notification else "MAINTENANCE_COMPLETED"
+        )
         self.connection.reset_tmp_settings(reset_relaxed_timeout=True)
         # Maintenance completed - reset the connection
         # timeouts by providing -1 as the relaxed timeout
@@ -1016,8 +1021,7 @@ def handle_maintenance_completed_notification(self, **kwargs):
         # notifications and skipped end maint notifications
         self.connection.reset_received_notifications()
 
-        if kwargs.get("notification"):
-            notification = kwargs["notification"]
+        if notification:
             record_connection_relaxed_timeout(
                 connection_name=repr(self.connection),
                 maint_notification=notification.__class__.__name__,
@@ -1076,7 +1080,8 @@ def handle_oss_maintenance_completed_notification(
                 # process the same notification twice
                 return
 
-            logger.debug(f"Handling SMIGRATED notification: {notification}")
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug(f"Handling SMIGRATED notification: {notification}")
             self._in_progress.add(notification)
 
             # Extract the information about the src and destination nodes that are affected
@@ -1130,7 +1135,16 @@ def handle_oss_maintenance_completed_notification(
                         # Some of them might be used by sub sub and we don't know which ones - so we disconnect
                         # all in flight connections after they are done with current command execution
                         for conn in current_node.redis_connection.connection_pool._get_in_use_connections():
+                            add_debug_log_for_notification(
+                                conn, "SMIGRATED - mark for reconnect"
+                            )
                             conn.mark_for_reconnect()
+                    else:
+                        if logger.isEnabledFor(logging.DEBUG):
+                            logger.debug(
+                                f"SMIGRATED: Node {current_node.name} not affected by maintenance, "
+                                f"skipping mark for reconnect"
+                            )
 
                     if (
                         current_node

tests/test_asyncio/test_cluster.py

@@ -3162,6 +3162,44 @@ def cmd_init_mock(
 class TestClusterPipeline:
     """Tests for the ClusterPipeline class."""
 
+    async def test_pipeline_nodes_manager_property(self) -> None:
+        """
+        Test that ClusterPipeline exposes nodes_manager property
+        that delegates to the cluster client's nodes_manager.
+        """
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            pipeline = r.pipeline()
+            # Verify that nodes_manager property exists and returns the same object
+            # as the cluster client's nodes_manager
+            assert pipeline.nodes_manager is r.nodes_manager
+            # Verify that we can access nodes_manager attributes
+            assert pipeline.nodes_manager.default_node is not None
+        finally:
+            await r.aclose()
+
+    async def test_pipeline_set_response_callback(self) -> None:
+        """
+        Test that ClusterPipeline exposes set_response_callback method
+        that delegates to the cluster client's set_response_callback.
+        """
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            pipeline = r.pipeline()
+
+            # Define a custom callback
+            def custom_callback(response):
+                return f"custom_{response}"
+
+            # Set the callback via the pipeline
+            pipeline.set_response_callback("CUSTOM_CMD", custom_callback)
+
+            # Verify that the callback was set on the cluster client
+            assert "CUSTOM_CMD" in r.response_callbacks
+            assert r.response_callbacks["CUSTOM_CMD"] is custom_callback
+        finally:
+            await r.aclose()
+
     async def test_blocked_arguments(self, r: RedisCluster) -> None:
         """Test handling for blocked pipeline arguments."""
 

tests/test_asyncio/test_pipeline.py

@@ -466,6 +466,26 @@ async def test_pipeline_with_msetex(self, r):
                 {"key1_transaction": "value1", "key2_transaction": "value2"}, ex=10
             )
 
+    async def test_pipeline_json_module_access(self, r):
+        """
+        Test that pipeline can access the json() module.
+        The JSON module requires nodes_manager (for cluster) and set_response_callback
+        on the client during initialization.
+
+        """
+        pipeline = r.pipeline()
+
+        # This should not raise an AttributeError
+        json_pipeline = pipeline.json()
+
+        # Verify the JSON module was created successfully
+        assert json_pipeline is not None
+        assert json_pipeline.client is pipeline
+
+        # Verify that JSON callbacks were registered
+        assert "JSON.SET" in r.response_callbacks
+        assert "JSON.GET" in r.response_callbacks
+
 
 @pytest.mark.asyncio
 class TestAsyncPipelineOperationDurationMetricsRecording:

tests/test_scenario/conftest.py

@@ -226,6 +226,39 @@ def extract_cluster_fqdn(url):
     return f"https://{cleaned_hostname}"
 
 
+def _prepare_ssl_certificates(cert_chain: bool) -> dict:
+    """
+    Prepare SSL certificates for Redis cluster connection.
+
+    Args:
+        cert_chain: PEM-encoded certificate chain containing client cert + intermediate + CA cert.
+                   This is the full certificate chain that will be used to validate the server.
+
+    Returns:
+        dict: SSL configuration kwargs for RedisCluster
+    """
+    certs_config_path = os.environ.get("MTLS_CONFIG_PATH", None)
+
+    if not cert_chain:
+        return {
+            "ssl_cert_reqs": "none",
+            "ssl_check_hostname": False,
+        }
+
+    if not certs_config_path:
+        raise ValueError(
+            "MTLS enabled test is triggered but MTLS_CONFIG_PATH environment variable not set"
+        )
+
+    # The cert_chain contains the full chain (client cert + intermediate + root CA)
+    # Use it as CA data for validating the server's certificate
+    return {
+        "ssl_cert_reqs": "none",
+        "ssl_keyfile": os.path.join(certs_config_path, "client.key"),
+        "ssl_certfile": os.path.join(certs_config_path, "client.crt"),
+    }
+
+
 @pytest.fixture()
 def client_maint_notifications(endpoints_config):
     return _get_client_maint_notifications(endpoints_config)
@@ -307,8 +340,8 @@ def get_cluster_client_maint_notifications(
     enable_relaxed_timeout: bool = True,
     enable_proactive_reconnect: bool = True,
     disable_retries: bool = False,
+    auth_ssl_client_certs: bool = False,
     socket_timeout: Optional[float] = None,
-    host_config: Optional[str] = None,
 ):
     """Create Redis cluster client with maintenance notifications enabled."""
     # Get credentials from the configuration
@@ -337,6 +370,13 @@ def get_cluster_client_maint_notifications(
     tls_enabled = True if parsed.scheme == "rediss" else False
     logging.info(f"TLS enabled: {tls_enabled}")
 
+    tls_kwargs = {"ssl": tls_enabled}
+
+    if tls_enabled:
+        # Prepare SSL certificate configuration
+        ssl_config = _prepare_ssl_certificates(auth_ssl_client_certs)
+        tls_kwargs.update(ssl_config)
+
     # Configure maintenance notifications
     maintenance_config = MaintNotificationsConfig(
         enabled=enable_maintenance_notifications,
@@ -352,12 +392,10 @@ def get_cluster_client_maint_notifications(
         socket_timeout=CLIENT_TIMEOUT if socket_timeout is None else socket_timeout,
         username=username,
         password=password,
-        ssl=tls_enabled,
-        ssl_cert_reqs="none",
-        ssl_check_hostname=False,
         protocol=protocol,  # RESP3 required for push notifications
         maint_notifications_config=maintenance_config,
         retry=retry,
+        **tls_kwargs,
     )
     logging.info("Redis cluster client created with maintenance notifications enabled")
     logging.info(