feat: redis support keepalive and retry more error (#38973)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/.env.example

@@ -86,6 +86,10 @@ REDIS_RETRY_BACKOFF_CAP=10.0
 REDIS_SOCKET_TIMEOUT=5.0
 REDIS_SOCKET_CONNECT_TIMEOUT=5.0
 REDIS_HEALTH_CHECK_INTERVAL=30
+REDIS_KEEPALIVE_IDLE=30
+REDIS_KEEPALIVE_INTERVAL=10
+REDIS_KEEPALIVE_COUNT=10
+REDIS_KEEPALIVE=true
 
 # celery configuration
 CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1

api/configs/middleware/cache/redis_config.py

@@ -153,6 +153,11 @@ class RedisConfig(BaseSettings):
         default=30,
     )
 
+    REDIS_KEEPALIVE: bool = Field(default=False, description="Keepalive for Redis connections")
+    REDIS_KEEPALIVE_IDLE: PositiveInt = Field(default=30, description="redis keepalive idle timeout")
+    REDIS_KEEPALIVE_INTERVAL: PositiveInt = Field(default=10, description="redis keepalive interval")
+    REDIS_KEEPALIVE_COUNT: PositiveInt = Field(default=10, description="redis keepalive count")
+
     @field_validator("REDIS_MAX_CONNECTIONS", mode="before")
     @classmethod
     def _empty_string_to_none_for_max_conns(cls, v):

api/extensions/ext_redis.py

@@ -1,6 +1,8 @@
 import functools
 import logging
+import socket
 import ssl
+import sys
 from collections.abc import Callable
 from datetime import timedelta
 from typing import Any, Union, cast
@@ -12,6 +14,7 @@ from redis.cache import CacheConfig
 from redis.client import PubSub
 from redis.cluster import ClusterNode, RedisCluster
 from redis.connection import Connection, SSLConnection
+from redis.exceptions import ConnectionError, TimeoutError
 from redis.retry import Retry
 from redis.sentinel import Sentinel
 from typing_extensions import TypedDict
@@ -234,12 +237,16 @@ class RedisHealthParamsDict(TypedDict):
     socket_timeout: float | None
     socket_connect_timeout: float | None
     health_check_interval: int | None
+    socket_keepalive: bool
+    socket_keepalive_options: dict[int, int]
 
 
 class RedisClusterHealthParamsDict(TypedDict):
     retry: Retry
     socket_timeout: float | None
     socket_connect_timeout: float | None
+    socket_keepalive: bool
+    socket_keepalive_options: dict[int, int]
 
 
 class RedisBaseParamsDict(TypedDict):
@@ -255,6 +262,8 @@ class RedisBaseParamsDict(TypedDict):
     socket_timeout: float | None
     socket_connect_timeout: float | None
     health_check_interval: int | None
+    socket_keepalive: bool
+    socket_keepalive_options: dict[int, int]
 
 
 def _get_ssl_configuration() -> tuple[type[Union[Connection, SSLConnection]], dict[str, Any]]:
@@ -299,16 +308,32 @@ def _get_retry_policy() -> Retry:
             cap=dify_config.REDIS_RETRY_BACKOFF_CAP,
         ),
         retries=dify_config.REDIS_RETRY_RETRIES,
+        supported_errors=(
+            ConnectionError,
+            TimeoutError,
+            BrokenPipeError,
+            OSError,
+        ),
     )
 
 
 def _get_connection_health_params() -> RedisHealthParamsDict:
     """Get connection health and retry parameters for standalone and Sentinel Redis clients."""
+    socket_keepalive_options: dict[int, int] = {}
+    if sys.platform == "linux":
+        socket_keepalive_options[socket.TCP_KEEPIDLE] = dify_config.REDIS_KEEPALIVE_IDLE
+        socket_keepalive_options[socket.TCP_KEEPINTVL] = dify_config.REDIS_KEEPALIVE_INTERVAL
+        socket_keepalive_options[socket.TCP_KEEPCNT] = dify_config.REDIS_KEEPALIVE_COUNT
+    elif sys.platform == "darwin":
+        socket_keepalive_options[socket.TCP_KEEPALIVE] = dify_config.REDIS_KEEPALIVE_IDLE
+
     return RedisHealthParamsDict(
         retry=_get_retry_policy(),
         socket_timeout=dify_config.REDIS_SOCKET_TIMEOUT,
         socket_connect_timeout=dify_config.REDIS_SOCKET_CONNECT_TIMEOUT,
         health_check_interval=dify_config.REDIS_HEALTH_CHECK_INTERVAL,
+        socket_keepalive=dify_config.REDIS_KEEPALIVE,
+        socket_keepalive_options=socket_keepalive_options,
     )
 
 
