Merge pull request #5495 from FinnStutzenstein/redisConnectionError

Ping redis connection when popped from pool

.gitignore

@@ -18,6 +18,7 @@ bower_components/*
 
 # OS3+
 /server/
+/haproxy/
 
 # Local user data (settings, database, media, search index, static files)
 personal_data/*

openslides/utils/redis_connection_pool.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 import aioredis
 from channels_redis.core import ConnectionPool as ChannelRedisConnectionPool
@@ -13,6 +13,10 @@ connection_pool_limit = getattr(settings, "CONNECTION_POOL_LIMIT", 100)
 logger.info(f"CONNECTION_POOL_LIMIT={connection_pool_limit}")
 
 
+class InvalidConnection(Exception):
+    pass
+
+
 class ConnectionPool(ChannelRedisConnectionPool):
     """ Adds a trivial, soft limit for the pool """
 
@@ -27,7 +31,34 @@ class ConnectionPool(ChannelRedisConnectionPool):
             await asyncio.sleep(0.1)
         self.counter += 1
 
-        return await super().pop(*args, **kwargs)
+        return await self.pop_ensured_connection(*args, **kwargs)
+
+    async def pop_ensured_connection(
+        self, *args: List[Any], **kwargs: Dict[str, Any]
+    ) -> aioredis.commands.Redis:
+        redis: Optional[aioredis.commands.Redis] = None
+
+        while redis is None:
+            redis = await super().pop(*args, **kwargs)
+
+            try:
+                await self.try_ping(redis)
+            except InvalidConnection:
+                if redis is not None:
+                    super().conn_error(redis)
+                redis = None
+
+        return redis
+
+    async def try_ping(self, redis: aioredis.commands.Redis) -> None:
+        try:
+            pong = await redis.ping()
+            if pong != b"PONG":
+                logger.info("Redis connection invalid, did not recieve PONG")
+                raise InvalidConnection()
+        except (ConnectionRefusedError, ConnectionResetError):
+            logger.info("Redis connection invalid, connection is bad")
+            raise InvalidConnection()
 
     def push(self, conn: aioredis.commands.Redis) -> None:
         super().push(conn)