mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-12 08:42:18 +00:00
fix(redis): treat RedisClusterException as a health failure so a dead cluster cannot hang the worker
redis-py wraps CLUSTER SLOTS connect timeouts as RedisClusterException (subclasses Exception, not TimeoutError). The circuit breaker ignored them, so every cache write retried an unreachable ElastiCache node and the 2026-08-13 staging e2e suite died on 60s ALB read timeouts. Count that exception as unhealthy, and apply fail-fast reconnect defaults (connect timeout, keepalive, health check) on both sync and async RedisCluster clients.
This commit is contained in:
parent
5f2986a1f3
commit
342cdf628e
4 changed files with 92 additions and 5 deletions
|
|
@ -478,6 +478,20 @@ def _get_redis_client_logic(**env_overrides):
|
|||
return redis_kwargs
|
||||
|
||||
|
||||
def _apply_redis_cluster_reconnect_defaults(cluster_kwargs: dict) -> None:
|
||||
"""Fail-fast reconnect defaults so a dropped cluster node cannot stall callers.
|
||||
|
||||
redis-py rediscovers CLUSTER SLOTS on a dead connection. Without a connect
|
||||
timeout, health check, and keepalive that rediscovery blocks for the OS TCP
|
||||
timeout (tens of seconds) — the 2026-08-13 staging e2e 60s ALB hang. Explicit
|
||||
kwargs from config still win.
|
||||
"""
|
||||
cluster_kwargs.setdefault("health_check_interval", REDIS_CLUSTER_HEALTH_CHECK_INTERVAL)
|
||||
cluster_kwargs.setdefault("socket_keepalive", True)
|
||||
cluster_kwargs.setdefault("socket_timeout", REDIS_SOCKET_TIMEOUT)
|
||||
cluster_kwargs.setdefault("socket_connect_timeout", cluster_kwargs["socket_timeout"])
|
||||
|
||||
|
||||
def init_redis_cluster(redis_kwargs) -> redis.RedisCluster:
|
||||
_redis_cluster_nodes_in_env: Final[str | None] = get_secret("REDIS_CLUSTER_NODES")
|
||||
if _redis_cluster_nodes_in_env is not None:
|
||||
|
|
@ -503,6 +517,7 @@ def init_redis_cluster(redis_kwargs) -> redis.RedisCluster:
|
|||
new_startup_nodes.append(ClusterNode(**item))
|
||||
|
||||
cluster_kwargs.pop("startup_nodes", None)
|
||||
_apply_redis_cluster_reconnect_defaults(cluster_kwargs)
|
||||
return redis.RedisCluster(startup_nodes=new_startup_nodes, **cluster_kwargs)
|
||||
|
||||
|
||||
|
|
@ -630,8 +645,7 @@ def get_redis_async_client(
|
|||
# by a cluster restart (e.g. ElastiCache Serverless maintenance) is revalidated and
|
||||
# reconnected before reuse instead of stalling in re-initialization; an explicit value
|
||||
# from config still wins.
|
||||
cluster_kwargs.setdefault("health_check_interval", REDIS_CLUSTER_HEALTH_CHECK_INTERVAL)
|
||||
cluster_kwargs.setdefault("socket_keepalive", True)
|
||||
_apply_redis_cluster_reconnect_defaults(cluster_kwargs)
|
||||
|
||||
# Create async RedisCluster with IAM token as password if available
|
||||
cluster_client: Final = async_redis.RedisCluster(
|
||||
|
|
|
|||
|
|
@ -184,14 +184,28 @@ def _redis_health_error_types() -> tuple[type, ...]:
|
|||
a caller trip the shared breaker on demand, dropping rate limiting to per-process
|
||||
counters that spreading traffic across replicas can outrun.
|
||||
|
||||
RedisClusterException is the wrapper redis-py raises when CLUSTER SLOTS / startup
|
||||
nodes are unreachable (``Redis Cluster cannot be connected ... Timeout connecting
|
||||
to server``). It subclasses ``Exception`` directly, not TimeoutError/ConnectionError,
|
||||
so omitting it leaves every cache write retrying a dead cluster and blocking the
|
||||
request for the connect timeout — the 2026-08-13 staging e2e hang.
|
||||
|
||||
Imported lazily because this module is reachable from a base ``import litellm`` while
|
||||
redis is not a base dependency.
|
||||
"""
|
||||
from redis.exceptions import BusyLoadingError, ClusterDownError
|
||||
from redis.exceptions import BusyLoadingError, ClusterDownError, RedisClusterException
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
return (RedisConnectionError, RedisTimeoutError, BusyLoadingError, ClusterDownError, OSError, asyncio.TimeoutError)
|
||||
return (
|
||||
RedisConnectionError,
|
||||
RedisTimeoutError,
|
||||
BusyLoadingError,
|
||||
ClusterDownError,
|
||||
RedisClusterException,
|
||||
OSError,
|
||||
asyncio.TimeoutError,
|
||||
)
|
||||
|
||||
|
||||
def _is_redis_health_failure(exc: BaseException) -> bool:
|
||||
|
|
|
|||
|
|
@ -582,6 +582,7 @@ async def test_concurrent_success_is_not_cancelled_by_another_calls_failure():
|
|||
pytest.param("ConnectionError", True, id="connection_refused_is_unhealthy"),
|
||||
pytest.param("TimeoutError", True, id="timeout_is_unhealthy"),
|
||||
pytest.param("BusyLoadingError", True, id="loading_is_unhealthy"),
|
||||
pytest.param("RedisClusterException", True, id="cluster_unreachable_is_unhealthy"),
|
||||
pytest.param("ResponseError", False, id="wrong_type_command_is_not"),
|
||||
pytest.param("DataError", False, id="bad_data_is_not"),
|
||||
],
|
||||
|
|
@ -608,3 +609,42 @@ async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker)
|
|||
await _run_under_circuit_breaker(breaker, "op", failing_call)
|
||||
|
||||
assert breaker.is_open() is opens_breaker
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cluster_connect_timeout_swallowed_set_opens_breaker():
|
||||
"""Staging 2026-08-13: redis-py wraps connect timeouts as RedisClusterException.
|
||||
|
||||
``async_set_cache`` swallows the error so a dead cluster degrades instead of
|
||||
500ing. The breaker must still open — RedisClusterException subclasses
|
||||
Exception, not TimeoutError — otherwise every later request retries CLUSTER
|
||||
SLOTS and the worker hangs for the connect timeout (the 60s ALB read timeout).
|
||||
"""
|
||||
from redis.exceptions import RedisClusterException
|
||||
|
||||
from litellm.caching.redis_cache import (
|
||||
RedisCircuitBreaker,
|
||||
_is_redis_health_failure,
|
||||
_record_swallowed_redis_failure,
|
||||
_run_under_circuit_breaker,
|
||||
)
|
||||
|
||||
production = RedisClusterException(
|
||||
"Redis Cluster cannot be connected. Please provide at least one reachable node: "
|
||||
"Timeout connecting to server"
|
||||
)
|
||||
assert _is_redis_health_failure(production) is True
|
||||
|
||||
breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
|
||||
|
||||
async def swallowed_cluster_timeout():
|
||||
_record_swallowed_redis_failure(breaker, production)
|
||||
return None
|
||||
|
||||
for _ in range(breaker.failure_threshold):
|
||||
await _run_under_circuit_breaker(breaker, "async_set_cache", swallowed_cluster_timeout)
|
||||
|
||||
assert breaker.is_open() is True
|
||||
|
||||
with pytest.raises(Exception, match="circuit breaker is open"):
|
||||
await _run_under_circuit_breaker(breaker, "async_set_cache", swallowed_cluster_timeout)
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ from litellm._redis import (
|
|||
get_redis_connection_pool,
|
||||
get_redis_url_from_environment,
|
||||
)
|
||||
from litellm.constants import REDIS_CLUSTER_HEALTH_CHECK_INTERVAL
|
||||
from litellm.constants import REDIS_CLUSTER_HEALTH_CHECK_INTERVAL, REDIS_SOCKET_TIMEOUT
|
||||
from litellm._redis_credential_provider import (
|
||||
GCPIAMCredentialProvider,
|
||||
_token_cache,
|
||||
|
|
@ -195,6 +195,25 @@ def test_async_cluster_sets_reconnect_defaults(mock_cluster_cls):
|
|||
assert call_kwargs["health_check_interval"] == REDIS_CLUSTER_HEALTH_CHECK_INTERVAL
|
||||
assert call_kwargs["health_check_interval"] > 0
|
||||
assert call_kwargs["socket_keepalive"] is True
|
||||
assert call_kwargs["socket_timeout"] == REDIS_SOCKET_TIMEOUT
|
||||
assert call_kwargs["socket_connect_timeout"] == REDIS_SOCKET_TIMEOUT
|
||||
|
||||
|
||||
@patch("litellm._redis.redis.RedisCluster")
|
||||
def test_sync_cluster_sets_reconnect_defaults(mock_cluster_cls):
|
||||
"""Sync RedisCluster must get the same fail-fast reconnect defaults as async.
|
||||
|
||||
init_redis_cluster used to skip them, so a dropped ElastiCache Serverless
|
||||
connection stalled the (often event-loop) caller in CLUSTER SLOTS rediscovery.
|
||||
"""
|
||||
get_redis_client(startup_nodes=[{"host": "cluster-node", "port": 6379}])
|
||||
|
||||
mock_cluster_cls.assert_called_once()
|
||||
call_kwargs = mock_cluster_cls.call_args[1]
|
||||
assert call_kwargs["health_check_interval"] == REDIS_CLUSTER_HEALTH_CHECK_INTERVAL
|
||||
assert call_kwargs["socket_keepalive"] is True
|
||||
assert call_kwargs["socket_timeout"] == REDIS_SOCKET_TIMEOUT
|
||||
assert call_kwargs["socket_connect_timeout"] == REDIS_SOCKET_TIMEOUT
|
||||
|
||||
|
||||
@patch("litellm._redis.async_redis.RedisCluster")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue