"""Как клиент находит узел: два совершенно разных механизма — блоки 1-4.

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

ПОЧЕМУ ЭТО ВАЖНЕЕ ВСЕХ ОСТАЛЬНЫХ ОТЛИЧИЙ. Из него следует и разный протокол
подключения, и разное поведение при отказе, и разный набор доступных команд.
Все три разбираются дальше, в других блоках; здесь показан корень.

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ, А ЧТО ПРОСТО ПЕЧАТАЕТСЯ. Дословные ответы сервера —
печатаются (это поведение, а не измерение, и воспроизводится точно). А блоки 3
и 4 именно МЕРЯЮТ: по счётчикам самих узлов видно, сколько обращений и к кому
сделал настоящий клиент, пока выполнял одну-единственную команду.

ЗАПУСК: python3 bench/valkey-cluster/discovery.py
Нужен собранный Valkey (см. README), около минуты времени.
"""

from __future__ import annotations

import re
import time

import redis
from redis.cluster import RedisCluster
from redis.sentinel import Sentinel

from common import (
    Cluster,
    cli,
    free_port,
    head,
    master_addr,
    note,
    preamble,
    role_of,
    row,
)

SENTINEL_NAME = "cache"


def commands_processed(port: int) -> int:
    """Сколько команд узел обработал с запуска.

    Берётся из INFO stats, а не из commandstats, по одной причине: у sentinel
    нет команды CONFIG, а значит и RESETSTAT, — зато INFO есть у обоих. Раз
    счётчик нельзя обнулить, считается РАЗНИЦА до и после.
    """
    out = cli(port, "info", "stats")
    m = re.search(r"total_commands_processed:(\d+)", out)
    return int(m.group(1)) if m else -1


def named_commands(port: int) -> dict[str, int]:
    """Какие именно команды узел получил — по именам, из INFO commandstats."""
    out = cli(port, "info", "commandstats")
    res: dict[str, int] = {}
    for line in out.splitlines():
        m = re.match(r"cmdstat_([\w|\-]+):calls=(\d+)", line.strip())
        if m:
            res[m.group(1)] = int(m.group(2))
    return res


def block1(cluster: Cluster, ports: list[int]) -> None:
    head(1, "КЛАСТЕР ОТВЕЧАЕТ НЕ ДАННЫМИ, А АДРЕСОМ")

    # Один и тот же ключ спрашивается у ВСЕХ шести узлов по очереди. Флага -c
    # нет: клиент нарочно «глупый», он никуда не переходит и показывает то,
    # что сервер сказал.
    key = "foo"
    slot = cli(ports[0], "cluster", "keyslot", key)
    row(f"CLUSTER KEYSLOT {key}", slot)
    print()
    for port in ports:
        answer = cli(port, "set", key, "1")
        role = "мастер" if role_of(port) == "master" else "реплика"
        row(f"узел {port} ({role}): SET {key} 1", answer)

    note(
        """
        Пять узлов из шести ответили одинаково и не данными. MOVED — это не
        ошибка в обычном смысле: узел не отказал, а сказал, КУДА идти, и
        назвал слот. Клиент кластера на таком ответе чинит свою карту слотов
        и повторяет команду по названному адресу; «глупый» клиент, как
        здесь, просто видит текст.

        Адресация тут вообще не про узлы, а про слоты: ключ отображается в
        один из 16 384 слотов, а уже слот принадлежит узлу. Поэтому
        переезд данных между узлами не меняет ни одного ключа — меняет
        владельца слота.
        """
    )


