Redis Cluster and Failover Testing: Patterns for Production Resilience
Redis Cluster is the native horizontal scaling solution for Redis. It partitions data across multiple nodes using consistent hashing (16384 slots), with each node responsible for a subset. When a primary fails, a replica is promoted. This guide covers how to test cluster behavior—slot assignment, redirects, failover, and reconnection—so you can verify your application handles these events correctly.
What Can Go Wrong in a Redis Cluster
Before writing tests, enumerate the failure modes your application must handle:
- MOVED redirect — A key's slot is owned by a different node. Client must retry against the correct node.
- ASK redirect — During slot migration, a key may temporarily be on either the old or new node.
- CLUSTERDOWN — The cluster rejects operations when too many nodes are unavailable.
- Replica failover — Primary node dies, replica is promoted. Brief unavailability window.
- Network partition — Nodes can't reach each other; split-brain risk.
- Hot slots — Uneven key distribution causes load imbalance.
Most client libraries (redis-py, ioredis, go-redis) handle MOVED/ASK automatically. Your job is to verify your application-level behavior: connection pooling, retry logic, timeout handling, and error propagation.
Testing MOVED Redirect Handling
# cluster_client.py
from redis.cluster import RedisCluster
from redis.exceptions import RedisClusterException, ClusterDownError
import json
import logging
logger = logging.getLogger(__name__)
class ClusterCacheClient:
def __init__(self, startup_nodes: list[dict]):
self.client = RedisCluster(
startup_nodes=startup_nodes,
decode_responses=True,
skip_full_coverage_check=True,
socket_timeout=2.0,
retry_on_timeout=True,
max_connections=20
)
def get(self, key: str) -> dict | None:
try:
value = self.client.get(key)
return json.loads(value) if value else None
except ClusterDownError as e:
logger.error(f"Cluster is down: {e}")
raise
except RedisClusterException as e:
logger.error(f"Cluster error on GET {key}: {e}")
raise
def set(self, key: str, value: dict, ttl: int = 300) -> bool:
try:
return self.client.setex(key, ttl, json.dumps(value))
except ClusterDownError as e:
logger.error(f"Cluster is down: {e}")
raise
except RedisClusterException as e:
logger.error(f"Cluster error on SET {key}: {e}")
raise
def pipeline_get_multi(self, keys: list[str]) -> dict:
"""
Get multiple keys. In cluster mode, keys on different slots
can't be in the same pipeline unless using hash tags.
"""
results = {}
for key in keys:
results[key] = self.get(key)
return results# test_cluster_client.py
import pytest
from unittest.mock import MagicMock, patch
from redis.exceptions import ClusterDownError, RedisClusterException
from cluster_client import ClusterCacheClient
import json
@pytest.fixture
def mock_cluster_client():
with patch('cluster_client.RedisCluster') as mock_cls:
mock_instance = MagicMock()
mock_cls.return_value = mock_instance
client = ClusterCacheClient([{'host': 'localhost', 'port': 6379}])
yield client, mock_instance
def test_get_returns_deserialized_value(mock_cluster_client):
client, mock_redis = mock_cluster_client
mock_redis.get.return_value = json.dumps({'id': 1, 'name': 'Alice'})
result = client.get('user:1')
assert result == {'id': 1, 'name': 'Alice'}
def test_get_returns_none_for_missing_key(mock_cluster_client):
client, mock_redis = mock_cluster_client
mock_redis.get.return_value = None
result = client.get('missing:key')
assert result is None
def test_get_propagates_cluster_down_error(mock_cluster_client):
client, mock_redis = mock_cluster_client
mock_redis.get.side_effect = ClusterDownError("CLUSTERDOWN")
with pytest.raises(ClusterDownError):
client.get('any:key')
def test_set_propagates_cluster_exception(mock_cluster_client):
client, mock_redis = mock_cluster_client
mock_redis.setex.side_effect = RedisClusterException("MOVED 1234 127.0.0.1:6380")
with pytest.raises(RedisClusterException):
client.set('any:key', {'data': 'value'})
def test_pipeline_get_multi_handles_keys_on_different_slots(mock_cluster_client):
client, mock_redis = mock_cluster_client
call_count = 0
def get_side_effect(key):
nonlocal call_count
call_count += 1
return json.dumps({'key': key})
mock_redis.get.side_effect = get_side_effect
keys = ['user:1', 'product:1', 'order:1']
results = client.pipeline_get_multi(keys)
assert len(results) == 3
assert call_count == 3 # One GET per key (no pipeline batching across slots)Testing Hash Tags for Multi-Key Operations
Redis Cluster routes keys based on their hash slot. Multi-key operations (MGET, MSET, pipelines) require all keys to be on the same slot. Hash tags ({...}) force a group of keys to the same slot.
# hash_tag_operations.py
from redis.cluster import RedisCluster
import json
r = RedisCluster(startup_nodes=[{'host': 'localhost', 'port': 6379}], decode_responses=True)
def get_user_with_profile(user_id: int) -> dict:
"""
Use hash tags to ensure user data and profile are on the same slot.
Keys: {user:1}:data and {user:1}:profile — both hash to same slot.
"""
pipe = r.pipeline()
pipe.get(f'{{user:{user_id}}}:data')
pipe.get(f'{{user:{user_id}}}:profile')
data_str, profile_str = pipe.execute()
return {
'data': json.loads(data_str) if data_str else None,
'profile': json.loads(profile_str) if profile_str else None
}
def set_user_with_profile(user_id: int, data: dict, profile: dict):
pipe = r.pipeline()
pipe.set(f'{{user:{user_id}}}:data', json.dumps(data))
pipe.set(f'{{user:{user_id}}}:profile', json.dumps(profile))
pipe.execute()# test_hash_tags.py
def test_hash_tagged_keys_are_on_same_slot():
"""Verify that hash-tagged keys hash to the same CRC16 slot."""
from redis.cluster import get_node_name
import binascii
def crc16(data: str) -> int:
return binascii.crc_hqx(data.encode(), 0) % 16384
def get_hash_slot(key: str) -> int:
# Redis uses the content between {} if present
start = key.find('{')
end = key.find('}', start + 1)
if start != -1 and end != -1 and end > start + 1:
hash_key = key[start + 1:end]
else:
hash_key = key
return crc16(hash_key)
user_id = 42
data_slot = get_hash_slot(f'{{user:{user_id}}}:data')
profile_slot = get_hash_slot(f'{{user:{user_id}}}:profile')
orders_slot = get_hash_slot(f'{{user:{user_id}}}:orders')
assert data_slot == profile_slot == orders_slot, \
"All hash-tagged keys should map to the same slot"
def test_untagged_keys_likely_on_different_slots():
"""
Without hash tags, keys for the same user land on different slots,
preventing multi-key operations.
"""
import binascii
def crc16(data: str) -> int:
return binascii.crc_hqx(data.encode(), 0) % 16384
slot1 = crc16('user:1:data')
slot2 = crc16('user:1:profile')
# These will almost certainly be on different slots
# (probability ~16383/16384 they differ)
# Document this behavior for developers
print(f"user:1:data → slot {slot1}")
print(f"user:1:profile → slot {slot2}")
# No hard assertion — this is a documentation testTesting Failover with Testcontainers
Setting up a real Redis Cluster requires multiple containers. Here's a minimal 3-node cluster setup:
# conftest_cluster.py
import subprocess
import time
import redis
from redis.cluster import RedisCluster
import pytest
def create_redis_cluster_docker():
"""
Use docker-compose or direct container management to create a 3-node cluster.
This is a simplified example — in practice use docker-compose.
"""
# Using the convenient redis/redis-stack or bitnami/redis-cluster image
result = subprocess.run([
'docker', 'run', '-d',
'--name', 'redis-cluster-test',
'-p', '7000-7005:7000-7005',
'-e', 'ALLOW_EMPTY_PASSWORD=yes',
'-e', 'REDIS_CLUSTER_ENABLED=yes',
'bitnami/redis-cluster:latest'
], capture_output=True, text=True)
return result.stdout.strip()
@pytest.fixture(scope='session')
def redis_cluster():
container_id = create_redis_cluster_docker()
# Wait for cluster to be ready
deadline = time.time() + 30
cluster_client = None
while time.time() < deadline:
try:
cluster_client = RedisCluster(
startup_nodes=[{'host': 'localhost', 'port': 7000}],
decode_responses=True,
skip_full_coverage_check=True
)
cluster_client.ping()
break
except Exception:
time.sleep(0.5)
if not cluster_client:
pytest.skip("Redis cluster not available")
yield cluster_client
subprocess.run(['docker', 'rm', '-f', container_id])Testing Application Behavior During Failover
The most important failover tests verify that your application retries correctly and doesn't fail the user request:
# test_failover_resilience.py
import pytest
from unittest.mock import MagicMock, patch, call
from redis.exceptions import ConnectionError, ClusterDownError, TimeoutError
import time
def test_client_retries_on_connection_error(mock_cluster_client):
client, mock_redis = mock_cluster_client
# Simulate: first call fails (node down), second succeeds (after failover)
mock_redis.get.side_effect = [
ConnectionError("Connection refused"),
'{"id": 1}'
]
# Application should retry internally
# This tests that your retry wrapper works
from cluster_client import ClusterCacheClient
call_count = 0
original_get = mock_redis.get
def retry_get(key, max_retries=3, delay=0.01):
for attempt in range(max_retries):
try:
return original_get(key)
except ConnectionError:
if attempt == max_retries - 1:
raise
time.sleep(delay)
result = retry_get('user:1')
assert result == '{"id": 1}'
assert mock_redis.get.call_count == 2
def test_circuit_breaker_opens_after_threshold(mock_cluster_client):
"""
Test that a circuit breaker prevents cascading failures
when Redis is down.
"""
client, mock_redis = mock_cluster_client
mock_redis.get.side_effect = ConnectionError("Redis down")
failure_count = 0
FAILURE_THRESHOLD = 5
for _ in range(FAILURE_THRESHOLD + 2):
try:
client.get('any:key')
except (ConnectionError, ClusterDownError):
failure_count += 1
# After threshold, should open circuit (stop hitting Redis)
# This tests your circuit breaker implementation
assert failure_count >= FAILURE_THRESHOLD
def test_timeout_handled_gracefully(mock_cluster_client):
client, mock_redis = mock_cluster_client
mock_redis.get.side_effect = TimeoutError("Timeout")
with pytest.raises(TimeoutError):
client.get('slow:key')
def test_fallback_to_database_when_cluster_down(monkeypatch):
"""
Your application should fall back to the database when Redis is unavailable.
This is the most critical resilience test.
"""
db_result = {'id': 1, 'name': 'Alice'}
db_fetch = MagicMock(return_value=db_result)
with patch('cluster_client.RedisCluster') as mock_cls:
mock_instance = MagicMock()
mock_instance.get.side_effect = ClusterDownError("CLUSTERDOWN")
mock_cls.return_value = mock_instance
from resilient_cache import get_with_fallback
monkeypatch.setattr('resilient_cache.fetch_from_db', db_fetch)
result = get_with_fallback('user:1')
assert result == db_result
db_fetch.assert_called_once()Testing Slot Distribution and Hot Key Detection
# test_slot_distribution.py
def test_key_distribution_across_slots():
"""
Verify that your key naming convention distributes across slots.
Poor distribution causes hot nodes and performance issues.
"""
import binascii
def crc16(data: str) -> int:
return binascii.crc_hqx(data.encode(), 0) % 16384
# Simulate 1000 keys
import random
keys = [f'user:{random.randint(1, 100000)}' for _ in range(1000)]
slots = [crc16(k) for k in keys]
# Divide into 3 equal ranges (3 nodes with ~5461 slots each)
node_counts = [0, 0, 0]
for slot in slots:
if slot < 5461:
node_counts[0] += 1
elif slot < 10923:
node_counts[1] += 1
else:
node_counts[2] += 1
total = sum(node_counts)
# Each node should handle roughly 33% of keys
# Allow 20% deviation from ideal
for count in node_counts:
percentage = count / total
assert 0.20 <= percentage <= 0.47, \
f"Node handling {percentage:.1%} of keys — distribution is uneven"
def test_hot_key_pattern_detected():
"""
Detect if a key pattern would create hot slots.
Example: all keys sharing same hash tag land on same slot.
"""
import binascii
def crc16(data: str) -> int:
return binascii.crc_hqx(data.encode(), 0) % 16384
def get_hash_slot(key: str) -> int:
start = key.find('{')
end = key.find('}', start + 1)
if start != -1 and end != -1 and end > start + 1:
return crc16(key[start + 1:end])
return crc16(key)
# BAD pattern: all session keys use same hash tag
bad_pattern_slots = set(
get_hash_slot(f'{{sessions}}:user:{i}') for i in range(1000)
)
# All map to the same slot — this is a hot slot
assert len(bad_pattern_slots) == 1, "Expected all to be on same slot (hot key risk)"
# GOOD pattern: include user ID in hash tag
good_pattern_slots = set(
get_hash_slot(f'{{user:{i}}}:session') for i in range(1000)
)
# Spread across many slots
assert len(good_pattern_slots) > 100, "Expected good distribution across slots"Chaos Testing for Redis Cluster
Chaos tests deliberately introduce failures to verify system behavior:
# test_chaos_redis.py
import threading
import time
import random
import fakeredis
import pytest
from unittest.mock import patch, MagicMock
from redis.exceptions import ConnectionError, TimeoutError
class ChaosRedis:
"""Wraps a Redis client and randomly injects failures."""
def __init__(self, real_client, failure_rate: float = 0.1):
self._client = real_client
self.failure_rate = failure_rate
self.injected_failures = 0
def __getattr__(self, name):
attr = getattr(self._client, name)
if callable(attr):
def wrapper(*args, **kwargs):
if random.random() < self.failure_rate:
self.injected_failures += 1
raise ConnectionError(f"Chaos: simulated failure on {name}")
return attr(*args, **kwargs)
return wrapper
return attr
def test_application_survives_10_percent_failure_rate(monkeypatch):
"""
Under 10% Redis failure rate, application should succeed
on at least 80% of requests (accounting for retries).
"""
fake_r = fakeredis.FakeRedis(decode_responses=True)
chaos = ChaosRedis(fake_r, failure_rate=0.1)
import cache_service
monkeypatch.setattr(cache_service, 'r', chaos)
db_fetch = MagicMock(return_value={'data': 'value'})
monkeypatch.setattr(cache_service, 'fetch_from_db', db_fetch)
successes = 0
failures = 0
for i in range(100):
try:
cache_service.get_user(i % 10)
successes += 1
except Exception:
failures += 1
success_rate = successes / 100
assert success_rate >= 0.80, \
f"Success rate {success_rate:.1%} below acceptable threshold"
print(f"Injected {chaos.injected_failures} failures, "
f"success rate: {success_rate:.1%}")Key Takeaways
- Test MOVED/ASK handling by verifying your client library is configured with
retry_on_errorand watching that it doesn't surface these as user-visible errors. - Test failover recovery by simulating
ConnectionErrorfrom one node and verifying your application retries or falls back gracefully. - Test hash tag correctness with pure slot calculation—no Redis needed. Verify that keys that must be co-located actually hash to the same slot.
- Test key distribution to detect hot spots before they hit production.
- Chaos tests with controlled failure injection rates give you confidence in your retry and fallback logic without requiring a full cluster teardown.
Redis Cluster adds operational complexity. Your tests should document and verify every assumption your application makes about cluster topology, failure modes, and recovery behavior.
