Merge pull request #5290 from FinnStutzenstein/redisWaitForReplication

Redis: Wait for replication on writes

openslides/utils/cache_providers.py

@@ -8,7 +8,7 @@ from django.core.exceptions import ImproperlyConfigured
 from typing_extensions import Protocol
 
 from . import logging
-from .redis import use_redis
+from .redis import read_only_redis_amount_replicas, use_redis
 from .schema_version import SchemaVersion
 from .utils import split_element_id, str_dict_to_bytes
 
@@ -435,14 +435,23 @@ class RedisCacheProvider:
 
         async with get_connection(read_only=read_only) as redis:
             try:
-                return await redis.evalsha(hash, keys, args)
+                result = await redis.evalsha(hash, keys, args)
             except aioredis.errors.ReplyError as e:
                 if str(e).startswith("NOSCRIPT"):
-                    return await self._eval(redis, script_name, keys=keys, args=args)
+                    result = await self._eval(redis, script_name, keys=keys, args=args)
                 elif str(e) == "cache_reset":
                     raise CacheReset()
                 else:
                     raise e
+            if not read_only and read_only_redis_amount_replicas is not None:
+                reported_amount = await redis.wait(
+                    read_only_redis_amount_replicas, 1000
+                )
+                if reported_amount != read_only_redis_amount_replicas:
+                    logger.warn(
+                        f"WAIT reported {reported_amount} replicas of {read_only_redis_amount_replicas} requested!"
+                    )
+            return result
 
     async def _eval(
         self, redis: Any, script_name: str, keys: List[str] = [], args: List[Any] = []

openslides/utils/redis.py

@@ -7,11 +7,15 @@ from . import logging
 
 logger = logging.getLogger(__name__)
 
+# Defaults
+use_redis = False
+use_read_only_redis = False
+read_only_redis_amount_replicas = None
+
 try:
     import aioredis
 except ImportError:
-    use_redis = False
-    use_read_only_redis = False
+    pass
 else:
     from .redis_connection_pool import ConnectionPool
 
@@ -28,6 +32,9 @@ else:
         if use_read_only_redis:
             logger.info(f"Redis read only address {redis_read_only_address}")
             read_only_pool = ConnectionPool({"address": redis_read_only_address})
+
+            read_only_redis_amount_replicas = getattr(settings, "AMOUNT_REPLICAS", 1)
+            logger.info(f"AMOUNT_REPLICAS={read_only_redis_amount_replicas}")
     else:
         logger.info("Redis is not configured.")
 

openslides/utils/settings.py.tpl

@@ -103,6 +103,8 @@ if use_redis:
     # a (host, port) tuple — ('localhost', 6379);
     # or a unix domain socket path string — "/path/to/redis.sock".
     REDIS_ADDRESS = "redis://127.0.0.1"
+    # REDIS_READ_ONLY_ADDRESS
+    AMOUNT_REPLICAS = 1
 
     # Session backend