def block2(cluster: Cluster, master: int, replica: int, sentinels: list[int]) -> None:
    head(2, "В РЕЖИМЕ SENTINEL УЗЕЛ ДАННЫХ ПРО ТОПОЛОГИЮ НЕ ЗНАЕТ НИЧЕГО")

    row(f"узел данных {master}: CLUSTER INFO", cli(master, "cluster", "info").splitlines()[0])
    row(f"узел данных {master}: роль", role_of(master))
    row(f"узел данных {replica}: роль", role_of(replica))
    print()
    row(f"sentinel {sentinels[0]}: SENTINEL get-master-addr-by-name",
        cli(sentinels[0], "sentinel", "get-master-addr-by-name", SENTINEL_NAME).replace("\n", ":"))
    row(f"sentinel {sentinels[0]}: сколько реплик знает",
        str(len([1 for line in cli(sentinels[0], "sentinel", "replicas", SENTINEL_NAME).splitlines()
                 if line.strip() == "name"])))
    row(f"sentinel {sentinels[0]}: сколько других sentinel знает",
        str(len([1 for line in cli(sentinels[0], "sentinel", "sentinels", SENTINEL_NAME).splitlines()
                 if line.strip() == "name"])))
    print()
    row(f"sentinel {sentinels[0]}: CONFIG RESETSTAT", cli(sentinels[0], "config", "resetstat"))

    note(
        """
        Первая строка — весь ответ на вопрос «а нельзя ли спросить у самого
        узла». Нельзя: узел данных вне кластера про кластер не знает и
        честно об этом говорит. Он не знает и того, что за ним кто-то
        наблюдает: роль он покажет свою (master или slave), а вот кто
        мастер С ТОЧКИ ЗРЕНИЯ СИСТЕМЫ — вопрос не к нему.

        Отвечает на него отдельная служба, и она не хранит данных. Последняя
        строка показывает, насколько отдельная: у sentinel даже нет команды
        CONFIG. Это не урезанный узел, это другой набор команд на том же
        двоичном файле.
        """
    )


def observe(ports: list[int], seconds: float, action=None) -> dict[int, tuple[int, list[str]]]:
    """Что узлы получили за окно времени: сколько команд и какие именно.

    Окно, а не «до и после команды», — потому что в обоих режимах есть
    постоянный фон: узлы кластера шлют друг другу heartbeat по шине, sentinel
    опрашивает данные и друг друга. Без контрольного окна той же длины
    клиентские команды не отличить от фоновых.
    """
    before = {p: commands_processed(p) for p in ports}
    named_before = {p: named_commands(p) for p in ports}
    started = time.time()
    if action is not None:
        action()
    left = seconds - (time.time() - started)
    if left > 0:
        time.sleep(left)
    out: dict[int, tuple[int, list[str]]] = {}
    for port in ports:
        delta = commands_processed(port) - before[port]
        names = named_commands(port)
        added = sorted(
            f"{k}×{v - named_before[port].get(k, 0)}"
            for k, v in names.items()
            if v - named_before[port].get(k, 0) > 0 and not k.startswith("info")
        )
        out[port] = (delta, added)
    return out


WINDOW_S = 2.0


def block3(cluster: Cluster, ports: list[int]) -> None:
    head(3, "ЧТО ДЕЛАЕТ НАСТОЯЩИЙ КЛИЕНТ КЛАСТЕРА ЗА ОДНУ КОМАНДУ")

    seed = ports[-1]

    def act() -> None:
        # Клиенту даётся РОВНО ОДИН адрес из шести — и не мастер ключа.
        rc = RedisCluster(host="127.0.0.1", port=seed, socket_timeout=5)
        rc.set("foo", "1")
        rc.close()

    idle = observe(ports, WINDOW_S)
    work = observe(ports, WINDOW_S, act)

    row("клиенту дан один адрес из шести", f"127.0.0.1:{seed}", pad=40)
    row("окно наблюдения", f"{WINDOW_S:.0f} с, и столько же вхолостую для контроля", pad=40)
    print()
    row("узел", "вхолостую  с клиентом  что пришло за второе окно", pad=24)
    for port in ports:
        mark = " (адрес клиенту)" if port == seed else ""
        row(f"{port}{mark}",
            f"{idle[port][0]:>9}  {work[port][0]:>10}  {' '.join(work[port][1])}", pad=24)

    note(
        """
        Считают сами узлы: это их счётчик обработанных команд. Контрольное
        окно вхолостую нужно потому, что узлы кластера всё время говорят друг
        с другом по шине; клиентское видно только на его фоне.

        Видно, что клиент сделал на самом деле: сходил к единственному
        известному адресу за картой слотов (cluster|slots) и уже по карте
        отправил запись туда, куда надо (set). Команды hello и
        client|setinfo — это клиент представляется; они к поиску узла
        отношения не имеют.

        Два лишних на первый взгляд следа стоит прочитать правильно. Пара
        replconf — это разговор мастера с его репликой, он идёт и без
        клиента. А select с set на ЕЩЁ ОДНОМ узле — это та же самая запись,
        доехавшая до реплики по репликации: клиент к этому узлу не
        подключался вовсе. Команды в счётчике узла — не обязательно команды
        от клиента, и здесь это видно прямо.

        Отсюда свойство, которое обычно и имеют в виду под «кластер сам
        разбирается»: знать все узлы клиенту не нужно, достаточно одного
        живого. Карту он получит от него и дальше будет ходить напрямую —
        посредника между клиентом и данными в этом режиме нет вообще.
        """
    )


