Redis Cluster 把键分到不同的槽位

redis_lab=$(mktemp -d)
curl --fail --location --output "$redis_lab/redis.tar.gz" https://download.redis.io/releases/redis-8.10.1.tar.gz
tar -xzf "$redis_lab/redis.tar.gz" -C "$redis_lab"
make -C "$redis_lab/redis-8.10.1" build redis -j2 MALLOC=libc
export REDIS_SERVER="$redis_lab/redis-8.10.1/src/redis-server"
python3 -m venv "$redis_lab/venv"
"$redis_lab/venv/bin/python" -m pip install redis==8.1.0

from contextlib import contextmanager, ExitStack
import os
from pathlib import Path
import signal
import socket
import subprocess
from tempfile import TemporaryDirectory
import time

import redis
from redis.backoff import NoBackoff
from redis.crc import key_slot
from redis.exceptions import ClusterCrossSlotError, ClusterDownError, MovedError
from redis.retry import Retry


def until(check):
    deadline = time.monotonic() + 15
    while not check():
        assert time.monotonic() < deadline, "cluster state did not converge"
        time.sleep(0.05)


@contextmanager
def node(directory, port, bus):
    directory.mkdir()
    endpoint = str(directory / "redis.sock")
    process = subprocess.Popen([
        os.environ["REDIS_SERVER"], "--bind", "127.0.0.1", "--port", str(port),
        "--protected-mode", "yes", "--unixsocket", endpoint,
        "--unixsocketperm", "700", "--dir", str(directory),
        "--save", "", "--appendonly", "no", "--cluster-enabled", "yes",
        "--cluster-config-file", "nodes.conf", "--cluster-port", str(bus),
        "--cluster-announce-ip", "127.0.0.1", "--cluster-node-timeout", "1000",
        "--cluster-require-full-coverage", "yes",
    ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
    client = redis.Redis(unix_socket_path=endpoint, decode_responses=False, protocol=2,
                         socket_timeout=2, retry=Retry(NoBackoff(), 0))
    try:
        deadline = time.monotonic() + 5
        while True:
            assert process.poll() is None, "Redis exited during startup"
            try:
                if client.ping():
                    break
            except redis.ConnectionError:
                assert time.monotonic() < deadline, "Redis did not start"
                time.sleep(0.02)
        assert client.info("server")["process_id"] == process.pid
        yield client, process
    finally:
        client.close()
        if process.poll() is None:
            process.terminate()
        try:
            _, errors = process.communicate(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
            process.communicate()
            raise
        assert process.returncode in (0, -signal.SIGKILL), errors.decode()


with TemporaryDirectory(prefix="redis-cluster-", dir="/tmp") as directory, ExitStack() as stack:
    leases = [stack.enter_context(socket.socket()) for _ in range(6)]
    for lease in leases:
        lease.bind(("127.0.0.1", 0))
    ports = [lease.getsockname()[1] for lease in leases]
    nodes = []
    for index in range(3):
        leases[2 * index].close()
        leases[2 * index + 1].close()
        nodes.append(stack.enter_context(node(
            Path(directory) / str(index), ports[2 * index], ports[2 * index + 1],
        )))
    clients = [item[0] for item in nodes]
    ranges = [[(0, 4095), (8192, 12287)], [(4096, 8191)], [(12288, 16383)]]
    for client, owned in zip(clients, ranges):
        bounds = [value for interval in owned for value in interval]
        assert client.execute_command("CLUSTER", "ADDSLOTSRANGE", *bounds) == b"OK"
    for index in (1, 2):
        assert clients[0].execute_command(
            "CLUSTER", "MEET", "127.0.0.1", ports[2 * index], ports[2 * index + 1],
        ) == b"OK"
    until(lambda: all(b"cluster_state:ok" in client.execute_command("CLUSTER", "INFO")
                      for client in clients))
    shards = [dict(zip(item[::2], item[1::2], strict=True))
              for item in clients[0].execute_command("CLUSTER", "SHARDS")]
    assert len(shards) == 3
    assert sorted(interval for shard in shards
                  for interval in zip(shard[b"slots"][::2], shard[b"slots"][1::2])) == [
        (0, 4095), (4096, 8191), (8192, 12287), (12288, 16383),
    ]
    first_id = clients[0].execute_command("CLUSTER", "MYID")
    first_shard = next(shard for shard in shards
                       if first_id in [dict(zip(item[::2], item[1::2]))[b"id"]
                                       for item in shard[b"nodes"]])
    assert first_shard[b"slots"] == [0, 4095, 8192, 12287]
    print("16384 slots are covered; one primary owns two noncontiguous ranges")

    keys = [b"profile:{42}", b"quota:{42}"]
    slot = key_slot(keys[0])
    assert key_slot(keys[1]) == slot
    assert clients[0].execute_command("CLUSTER", "KEYSLOT", keys[0]) == slot
    owner = next(index for index, owned in enumerate(ranges)
                 if any(start <= slot <= end for start, end in owned))
    wrong = clients[(owner + 1) % 3]
    try:
        wrong.get(keys[0])
    except MovedError as error:
        assert error.slot_id == slot
        assert error.node_addr == ("127.0.0.1", ports[2 * owner])
    else:
        raise AssertionError("a plain client on the wrong primary must receive MOVED")

    routed = redis.RedisCluster(host="127.0.0.1", port=ports[0],
                                socket_timeout=2, retry=Retry(NoBackoff(), 0))
    try:
        assert routed.mset({keys[0]: "Ada", keys[1]: "10"})
        assert routed.mget(keys) == [b"Ada", b"10"]
        print("MOVED names the owner; a cluster client routes a same-slot MSET and MGET")
        other = b"unrelated:2"
        assert key_slot(other) != slot
        assert any(start <= key_slot(other) <= end for start, end in ranges[owner])
        try:
            clients[owner].execute_command("MSET", keys[0], "bad", other, "bad")
        except ClusterCrossSlotError:
            pass
        else:
            raise AssertionError("cross-slot MSET must fail")
        assert routed.get(keys[0]) == b"Ada"
        assert routed.get(other) is None
        print("cross-slot MSET fails even on the same primary, without changing either key")
    finally:
        routed.close()

    assert nodes[0][1].poll() is None
    nodes[0][1].kill()
    assert nodes[0][1].wait(timeout=5) == -signal.SIGKILL
    until(lambda: all(b"cluster_state:fail" in client.execute_command("CLUSTER", "INFO")
                      for client in clients[1:]))
    try:
        clients[1].get(b"any-key")
    except ClusterDownError:
        print("without replicas, a missing primary stops key service under full coverage")
    else:
        raise AssertionError("an uncovered cluster must reject key commands")