@@ -325,6 +350,8 @@ def _get_cluster_connection_health_params() -> RedisClusterHealthParamsDict:
         "retry": health_params["retry"],
         "socket_timeout": health_params["socket_timeout"],
         "socket_connect_timeout": health_params["socket_connect_timeout"],
+        "socket_keepalive": health_params["socket_keepalive"],
+        "socket_keepalive_options": health_params["socket_keepalive_options"],
     }
     return result
 
@@ -354,10 +381,14 @@ def _create_sentinel_client(redis_params: RedisBaseParamsDict) -> Union[redis.Re
 
     sentinel_hosts = [(node.split(":")[0], int(node.split(":")[1])) for node in dify_config.REDIS_SENTINELS.split(",")]
 
+    health_params = _get_connection_health_params()
+
     sentinel_kwargs = {
         "socket_timeout": dify_config.REDIS_SENTINEL_SOCKET_TIMEOUT,
         "username": dify_config.REDIS_SENTINEL_USERNAME,
         "password": dify_config.REDIS_SENTINEL_PASSWORD,
+        "socket_keepalive": health_params["socket_keepalive"],
+        "socket_keepalive_options": health_params["socket_keepalive_options"],
     }
 
     if dify_config.REDIS_MAX_CONNECTIONS:

api/tests/unit_tests/extensions/test_redis.py

@@ -69,6 +69,10 @@ class TestGetBaseRedisParams:
         mock_config.REDIS_SOCKET_TIMEOUT = 5.0
         mock_config.REDIS_SOCKET_CONNECT_TIMEOUT = 5.0
         mock_config.REDIS_HEALTH_CHECK_INTERVAL = 30
+        mock_config.REDIS_KEEPALIVE = True
+        mock_config.REDIS_KEEPALIVE_IDLE = 60
+        mock_config.REDIS_KEEPALIVE_INTERVAL = 10
+        mock_config.REDIS_KEEPALIVE_COUNT = 3
 
         params = _get_base_redis_params()
 
@@ -77,6 +81,8 @@ class TestGetBaseRedisParams:
         assert params["socket_timeout"] == 5.0
         assert params["socket_connect_timeout"] == 5.0
         assert params["health_check_interval"] == 30
+        assert params["socket_keepalive"] is True
+        assert isinstance(params["socket_keepalive_options"], dict)
         # Existing params still present
         assert params["db"] == 0
         assert params["encoding"] == "utf-8"

docker/.env.example

@@ -117,6 +117,10 @@ REDIS_RETRY_BACKOFF_CAP=10.0
 REDIS_SOCKET_TIMEOUT=5.0
 REDIS_SOCKET_CONNECT_TIMEOUT=5.0
 REDIS_HEALTH_CHECK_INTERVAL=30
+REDIS_KEEPALIVE=true
+REDIS_KEEPALIVE_IDLE=30
+REDIS_KEEPALIVE_INTERVAL=10
+REDIS_KEEPALIVE_COUNT=10
 CELERY_BROKER_URL=redis://:difyai123456@redis:6379/1
 CELERY_BACKEND=redis
 BROKER_USE_SSL=false

docker/envs/databases/redis.env.example

@@ -29,6 +29,10 @@ REDIS_RETRY_BACKOFF_CAP=10.0
 REDIS_SOCKET_TIMEOUT=5.0
 REDIS_SOCKET_CONNECT_TIMEOUT=5.0
 REDIS_HEALTH_CHECK_INTERVAL=30
+REDIS_KEEPALIVE=true
+REDIS_KEEPALIVE_IDLE=30
+REDIS_KEEPALIVE_INTERVAL=10
+REDIS_KEEPALIVE_COUNT=10
 EVENT_BUS_REDIS_URL=
 EVENT_BUS_REDIS_CHANNEL_TYPE=pubsub
 EVENT_BUS_REDIS_USE_CLUSTERS=false