Cache Replication: High Availability with Redis Sentinel

import redis

class SentinelCache:
    def __init__(self, sentinel_hosts, master_name="mymaster"):
        self.sentinel = redis.Sentinel(
            sentinel_hosts,
            socket_timeout=0.1,
            sentinel_kwargs={"password": None}
        )
        self.master_name = master_name
        self._master = None
        self._replica = None

    @property
    def master(self):
        """Get the current master connection."""
        if self._master is None:
            self._master = self.sentinel.master_for(self.master_name)
        return self._master

    @property
    def replica(self):
        """Get a replica connection for reads."""
        if self._replica is None:
            self._replica = self.sentinel.slave_for(self.master_name)
        return self._replica

    def set(self, key, value, ttl=3600):
        """Write to master."""
        return self.master.setex(key, ttl, value)

    def get(self, key, allow_stale=True):
        """Read from replica if allowed, otherwise from master."""
        try:
            if allow_stale:
                return self.replica.get(key)
            return self.master.get(key)
        except redis.exceptions.ConnectionError:
            return self.master.get(key)

    def get_master_info(self):
        """Get info about the current master."""
        try:
            info = self.master.info()
            return {
                "role": info["role"],
                "connected_slaves": info.get("connected_slaves", 0),
                "used_memory_human": info.get("used_memory_human"),
                "uptime_in_seconds": info.get("uptime_in_seconds"),
            }
        except Exception as e:
            return {"error": str(e)}

cache = SentinelCache(
    sentinel_hosts=[("127.0.0.1", 26379)],
    master_name="cache-cluster"
)

info = cache.get_master_info()
print(f"Master role: {info.get('role', 'unknown')}")
print(f"Connected replicas: {info.get('connected_slaves', 0)}")
print(f"Memory: {info.get('used_memory_human', 'unknown')}")

cache.set("replication:test", "cache replication works", ttl=300)
value = cache.get("replication:test")
print(f"Read from replica: {value}")

import redis
import random
import time

class ReadScalingCache:
    def __init__(self, replicas):
        """replicas is a list of (host, port) tuples."""
        self.replicas = [redis.Redis(host=h, port=p, decode_responses=True)
                        for h, p in replicas]

    def get(self, key):
        """Read from a random replica for load distribution."""
        replica = random.choice(self.replicas)
        start = time.perf_counter()
        value = replica.get(key)
        elapsed = time.perf_counter() - start
        return {
            "value": value,
            "from": f"{replica.connection_pool.connection_kwargs['host']}:"
                    f"{replica.connection_pool.connection_kwargs['port']}",
            "latency_ms": round(elapsed * 1000, 2)
        }

    def check_consistency(self, key, expected_value):
        """Check if all replicas have the same value for a key."""
        results = {}
        consistent = True
        for replica in self.replicas:
            value = replica.get(key)
            host = replica.connection_pool.connection_kwargs['host']
            port = replica.connection_pool.connection_kwargs['port']
            results[f"{host}:{port}"] = value
            if value != expected_value:
                consistent = False

        return {
            "consistent": consistent,
            "replicas": results,
        }

replicas = [
    ("127.0.0.1", 6380),
    ("127.0.0.1", 6381),
    ("127.0.0.1", 6382),
]

cache = ReadScalingCache(replicas)

for _ in range(5):
    result = cache.get("scaling:test")
    print(f"Read from {result['from']} ({result['latency_ms']}ms): {result['value']}")

consistency = cache.check_consistency("scaling:test", "replicated_value")
print(f"\nReplicas consistent: {consistency['consistent']}")
for replica, value in consistency['replicas'].items():
    print(f"  {replica}: {value}")

import redis
import time

class LagMonitor:
    def __init__(self, master, replicas):
        self.master = redis.Redis(host=master[0], port=master[1])
        self.replicas = [
            redis.Redis(host=h, port=p) for h, p in replicas
        ]

    def measure_lag(self):
        """Measure replication lag for each replica."""
        master_time = self.master.time()
        master_time_seconds = master_time[0] + master_time[1] / 1_000_000

        results = []
        for replica in self.replicas:
            info = replica.info("replication")
            lag = info.get("master_last_io_seconds_ago", -1)

            replica_info = replica.info("server")
            uptime = replica_info.get("uptime_in_seconds", 0)

            results.append({
                "host": replica.connection_pool.connection_kwargs['host'],
                "port": replica.connection_pool.connection_kwargs['port'],
                "lag_seconds": lag,
                "master_link_status": info.get("master_link_status", "unknown"),
                "uptime": uptime,
            })

        return results

    def report(self):
        """Generate a replication health report."""
        results = self.measure_lag()
        print("Replication Lag Report:")
        print(f"{'Host':20s} {'Lag(s)':8s} {'Status':12s} {'Uptime(s)':10s}")
        print("-" * 50)

        for r in results:
            host = f"{r['host']}:{r['port']}"
            status = "UP" if r['master_link_status'] == b"up" else "DOWN"
            print(f"{host:20s} {r['lag_seconds']:8d} {status:12s} {r['uptime']:10d}")

        return results

monitor = LagMonitor(
    master=("127.0.0.1", 6379),
    replicas=[("127.0.0.1", 6380), ("127.0.0.1", 6381)]
)

results = monitor.report()
all_healthy = all(r['lag_seconds'] < 5 and r['master_link_status'] == b"up" for r in results)
print(f"\nAll replicas healthy: {all_healthy}")