RedisCluster support + use same redis client if already setup in redis_queue plugin

.github/workflows/tests.yml

@@ -5,22 +5,30 @@ on:
   pull_request:
 
 env:
-  POETRY_VERSION: 1.1.7
+  POETRY_VERSION: 1.1.11
 
 jobs:
   int:
     name: Integration Tests
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2
-    - name: Run integration tests
+    - name: Run integration tests - Redis host
       run: |
         docker-compose -f ./int/docker-compose.yml run tests
+    - name: Run integration tests - Redis cluster
+      run: |
+        docker network create --subnet=172.28.0.0/24 test_network
+        export REDIS_PASSWORD=test1234
+        export NETWORK_NAME=test_network
+        docker-compose -f ./int/docker-compose.yml down
+        docker-compose -f ./int/docker-compose.cluster.yml run tests
     - name: Print logs on failure
       if: failure()
       run: |
         docker-compose -f ./int/docker-compose.yml logs
+        docker-compose -f ./int/docker-compose.cluster.yml logs
     - name: Clean up integration tests
       if: always()
       run: |
-        docker-compose -f ./int/docker-compose.yml down
+        docker-compose -f ./int/docker-compose.cluster.yml down

.gitignore

@@ -27,6 +27,12 @@ share/python-wheels/
 *.egg
 MANIFEST
 
+###
+### Visual Studio Code
+###
+
+.vscode/
+
 # PyInstaller
 #  Usually these files are written by a python script from a template
 #  before PyInstaller builds the exe, so as to inject date/other infos into it.

README.md

@@ -30,9 +30,21 @@ it is as simple as:
 $ docker-compose up --build
 ```
 
+To launch ACA-Py with an accompanying redis cluster of 6 nodes [3 primararies and 3 replicas], please refer to example [docker-compose.cluster.yml](./docker-compose.cluster.yml) and run the following:
+
+Note: Cluster requires external docker network with specified subnet
+
+```sh
+$ docker network create --subnet=172.28.0.0/24 `network_name`
+$ export REDIS_PASSWORD=" ... As specified in redis_cluster.conf ... "
+$ export NETWORK_NAME="`network_name`"
+$ docker-compose -f docker-compose.cluster.yml up --build
+```
+
 If you are looking to integrate the plugin with your own projects, it is highly
 recommended to take a look at both [docker-compose.yml](./docker-compose.yml)
-and the [ACA-Py default.yml](./docker/default.yml) files to help kickstart your
+and the [ACA-Py default.yml](./docker/default.yml) files for a single redis host setup or at both [docker-compose.cluster.yml](./docker-compose.cluster.yml.yml)
+and the [ACA-Py default_cluster.yml](./docker/default_cluster.yml) files for a redis cluster setup to help kickstart your
 project.
 
 ### Without Docker
@@ -58,6 +70,8 @@ parameters. *Note: You may need to change the redis hostname*
 $ aca-py start --arg-file ./docker/default.yml
 ```
 
