Deep Dive: 2026 Redis 8 Hash Slot Internals vs. Dragonfly 1.0 for Clustering

import redis
import time
import logging
from redis.cluster import RedisCluster, ClusterNode
from redis.exceptions import RedisClusterException, ClusterDownException

# Configure logging for migration audit trail
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class Redis8SlotMigrator:
    def __init__(self, source_nodes: list[ClusterNode], target_nodes: list[ClusterNode]):
        '''Initialize migrator with source and target cluster nodes.

        Args:
            source_nodes: List of ClusterNode objects for the source cluster
            target_nodes: List of ClusterNode objects for the target cluster
        '''
        self.source = RedisCluster(startup_nodes=source_nodes, decode_responses=True)
        self.target = RedisCluster(startup_nodes=target_nodes, decode_responses=True)
        self.migration_log = []

    def get_slot_for_key(self, key: str) -> int:
        '''Calculate the hash slot for a given key using Redis's CRC16 implementation.'''
        return self.source.cluster_keyslot(key)

    def migrate_slot(self, slot: int, batch_size: int = 100) -> bool:
        '''Migrate a single hash slot from source to target cluster with batching.

        Args:
            slot: Hash slot to migrate (0-16383)
            batch_size: Number of keys to migrate per batch

        Returns:
            bool: True if migration succeeded, False otherwise
        '''
        try:
            # Get all keys in the slot from source
            keys = self.source.cluster_get_keys_in_slot(slot, batch_size * 10)
            if not keys:
                logger.info(f'Slot {slot} has no keys, skipping migration')
                return True

            logger.info(f'Migrating slot {slot} with {len(keys)} keys to target cluster')
            migrated_count = 0

            # Migrate keys in batches to avoid OOM
            for i in range(0, len(keys), batch_size):
                batch = keys[i:i+batch_size]
                # Use MIGRATE command with COPY and REPLACE flags
                for key in batch:
                    try:
                        # Get key value and TTL from source
                        ttl = self.source.ttl(key)
                        value = self.source.dump(key)
                        if value is None:
                            continue  # Key expired during migration
                        # Restore key on target with original TTL
                        self.target.restore(key, ttl, value, replace=True)
                        # Delete key from source after successful restore
                        self.source.delete(key)
                        migrated_count += 1
                    except Exception as e:
                        logger.error(f'Failed to migrate key {key} in slot {slot}: {str(e)}')
                        self.migration_log.append({'slot': slot, 'key': key, 'error': str(e)})
                        return False

            # Verify slot ownership transferred to target
            target_slots = self.target.cluster_slots()
            for node, slots in target_slots.items():
                if slot in slots:
                    logger.info(f'Slot {slot} successfully migrated to target node {node}')
                    return True

            logger.error(f'Slot {slot} not found on target cluster after migration')
            return False

        except ClusterDownException as e:
            logger.error(f'Cluster down during slot {slot} migration: {str(e)}')
            return False
        except RedisClusterException as e:
            logger.error(f'Redis cluster error during slot {slot} migration: {str(e)}')
            return False
        except Exception as e:
            logger.error(f'Unexpected error migrating slot {slot}: {str(e)}')
            return False

if __name__ == '__main__':
    # Benchmark environment: 3-node Redis 8 cluster on i4i.4xlarge (16 vCPU, 128GB RAM)
    source_nodes = [ClusterNode('10.0.1.10', 6379), ClusterNode('10.0.1.11', 6379), ClusterNode('10.0.1.12', 6379)]
    target_nodes = [ClusterNode('10.0.2.10', 6379), ClusterNode('10.0.2.11', 6379), ClusterNode('10.0.2.12', 6379)]

    migrator = Redis8SlotMigrator(source_nodes, target_nodes)

    # Migrate slots 0-100 as a test batch
    start_time = time.time()
    success_count = 0
    for slot in range(0, 101):
        if migrator.migrate_slot(slot):
            success_count += 1
    end_time = time.time()

    logger.info(f'Migrated {success_count}/101 slots in {end_time - start_time:.2f}s')
    logger.info(f'Migration error log: {migrator.migration_log}')

import requests
import time
import logging
from typing import List, Dict

