"""Один кластер на два датацентра: что делает каждая половина при разрыве.

ПОЧЕМУ ЭТО ГЛАВНЫЙ ЗАМЕР СТАТЬИ. Просьба звучит как «разложите узлы поровну
между ДЦ, чтобы пережить потерю любого из них». Звучит симметрично и потому
убедительно. Замер показывает, что именно симметрия и убивает: решение о
том, кто продолжает работать, кластер принимает большинством голосов
мастеров, а у ровно поделённого пополам кластера большинства нет НИ У КОГО.

ЧТО ИМЕННО ПРОВЕРЯЕТСЯ. Три расклада на шести узлах:
  A. 3 мастера в ДЦ-1, 3 реплики в ДЦ-2 — «реплики про запас»;
  B. по 3 узла в каждом ДЦ так, что мастера поделены 2:1, и реплика третьего
     мастера осталась при нём;
  C. то же деление узлов, но реплика третьего мастера перенесена в ДЦ-1.

КАЖДАЯ СТОРОНА МЕРЯЕТСЯ НА СВОЁМ КЛАСТЕРЕ, И ЭТО ИСПРАВЛЕНИЕ. В первой версии
замера обе половины опрашивались по очереди на одном кластере: остановили
дальнюю, спросили ближнюю, отпустили, остановили ближнюю, спросили дальнюю.
Разбор снаружи указал на дыру, и указал верно. После первой проверки топология
не обязана вернуться к исходной: в раскладе C первая половина успевает
ПОВЫСИТЬ реплику, сменить configEpoch и переназначить слоты, — и вторая
половина меряется уже на другом кластере, чем задумано. Две последовательные
проверки нельзя называть двумя сторонами одного разрыва.

Поэтому здесь на каждую сторону поднимается свой кластер с той же топологией,
и топология печатается ДО разрыва и ПОСЛЕ него — чтобы смену ролей было видно,
а не приходилось предполагать.

ТОПОЛОГИЯ ЗАДАЁТСЯ РУКАМИ, А НЕ `--cluster-replicas`. Иначе расклады B и C
неповторимы: кто чья реплика, решает утилита, и на втором кластере решение
может быть другим. Здесь слоты раздаются через `CLUSTER ADDSLOTS`, а пары —
через `CLUSTER REPLICATE`, поэтому «тот же расклад» означает буквально тот же.

ЗАПУСК: python3 bench/stretched-cache/cluster.py
Нужны redis-server и redis-cli в PATH (или в REDIS_SERVER_BIN/REDIS_CLI_BIN).
Занимает несколько минут: кластеров поднимается шесть, и каждому надо дать
время на выборы (cluster-node-timeout ниже).
"""

from __future__ import annotations

import os
import shutil
import socket
import subprocess
import sys
import tempfile
import time

import redis

# Какой сервер запускать. По умолчанию тот, что в PATH; REDIS_SERVER_BIN и
# REDIS_CLI_BIN позволяют прогнать тот же замер на другой сборке — так снята
# таблица регрессии на Redis 8.10.1 и smoke-прогон на Valkey.
SERVER = os.environ.get("REDIS_SERVER_BIN", "redis-server")
CLI = os.environ.get("REDIS_CLI_BIN", "redis-cli")

NODE_TIMEOUT_MS = 2000
SETTLE_S = 8.0
SLOT_RANGES = ((0, 5460), (5461, 10922), (10923, 16383))
PROBE = "dc:probe"


def free_port() -> int:
    """Свободный порт НИЖЕ 55535 — и это не придирка.

    В режиме кластера узел слушает два порта: обычный и шинный, ровно на
    десять тысяч больше. Порт из эфемерного диапазона (на этой машине он
    доходит до 60999) даёт шинный порт больше 65535, и узел не поднимается.
    Поймано именно так: замер падал на «узел не поднялся», а в журнале узла
    стояло `Could not create server TCP listening socket`.
    """
    for _ in range(500):
        s = socket.socket()
        s.bind(("127.0.0.1", 0))
        port = s.getsockname()[1]
        s.close()
        if port < 55535:
            return port
    raise RuntimeError("не нашлось свободного порта ниже 55535")


def conn(port: int, timeout: float = 2.0) -> redis.Redis:
    return redis.Redis(port=port, socket_timeout=timeout, decode_responses=True)


