KindClustersDashboard/app/kubeconfig_patch.py

"""Правка kubeconfig kind для доступа к API с хоста (после create из контейнера).

Kind внутри Docker видит другой адрес apiserver; на хосте нужен 127.0.0.1:<порт>
из проброса `docker port <cluster>-control-plane 6443/tcp` (Podman — тот же клиент
`docker` к docker-совместимому сокету).

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

from __future__ import annotations

import logging
import os
import subprocess
from pathlib import Path

logger = logging.getLogger("kind_k8s.kubeconfig_patch")


def _container_cli() -> str:
    """CLI для `port` (обычно docker к сокету Docker или Podman)."""
    return (os.environ.get("CONTAINER_CLI") or "docker").strip() or "docker"


def _control_plane_container_name(cluster_name: str) -> str:
    return f"{cluster_name}-control-plane"


def _parse_docker_port_line(line: str) -> tuple[str, str] | None:
    """Строка вида '0.0.0.0:32768' или '127.0.0.1:32768' -> (host, port)."""
    line = line.strip()
    if ":" not in line:
        return None
    # IPv6 [::]:port
    if line.startswith("["):
        rb = line.rfind("]")
        if rb == -1:
            return None
        host = line[1:rb]
        rest = line[rb + 1 :].lstrip(":")
        if not rest.isdigit():
            return None
        return host, rest
    host, _, port = line.rpartition(":")
    if not port.isdigit():
        return None
    host = host.strip()
    return host, port


def _host_bind_for_kubeconfig(host: str) -> str:
    if host in ("0.0.0.0", "::", ""):
        return "127.0.0.1"
    if host == "[::]":
        return "127.0.0.1"
    return host


def get_apiserver_host_port(cluster_name: str) -> tuple[str, str] | None:
    """Узнать (host, port) с хоста для доступа к apiserver."""
    cli = _container_cli()
    ctr = _control_plane_container_name(cluster_name)
    p = subprocess.run(
        [cli, "port", ctr, "6443/tcp"],
        capture_output=True,
        text=True,
    )
    if p.returncode != 0:
        logger.info(
            "Не удалось %s port %s 6443/tcp: %s",
            cli,
            ctr,
            (p.stderr or p.stdout or "").strip(),
        )
        return None
    for raw in (p.stdout or "").splitlines():
        parsed = _parse_docker_port_line(raw)
        if not parsed:
            continue
        h, port = parsed
        return _host_bind_for_kubeconfig(h), port
    return None


def _kubeconfig_cluster_name(kube_path: Path, logical_name: str) -> str:
    """Имя блока cluster в kubeconfig (обычно kind-<имя>)."""
    p = subprocess.run(
        [
            "kubectl",
            "--kubeconfig",
            str(kube_path),
            "config",
            "view",
            "-o",
            'jsonpath={range .clusters[*]}{.name}{"\n"}{end}',
        ],
        capture_output=True,
        text=True,
    )
    names = [x.strip() for x in (p.stdout or "").splitlines() if x.strip()]
    want = f"kind-{logical_name}"
    if want in names:
        return want
    for n in names:
        if n.startswith("kind-"):
            return n
    return want


def patch_kubeconfig_server_for_host(
    *,
    cluster_name: str,
    kube_path: Path,
) -> bool:
    """
    Подставить в kubeconfig server=https://<хост>:<порт> для доступа с хоста.

    Порт берётся из ``docker port`` / аналога к сокету; для Podman — тот же клиент
    при ``DOCKER_HOST`` на podman.sock.
    """
    hp = get_apiserver_host_port(cluster_name)
    if not hp:
        print(
            "Предупреждение: не удалось получить порт apiserver с хоста; "
            "kubeconfig оставлен как выдал kind (с хоста может не открываться).",
        )
        return False
    host, port = hp
    server = f"https://{host}:{port}"
    cluster_id = _kubeconfig_cluster_name(kube_path, cluster_name)
    p = subprocess.run(
        [
            "kubectl",
            "--kubeconfig",
            str(kube_path),
            "config",
            "set-cluster",
            cluster_id,
            f"--server={server}",
        ],
        capture_output=True,
        text=True,
    )
    if p.returncode != 0:
        err = (p.stderr or p.stdout or "").strip()
        print(f"Предупреждение: kubectl config set-cluster не удался: {err}")
        return False
    print(f"Kubeconfig для хоста: apiserver → {server}")
    return True


def should_patch_after_create() -> bool:
    """Патчить после create, если задано явно или kind шёл из контейнера."""
    if os.environ.get("KIND_K8S_PATCH_KUBECONFIG", "").strip().lower() in ("1", "true", "yes", "да"):
        return True
    return os.environ.get("KIND_K8S_IN_CONTAINER", "").strip() == "1"