# Dragonfly 1.0 cluster management client
# Test environment: 3-node Dragonfly cluster on c7g.4xlarge (16 vCPU, 128GB RAM)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DragonflyClusterManager:
    def __init__(self, seed_nodes: List[str]):
        '''Initialize with seed nodes (host:port format).'''
        self.seed_nodes = seed_nodes
        self.cluster_id = None
        self.nodes = []

    def discover_nodes(self) -> List[str]:
        '''Discover all nodes in the Dragonfly cluster via Raft API.'''
        for node in self.seed_nodes:
            try:
                resp = requests.get(f'http://{node}/v1/cluster/nodes', timeout=5)
                resp.raise_for_status()
                nodes = resp.json().get('nodes', [])
                self.nodes = [f'{n["host"]}:{n["port"]}' for n in nodes]
                logger.info(f'Discovered {len(self.nodes)} Dragonfly nodes: {self.nodes}')
                return self.nodes
            except Exception as e:
                logger.warning(f'Failed to discover nodes from {node}: {str(e)}')
        raise RuntimeError('No reachable Dragonfly seed nodes')

    def create_cluster(self, cluster_id: str) -> bool:
        '''Initialize a new Dragonfly cluster with Raft consensus.'''
        if not self.nodes:
            self.discover_nodes()
        try:
            # Use first node as bootstrap node
            resp = requests.post(
                f'http://{self.nodes[0]}/v1/cluster/create',
                json={'cluster_id': cluster_id, 'nodes': self.nodes},
                timeout=10
            )
            resp.raise_for_status()
            self.cluster_id = cluster_id
            logger.info(f'Created Dragonfly cluster {cluster_id} with {len(self.nodes)} nodes')
            return True
        except Exception as e:
            logger.error(f'Failed to create cluster: {str(e)}')
            return False

    def allocate_slots(self, slot_ranges: List[Dict[int, int]]) -> bool:
        '''Allocate hash slot ranges to cluster nodes.

        Args:
            slot_ranges: List of dicts mapping node host:port to (start_slot, end_slot)
        '''
        try:
            payload = []
            for node, (start, end) in slot_ranges.items():
                payload.append({
                    'node': node,
                    'slot_start': start,
                    'slot_end': end
                })
            resp = requests.post(
                f'http://{self.nodes[0]}/v1/cluster/slots/allocate',
                json={'allocations': payload},
                timeout=10
            )
            resp.raise_for_status()
            logger.info(f'Allocated slot ranges: {slot_ranges}')
            return True
        except Exception as e:
            logger.error(f'Failed to allocate slots: {str(e)}')
            return False

    def check_slot_availability(self, slot: int) -> bool:
        '''Verify a hash slot is available and owned by a cluster node.'''
        try:
            resp = requests.get(
                f'http://{self.nodes[0]}/v1/cluster/slots/{slot}',
                timeout=5
            )
            resp.raise_for_status()
            data = resp.json()
            return data.get('owned', False)
        except Exception as e:
            logger.error(f'Failed to check slot {slot} availability: {str(e)}')
            return False

    def simulate_node_failure(self, node: str) -> bool:
        '''Simulate node failure by stopping the Dragonfly process (requires SSH access).'''
        # Note: This is a test utility, requires passwordless SSH to nodes
        import paramiko
        try:
            host, port = node.split(':')
            ssh = paramiko.SSHClient()
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(host, username='ec2-user', timeout=5)
            stdin, stdout, stderr = ssh.exec_command('sudo systemctl stop dragonfly')
            exit_code = stdout.channel.recv_exit_status()
            if exit_code == 0:
                logger.info(f'Stopped Dragonfly on {node} to simulate failure')
                return True
            logger.error(f'Failed to stop Dragonfly on {node}: {stderr.read().decode()}')
            return False
        except Exception as e:
            logger.error(f'SSH failure to {node}: {str(e)}')
            return False

if __name__ == '__main__':
    # 3-node Dragonfly cluster setup
    seed_nodes = ['10.0.3.10:6379', '10.0.3.11:6379', '10.0.3.12:6379']
    manager = DragonflyClusterManager(seed_nodes)

    # Discover existing nodes or create new cluster
    try:
        manager.discover_nodes()
    except RuntimeError:
        manager.create_cluster('df-cluster-01')
        # Allocate 16384 slots evenly across 3 nodes
        slot_ranges = {
            '10.0.3.10:6379': (0, 5461),
            '10.0.3.11:6379': (5462, 10922),
            '10.0.3.12:6379': (10923, 16383)
        }
        manager.allocate_slots(slot_ranges)

    # Verify slot 5461 is owned by first node
    if manager.check_slot_availability(5461):
        logger.info('Slot 5461 is available and owned by cluster node')

    # Simulate node failure and check slot redistribution
    manager.simulate_node_failure('10.0.3.10:6379')
    time.sleep(10)  # Wait for Raft failover
    if not manager.check_slot_availability(5461):
        logger.info('Slot 5461 redistributed after node failure')
    else:
        logger.warning('Slot 5461 not redistributed after node failure')

import redis
import dragonflydb  # Dragonfly Python client: https://github.com/dragonflydb/dragonflydb-py
import time
import statistics
from typing import List, Dict

# Benchmark configuration
BENCH_DURATION = 300  # 5 minutes per test
KEY_COUNT = 1000000
VALUE_SIZE = 1024  # 1KB values
CONCURRENT_CLIENTS = 50

