KindClustersDashboard/app/api/v1/endpoints/clusters.py

"""CRUD-операции над кластерами kind и сводная статистика.

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

from __future__ import annotations

import asyncio
import logging
from typing import Any

from fastapi import APIRouter, BackgroundTasks, HTTPException, Query
from fastapi.responses import FileResponse, JSONResponse

from core.cluster_lifecycle import (
    KindClusterError,
    cluster_summary_for_api,
    create_cluster_non_interactive,
    delete_kind_cluster_and_data,
    kubectl_nodes_wide,
    kubectl_pods_all_namespaces,
    list_registered_kind_clusters,
    read_meta_json,
    start_kind_cluster_containers,
    stop_kind_cluster_containers,
    validate_cluster_name,
)
from core.container_resource_stats import collect_kind_clusters_resource_stats
from core.job_store import (
    JobRecord,
    end_job_tracking,
    get_logs_snapshot_sync,
    get_progress_sync,
    job_store,
    request_cancel_sync,
)
from core.kind_guard import kind_cluster_lock
from kind_k8s_paths import clusters_dir
from models.schemas import (
    ClusterCreateAccepted,
    ClusterCreateRequest,
    ClusterSummary,
    ClusterWorkloadsResponse,
    JobView,
    StatsResponse,
)

logger = logging.getLogger("kind_k8s.api.clusters")

router = APIRouter(tags=["clusters"])


def _record_to_job_view(rec: JobRecord) -> JobView:
    """JobRecord → JobView с полями прогресса из потокобезопасного снимка."""
    prog = get_progress_sync(rec.job_id)
    stage, pct = (None, None)
    if prog is not None:
        stage, pct = prog[0], prog[1]
    if rec.status in ("queued", "running"):
        log_tail = get_logs_snapshot_sync(rec.job_id)
    else:
        log_tail = list(rec.log_lines or [])
    max_log = 400
    if len(log_tail) > max_log:
        log_tail = log_tail[-max_log:]
    return JobView(
        job_id=rec.job_id,
        kind=rec.kind,
        status=rec.status,
        cluster_name=rec.cluster_name,
        created_at_utc=rec.created_at_utc,
        message=rec.message,
        result=rec.result,
        progress_stage=stage,
        progress_percent=pct,
        progress_log=log_tail,
    )


def _stats_sync() -> StatsResponse:
    """Собрать статистику (синхронно; вызывать из thread при необходимости)."""
    kind_names = list_registered_kind_clusters()
    cdir = clusters_dir()
    subdirs: list[str] = []
    if cdir.is_dir():
        subdirs = sorted(p.name for p in cdir.iterdir() if p.is_dir() and not p.name.startswith("."))

    total_workers = 0
    counted = False
    for name in subdirs:
        meta = read_meta_json(name)
        if not meta:
            continue
        w = meta.get("worker_nodes")
        if w is None:
            continue
        try:
            total_workers += int(w)
            counted = True
        except (TypeError, ValueError):
            continue

    jobs = job_store.snapshot_all()
    failed = sum(1 for j in jobs if j.status == "failed")

    res_blocks, res_err = collect_kind_clusters_resource_stats()

    return StatsResponse(
        kind_clusters_count=len(kind_names),
        local_cluster_dirs_count=len(subdirs),
        total_workers_from_meta=total_workers if counted else None,
        jobs_total=len(jobs),
        jobs_recent_failed=failed,
        cluster_resources=res_blocks,
        cluster_resources_error=res_err,
    )


@router.get("/stats", response_model=StatsResponse, summary="Статистика")
async def get_stats() -> StatsResponse:
    """Число кластеров kind, локальных каталогов, сумма workers из meta (если есть), счётчики заданий."""
    return await asyncio.to_thread(_stats_sync)


@router.get("/jobs", response_model=list[JobView], summary="Список заданий")
async def list_jobs(limit: int = Query(30, ge=1, le=200, description="Сколько последних заданий")) -> list[JobView]:
    """История создания кластеров (в памяти процесса; после перезапуска контейнера пусто)."""
    items = job_store.snapshot_recent_sorted(limit=limit)
    return [_record_to_job_view(r) for r in items]


@router.get("/clusters", response_model=list[ClusterSummary], summary="Список кластеров")
async def list_clusters() -> list[ClusterSummary]:
    """Объединение: зарегистрированные в kind + каталоги в ``clusters/`` (без дубликатов в выдаче)."""
    kind_names = set(list_registered_kind_clusters())
    cdir = clusters_dir()
    dir_names: set[str] = set()
    if cdir.is_dir():
        dir_names = {p.name for p in cdir.iterdir() if p.is_dir() and not p.name.startswith(".")}
    all_names = sorted(kind_names | dir_names)
    out: list[ClusterSummary] = []
    for name in all_names:
        summary = cluster_summary_for_api(name)
        out.append(
            ClusterSummary(
                name=str(summary["name"]),
                registered_in_kind=bool(summary["registered_in_kind"]),
                has_local_kubeconfig=bool(summary["has_local_kubeconfig"]),
                meta=dict(summary["meta"]) if isinstance(summary.get("meta"), dict) else {},
            )
        )
    logger.debug("list_clusters: %s записей", len(out))
    return out


@router.get(
    "/clusters/{name}/kubeconfig",
    summary="Скачать kubeconfig",
    responses={404: {"description": "Файл не найден"}},
)
async def download_kubeconfig(name: str) -> FileResponse:
    """Файл ``clusters/<имя>/kubeconfig`` для использования с хоста (локальная dev-среда)."""
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    path = clusters_dir() / name / "kubeconfig"
    if not path.is_file():
        raise HTTPException(status_code=404, detail="kubeconfig не найден")
    logger.info("Отдача kubeconfig для кластера %s", name)
    return FileResponse(
        path=path,
        filename=f"kubeconfig-{name}.yaml",
        media_type="application/x-yaml",
    )


@router.get(
    "/clusters/{name}/workloads",
    response_model=ClusterWorkloadsResponse,
    summary="Узлы и поды (kubectl)",
)
async def cluster_workloads(name: str) -> ClusterWorkloadsResponse:
    """``kubectl get nodes`` и ``kubectl get pods -A`` по сохранённому kubeconfig."""
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    kc = clusters_dir() / name / "kubeconfig"
    if not kc.is_file():
        return ClusterWorkloadsResponse(cluster_name=name, error="Нет сохранённого kubeconfig в clusters/<имя>/")

    nodes_rc, nodes_out = await asyncio.to_thread(kubectl_nodes_wide, kubeconfig=kc)
    pods_rc, pods_out = await asyncio.to_thread(kubectl_pods_all_namespaces, kubeconfig=kc)
    return ClusterWorkloadsResponse(
        cluster_name=name,
        nodes_rc=nodes_rc,
        nodes_output=nodes_out,
        pods_rc=pods_rc,
        pods_output=pods_out,
    )


@router.get("/clusters/{name}", summary="Детали кластера")
async def get_cluster(name: str) -> dict[str, object]:
    """Сводка и попытка ``kubectl get nodes`` (предпочтительно сохранённый kubeconfig)."""
    summary = cluster_summary_for_api(name)
    saved = clusters_dir() / name / "kubeconfig"
    kubectl_rc: int | None = None
    kubectl_msg: str | None = None
    if saved.is_file():
        rc, msg = await asyncio.to_thread(kubectl_nodes_wide, kubeconfig=saved)
        kubectl_rc = rc
        kubectl_msg = msg
    return {
        **summary,
        "kubectl_get_nodes_rc": kubectl_rc,
        "kubectl_get_nodes": kubectl_msg,
    }


async def _run_create_job(job_id: str, body: ClusterCreateRequest) -> None:
    try:
        async with kind_cluster_lock:
            await job_store.set_running(job_id)
            try:
                result = await asyncio.to_thread(
                    create_cluster_non_interactive,
                    name=body.name.strip(),
                    kubernetes_version_tag=body.kubernetes_version.strip(),
                    workers=body.workers,
                    job_id=job_id,
                )
            except KindClusterError as e:
                msg = str(e)
                if "отменено" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("create job %s: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("create job %s: непредвиденная ошибка", job_id)
                return

            payload: dict[str, Any] = {
                "cluster_name": result.cluster_name,
                "kubernetes_version_tag": result.ver_tag,
                "node_image": result.node_image,
                "workers": result.workers,
                "kubeconfig_path": str(result.kubeconfig_path),
                "kubeconfig_patched_for_host": result.kubeconfig_patched_for_host,
                "nodes_ready": result.nodes_ready,
                "nodes_ready_message": result.nodes_ready_message,
            }
            await job_store.set_success(job_id, result=payload, message="Кластер создан")
            logger.info("create job %s: успех, кластер %s", job_id, result.cluster_name)
    finally:
        end_job_tracking(job_id)


async def _run_start_cluster_job(job_id: str, name: str, kubernetes_version_tag: str, workers: int) -> None:
    """Фоновое создание кластера по уже сохранённому ``kind-config.yaml`` (без kind в списке)."""
    try:
        async with kind_cluster_lock:
            await job_store.set_running(job_id)
            try:
                result = await asyncio.to_thread(
                    create_cluster_non_interactive,
                    name=name.strip(),
                    kubernetes_version_tag=kubernetes_version_tag.strip(),
                    workers=workers,
                    job_id=job_id,
                    use_existing_config=True,
                )
            except KindClusterError as e:
                msg = str(e)
                if "отменено" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("start_cluster job %s: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("start_cluster job %s: непредвиденная ошибка", job_id)
                return

            payload: dict[str, Any] = {
                "cluster_name": result.cluster_name,
                "kubernetes_version_tag": result.ver_tag,
                "node_image": result.node_image,
                "workers": result.workers,
                "kubeconfig_path": str(result.kubeconfig_path),
                "kubeconfig_patched_for_host": result.kubeconfig_patched_for_host,
                "nodes_ready": result.nodes_ready,
                "nodes_ready_message": result.nodes_ready_message,
            }
            await job_store.set_success(job_id, result=payload, message="Кластер поднят по сохранённому конфигу")
            logger.info("start_cluster job %s: успех, кластер %s", job_id, result.cluster_name)
    finally:
        end_job_tracking(job_id)


@router.post(
    "/clusters",
    response_model=ClusterCreateAccepted,
    status_code=202,
    summary="Создать кластер (фон)",
)
async def post_create_cluster(
    body: ClusterCreateRequest,
    background_tasks: BackgroundTasks,
) -> ClusterCreateAccepted:
    """Поставить создание кластера в фон; идентификатор задания — в ответе."""
    if not validate_cluster_name(body.name.strip()):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    existing = await asyncio.to_thread(list_registered_kind_clusters)
    if body.name.strip() in existing:
        raise HTTPException(status_code=409, detail="Кластер с таким именем уже есть в kind")

    rec = await job_store.create_job("create_cluster", cluster_name=body.name.strip())
    background_tasks.add_task(_run_create_job, rec.job_id, body)
    logger.info("Принят запрос на создание кластера %s, job_id=%s", body.name, rec.job_id)
    return ClusterCreateAccepted(job_id=rec.job_id)


@router.delete("/clusters/{name}", summary="Удалить кластер")
async def delete_cluster(name: str) -> dict[str, object]:
    """``kind delete`` и удаление локальной папки ``clusters/<имя>/``."""
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    async with kind_cluster_lock:

        def _do() -> tuple[bool, str]:
            return delete_kind_cluster_and_data(name=name, log_to_stdout=False)

        try:
            kind_ok, summary = await asyncio.to_thread(_do)
        except KindClusterError as e:
            raise HTTPException(status_code=500, detail=str(e)) from e

    logger.info("Удаление кластера %s: kind_ok=%s", name, kind_ok)
    return {"name": name, "kind_delete_ok": kind_ok, "summary": summary}


@router.post(
    "/clusters/{name}/stop",
    summary="Остановить узлы кластера (docker stop)",
    responses={400: {"description": "Некорректное имя"}},
)
async def stop_cluster_nodes(name: str) -> dict[str, object]:
    """
    Остановить контейнеры узлов kind; запись кластера в kind сохраняется.

    После этого API «Старт» запустит те же контейнеры без ``kind create``.
    """
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    async with kind_cluster_lock:

        def _do() -> tuple[bool, str]:
            return stop_kind_cluster_containers(name=name)

        try:
            ok, summary = await asyncio.to_thread(_do)
        except KindClusterError as e:
            raise HTTPException(status_code=500, detail=str(e)) from e

    logger.info("Остановка узлов %s: ok=%s", name, ok)
    return {"name": name, "containers_stopped_ok": ok, "summary": summary}


@router.post(
    "/clusters/{name}/start",
    summary="Запустить кластер (контейнеры или kind create по конфигу)",
    responses={400: {"description": "Нет kind и нет kind-config.yaml"}},
)
async def start_cluster_nodes(
    name: str,
    background_tasks: BackgroundTasks,
) -> JSONResponse:
    """
    Если кластер есть в ``kind get clusters`` — ``docker start`` всех узлов.

    Если в kind нет, но есть ``clusters/<имя>/kind-config.yaml`` — фоновое ``kind create``
    (как при создании, с журналом в GET /jobs/{job_id}).
    """
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    n = name.strip()

    async with kind_cluster_lock:
        in_kind = n in await asyncio.to_thread(list_registered_kind_clusters)
        if in_kind:

            def _start() -> tuple[bool, str]:
                return start_kind_cluster_containers(name=n)

            try:
                ok, summary = await asyncio.to_thread(_start)
            except KindClusterError as e:
                raise HTTPException(status_code=500, detail=str(e)) from e
            logger.info("Запуск контейнеров кластера %s: ok=%s", n, ok)
            return JSONResponse(
                status_code=200,
                content={
                    "name": n,
                    "mode": "containers",
                    "containers_started_ok": ok,
                    "summary": summary,
                },
            )

    cfg = clusters_dir() / n / "kind-config.yaml"
    if not cfg.is_file():
        raise HTTPException(
            status_code=400,
            detail="Кластер не в kind и нет файла clusters/<имя>/kind-config.yaml — создайте кластер или восстановите конфиг.",
        )

    meta = read_meta_json(n) or {}
    ver_raw = str(meta.get("kubernetes_version_tag") or "v1.29.4").strip() or "v1.29.4"
    w_raw = meta.get("worker_nodes")
    try:
        w = int(w_raw) if w_raw is not None else 0
    except (TypeError, ValueError):
        w = 0

    rec = await job_store.create_job("start_cluster", cluster_name=n)
    background_tasks.add_task(_run_start_cluster_job, rec.job_id, n, ver_raw, w)
    logger.info("Фоновый старт кластера %s по конфигу, job_id=%s", n, rec.job_id)
    return JSONResponse(
        status_code=202,
        content={
            "job_id": rec.job_id,
            "status": "queued",
            "message": "Подъём кластера по kind-config.yaml; опросите GET /api/v1/jobs/{job_id}",
        },
    )


@router.post(
    "/jobs/{job_id}/cancel",
    summary="Запросить отмену создания кластера",
    responses={400: {"description": "Задание уже завершено"}, 404: {"description": "Нет задания"}},
)
async def cancel_create_job(job_id: str) -> dict[str, object]:
    """
    Установить флаг отмены для задания ``create_cluster`` или ``start_cluster``.

    Этап ``kind create cluster`` нельзя прервать до его завершения; после него отмена удалит
    кластер и данные (если успели создать).
    """
    rec = await job_store.get(job_id)
    if not rec:
        raise HTTPException(status_code=404, detail="Задание не найдено")
    if rec.status not in ("queued", "running"):
        raise HTTPException(status_code=400, detail="Задание уже завершено; отмена невозможна")
    if not request_cancel_sync(job_id):
        raise HTTPException(status_code=404, detail="Задание не найдено")
    logger.info("Принят запрос отмены задания %s", job_id)
    return {
        "job_id": job_id,
        "cancel_requested": True,
        "message": "Отмена обрабатывается между этапами; во время kind create дождитесь окончания шага",
    }


@router.get("/jobs/{job_id}", response_model=JobView, summary="Статус одного задания")
async def get_job(job_id: str) -> JobView:
    """Узнать состояние фонового создания кластера."""
    rec = await job_store.get(job_id)
    if not rec:
        raise HTTPException(status_code=404, detail="Задание не найдено")
    return _record_to_job_view(rec)