+For redis cluster, please review `redis_cluster.conf` and `default_cluster.yml`. Basically `defualt_cluster.yml` includes connection string of a cluster node as `redis_cache.connection`.
+
 For manual testing with a second ACA-Py instance, you can run the following.
 
 ```sh
@@ -83,7 +97,18 @@ configuration options are defined as follows:
 
 ## Running The Integration Tests
 
+### Single redis host
 ```sh
 $ docker-compose -f int/docker-compose.yml build
 $ docker-compose -f int/docker-compose.yml run tests
 ```
+
+### Redis cluster
+Cluster requires external docker network with specified subnet
+```sh
+$ docker network create --subnet=172.28.0.0/24 `network_name`
+$ export REDIS_PASSWORD=" ... As specified in redis_cluster.conf ... "
+$ export NETWORK_NAME="`network_name`"
+$ docker-compose -f int/docker-compose.cluster.yml build
+$ docker-compose -f int/docker-compose.cluster.yml run tests
+```

acapy_cache_redis/v0_1/__init__.py

@@ -16,7 +16,9 @@
 async def setup(context: InjectionContext):
     """Load Redis Base Cache Plugin."""
     LOGGER.debug("Loading Redis Base Cache Plugin")
-    context.injector.bind_instance(BaseCache, RedisBaseCache(context))
+    redis_base_cache_inst = RedisBaseCache(context)
+    await redis_base_cache_inst.check_for_redis_cluster()
+    context.injector.bind_instance(BaseCache, redis_base_cache_inst)
 
 
 __all__ = ["ProblemReport"]

acapy_cache_redis/v0_1/redis_base_cache.py

@@ -1,11 +1,14 @@
+import aioredis
 import json
 import logging
-from typing import Any, Sequence, Text, Union
 
-import aioredis
 from aries_cloudagent.cache.base import BaseCache, CacheKeyLock
 from aries_cloudagent.core.profile import Profile
 from aries_cloudagent.core.error import BaseError
+from redis.asyncio import RedisCluster
+from redis.exceptions import RedisError, RedisClusterException
+from typing import Any, Sequence, Text, Union
+from uuid import uuid4
 
 LOGGER = logging.getLogger(__name__)
 
@@ -22,13 +25,14 @@ def __init__(self, root_profile: Profile) -> None:
         super().__init__()
         # looks like { "key": { "expires": <epoch timestamp>, "value": <val> } }
         """Set initial state."""
-        username = None
-        password = None
-        ca_cert = None
+        self.username = None
+        self.password = None
+        self.ca_cert = None
+        self.root_profile = root_profile
 
         # Get the connection string
         try:
-            plugin_config = root_profile.settings["plugin_config"] or {}
+            plugin_config = self.root_profile.settings["plugin_config"] or {}
             config = plugin_config[self.config_key]
             self.connection = config["connection"]
         except KeyError as error:
@@ -39,22 +43,22 @@ def __init__(self, root_profile: Profile) -> None:
         # Get the credentials for the redis server (for those with ACL enabled)
         try:
             credentials = config["credentials"]
-            username = credentials["username"]
-            password = credentials["password"]
+            self.username = credentials["username"]
+            self.password = credentials["password"]
         except KeyError as error:
             pass
 
         # Get the connection string
         try:
-            max_connections = int(config["max_connections"])
+            self.max_connections = int(config["max_connections"])
         except:
-            max_connections = 50
-        LOGGER.debug(f"Max Redis Cache Pool connections set to: {max_connections}")
+            self.max_connections = 50
+        LOGGER.debug(f"Max Redis Cache Pool connections set to: {self.max_connections}")
 
         # Get the SSL CA Cert information (special redis SSL implementations only)
         try:
             lssl = config["ssl"]
-            ca_cert = lssl["cacerts"]
+            self.ca_cert = lssl["cacerts"]
         except KeyError as error:
             pass
 
@@ -65,13 +69,47 @@ def __init__(self, root_profile: Profile) -> None:
         # Setup the aioredis instance
         self.pool = aioredis.ConnectionPool.from_url(
             self.connection,
-            max_connections=max_connections,
-            username=username,
-            password=password,
-            # ssl_ca_certs=ca_cert,
+            max_connections=self.max_connections,
+            username=self.username,
+            password=self.password,
+            # ssl_ca_certs=self.ca_cert,
         )
         self.redis = aioredis.Redis(connection_pool=self.pool)
 
+    async def check_for_redis_cluster(self):
+        """
+        Check if connection corresponds to a cluster node. If so,
+        reassign redis to redis.asyncio.RedisCluster client.
+
+        """
+        try:
+            # Execute a redis SET command on a fake test_key prefix with b""
+            # value. In case, connection string is that of a single redis
+            # host then it will return None as it doesn't exists. Otherwise,
+            # it will raise a MOVED error.
+            fake_test_key = f"test_key_{str(uuid4())}"
+            await self.redis.set(fake_test_key, b"")
+        except aioredis.exceptions.ResponseError as err:
+            if "MOVED" in str(err):
+                self.redis = self.root_profile.inject_or(RedisCluster)
+                if not self.redis:
+                    self.redis = RedisCluster.from_url(
+                        self.connection,
+                        max_connections=self.max_connections,
+                        username=self.username,
+                        password=self.password,
+                    )
+                    # Binds RedisCluster cluster instance, so that it is
+                    # accessible to redis_queue plugin.
+                    LOGGER.info(
+                        "Found redis connection string correspond to a cluster node,"
+                        " reassigning redis to redis.asyncio.RedisCluster client."
+                    )
+                    self.root_profile.injector.bind_instance(RedisCluster, self.redis)
+                    await self.redis.ping(target_nodes=RedisCluster.PRIMARIES)
+                else:
+                    LOGGER.info("Using an existing provided instance of RedisCluster.")
+
     def _getKey(self, key: Text) -> Text:
         return f"{self.prefix}:{key}"
 
@@ -106,7 +144,11 @@ async def set(self, keys: Union[Text, Sequence[Text]], value: Any, ttl: int = No
             for key in [keys] if isinstance(keys, Text) else keys:
                 # self._cache[key] = {"expires": expires_ts, "value": value}
                 await self.redis.set(self._getKey(key), json.dumps(value), ex=ttl)
-        except aioredis.RedisError as error:
+        except (
+            aioredis.RedisError,
+            RedisClusterException,
+            RedisError,
+        ) as error:
             raise RedisCacheSetKeyValueError(
                 "Unexpected redis client exception"
             ) from error

docker-compose.cluster.yml

@@ -0,0 +1,144 @@
+version: "3"
+services:
+  tunnel:
+    image: dbluhm/agent-tunnel
+    command: -s reverse-proxy:80 -p 4040 -h ${AGENT_TUNNEL_HOST}
+    ports:
+      - 4040:4040
+    networks:
+      - acapy_default
+  agent:
+    image: acapy-cache-redis
+    build:
+      context: ./
+      dockerfile: ./docker/Dockerfile
+    depends_on:
+      - redis-cluster
+    links:
+      - redis-cluster
+      - reverse-proxy
+    ports:
+      - 3000:3000
+      - 3001:3001
+    volumes:
+      - ./acapy-endpoint.sh:/acapy-endpoint.sh:ro,z
+      - ./acapy_cache_redis:/home/indy/acapy_cache_redis:ro,z
+      - ./docker/default_cluster.yml:/home/indy/default.yml:ro,z
+    environment:
+      TUNNEL_ENDPOINT: http://tunnel:4040
+    entrypoint: >
+      /bin/sh -c '/acapy-endpoint.sh poetry run aca-py "$$@"' --
+    command: >
+      start --arg-file default.yml
+    networks:
+      - acapy_default
+  reverse-proxy:
+    image: nginx:alpine
+    restart: unless-stopped
+    environment:
+      AGENT_HTTP: "agent:3000"
+      AGENT_WS: "agent:3002"
+    ports:
+      - 80:80
+    volumes:
+      - ./nginx.conf.template:/etc/nginx/templates/default.conf.template:z
+    networks:
+      - acapy_default
+  redis-cluster:
+    image: redis:latest
+    container_name: cluster
+    command: redis-cli --cluster create 172.28.0.101:6377 172.28.0.102:6378 172.28.0.103:6379 172.28.0.104:6380 172.28.0.105:6381 172.28.0.106:6382 --cluster-replicas 1 --cluster-yes
+    environment:
+      - REDISCLI_AUTH=${REDIS_PASSWORD}
+    networks:
+      acapy_default:
+        ipv4_address: 172.28.0.107
+    depends_on:
+      - redis-node-1
+      - redis-node-2
+      - redis-node-3
+      - redis-node-4
+      - redis-node-5
+      - redis-node-6
+  redis-node-1:
+    image: redis:latest
+    container_name: node1
+    command: [ "redis-server", "/conf/redis.conf", "--port 6377" ]
+    environment:
+      - REDISCLI_AUTH=${REDIS_PASSWORD}
+    ports:
+      - 6377:6377
+    volumes:
+      - ./redis_cluster.conf:/conf/redis.conf
+    networks:
+      acapy_default:
+        ipv4_address: 172.28.0.101
+  redis-node-2:
+    image: redis:latest
+    container_name: node2
+    command: [ "redis-server", "/conf/redis.conf", "--port 6378" ]
+    environment:
+      - REDISCLI_AUTH=${REDIS_PASSWORD}
+    ports:
+      - 6378:6378
+    volumes:
+      - ./redis_cluster.conf:/conf/redis.conf
+    networks:
+      acapy_default:
+        ipv4_address: 172.28.0.102
+  redis-node-3:
+    image: redis:latest
+    container_name: node3
+    command: [ "redis-server", "/conf/redis.conf", "--port 6379" ]
+    environment:
+      - REDISCLI_AUTH=${REDIS_PASSWORD}
+    ports:
+      - 6379:6379
+    volumes:
+      - ./redis_cluster.conf:/conf/redis.conf
+    networks:
+      acapy_default:
+        ipv4_address: 172.28.0.103
+  redis-node-4:
+    image: redis:latest
+    container_name: node4
+    command: [ "redis-server", "/conf/redis.conf", "--port 6380" ]
+    environment:
+      - REDISCLI_AUTH=${REDIS_PASSWORD}
+    ports:
+      - 6380:6380
+    volumes:
+      - ./redis_cluster.conf:/conf/redis.conf
+    networks:
+      acapy_default:
+        ipv4_address: 172.28.0.104
+  redis-node-5:
+    image: redis:latest
+    container_name: node5
+    command: [ "redis-server", "/conf/redis.conf", "--port 6381" ]
+    environment:
+      - REDISCLI_AUTH=${REDIS_PASSWORD}
+    ports:
+      - 6381:6381
+    volumes:
+      - ./redis_cluster.conf:/conf/redis.conf
+    networks:
+      acapy_default:
+        ipv4_address: 172.28.0.105
+  redis-node-6:
+    image: redis:latest
+    container_name: node6
+    command: [ "redis-server", "/conf/redis.conf", "--port 6382" ]
+    environment:
+      - REDISCLI_AUTH=${REDIS_PASSWORD}
+    ports:
+      - 6382:6382
+    volumes:
+      - ./redis_cluster.conf:/conf/redis.conf
+    networks:
+      acapy_default:
+        ipv4_address: 172.28.0.106
+networks:
+  acapy_default:
+    external: true
+    name: ${NETWORK_NAME}

docker/Dockerfile

@@ -5,7 +5,9 @@ ENV POETRY_HOME=/opt/poetry \
     VENV=/usr/src/app/.venv
 ENV PATH="$POETRY_HOME/bin:$VENV/bin:$PATH"
 
-RUN curl -sSL https://install.python-poetry.org | python -
+# RUN curl -sSL https://install.python-poetry.org | python -
+# Fixes installation failure
+RUN pip3 install --no-cache-dir poetry
 USER indy
 RUN poetry config virtualenvs.create true; poetry config virtualenvs.in-project true