KindClustersDashboard/app/cluster_status.py

#!/usr/bin/env python3
"""Проверка статуса кластеров kind: регистрация, kubeconfig, узлы (kubectl).

Запуск без аргументов — все кластеры из `kind get clusters`.
Один аргумент — только указанное имя.

Автор: Сергей Антропов
Сайт: https://devops.org.ru

Требования: kind и kubectl в PATH (в образе kind-k8s-tools уже есть).
"""

from __future__ import annotations

import argparse
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

from kind_k8s_paths import clusters_dir


def _kind_cluster_names() -> list[str]:
    p = subprocess.run(["kind", "get", "clusters"], capture_output=True, text=True)
    if p.returncode != 0:
        return []
    lines = [x.strip() for x in (p.stdout or "").splitlines() if x.strip()]
    return [x for x in lines if "no kind" not in x.lower()]


def _kubeconfig_for_cluster(name: str) -> tuple[str | None, str]:
    """Временный файл kubeconfig или None; второе значение — пояснение."""
    p = subprocess.run(
        ["kind", "get", "kubeconfig", "--name", name],
        capture_output=True,
        text=True,
    )
    if p.returncode != 0:
        err = (p.stderr or p.stdout or "").strip()
        return None, err or "kind get kubeconfig не удался"
    data = (p.stdout or "").strip()
    if not data:
        return None, "пустой kubeconfig"
    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".kubeconfig",
        prefix=f"kind-{name}-",
        delete=False,
        encoding="utf-8",
    ) as f:
        f.write(data)
        return f.name, ""


def _kubectl_nodes(kubeconfig: str) -> tuple[int, str]:
    p = subprocess.run(
        [
            "kubectl",
            "--kubeconfig",
            kubeconfig,
            "get",
            "nodes",
            "-o",
            "wide",
            "--request-timeout=10s",
        ],
        capture_output=True,
        text=True,
    )
    out = (p.stdout or "").strip()
    err = (p.stderr or "").strip()
    msg = out if out else err
    return p.returncode, msg


def _local_meta(name: str) -> dict[str, str] | None:
    meta = CLUSTERS_DIR / name / "meta.json"
    if not meta.is_file():
        return None
    try:
        import json

        raw = json.loads(meta.read_text(encoding="utf-8"))
        if isinstance(raw, dict):
            return {str(k): str(v) for k, v in raw.items() if isinstance(k, str)}
    except Exception:
        pass
    return None


def _print_cluster(name: str, *, kube_path_saved: Path | None) -> None:
    print(f"── Кластер: {name} ──")
    if kube_path_saved and kube_path_saved.is_file():
        print(f"  Сохранённый kubeconfig: {kube_path_saved}")
    meta = _local_meta(name)
    if meta:
        ver = meta.get("kubernetes_version_tag") or meta.get("node_image", "—")
        wn = meta.get("worker_nodes", "—")
        print(f"  meta.json: версия={ver}, workers={wn}")

    # С хоста удобнее тот же файл, что после create (в т.ч. с патчем 127.0.0.1:порт).
    use_path: str | None = None
    if kube_path_saved and kube_path_saved.is_file():
        use_path = str(kube_path_saved)
        print("  Проверка API: kubectl с сохранённым kubeconfig (тот же файл, что пишет create / UI).")

    tmp_kc: str | None = None
    if not use_path:
        tmp_kc, kerr = _kubeconfig_for_cluster(name)
        if not tmp_kc:
            print(f"  Статус API: недоступен ({kerr})")
            return
        use_path = tmp_kc

    try:
        rc, msg = _kubectl_nodes(use_path)
        if rc == 0:
            print("  Узлы (kubectl get nodes -o wide):")
            for line in msg.splitlines():
                print(f"    {line}")
        else:
            print(f"  kubectl: код {rc}")
            for line in msg.splitlines()[:20]:
                print(f"    {line}")
    finally:
        if tmp_kc:
            Path(tmp_kc).unlink(missing_ok=True)


def main() -> None:
    CLUSTERS_DIR = clusters_dir()
    parser = argparse.ArgumentParser(description="Статус кластеров kind")
    parser.add_argument(
        "cluster",
        nargs="?",
        help="Имя кластера (если не указано — все из kind get clusters)",
    )
    args = parser.parse_args()

    if not shutil.which("kind"):
        print("Не найден kind.", file=sys.stderr)
        print("  Статус узлов — в веб-интерфейсе (make docker up) или внутри контейнера kind-k8s-web.", file=sys.stderr)
        print(
            "  Либо установите kind на хост: https://kind.sigs.k8s.io/docs/user/quick-start/#installation",
            file=sys.stderr,
        )
        sys.exit(127)
    if not shutil.which("kubectl"):
        print("Не найден kubectl в PATH.", file=sys.stderr)
        print(
            "  Запустите скрипт внутри контейнера приложения (там kubectl в образе), например:",
            file=sys.stderr,
        )
        print(
            "    docker compose exec kind-k8s-web python3 /opt/kind-k8s/app/cluster_status.py <имя>",
            file=sys.stderr,
        )
        print("  Либо смотрите узлы в веб-интерфейсе; на хост kubectl не обязателен.", file=sys.stderr)
        sys.exit(127)

    names = _kind_cluster_names()
    if args.cluster:
        if args.cluster not in names:
            print(f"Кластер «{args.cluster}» не найден в kind get clusters.", file=sys.stderr)
            print("Известные:", ", ".join(names) if names else "(пусто)", file=sys.stderr)
            sys.exit(1)
        names = [args.cluster]

    if not names:
        print("Нет кластеров kind (kind get clusters).")
        return

    for name in names:
        saved = CLUSTERS_DIR / name / "kubeconfig"
        kube_saved = saved if saved.is_file() else None
        _print_cluster(name, kube_path_saved=kube_saved)
        print()


if __name__ == "__main__":
    main()