KindClustersDashboard/app/core/helm_addons.py

"""Установка и удаление типовых Helm-чартов в выбранном кластере (kubeconfig с диска).

Используются бинарники ``helm`` и ``kubectl`` из PATH; ``KUBECONFIG`` — путь из
:func:`kubeconfig_patch.kubeconfig_path_for_container_kubectl` (как для kubectl в приложении).

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

from __future__ import annotations

import json
import logging
import os
import shutil
import subprocess
import tempfile
import threading
import time
from pathlib import Path
from typing import Any

import yaml

from core.cluster_lifecycle import validate_cluster_name
from kind_k8s_paths import clusters_dir
from kubeconfig_patch import kubeconfig_path_for_container_kubectl

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

# Официальные репозитории чартов (имя → URL).
HELM_REPOS: dict[str, str] = {
    "ingress-nginx": "https://kubernetes.github.io/ingress-nginx",
    "prometheus-community": "https://prometheus-community.github.io/helm-charts",
    "metrics-server": "https://kubernetes-sigs.github.io/metrics-server/",
    "istio": "https://istio-release.storage.googleapis.com/charts",
    "kiali": "https://kiali.org/helm-charts",
}

# Идентификаторы аддонов, доступных для установки на странице «Аддоны» и через API ``/clusters/{name}/addons/…``.
# Должен совпадать с набором в ``HelmComposeValuesRequest.addon`` и кнопками в ``cluster_addons.html``.
HELM_INSTALLABLE_ADDON_IDS: tuple[str, ...] = (
    "ingress-nginx",
    "kube-prometheus-stack",
    "metrics-server",
    "istio-kiali",
)

# Ссылки для ``helm search repo <ref> --versions`` (ключи совпадают с полями API версий чартов).
HELM_ADDON_CHART_REFS: dict[str, str] = {
    "ingress_nginx": "ingress-nginx/ingress-nginx",
    "kube_prometheus_stack": "prometheus-community/kube-prometheus-stack",
    "metrics_server": "metrics-server/metrics-server",
    # Для base и istiod обычно берут одну версию чарта Istio.
    "istio": "istio/base",
    "kiali_server": "kiali/kiali-server",
}

_versions_cache: dict[str, tuple[float, list[str]]] = {}
_versions_lock = threading.Lock()


class HelmAddonError(Exception):
    """Ошибка helm/kubectl (сообщение для API)."""

    def __init__(self, message: str, *, exit_code: int = 1) -> None:
        super().__init__(message)
        self.exit_code = exit_code


def dump_helm_values_yaml(data: dict[str, Any]) -> str:
    """Сериализация mapping в YAML для UI и временных файлов Helm."""
    return yaml.safe_dump(
        data,
        allow_unicode=True,
        default_flow_style=False,
        sort_keys=False,
    )


def helm_show_values_dict(chart_ref: str, chart_version: str | None = None) -> dict[str, Any]:
    """
    Словарь дефолтных values чарта: ``helm show values <ref> [--version …]``.

    При пустом stdout или ошибке разбора возвращает ``{}`` (ошибка helm — :class:`HelmAddonError`).
    """
    require_helm_binary()
    cref = chart_ref.strip()
    if not cref:
        return {}
    cmd = ["helm", "show", "values", cref]
    ver = (chart_version or "").strip()
    if ver:
        cmd.extend(["--version", ver])
    rc, out, err = _run_cmd(cmd, timeout=180)
    if rc != 0:
        raise HelmAddonError(
            (err or out or f"helm show values {cref} код {rc}")[:4000],
            exit_code=rc,
        )
    raw = (out or "").strip()
    if not raw:
        return {}
    try:
        data = yaml.safe_load(raw)
    except yaml.YAMLError as e:
        logger.warning("YAML из helm show values %s: %s", cref, e)
        return {}
    return data if isinstance(data, dict) else {}


def _ingress_kind_service_overlay() -> dict[str, Any]:
    """То же, что передаётся через ``--set`` при установке ingress для kind."""
    return {
        "controller": {
            "service": {
                "type": "NodePort",
                "nodePorts": {"http": 30080},
            }
        }
    }


def compose_ingress_effective_values(chart_version: str | None = None) -> dict[str, Any]:
    """Дефолты чарта ingress-nginx + сервис NodePort (как в install). Репозитории — см. :func:`helm_repo_ensure`."""
    base = helm_show_values_dict("ingress-nginx/ingress-nginx", chart_version)
    return deep_merge_values(base, _ingress_kind_service_overlay())


def compose_kube_prometheus_effective_values(
    chart_version: str | None,
    grafana_admin_user: str,
    grafana_admin_password: str,
) -> dict[str, Any]:
    """Дефолты kube-prometheus-stack + учётные данные Grafana."""
    base = helm_show_values_dict("prometheus-community/kube-prometheus-stack", chart_version)
    overlay: dict[str, Any] = {
        "grafana": {
            "adminUser": grafana_admin_user.strip(),
            "adminPassword": grafana_admin_password,
        }
    }
    return deep_merge_values(base, overlay)


def compose_metrics_server_effective_values(chart_version: str | None = None) -> dict[str, Any]:
    """Дефолты metrics-server + args для kind (как в install)."""
    base = helm_show_values_dict("metrics-server/metrics-server", chart_version)
    overlay: dict[str, Any] = {
        "args": [
            "--kubelet-insecure-tls",
            "--kubelet-preferred-address-types=InternalIP,Hostname,ExternalIP",
        ]
    }
    return deep_merge_values(base, overlay)


def compose_istio_mesh_preview_values(
    istio_chart_version: str | None = None,
    kiali_chart_version: str | None = None,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
    """
    Три словаря values: ``kiali-server``, ``istio/base``, ``istio/istiod``
    (чистые ``helm show values`` для редактирования в UI).
    """
    kiali = helm_show_values_dict("kiali/kiali-server", kiali_chart_version)
    ib = helm_show_values_dict("istio/base", istio_chart_version)
    ird = helm_show_values_dict("istio/istiod", istio_chart_version)
    return kiali, ib, ird


def deep_merge_values(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
    """
    Рекурсивное объединение mapping: ключи из ``override`` перекрывают ``base``;
    для вложенных dict сливаем дальше (как типичный merge для Helm values).
    """
    out: dict[str, Any] = dict(base)
    for k, v in override.items():
        if k in out and isinstance(out[k], dict) and isinstance(v, dict):
            out[k] = deep_merge_values(out[k], v)
        else:
            out[k] = v
    return out


def _helm_timeout_sec() -> int:
    raw = (os.environ.get("KIND_K8S_HELM_TIMEOUT_SEC") or "900").strip()
    try:
        return max(60, min(int(raw), 7200))
    except ValueError:
        return 900


def _helm_versions_cache_sec() -> int:
    """TTL кэша списков версий (``helm search repo`` после ``repo update``)."""
    raw = (os.environ.get("KIND_K8S_HELM_VERSIONS_CACHE_SEC") or "600").strip()
    try:
        return max(30, min(int(raw), 86400))
    except ValueError:
        return 600


def _helm_versions_max_list() -> int:
    raw = (os.environ.get("KIND_K8S_HELM_VERSIONS_MAX") or "80").strip()
    try:
        return max(10, min(int(raw), 200))
    except ValueError:
        return 80


def search_repo_chart_versions(chart_ref: str) -> list[str]:
    """
    Уникальные версии чарта (новые первыми), через ``helm search repo … --versions -o json``.

    Перед вызовом рекомендуется :func:`helm_repo_ensure`; здесь кэш по ``chart_ref``.
    """
    ref = chart_ref.strip()
    if not ref:
        return []
    now = time.monotonic()
    ttl = _helm_versions_cache_sec()
    with _versions_lock:
        hit = _versions_cache.get(ref)
        if hit is not None:
            ts, vers = hit
            if now - ts < ttl:
                logger.debug("Версии чарта %s из кэша (%s шт.)", ref, len(vers))
                return list(vers)

    require_helm_binary()
    rc, out, err = _run_cmd(
        ["helm", "search", "repo", ref, "--versions", "-o", "json"],
        timeout=180,
    )
    if rc != 0:
        logger.warning("helm search repo %s: %s", ref, err or out or rc)
        with _versions_lock:
            _versions_cache[ref] = (now, [])
        return []

    versions: list[str] = []
    seen: set[str] = set()
    try:
        rows = json.loads(out or "[]")
    except json.JSONDecodeError:
        rows = []
    cap = _helm_versions_max_list()
    if isinstance(rows, list):
        for row in rows:
            if not isinstance(row, dict):
                continue
            v = str(row.get("version", "")).strip()
            if not v or v in seen:
                continue
            seen.add(v)
            versions.append(v)
            if len(versions) >= cap:
                break

    with _versions_lock:
        _versions_cache[ref] = (now, versions)
    logger.info("Получены версии для %s: %s записей", ref, len(versions))
    return versions


def helm_addon_chart_versions_all() -> dict[str, list[str]]:
    """
    Версии для всех аддонов UI (после обновления репозиториев — один раз).

    Ключи: ``ingress_nginx``, ``kube_prometheus_stack``, ``metrics_server``, ``istio``, ``kiali_server``.
    """
    helm_repo_ensure()
    out: dict[str, list[str]] = {}
    for key, cref in HELM_ADDON_CHART_REFS.items():
        out[key] = search_repo_chart_versions(cref)
    return out


def require_helm_binary() -> None:
    if not shutil.which("helm"):
        raise HelmAddonError("В образе не найден helm в PATH. Пересоберите образ с установленным Helm.", exit_code=127)


def require_kubectl_binary() -> None:
    if not shutil.which("kubectl"):
        raise HelmAddonError("Не найден kubectl в PATH.", exit_code=127)


def _kubeconfig_runtime_path(cluster_name: str) -> Path:
    if not validate_cluster_name(cluster_name):
        raise HelmAddonError("Некорректное имя кластера")
    src = clusters_dir() / cluster_name / "kubeconfig"
    if not src.is_file():
        raise HelmAddonError(f"Нет kubeconfig: {src}")
    return kubeconfig_path_for_container_kubectl(cluster_name=cluster_name, kube_source_path=src)


def _run_cmd(
    cmd: list[str],
    *,
    timeout: int | None = None,
    env: dict[str, str] | None = None,
) -> tuple[int, str, str]:
    """Выполнить команду; возвращает (код, stdout, stderr)."""
    t = timeout if timeout is not None else _helm_timeout_sec()
    logger.info("Выполнение: %s (timeout=%s)", " ".join(cmd[:12]) + (" …" if len(cmd) > 12 else ""), t)
    try:
        p = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=t,
            env={**os.environ, **(env or {})},
        )
    except subprocess.TimeoutExpired:
        logger.warning("Таймаут команды: %s", cmd[0])
        raise HelmAddonError(f"Превышено время ожидания ({t} с) для команды «{cmd[0]}»", exit_code=124) from None
    out = (p.stdout or "").strip()
    err = (p.stderr or "").strip()
    return p.returncode, out, err


def helm_repo_ensure() -> tuple[int, str]:
    """Добавить недостающие репозитории и выполнить ``helm repo update``."""
    require_helm_binary()
    lines: list[str] = []
    for name, url in HELM_REPOS.items():
        rc, o, e = _run_cmd(["helm", "repo", "add", name, url], timeout=120)
        if rc != 0 and "already exists" not in (e + o).lower():
            raise HelmAddonError(f"helm repo add {name}: {e or o or rc}", exit_code=rc)
        if o:
            lines.append(o)
        if e and "already exists" not in e.lower():
            lines.append(e)
    rc2, o2, e2 = _run_cmd(["helm", "repo", "update"], timeout=300)
    if rc2 != 0:
        raise HelmAddonError(f"helm repo update: {e2 or o2 or rc2}", exit_code=rc2)
    lines.append(o2 or e2)
    return 0, "\n".join(x for x in lines if x).strip()


def helm_list_all(cluster_name: str) -> list[dict[str, Any]]:
    """Список релизов ``helm list -A -o json``."""
    kcfg = str(_kubeconfig_runtime_path(cluster_name))
    rc, out, err = _run_cmd(
        ["helm", "list", "-A", "-o", "json", "--kubeconfig", kcfg],
        timeout=120,
    )
    if rc != 0:
        logger.warning("helm list: %s", err or out)
        return []
    if not out.strip():
        return []
    try:
        data = json.loads(out)
        return data if isinstance(data, list) else []
    except json.JSONDecodeError:
        return []


def helm_release_entry(
    lst: list[dict[str, Any]],
    release: str,
    namespace: str,
) -> dict[str, Any] | None:
    """Одна запись ``helm list -o json`` по имени релиза и namespace."""
    for x in lst:
        if not isinstance(x, dict):
            continue
        if str(x.get("name", "")) == release and str(x.get("namespace", "")) == namespace:
            return x
    return None


def _chart_version_from_list_field(chart: str, release_name: str) -> str | None:
    """
    Поле ``chart`` в выводе ``helm list`` имеет вид ``<имя_чарта>-<версия>`` (например ``ingress-nginx-4.11.2``).
    Для релиза Istio base в списке часто ``base-1.22.3``.
    """
    chart = (chart or "").strip()
    if not chart:
        return None
    prefixes: dict[str, str] = {
        "ingress-nginx": "ingress-nginx-",
        "kube-prometheus-stack": "kube-prometheus-stack-",
        "metrics-server": "metrics-server-",
        "istio-base": "base-",
        "istiod": "istiod-",
        "kiali-server": "kiali-server-",
    }
    p = prefixes.get(release_name)
    if p and chart.startswith(p):
        v = chart[len(p) :].strip()
        return v or None
    return chart


def helm_get_values_all_yaml(cluster_name: str, release: str, namespace: str) -> str:
    """
    Эффективные values релиза: ``helm get values --all -o yaml`` (для правки при upgrade).
    """
    kcfg = str(_kubeconfig_runtime_path(cluster_name))
    rc, out, err = _run_cmd(
        [
            "helm",
            "get",
            "values",
            release,
            "-n",
            namespace,
            "--all",
            "-o",
            "yaml",
            "--kubeconfig",
            kcfg,
        ],
        timeout=120,
    )
    if rc != 0:
        raise HelmAddonError(
            (err or out or f"helm get values {release}/{namespace} код {rc}")[:8000],
            exit_code=rc,
        )
    return (out or "").strip()


_VALUES_YAML_API_MAX = 131_072


def trim_values_yaml_for_api(raw: str) -> str:
    """Ограничение размера YAML в JSON-ответе API."""
    if len(raw) <= _VALUES_YAML_API_MAX:
        return raw
    return raw[: _VALUES_YAML_API_MAX - 120] + "\n\n# … YAML обрезан (лимит ответа API).\n"


def helm_addon_snapshot(cluster_name: str) -> dict[str, Any]:
    """
    Флаги установки, версии чартов из ``helm list`` и признак полного mesh Istio+Kiali.
    """
    lst = helm_list_all(cluster_name)

    def pair(rel: str, ns: str) -> tuple[bool, str | None]:
        e = helm_release_entry(lst, rel, ns)
        if not e:
            return False, None
        ch = str(e.get("chart") or "")
        ver = _chart_version_from_list_field(ch, rel)
        return True, ver

    ing_on, ing_v = pair("ingress-nginx", "ingress-nginx")
    prom_on, prom_v = pair("kube-prometheus-stack", "monitoring")
    ms_on, ms_v = pair("metrics-server", "kube-system")
    base_on, base_v = pair("istio-base", "istio-system")
    istiod_on, istiod_v = pair("istiod", "istio-system")
    kiali_on, kiali_v = pair("kiali-server", "istio-system")
    mesh = bool(base_on and istiod_on and kiali_on)
    return {
        "ingress_nginx": ing_on,
        "ingress_nginx_chart_version": ing_v if ing_on else None,
        "kube_prometheus_stack": prom_on,
        "kube_prometheus_stack_chart_version": prom_v if prom_on else None,
        "metrics_server": ms_on,
        "metrics_server_chart_version": ms_v if ms_on else None,
        "istio_base": base_on,
        "istiod": istiod_on,
        "kiali_server": kiali_on,
        "istio_base_chart_version": base_v if base_on else None,
        "istiod_chart_version": istiod_v if istiod_on else None,
        "kiali_server_chart_version": kiali_v if kiali_on else None,
        "istio_mesh_ready": mesh,
    }


def addon_status(cluster_name: str) -> dict[str, bool]:
    """Какие аддоны установлены (по имени релиза и namespace). Совместимость со старым кодом."""
    s = helm_addon_snapshot(cluster_name)
    return {
        "ingress_nginx": bool(s["ingress_nginx"]),
        "kube_prometheus_stack": bool(s["kube_prometheus_stack"]),
        "metrics_server": bool(s["metrics_server"]),
        "istio_base": bool(s["istio_base"]),
        "istiod": bool(s["istiod"]),
        "kiali": bool(s["kiali_server"]),
    }


def collect_installed_addons_values(cluster_name: str) -> dict[str, Any]:
    """
    Для установленных релизов — версия чарта (как в snapshot) и YAML ``helm get values --all``.
    Ключи совпадают с логическими аддонами страницы UI.
    """
    s = helm_addon_snapshot(cluster_name)
    out: dict[str, Any] = {}

    if s["ingress_nginx"]:
        try:
            y = trim_values_yaml_for_api(
                helm_get_values_all_yaml(cluster_name, "ingress-nginx", "ingress-nginx"),
            )
        except HelmAddonError as e:
            y = f"# Не удалось прочитать values: {e}\n"
        out["ingress_nginx"] = {"chart_version": s["ingress_nginx_chart_version"], "values_yaml": y}

    if s["kube_prometheus_stack"]:
        try:
            y = trim_values_yaml_for_api(
                helm_get_values_all_yaml(cluster_name, "kube-prometheus-stack", "monitoring"),
            )
        except HelmAddonError as e:
            y = f"# Не удалось прочитать values: {e}\n"
        out["kube_prometheus_stack"] = {"chart_version": s["kube_prometheus_stack_chart_version"], "values_yaml": y}

    if s["metrics_server"]:
        try:
            y = trim_values_yaml_for_api(
                helm_get_values_all_yaml(cluster_name, "metrics-server", "kube-system"),
            )
        except HelmAddonError as e:
            y = f"# Не удалось прочитать values: {e}\n"
        out["metrics_server"] = {"chart_version": s["metrics_server_chart_version"], "values_yaml": y}

    if s["istio_mesh_ready"]:
        try:
            kb = trim_values_yaml_for_api(
                helm_get_values_all_yaml(cluster_name, "istio-base", "istio-system"),
            )
            ki = trim_values_yaml_for_api(helm_get_values_all_yaml(cluster_name, "istiod", "istio-system"))
            kk = trim_values_yaml_for_api(
                helm_get_values_all_yaml(cluster_name, "kiali-server", "istio-system"),
            )
        except HelmAddonError as e:
            kb = ki = kk = f"# Не удалось прочитать values: {e}\n"
        out["istio_kiali"] = {
            "istio_chart_version": s["istiod_chart_version"] or s["istio_base_chart_version"],
            "kiali_chart_version": s["kiali_server_chart_version"],
            "istio_base_values_yaml": kb,
            "istiod_values_yaml": ki,
            "kiali_values_yaml": kk,
        }

    return out


def _kubectl_apply_manifest(cluster_name: str, manifest: dict[str, Any]) -> tuple[int, str]:
    require_kubectl_binary()
    kcfg = str(_kubeconfig_runtime_path(cluster_name))
    yml = yaml.safe_dump(manifest, allow_unicode=True, default_flow_style=False)
    p = subprocess.run(
        ["kubectl", "apply", "-f", "-", "--kubeconfig", kcfg, "--request-timeout=120s"],
        input=yml,
        capture_output=True,
        text=True,
        timeout=180,
    )
    msg = ((p.stdout or "").strip() + "\n" + (p.stderr or "").strip()).strip()
    return p.returncode, msg


def _ensure_namespace(cluster_name: str, namespace: str) -> None:
    manifest = {
        "apiVersion": "v1",
        "kind": "Namespace",
        "metadata": {"name": namespace},
    }
    rc, msg = _kubectl_apply_manifest(cluster_name, manifest)
    if rc != 0:
        raise HelmAddonError(f"Создание namespace {namespace}: {msg}", exit_code=rc)


def _helm_upgrade_install(
    cluster_name: str,
    release: str,
    chart: str,
    namespace: str,
    *,
    chart_version: str | None = None,
    extra_args: list[str] | None = None,
    values_file: Path | None = None,
    extra_values_files: list[Path] | None = None,
) -> tuple[int, str, str]:
    kcfg = str(_kubeconfig_runtime_path(cluster_name))
    cmd = [
        "helm",
        "upgrade",
        "--install",
        release,
        chart,
        "-n",
        namespace,
        "--create-namespace",
        "--kubeconfig",
        kcfg,
        "--wait",
        "--timeout",
        f"{_helm_timeout_sec()}s",
    ]
    ver = (chart_version or "").strip()
    if ver:
        cmd.extend(["--version", ver])
    if values_file is not None:
        cmd.extend(["-f", str(values_file)])
    for p in extra_values_files or []:
        cmd.extend(["-f", str(p)])
    if extra_args:
        cmd.extend(extra_args)
    return _run_cmd(cmd, timeout=_helm_timeout_sec() + 60)


def _helm_uninstall(cluster_name: str, release: str, namespace: str) -> tuple[int, str, str]:
    kcfg = str(_kubeconfig_runtime_path(cluster_name))
    return _run_cmd(
        ["helm", "uninstall", release, "-n", namespace, "--kubeconfig", kcfg, "--wait", "--timeout", "600s"],
        timeout=660,
    )


# --- Установка / удаление по типам ---


def install_ingress_nginx(
    cluster_name: str,
    chart_version: str | None,
    user_values: dict[str, Any] | None = None,
) -> tuple[bool, str]:
    """
    ingress-nginx: полный YAML values из UI (``helm show values`` + NodePort) или переданный словарь;
    после файла values остаются ``--set`` для NodePort (приоритет как у Helm).
    """
    helm_repo_ensure()
    eff = user_values if user_values is not None else compose_ingress_effective_values(chart_version)
    extra: list[str] = [
        "--set",
        "controller.service.type=NodePort",
        "--set",
        "controller.service.nodePorts.http=30080",
    ]
    user_tmp: Path | None = None
    try:
        fd, name = tempfile.mkstemp(suffix=".yaml", prefix="helm-ing-values-")
        os.close(fd)
        user_tmp = Path(name)
        user_tmp.write_text(dump_helm_values_yaml(eff), encoding="utf-8")
        rc, out, err = _helm_upgrade_install(
            cluster_name,
            "ingress-nginx",
            "ingress-nginx/ingress-nginx",
            "ingress-nginx",
            chart_version=chart_version,
            extra_args=extra,
            extra_values_files=[user_tmp],
        )
    finally:
        if user_tmp is not None:
            user_tmp.unlink(missing_ok=True)
    text = "\n".join(filter(None, [out, err])).strip()
    if rc != 0:
        raise HelmAddonError(text or f"helm exit {rc}", exit_code=rc)
    return True, text or "ingress-nginx установлен."


def uninstall_ingress_nginx(cluster_name: str) -> tuple[bool, str]:
    rc, out, err = _helm_uninstall(cluster_name, "ingress-nginx", "ingress-nginx")
    text = "\n".join(filter(None, [out, err])).strip()
    if rc != 0 and "not found" not in text.lower():
        raise HelmAddonError(text or f"helm exit {rc}", exit_code=rc)
    return True, text or "ingress-nginx удалён (или отсутствовал)."


def install_kube_prometheus_stack(
    cluster_name: str,
    *,
    grafana_admin_user: str,
    grafana_admin_password: str,
    chart_version: str | None = None,
    user_values: dict[str, Any] | None = None,
) -> tuple[bool, str]:
    """Полный values из UI или автосборка (``helm show values`` + Grafana из формы)."""
    helm_repo_ensure()
    if user_values is not None:
        values = user_values
    else:
        values = compose_kube_prometheus_effective_values(
            chart_version,
            grafana_admin_user,
            grafana_admin_password,
        )
    fd, tmp_name = tempfile.mkstemp(suffix=".yaml", prefix="helm-prom-stack-")
    os.close(fd)
    tmp_path = Path(tmp_name)
    try:
        tmp_path.write_text(dump_helm_values_yaml(values), encoding="utf-8")
        rc, out, err = _helm_upgrade_install(
            cluster_name,
            "kube-prometheus-stack",
            "prometheus-community/kube-prometheus-stack",
            "monitoring",
            chart_version=chart_version,
            values_file=tmp_path,
        )
    finally:
        tmp_path.unlink(missing_ok=True)
    text = "\n".join(filter(None, [out, err])).strip()
    if rc != 0:
        raise HelmAddonError(text or f"helm exit {rc}", exit_code=rc)
    return True, text or "kube-prometheus-stack установлен."


def uninstall_kube_prometheus_stack(cluster_name: str) -> tuple[bool, str]:
    rc, out, err = _helm_uninstall(cluster_name, "kube-prometheus-stack", "monitoring")
    text = "\n".join(filter(None, [out, err])).strip()
    if rc != 0 and "not found" not in text.lower():
        raise HelmAddonError(text or f"helm exit {rc}", exit_code=rc)
    return True, text or "kube-prometheus-stack удалён (или отсутствовал)."


def install_metrics_server(
    cluster_name: str,
    chart_version: str | None = None,
    user_values: dict[str, Any] | None = None,
) -> tuple[bool, str]:
    """Полный values из UI или автосборка (чарт + args для kind)."""
    helm_repo_ensure()
    if user_values is not None:
        values = user_values
    else:
        values = compose_metrics_server_effective_values(chart_version)
    fd_m, tmp_ms = tempfile.mkstemp(suffix=".yaml", prefix="helm-metrics-")
    os.close(fd_m)
    tmp_path = Path(tmp_ms)
    try:
        tmp_path.write_text(dump_helm_values_yaml(values), encoding="utf-8")
        rc, out, err = _helm_upgrade_install(
            cluster_name,
            "metrics-server",
            "metrics-server/metrics-server",
            "kube-system",
            chart_version=chart_version,
            values_file=tmp_path,
        )
    finally:
        tmp_path.unlink(missing_ok=True)
    text = "\n".join(filter(None, [out, err])).strip()
    if rc != 0:
        raise HelmAddonError(text or f"helm exit {rc}", exit_code=rc)
    return True, text or "metrics-server установлен."


def uninstall_metrics_server(cluster_name: str) -> tuple[bool, str]:
    rc, out, err = _helm_uninstall(cluster_name, "metrics-server", "kube-system")
    text = "\n".join(filter(None, [out, err])).strip()
    if rc != 0 and "not found" not in text.lower():
        raise HelmAddonError(text or f"helm exit {rc}", exit_code=rc)
    return True, text or "metrics-server удалён (или отсутствовал)."


def _kiali_user_values_define_auth_strategy(values: dict[str, Any] | None) -> bool:
    """
    True, если в пользовательских values kiali-server уже задан ``auth.strategy``
    (тогда Helm не подставляет ``anonymous`` поверх).
    """
    if not values:
        return False
    auth = values.get("auth")
    if not isinstance(auth, dict):
        return False
    strat = auth.get("strategy")
    return isinstance(strat, str) and bool(strat.strip())


def install_istio_and_kiali(
    cluster_name: str,
    *,
    istio_chart_version: str | None = None,
    kiali_chart_version: str | None = None,
    istio_base_values: dict[str, Any] | None = None,
    istiod_values: dict[str, Any] | None = None,
    kiali_user_values: dict[str, Any] | None = None,
) -> tuple[bool, str]:
    """
    Istio base + istiod (опциональные values из UI) + kiali-server.

    Kiali: стратегия ``login`` в актуальных версиях не поддерживается upstream. По умолчанию
    выставляется ``auth.strategy=anonymous`` (удобно для kind и port-forward); иначе
    задайте ``auth`` в YAML values (``token``, ``openid`` и т.д., см. kiali.io).

    Для каждого чарта: если словарь ``None`` — без файла ``-f`` (дефолты чарта);
    если непустой dict — полный values как в редакторе.
    """
    helm_repo_ensure()
    _ensure_namespace(cluster_name, "istio-system")

    log_parts: list[str] = []
    istio_ver = (istio_chart_version or "").strip() or None
    kiali_ver = (kiali_chart_version or "").strip() or None

    base_tmp: Path | None = None
    try:
        base_file: Path | None = None
        if istio_base_values:
            fd, name = tempfile.mkstemp(suffix=".yaml", prefix="helm-istio-base-")
            os.close(fd)
            base_tmp = Path(name)
            base_tmp.write_text(dump_helm_values_yaml(istio_base_values), encoding="utf-8")
            base_file = base_tmp
        rc, out, err = _helm_upgrade_install(
            cluster_name,
            "istio-base",
            "istio/base",
            "istio-system",
            chart_version=istio_ver,
            values_file=base_file,
        )
        log_parts.append(out or err)
        if rc != 0:
            raise HelmAddonError((err or out or f"istio-base exit {rc}")[:8000], exit_code=rc)
    finally:
        if base_tmp is not None:
            base_tmp.unlink(missing_ok=True)

    istiod_tmp: Path | None = None
    try:
        istiod_file: Path | None = None
        if istiod_values:
            fd, name = tempfile.mkstemp(suffix=".yaml", prefix="helm-istiod-")
            os.close(fd)
            istiod_tmp = Path(name)
            istiod_tmp.write_text(dump_helm_values_yaml(istiod_values), encoding="utf-8")
            istiod_file = istiod_tmp
        rc2, o2, e2 = _helm_upgrade_install(
            cluster_name,
            "istiod",
            "istio/istiod",
            "istio-system",
            chart_version=istio_ver,
            values_file=istiod_file,
        )
        log_parts.append(o2 or e2)
        if rc2 != 0:
            raise HelmAddonError((e2 or o2 or f"istiod exit {rc2}")[:8000], exit_code=rc2)
    finally:
        if istiod_tmp is not None:
            istiod_tmp.unlink(missing_ok=True)

    kiali_extra: list[Path] = []
    kiali_tmp: Path | None = None
    try:
        if kiali_user_values:
            fd, name = tempfile.mkstemp(suffix=".yaml", prefix="helm-kiali-user-")
            os.close(fd)
            kiali_tmp = Path(name)
            kiali_tmp.write_text(dump_helm_values_yaml(kiali_user_values), encoding="utf-8")
            kiali_extra.append(kiali_tmp)
        kiali_helm_extra: list[str] = []
        if not _kiali_user_values_define_auth_strategy(kiali_user_values):
            kiali_helm_extra.extend(["--set", "auth.strategy=anonymous"])
        rc3, o3, e3 = _helm_upgrade_install(
            cluster_name,
            "kiali-server",
            "kiali/kiali-server",
            "istio-system",
            chart_version=kiali_ver,
            extra_args=kiali_helm_extra or None,
            extra_values_files=kiali_extra or None,
        )
    finally:
        if kiali_tmp is not None:
            kiali_tmp.unlink(missing_ok=True)
    log_parts.append(o3 or e3)
    if rc3 != 0:
        raise HelmAddonError((e3 or o3 or f"kiali-server exit {rc3}")[:8000], exit_code=rc3)

    return True, "\n\n".join(x for x in log_parts if x).strip() or "Istio и Kiali установлены."


def uninstall_istio_and_kiali(cluster_name: str) -> tuple[bool, str]:
    """Обратный порядок: Kiali → istiod → istio-base."""
    parts: list[str] = []
    for rel in ("kiali-server", "istiod", "istio-base"):
        rc, out, err = _helm_uninstall(cluster_name, rel, "istio-system")
        line = "\n".join(filter(None, [out, err])).strip()
        if line:
            parts.append(line)
        if rc != 0 and "not found" not in line.lower() and "release: not found" not in line.lower():
            raise HelmAddonError(line or f"helm uninstall {rel} код {rc}", exit_code=rc)
    return True, "\n".join(parts).strip() or "Istio/Kiali удалены (или отсутствовали)."