class Cluster:
    """Шесть узлов с топологией, заданной вызывающим."""

    def __init__(self) -> None:
        self.tmp = tempfile.mkdtemp(prefix="cluster-")
        self.ports: list[int] = []
        self.procs: dict[int, subprocess.Popen] = {}

    def start(self, count: int = 6) -> list[int]:
        for _ in range(count):
            port = free_port()
            d = os.path.join(self.tmp, str(port))
            os.makedirs(d, exist_ok=True)
            proc = subprocess.Popen([
                SERVER,
                "--port", str(port),
                "--dir", d,
                "--cluster-enabled", "yes",
                "--cluster-config-file", f"nodes-{port}.conf",
                "--cluster-node-timeout", str(NODE_TIMEOUT_MS),
                "--save", "",
                "--appendonly", "no",
                "--logfile", os.path.join(d, "redis.log"),
            ])
            deadline = time.time() + 15
            while time.time() < deadline:
                try:
                    c = conn(port, 0.5)
                    if c.ping():
                        c.close()
                        break
                except Exception:
                    time.sleep(0.05)
            else:
                raise RuntimeError(f"узел {port} не поднялся")
            self.ports.append(port)
            self.procs[port] = proc
        return self.ports

    def form(self, pairing: list[tuple[int, int]]) -> None:
        """pairing — список пар (мастер, его реплика), ровно три."""
        masters = [m for m, _ in pairing]
        first = self.ports[0]
        for p in self.ports[1:]:
            conn(first).execute_command("CLUSTER", "MEET", "127.0.0.1", p)
        self._await(lambda: all(
            len(conn(p).execute_command("CLUSTER", "NODES").strip().splitlines()) == len(self.ports)
            for p in self.ports), 30, "узлы не увидели друг друга")

        for port, (lo, hi) in zip(masters, SLOT_RANGES):
            conn(port).execute_command("CLUSTER", "ADDSLOTS", *range(lo, hi + 1))

        ids = {p: conn(p).execute_command("CLUSTER", "MYID") for p in self.ports}
        self._await(lambda: all(
            len(conn(p).execute_command("CLUSTER", "SLOTS")) == 3 for p in self.ports),
            30, "слоты не разошлись по узлам")
        for master, replica in pairing:
            conn(replica).execute_command("CLUSTER", "REPLICATE", ids[master])

        # ЖДЁМ ДВА ПРИЗНАКА, А НЕ ОДИН. `cluster_state:ok` означает, что слоты
        # покрыты; `master_link_status:up` — что реплике есть чем повышаться.
        # Между ними несколько секунд, и мастер, упавший внутри этого окна, не
        # заменяется вовсе. Если ждать только первый признак, замер измеряет
        # это окно, а не расклад.
        self._await(lambda: all(state_of(p) == "ok" for p in self.ports),
                    60, "кластер не собрался")
        replicas = [r for _, r in pairing]
        self._await(lambda: all(
            conn(p).info("replication").get("master_link_status") == "up" for p in replicas),
            60, "реплики не подключились к мастерам")

    @staticmethod
    def _await(cond, timeout: float, message: str) -> None:
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                if cond():
                    return
            except Exception:
                pass
            time.sleep(0.2)
        raise RuntimeError(message)

    def stop(self, ports: list[int]) -> None:
        for p in ports:
            subprocess.run(["kill", "-STOP", str(self.procs[p].pid)], check=False)

    def destroy(self) -> None:
        for proc in self.procs.values():
            subprocess.run(["kill", "-CONT", str(proc.pid)], check=False)
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
        shutil.rmtree(self.tmp, ignore_errors=True)


def state_of(port: int) -> str:
    try:
        return conn(port, 1.0).cluster("INFO")["cluster_state"]
    except Exception:
        return "недоступен"


def can_read(port: int, key: str) -> str:
    """Что получает клиент на обычный GET: значение, отказ или молчание."""
    try:
        value = conn(port, 1.5).get(key)
        return "ответил" if value is not None else "ответил (пусто)"
    except redis.exceptions.ClusterDownError:
        return "CLUSTERDOWN"
    except (redis.exceptions.MovedError, redis.exceptions.AskError):
        # Слот не на этом узле. Для нашего вопроса это «жив и отвечает»:
        # клиент кластера пойдёт по указанному адресу.
        return "отдал слот другому узлу"
    except redis.exceptions.ResponseError as exc:
        return str(exc).split()[0]
    except Exception:
        return "таймаут"


def topology(ports: list[int], view_from: int | None = None) -> dict[int, str]:
    """Кто мастер, а кто чья реплика — по идентификаторам узлов кластера."""
    c = conn(view_from if view_from is not None else ports[0])
    raw = c.execute_command("CLUSTER", "NODES")
    by_id: dict[str, int] = {}
    rows = []
    for line in raw.strip().splitlines():
        f = line.split()
        node_id, addr, flags, master_id = f[0], f[1], f[2], f[3]
        port = int(addr.split("@")[0].split(":")[1])
        by_id[node_id] = port
        rows.append((port, flags, master_id))
    out = {}
    for port, flags, master_id in rows:
        if "master" in flags:
            out[port] = "мастер"
        else:
            out[port] = f"реплика {by_id.get(master_id, '?')}"
    return out