class ClusterBenchmark:
    def __init__(self, backend: str, nodes: List[str]):
        '''Initialize benchmark client for Redis or Dragonfly cluster.

        Args:
            backend: "redis" or "dragonfly"
            nodes: List of host:port strings for cluster nodes
        '''
        self.backend = backend
        self.nodes = nodes
        self.clients = []
        self.throughputs = []
        self.latencies = []

        # Initialize client pool
        if backend == 'redis':
            from redis.cluster import RedisCluster, ClusterNode
            cluster_nodes = [ClusterNode(n.split(':')[0], int(n.split(':')[1])) for n in nodes]
            for _ in range(CONCURRENT_CLIENTS):
                self.clients.append(RedisCluster(startup_nodes=cluster_nodes, decode_responses=False))
        elif backend == 'dragonfly':
            for _ in range(CONCURRENT_CLIENTS):
                # Dragonfly cluster client uses seed nodes for auto-discovery
                self.clients.append(dragonflydb.Dragonfly(host=nodes[0].split(':')[0], port=int(nodes[0].split(':')[1])))
        else:
            raise ValueError(f'Unsupported backend: {backend}')

    def generate_keys(self) -> List[bytes]:
        '''Generate test keys distributed across all hash slots.'''
        keys = []
        for i in range(KEY_COUNT):
            key = f'bench:key:{i}'.encode()
            keys.append(key)
        return keys

    def run_write_benchmark(self) -> Dict:
        '''Run write benchmark with SET commands across hash slots.'''
        keys = self.generate_keys()
        values = [b'x' * VALUE_SIZE for _ in range(KEY_COUNT)]
        start_time = time.time()
        completed = 0

        # Distribute keys to clients round-robin
        for i, (key, value) in enumerate(zip(keys, values)):
            client = self.clients[i % CONCURRENT_CLIENTS]
            try:
                start = time.perf_counter()
                client.set(key, value)
                latency = (time.perf_counter() - start) * 1000  # ms
                self.latencies.append(latency)
                completed += 1
            except Exception as e:
                print(f'Write error: {str(e)}')
            if time.time() - start_time > BENCH_DURATION:
                break

        end_time = time.time()
        duration = end_time - start_time
        throughput = completed / duration  # ops/s
        self.throughputs.append(throughput)

        return {
            'backend': self.backend,
            'ops_completed': completed,
            'duration_s': duration,
            'throughput_ops_s': throughput,
            'p50_latency_ms': statistics.median(self.latencies) if self.latencies else 0,
            'p99_latency_ms': statistics.quantiles(self.latencies, n=100)[98] if len(self.latencies) >= 100 else 0,
            'avg_latency_ms': statistics.mean(self.latencies) if self.latencies else 0
        }

    def run_slot_migration_benchmark(self, slot: int) -> float:
        '''Benchmark time to migrate a single hash slot with 1000 keys.'''
        if self.backend != 'redis':
            raise NotImplementedError('Slot migration only supported for Redis in this benchmark')

        # Populate slot with 1000 keys
        client = self.clients[0]
        for i in range(1000):
            key = f'migrate:slot:{slot}:{i}'.encode()
            client.set(key, b'x' * VALUE_SIZE)

        # Measure migration time
        start = time.perf_counter()
        # Use redis-trib to migrate slot (simplified for benchmark)
        import subprocess
        result = subprocess.run(
            ['redis-trib', 'migrate', '--from', self.nodes[0], '--to', self.nodes[1], '--slot', str(slot)],
            capture_output=True,
            text=True,
            timeout=60
        )
        end = time.perf_counter()

        if result.returncode == 0:
            return (end - start) * 1000  # ms
        else:
            raise RuntimeError(f'Migration failed: {result.stderr}')

if __name__ == '__main__':
    # Hardware: 16-core AMD EPYC 9754, 256GB RAM, 10Gbps network
    redis_nodes = ['10.0.1.10:6379', '10.0.1.11:6379', '10.0.1.12:6379']
    dragonfly_nodes = ['10.0.3.10:6379', '10.0.3.11:6379', '10.0.3.12:6379']

    # Benchmark Redis 8
    print('Running Redis 8 benchmark...')
    redis_bench = ClusterBenchmark('redis', redis_nodes)
    redis_results = redis_bench.run_write_benchmark()
    print(f'Redis 8 Results: {redis_results}')

    # Benchmark Dragonfly 1.0
    print('Running Dragonfly 1.0 benchmark...')
    df_bench = ClusterBenchmark('dragonfly', dragonfly_nodes)
    df_results = df_bench.run_write_benchmark()
    print(f'Dragonfly 1.0 Results: {df_results}')

    # Compare slot migration for Redis
    print('Running Redis slot migration benchmark...')
    migration_time = redis_bench.run_slot_migration_benchmark(1234)
    print(f'Redis 8 slot 1234 migration time: {migration_time:.2f}ms')

redis-cli --cluster check 10.0.1.10:6379 --cluster-search-slots

dragonfly --cluster_mode=yes --cluster_slot_count=4096 --bind=0.0.0.0 --port=6379

def idempotent_migrate_key(client_src, client_tgt, key):
    if client_tgt.exists(key):
        logger.warning(f'Key {key} already exists on target, skipping')
        return False
    value = client_src.dump(key)
    ttl = client_src.ttl(key)
    client_tgt.restore(key, ttl, value, replace=False)
    client_src.delete(key)
    return True