mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-11 08:38:24 +00:00
fix(redis): do not treat RedisClusterException subclasses as cluster-down
SlotNotCoveredError, CrossSlotTransactionError, and InvalidPipelineStack share that base but are command/slot/pipeline errors on a live cluster. Matching isinstance(RedisClusterException) opened the breaker and dropped shared cache and rate limits to per-process state. Detect the 2026-08-13 hang by walking __cause__ (redis-py raises RedisClusterException from TimeoutError) and the unreachable-node message when the cause is stripped.
This commit is contained in:
parent
342cdf628e
commit
7ee0b70d58
2 changed files with 88 additions and 13 deletions
|
|
@ -184,16 +184,20 @@ 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.
|
||||
RedisClusterException is *not* in this tuple. redis-py uses that base for
|
||||
command, slot-layout, and pipeline errors (SlotNotCoveredError,
|
||||
CrossSlotTransactionError, InvalidPipelineStack, "db must be 0", …) against
|
||||
a cluster that is still answering. Matching the base would open the breaker
|
||||
and drop shared cache / rate-limit state to per-process counters.
|
||||
|
||||
The CLUSTER SLOTS / startup-node hang is detected in ``_is_redis_health_failure``
|
||||
by walking ``__cause__`` (redis-py does ``raise RedisClusterException(...) from
|
||||
TimeoutError``) and by the unreachable-node message when the cause was lost.
|
||||
|
||||
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, RedisClusterException
|
||||
from redis.exceptions import BusyLoadingError, ClusterDownError
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
|
|
@ -202,16 +206,31 @@ def _redis_health_error_types() -> tuple[type, ...]:
|
|||
RedisTimeoutError,
|
||||
BusyLoadingError,
|
||||
ClusterDownError,
|
||||
RedisClusterException,
|
||||
OSError,
|
||||
asyncio.TimeoutError,
|
||||
)
|
||||
|
||||
|
||||
# redis-py NodesManager.initialize: raise RedisClusterException(
|
||||
# "Redis Cluster cannot be connected. Please provide at least one reachable node: …"
|
||||
# ) from exception
|
||||
_CLUSTER_UNREACHABLE_MARKER: Final = "redis cluster cannot be connected"
|
||||
|
||||
|
||||
def _is_redis_health_failure(exc: BaseException) -> bool:
|
||||
"""True when ``exc`` indicates Redis is unreachable rather than the request being bad."""
|
||||
try:
|
||||
return isinstance(exc, _redis_health_error_types())
|
||||
types: Final = _redis_health_error_types()
|
||||
seen: set[int] = set()
|
||||
current: BaseException | None = exc
|
||||
while current is not None and id(current) not in seen:
|
||||
if isinstance(current, types):
|
||||
return True
|
||||
if _CLUSTER_UNREACHABLE_MARKER in str(current).lower():
|
||||
return True
|
||||
seen.add(id(current))
|
||||
current = current.__cause__ if current.__cause__ is not None else current.__context__
|
||||
return False
|
||||
except ImportError:
|
||||
return True
|
||||
|
||||
|
|
|
|||
|
|
@ -582,9 +582,12 @@ 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"),
|
||||
pytest.param("SlotNotCoveredError", False, id="slot_layout_is_not_unhealthy"),
|
||||
pytest.param("CrossSlotTransactionError", False, id="cross_slot_is_not_unhealthy"),
|
||||
pytest.param("InvalidPipelineStack", False, id="pipeline_stack_is_not_unhealthy"),
|
||||
pytest.param("RedisClusterException", False, id="generic_cluster_command_error_is_not"),
|
||||
],
|
||||
)
|
||||
async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker):
|
||||
|
|
@ -599,7 +602,10 @@ async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker)
|
|||
from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker
|
||||
|
||||
breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
|
||||
raised = getattr(redis.exceptions, error)("boom")
|
||||
error_cls = getattr(redis.exceptions, error, None)
|
||||
if error_cls is None:
|
||||
pytest.skip(f"{error} is not in this redis-py")
|
||||
raised = error_cls("boom")
|
||||
|
||||
async def failing_call():
|
||||
raise raised
|
||||
|
|
@ -611,6 +617,41 @@ async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker)
|
|||
assert breaker.is_open() is opens_breaker
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_redis_cluster_exception_subclasses_do_not_open_the_breaker():
|
||||
"""Slot/pipeline/command errors share the RedisClusterException base.
|
||||
|
||||
Matching that base (isinstance) would let a covered-slot miss or a
|
||||
cross-slot pipeline trip the shared breaker and degrade rate limits to
|
||||
per-process counters while Redis is still answering.
|
||||
"""
|
||||
from redis.exceptions import RedisClusterException
|
||||
|
||||
from litellm.caching.redis_cache import (
|
||||
RedisCircuitBreaker,
|
||||
_is_redis_health_failure,
|
||||
_run_under_circuit_breaker,
|
||||
)
|
||||
|
||||
class SlotLayoutError(RedisClusterException):
|
||||
pass
|
||||
|
||||
class PipelineStackError(RedisClusterException):
|
||||
pass
|
||||
|
||||
for raised in (SlotLayoutError("slot"), PipelineStackError("pipeline")):
|
||||
assert _is_redis_health_failure(raised) is False
|
||||
breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
|
||||
|
||||
async def failing_call(exc=raised):
|
||||
raise exc
|
||||
|
||||
for _ in range(breaker.failure_threshold + 1):
|
||||
with pytest.raises(RedisClusterException):
|
||||
await _run_under_circuit_breaker(breaker, "op", failing_call)
|
||||
assert breaker.is_open() is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cluster_connect_timeout_swallowed_set_opens_breaker():
|
||||
"""Staging 2026-08-13: redis-py wraps connect timeouts as RedisClusterException.
|
||||
|
|
@ -629,16 +670,31 @@ async def test_cluster_connect_timeout_swallowed_set_opens_breaker():
|
|||
_run_under_circuit_breaker,
|
||||
)
|
||||
|
||||
production = RedisClusterException(
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
timeout = RedisTimeoutError("Timeout connecting to server")
|
||||
try:
|
||||
raise RedisClusterException(
|
||||
"Redis Cluster cannot be connected. Please provide at least one reachable node: "
|
||||
"Timeout connecting to server"
|
||||
) from timeout
|
||||
except RedisClusterException as production:
|
||||
wrapped = production
|
||||
|
||||
# Cause lost (message only) still counts — redis-py stringifies the timeout
|
||||
# into the RedisClusterException text even if __cause__ is stripped.
|
||||
message_only = 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
|
||||
assert _is_redis_health_failure(wrapped) is True
|
||||
assert _is_redis_health_failure(message_only) is True
|
||||
assert _is_redis_health_failure(RedisClusterException("Argument 'db' must be 0")) is False
|
||||
|
||||
breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
|
||||
|
||||
async def swallowed_cluster_timeout():
|
||||
_record_swallowed_redis_failure(breaker, production)
|
||||
_record_swallowed_redis_failure(breaker, wrapped)
|
||||
return None
|
||||
|
||||
for _ in range(breaker.failure_threshold):
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue