|
|
|
@@ -0,0 +1,280 @@
|
|
|
|
|
"""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
|
|
|
|
|
|
|
|
|
|
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,
|
|
|
|
|
validate_cluster_name,
|
|
|
|
|
)
|
|
|
|
|
from core.job_store import job_store
|
|
|
|
|
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 _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")
|
|
|
|
|
|
|
|
|
|
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,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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 [
|
|
|
|
|
JobView(
|
|
|
|
|
job_id=r.job_id,
|
|
|
|
|
kind=r.kind,
|
|
|
|
|
status=r.status,
|
|
|
|
|
cluster_name=r.cluster_name,
|
|
|
|
|
created_at_utc=r.created_at_utc,
|
|
|
|
|
message=r.message,
|
|
|
|
|
result=r.result,
|
|
|
|
|
)
|
|
|
|
|
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:
|
|
|
|
|
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,
|
|
|
|
|
)
|
|
|
|
|
except KindClusterError as e:
|
|
|
|
|
await job_store.set_failed(job_id, str(e))
|
|
|
|
|
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)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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.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 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,
|
|
|
|
|
)
|
Sergey Antropoff