def roles(ports: list[int]) -> dict[int, str]:
    out = {}
    for p in ports:
        try:
            out[p] = conn(p, 1.0).info("replication")["role"]
        except Exception:
            out[p] = "?"
    return out


def eligibility(replicas: list[int]) -> None:
    """Право реплики повыситься — то, чего мало кто проверяет заранее.

    Условие из спецификации («большинство мастеров плюс доступная реплика
    каждого потерянного») необходимо, но само по себе не достаточно: реплика
    ещё должна быть достаточно свежей. Свежесть проверяется настройкой
    `cluster-replica-validity-factor`, и здесь печатается то, из чего эта
    проверка считается.
    """
    for p in replicas:
        try:
            c = conn(p)
            info = c.info("replication")
            factor = c.config_get("cluster-replica-validity-factor")
            factor = list(factor.values())[0] if factor else "?"
            print(f"       узел {p}: канал {info.get('master_link_status')}, "
                  f"смещение {info.get('slave_repl_offset')}, "
                  f"validity-factor {factor}")
        except Exception as exc:  # pragma: no cover — диагностика
            print(f"       узел {p}: не опрошен ({exc})")


def observe(label: str, observed: list[int], probe: str) -> None:
    r = roles(observed)
    alive = sum(1 for v in r.values() if v == "master")
    print(f"     {label}: {alive} мастеров из {len(observed)} узлов")
    for p in observed:
        print(f"       узел {p}: состояние {state_of(p):<12} GET -> {can_read(p, probe)}")


def one_side(block_ports_factory, pairing_factory, dc_factory, observed_name: str,
             observed_index: int) -> None:
    """Поднять свежий кластер, оборвать канал и опросить ОДНУ половину."""
    cluster = Cluster()
    try:
        ports = cluster.start(6)
        pairing = pairing_factory(ports)
        cluster.form(pairing)
        dcs = dc_factory(ports, pairing)
        observed, stopped = dcs[observed_index], dcs[1 - observed_index]

        topo = topology(ports)
        print(f"  ── опыт на свежем кластере, наблюдается {observed_name}")
        print(f"     ДЦ-1: {dcs[0]}")
        for p in dcs[0]:
            print(f"       {p}: {topo[p]}")
        print(f"     ДЦ-2: {dcs[1]}")
        for p in dcs[1]:
            print(f"       {p}: {topo[p]}")
        # Кладём настоящий ключ, чтобы «GET -> ответил» означало «отдал
        # данные», а не «слот пуст». Делается ДО опроса реплик: иначе
        # смещение репликации у всех нулевое и смотреть на него бессмысленно.
        cc = redis.RedisCluster(host="127.0.0.1", port=ports[0], decode_responses=True)
        cc.set(probe_key(), "value")
        # Ещё немного ключей, чтобы данные достались всем трём мастерам: иначе
        # у двух реплик из трёх смещение репликации нулевое, и смотреть на него
        # в строке про право повышаться попросту не на что.
        for i in range(20):
            cc.set(f"spread:{i}", i)
        cc.close()
        time.sleep(0.5)

        replicas_here = [p for p in observed if topo[p] != "мастер"]
        if replicas_here:
            print("     право реплик наблюдаемой половины повыситься, до разрыва:")
            eligibility(replicas_here)

        # КАК ИЗОБРАЖАЕТСЯ РАЗРЫВ. Узлы дальней половины останавливаются
        # сигналом STOP. Это не то же самое, что их убить: процесс жив, порт
        # открыт, файл состояния на месте — но узел не отвечает и не шлёт
        # heartbeat. Для оставшейся половины он неотличим от узла за
        # оборванным каналом, а именно это и надо изобразить. Настоящее
        # разделение сети было бы точнее, но требует правил файрвола или
        # сетевых пространств имён, которых у замера может не быть.
        cluster.stop(stopped)
        time.sleep(SETTLE_S)

        observe(observed_name, observed, probe_key())
        after = topology(ports, view_from=observed[0])
        changed = [(p, topo[p], after[p]) for p in observed if topo[p] != after[p]]
        if changed:
            print("     что изменилось в ролях за время разрыва:")
            for p, before, now in changed:
                print(f"       узел {p}: {before} -> {now}")
        else:
            print("     роли на наблюдаемой половине не изменились")
        print()
    finally:
        cluster.destroy()


def probe_key() -> str:
    return PROBE


def pairing_a(ports: list[int]) -> list[tuple[int, int]]:
    """Мастера — первые три, каждая реплика при своём мастере."""
    return [(ports[0], ports[3]), (ports[1], ports[4]), (ports[2], ports[5])]


def dcs_a(ports, pairing):
    return ([ports[0], ports[1], ports[2]], [ports[3], ports[4], ports[5]])


