Adding maintenance notification configuration to cluster client. Integration with testing helper proxy server. (#3844)

* Adding maint_notifications_config to RedisCluster's NodesManager

* Adding Redis Proxy integration

* Migrating test proxy helper to use custom https client

Author: petyaslavova

.gitignore

@@ -27,3 +27,4 @@ docker/stunnel/keys
 /dockers/replica/
 /dockers/sentinel/
 /dockers/redis-stack/
+/experimenting/

docker-compose.yml

@@ -1,11 +1,14 @@
 ---
 # image tag 8.0-RC2-pre is the one matching the 8.0 GA release
 x-client-libs-stack-image: &client-libs-stack-image
-  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_STACK_IMAGE_TAG:-8.4-RC1-pre.2}"
+  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_STACK_IMAGE_TAG:-8.4-GA-pre.2}"
 
 x-client-libs-image: &client-libs-image
-  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_IMAGE_TAG:-8.4-RC1-pre.2}"
+  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_IMAGE_TAG:-8.4-GA-pre.2}"
 
+networks:
+  redis-net:
+    driver: bridge
 services:
 
   redis:
@@ -106,3 +109,42 @@ services:
       - standalone
       - all-stack
       - all
+
+  redis-proxied:
+    <<: *client-libs-image
+    container_name: redis-proxied
+    ports:
+      - "3000:3000"
+    networks:
+      - redis-net
+    healthcheck:
+      test: ["CMD", "redis-cli", "-p", "3000", "PING"]
+      interval: 10s
+      timeout: 3s
+      retries: 5
+
+  resp-proxy:
+    image: redislabs/client-resp-proxy
+    container_name: resp-proxy
+    environment:
+      LISTEN_HOST: "0.0.0.0"
+      LISTEN_PORT: "15379,15380,15381"
+      TARGET_HOST: "redis-proxied"
+      TARGET_PORT: "3000"
+      API_PORT: "4000"
+      ENABLE_LOGGING: true
+      SIMULATE_CLUSTER: true
+    ports:
+      - "15379:15379"
+      - "15380:15380"
+      - "15381:15381"
+      - "4000:4000"
+    depends_on:
+      - redis-proxied
+    networks:
+      - redis-net
+    healthcheck:
+      test: ["CMD", "sh", "-c", "wget -qO- http://localhost:4000/stats || exit 1"]
+      interval: 10s
+      timeout: 3s
+      retries: 5

redis/client.py

@@ -281,7 +281,7 @@ def __init__(
             if `True`, the response will be decoded to utf-8.
             Argument is ignored when connection_pool is provided.
         maint_notifications_config:
-            configuration the pool to support maintenance notifications - see
+            configures the pool to support maintenance notifications - see
             `redis.maint_notifications.MaintNotificationsConfig` for details.
             Only supported with RESP3
             If not provided and protocol is RESP3, the maintenance notifications

redis/cluster.py

@@ -196,6 +196,7 @@ def parse_cluster_myshardid(resp, **options):
     "username",
     "cache",
     "cache_config",
+    "maint_notifications_config",
 )
 KWARGS_DISABLED_KEYS = ("host", "port", "retry")
 
@@ -535,6 +536,7 @@ def __init__(
         cache_config: Optional[CacheConfig] = None,
         event_dispatcher: Optional[EventDispatcher] = None,
         policy_resolver: PolicyResolver = StaticPolicyResolver(),
+        maint_notifications_config: Optional[MaintNotificationsConfig] = None,
         **kwargs,
     ):
         """
@@ -605,6 +607,13 @@ def __init__(
             which the nodes _think_ they are, to addresses at which a client may
             reach them, such as when they sit behind a proxy.
 
+        :param maint_notifications_config:
+            Configures the nodes connections to support maintenance notifications - see
+            `redis.maint_notifications.MaintNotificationsConfig` for details.
+            Only supported with RESP3.
+            If not provided and protocol is RESP3, the maintenance notifications
+            will be enabled by default (logic is included in the NodesManager
+            initialization).
          :**kwargs:
              Extra arguments that will be sent into Redis instance when created
              (See Official redis-py doc for supported kwargs - the only limitation
@@ -709,6 +718,7 @@ def __init__(
             cache=cache,
             cache_config=cache_config,
             event_dispatcher=self._event_dispatcher,
+            maint_notifications_config=maint_notifications_config,
             **kwargs,
         )
 
@@ -1622,6 +1632,9 @@ def __init__(
         cache_config: Optional[CacheConfig] = None,
         cache_factory: Optional[CacheFactoryInterface] = None,
         event_dispatcher: Optional[EventDispatcher] = None,
+        maint_notifications_config: Optional[
+            MaintNotificationsConfig
+        ] = MaintNotificationsConfig(),
         **kwargs,
     ):
         self.nodes_cache: Dict[str, Redis] = {}
@@ -1650,6 +1663,7 @@ def __init__(
         self._credential_provider = self.connection_kwargs.get(
             "credential_provider", None
         )
+        self.maint_notifications_config = maint_notifications_config
         self.initialize()
 
     def get_node(self, host=None, port=None, node_name=None):
@@ -1797,7 +1811,10 @@ def create_redis_connections(self, nodes):
         for node in nodes:
             if node.redis_connection is None:
                 node.redis_connection = self.create_redis_node(
-                    host=node.host, port=node.port, **self.connection_kwargs
+                    host=node.host,
+                    port=node.port,
+                    maint_notifications_config=self.maint_notifications_config,
+                    **self.connection_kwargs,
                 )
                 connection_pools.append(node.redis_connection.connection_pool)
 
@@ -1807,7 +1824,12 @@ def create_redis_connections(self, nodes):
             )
         )
 
-    def create_redis_node(self, host, port, **kwargs):
+    def create_redis_node(
+        self,
+        host,
+        port,
+        **kwargs,
+    ):
         # We are configuring the connection pool not to retry
         # connections on lower level clients to avoid retrying
         # connections to nodes that are not reachable
@@ -1821,13 +1843,8 @@ def create_redis_node(self, host, port, **kwargs):
             backoff=NoBackoff(), retries=0, supported_errors=(ConnectionError,)
         )
 
-        protocol = kwargs.get("protocol", None)
-        if protocol in [3, "3"]:
-            kwargs.update(
-                {"maint_notifications_config": MaintNotificationsConfig(enabled=False)}
-            )
         if self.from_url:
-            # Create a redis node with a costumed connection pool
+            # Create a redis node with a custom connection pool
             kwargs.update({"host": host})
             kwargs.update({"port": port})
             kwargs.update({"cache": self._cache})
@@ -1885,7 +1902,10 @@ def initialize(self):
                 else:
                     # Create a new Redis connection
                     r = self.create_redis_node(
-                        startup_node.host, startup_node.port, **kwargs
+                        startup_node.host,
+                        startup_node.port,
+                        maint_notifications_config=self.maint_notifications_config,
+                        **kwargs,
                     )
                     self.startup_nodes[startup_node.name].redis_connection = r
                 # Make sure cluster mode is enabled on this node