def block4(cluster: Cluster, master: int, replicas: list[int], sentinels: list[int]) -> None:
    head(4, "ЧТО ДЕЛАЕТ НАСТОЯЩИЙ КЛИЕНТ SENTINEL ЗА ОДНУ КОМАНДУ")

    watched = [master] + replicas + sentinels

    def act() -> None:
        # Клиенту даётся адрес ОДНОГО sentinel и имя набора — и ни одного
        # адреса данных. Это и есть весь конфиг такого клиента в жизни.
        sen = Sentinel([("127.0.0.1", sentinels[0])], socket_timeout=5)
        conn = sen.master_for(SENTINEL_NAME, socket_timeout=5)
        conn.set("foo", "1")
        conn.close()

    idle = observe(watched, WINDOW_S)
    work = observe(watched, WINDOW_S, act)

    row("клиенту дан один sentinel и имя набора",
        f"127.0.0.1:{sentinels[0]} / {SENTINEL_NAME}", pad=40)
    row("окно наблюдения", f"{WINDOW_S:.0f} с, и столько же вхолостую для контроля", pad=40)
    print()
    row("узел", "вхолостую  с клиентом  что пришло за второе окно", pad=32)
    for port in watched:
        kind = ("sentinel" if port in sentinels else
                "мастер" if port == master else "реплика")
        mark = " (адрес клиенту)" if port == sentinels[0] else ""
        row(f"{kind} {port}{mark}",
            f"{idle[port][0]:>9}  {work[port][0]:>10}  {' '.join(work[port][1])}", pad=32)

    note(
        """
        Порядок обратный третьему блоку: сначала вопрос СЛУЖБЕ, потом
        команда данным. Узел данных не участвовал в поиске самого себя и не
        мог бы: у него для этого нет ни команды, ни сведений.

        Фон здесь заметнее, чем в кластере, и в нём же видно устройство
        службы. Sentinel непрерывно опрашивает данные (ping) — отсюда
        колонка «вхолостую». А publish на узлах ДАННЫХ — это как раз то, чем
        sentinel находит других sentinel: они публикуют себя в канал самого
        узла и там же друг друга слышат. Отдельного реестра у них нет;
        общая точка встречи — наблюдаемый узел.

        Имён команд у sentinel нет ни в одной колонке по той же причине, что
        и CONFIG в блоке 2: у него нет commandstats, только общий счётчик.

        Практическое следствие, которое из этого растёт: в режиме Sentinel
        клиент обязан ходить через библиотеку, умеющую sentinel, и держать
        адреса sentinel в конфиге. Если он подключится прямо к адресу
        данных — он будет работать ровно до первого переключения, а потом
        молча начнёт писать не туда. Что именно при этом происходит, видно
        в блоке про возвращение старого мастера.
        """
    )


def main() -> None:
    preamble("нативный кластер (6 узлов) и связка Sentinel (1 мастер + 2 реплики + 3 sentinel)")

    cluster = Cluster()
    try:
        # --- нативный кластер ---
        cports = [free_port() for _ in range(6)]
        for p in cports:
            cluster.start_data(p, cluster=True)
        cluster.form_cluster(cports, replicas=1)

        # --- связка Sentinel ---
        m = free_port()
        reps = [free_port(), free_port()]
        sents = [free_port() for _ in range(3)]
        cluster.start_data(m)
        for r in reps:
            cluster.start_data(r, replicaof=m)
        for s in sents:
            cluster.start_sentinel(s, m, quorum=2, name=SENTINEL_NAME)
        time.sleep(3)

        # Оба хозяйства подняты сразу, и блоки печатаются по порядку: файл
        # прогона читается сверху вниз, а не в порядке, удобном скрипту.
        block1(cluster, cports)
        block2(cluster, m, reps[0], sents)
        block3(cluster, cports)
        block4(cluster, m, reps, sents)
    finally:
        cluster.stop_all()


if __name__ == "__main__":
    main()