def dcs_b(ports, pairing):
    # ДЦ-1: два мастера и реплика первого. ДЦ-2: третий мастер и его реплика,
    # плюс реплика второго мастера. То есть реплика третьего — в ДЦ-2.
    return ([ports[0], ports[1], ports[3]], [ports[2], ports[4], ports[5]])


def dcs_c(ports, pairing):
    # То же деление по числу узлов, но домой берём реплику ТРЕТЬЕГО мастера.
    return ([ports[0], ports[1], ports[5]], [ports[2], ports[3], ports[4]])


def main() -> None:
    probe = Cluster()
    try:
        probe.start(1)
        version = conn(probe.ports[0]).info("server")["redis_version"]
    finally:
        probe.destroy()

    print(f"Redis {version} · кластер из шести узлов, 3 мастера и 3 реплики")
    print(f"cluster-node-timeout {NODE_TIMEOUT_MS} мс; после разрыва ждём "
          f"{SETTLE_S:.0f} с на выборы")
    print("каждая половина меряется на СВОЁМ кластере с той же топологией")
    print()

    print("4. РАСКЛАД A: ВСЕ МАСТЕРА В ОДНОМ ДЦ, РЕПЛИКИ ВО ВТОРОМ")
    print("-" * 66)
    one_side(None, pairing_a, dcs_a, "ДЦ-1 (мастера)", 0)
    one_side(None, pairing_a, dcs_a, "ДЦ-2 (реплики)", 1)
    print("  ДЦ-1 работает, ДЦ-2 не отвечает ничем. Реплики не стали")
    print("  мастерами: для повышения нужен голос большинства мастеров, а")
    print("  все мастера остались по ту сторону обрыва. То есть вторая")
    print("  половина не запасная — при потере первой она не заменит её.")
    print()

    print("5. РАСКЛАД B: УЗЛЫ ПОРОВНУ, МАСТЕРА ПОДЕЛЕНЫ 2 И 1")
    print("-" * 66)
    one_side(None, pairing_a, dcs_b, "ДЦ-1 (2 мастера)", 0)
    one_side(None, pairing_a, dcs_b, "ДЦ-2 (1 мастер)", 1)
    print("  ОТКАЗАЛИ ОБЕ ПОЛОВИНЫ, и это главный результат замера. Теперь это")
    print("  два независимых опыта на двух свежих кластерах, а не две проверки")
    print("  подряд на одном: вторая половина не могла достаться второму опыту")
    print("  в изменённом виде, потому что первого опыта он не видел.")
    print()
    print("  У ДЦ-1 большинство мастеров, и голосов на повышение реплики")
    print("  ему хватает. Но повышать некого: реплика мастера, оставшегося")
    print("  в ДЦ-2, тоже осталась в ДЦ-2 — смотрите карту выше. Слоты")
    print("  этого мастера не покрыты никем, а кластер по умолчанию")
    print("  отказывается работать с неполным покрытием слотов целиком, а")
    print("  не только по недостающим ключам.")
    print()
    print("  Отсюда правило, ради которого замер и поставлен: большинство")
    print("  голосов необходимо, но НЕ достаточно. Нужна ещё и реплика")
    print("  каждого потерянного мастера на своей стороне обрыва — и она")
    print("  должна быть достаточно свежей, чтобы ей разрешили повышаться.")
    print("  Свежесть в этих опытах не мешала: канал у всех реплик `up` ещё")
    print("  до разрыва, что и напечатано выше. Но условие из спецификации")
    print("  без этой третьей части неполно.")
    print()

    print("6. РАСКЛАД C: ТО ЖЕ ДЕЛЕНИЕ, НО РЕПЛИКА ЧУЖОГО МАСТЕРА ДОМА")
    print("-" * 66)
    one_side(None, pairing_a, dcs_c, "ДЦ-1 (2 мастера + реплика третьего)", 0)
    one_side(None, pairing_a, dcs_c, "ДЦ-2 (1 мастер)", 1)
    print("  Вот теперь ДЦ-1 выживает: и голосов хватило, и было кого")
    print("  повысить. Строка «что изменилось в ролях» показывает это прямо:")
    print("  реплика, привезённая домой, стала мастером.")
    print()
    print("  Но симметрии не появилось — в ДЦ-2 кэша по-прежнему нет. Такой")
    print("  раскладкой выбирают, какой датацентр переживёт разрыв; сделать")
    print("  так, чтобы пережили оба, нельзя.")
    print()
    print("  И ещё одно, ради чего понадобились свежие кластеры. В прежней")
    print("  версии замера обе половины проверялись подряд на одном кластере,")
    print("  и повышение реплики в первой проверке меняло топологию для")
    print("  второй. Здесь этого не может быть по устройству опыта, и разница")
    print("  между блоками 5 и 6 — ровно одна перестановка реплики, а не она")
    print("  же плюс последствия предыдущего опыта.")
    print()


if __name__ == "__main__":
    sys.exit(main())