Cache Clustering: Horizontal Scaling with Redis Cluster

import redis
import hashlib

class ClusterKeyDistribution:
    def __init__(self):
        self.hash_slots = 16384

    def get_slot(self, key):
        """Calculate the hash slot for a given key."""
        if "{" in key and "}" in key:
            key = key[key.index("{") + 1:key.index("}")]
        crc = hashlib.sha1(key.encode()).digest()
        slot = int.from_bytes(crc[:2], "big") % self.hash_slots
        return slot

    def simulate_distribution(self, keys, num_nodes=6):
        """Simulate how keys distribute across cluster nodes."""
        slots_per_node = self.hash_slots // num_nodes
        distribution = {i: 0 for i in range(num_nodes)}

        for key in keys:
            slot = self.get_slot(key)
            node = slot // slots_per_node
            distribution[node] += 1

        return distribution

dist = ClusterKeyDistribution()

keys = [f"user:{i}" for i in range(100000)]
distribution = dist.simulate_distribution(keys, num_nodes=6)

total = sum(distribution.values())
print(f"Distribution of {total:,} keys across 6 nodes:")
for node, count in sorted(distribution.items()):
    pct = count / total * 100
    print(f"  Node {node}: {count:>6,} keys ({pct:.1f}%)")

print(f"\nMax imbalance: {max(distribution.values()) - min(distribution.values())}")
print(f"Standard deviation in check.")

from redis.cluster import RedisCluster as RedisClusterClient
import random
import string

rc = RedisClusterClient(
    host="127.0.0.1",
    port=7000,
    skip_full_coverage_check=True
)

class ClusterCache:
    def __init__(self, cluster_client):
        self.rc = cluster_client

    def set(self, key, value, ttl=3600):
        """Set a key in the cluster. The client routes automatically."""
        self.rc.setex(key, ttl, value)
        node = self.rc.get_node_from_key(key)
        return {"key": key, "node": node}

    def get(self, key):
        """Get a key from the cluster."""
        value = self.rc.get(key)
        if value is not None:
            node = self.rc.get_node_from_key(key)
            return {"value": value, "node": node}
        return None

    def check_slot_distribution(self, num_keys=10000):
        """Check how keys are distributed across cluster nodes."""
        node_counts = {}
        for i in range(num_keys):
            key = f"key:{random.randint(0, 1000000)}"
            self.rc.setex(key, 3600, f"value_{i}")
            node = self.rc.get_node_from_key(key)
            node_counts[node] = node_counts.get(node, 0) + 1

        total = sum(node_counts.values())
        print(f"\nKey distribution ({total} keys):")
        for node, count in sorted(node_counts.items()):
            print(f"  {node}: {count} keys ({count/total*100:.1f}%)")

cache = ClusterCache(rc)

result = cache.set("cluster:test", "hello cluster")
print(f"Set key on node: {result['node']}")

result = cache.get("cluster:test")
if result:
    print(f"Got value: {result['value']} from node: {result['node']}")

print(f"\nCluster info:")
print(f"  Nodes: {rc.get_nodes()}")
print(f"  Keys: {rc.dbsize()}")

import redis
from redis.cluster import RedisCluster as RedisClusterClient

rc = RedisClusterClient(
    host="127.0.0.1",
    port=7000,
    skip_full_coverage_check=True
)

class TaggedCache:
    def __init__(self, cluster_client):
        self.rc = cluster_client

    def tag_key(self, user_id, resource):
        """Create a hash-tagged key that forces co-location."""
        return f"{{{user_id}}}:{resource}"

    def set_user_data(self, user_id, profile, settings, cart):
        """Store all user-related data on the same cluster node."""
        profile_key = self.tag_key(user_id, "profile")
        settings_key = self.tag_key(user_id, "settings")
        cart_key = self.tag_key(user_id, "cart")

        self.rc.setex(profile_key, 3600, profile)
        self.rc.setex(settings_key, 3600, settings)
        self.rc.setex(cart_key, 300, cart)

        node = self.rc.get_node_from_key(profile_key)
        return {"user_id": user_id, "node": node}

    def verify_co_location(self, user_id):
        """Verify that all keys for a user are on the same node."""
        keys = [
            self.tag_key(user_id, "profile"),
            self.tag_key(user_id, "settings"),
            self.tag_key(user_id, "cart"),
        ]
        nodes = {k: self.rc.get_node_from_key(k) for k in keys}
        unique_nodes = set(nodes.values())
        return {
            "all_same_node": len(unique_nodes) == 1,
            "nodes": nodes,
        }

cache = TaggedCache(rc)

result = cache.set_user_data(
    "user_12345",
    '{"name": "Alice"}',
    '{"theme": "dark"}',
    '{"items": []}'
)
print(f"Data stored on node: {result['node']}")

co_location = cache.verify_co_location("user_12345")
print(f"All keys on same node: {co_location['all_same_node']}")

for key, node in co_location['nodes'].items():
    print(f"  {key} -> {node}")