Sergey Antropoff
af0d1705cc
- Страница /cluster/<имя>: сводка, ресурсы узлов полосами, отдельные карточки на каждый kubectl get … -o json, рестарт пода. - API: overview с блоками k8s_*, POST pods/restart, расширенный набор ресурсов (-A). - Панель: спиннер загрузки, правки дашборда и стилей; документация api_routes, compose и прочие сопутствующие изменения.
894 lines
38 KiB
Python
894 lines
38 KiB
Python
"""CRUD-операции над кластерами kind и сводная статистика.
|
||
|
||
Автор: Сергей Антропов
|
||
Сайт: https://devops.org.ru
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import shutil
|
||
import tempfile
|
||
from functools import partial
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query
|
||
from fastapi.responses import FileResponse, JSONResponse
|
||
from starlette.background import BackgroundTask
|
||
|
||
from core.cluster_lifecycle import (
|
||
KindClusterError,
|
||
cluster_summary_for_api,
|
||
configured_container_cli,
|
||
create_cluster_non_interactive,
|
||
delete_kind_cluster_and_data,
|
||
kubectl_delete_pod,
|
||
kubectl_get_json,
|
||
kubectl_get_all_namespaces_wide,
|
||
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 (
|
||
aggregate_kind_cluster_resources,
|
||
collect_kind_clusters_resource_stats,
|
||
collect_single_cluster_resource_stats,
|
||
running_kind_clusters_mask,
|
||
)
|
||
from core.job_store import (
|
||
JobRecord,
|
||
end_job_tracking,
|
||
get_logs_snapshot_sync,
|
||
get_progress_sync,
|
||
job_store,
|
||
register_uncapped_job_log,
|
||
request_cancel_sync,
|
||
take_uncapped_log_finalize,
|
||
)
|
||
from core.provision_log import provision_log_file_path, write_cluster_provision_log
|
||
from core.kind_guard import kind_cluster_lock
|
||
from kind_k8s_paths import clusters_dir
|
||
from kubeconfig_patch import kubeconfig_host_file, patch_kubeconfig_server_for_host
|
||
from models.schemas import (
|
||
AggregateResourcesSummary,
|
||
ClusterCreateAccepted,
|
||
ClusterCreateRequest,
|
||
ClusterOverviewResponse,
|
||
ClusterSummary,
|
||
ClusterWorkloadsResponse,
|
||
JobView,
|
||
K8sListJsonBlock,
|
||
PodRestartRequest,
|
||
PodRestartResponse,
|
||
StatsResponse,
|
||
)
|
||
|
||
logger = logging.getLogger("kind_k8s.api.clusters")
|
||
|
||
# Безопасные имена namespace/pod для kubectl (без shell-метасимволов).
|
||
_K8S_NAME_SAFE = re.compile(r"^[a-zA-Z0-9][-a-zA-Z0-9.]*[a-zA-Z0-9]$|^[a-zA-Z0-9]$")
|
||
|
||
|
||
def _k8s_ns_pod_valid(s: str) -> bool:
|
||
if not s or len(s) > 253:
|
||
return False
|
||
return bool(_K8S_NAME_SAFE.match(s))
|
||
|
||
|
||
def _k8s_list_from_kubectl(
|
||
rc: int,
|
||
items: list[dict[str, Any]] | None,
|
||
err: str,
|
||
) -> K8sListJsonBlock:
|
||
"""Собрать блок JSON для UI из результата :func:`kubectl_get_json`."""
|
||
if items is not None and rc == 0:
|
||
return K8sListJsonBlock(ok=True, rc=0, items=items, message=None)
|
||
return K8sListJsonBlock(ok=False, rc=rc, items=[], message=err or "ошибка kubectl")
|
||
|
||
|
||
def _unlink_temp_kubeconfig(path: Path) -> None:
|
||
"""Удалить временный файл после отдачи ответа (скачивание kubeconfig)."""
|
||
try:
|
||
path.unlink(missing_ok=True)
|
||
except OSError as e:
|
||
logger.debug("Не удалось удалить временный kubeconfig %s: %s", path, e)
|
||
|
||
router = APIRouter(tags=["clusters"])
|
||
|
||
|
||
def _api_job_log_limit_detail() -> int:
|
||
"""Сколько строк журнала отдавать в GET /jobs/{id} (полный хвост буфера задания)."""
|
||
raw = (os.environ.get("KIND_K8S_JOB_API_LOG_MAX_LINES") or "5000").strip()
|
||
try:
|
||
return max(200, min(int(raw), 20000))
|
||
except ValueError:
|
||
return 5000
|
||
|
||
|
||
def _record_to_job_view(rec: JobRecord, *, include_progress_log: bool = True) -> 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 not include_progress_log:
|
||
log_tail: list[str] = []
|
||
else:
|
||
if rec.status in ("queued", "running"):
|
||
log_tail = get_logs_snapshot_sync(rec.job_id)
|
||
else:
|
||
log_tail = list(rec.log_lines or [])
|
||
cap = _api_job_log_limit_detail()
|
||
if len(log_tail) > cap:
|
||
log_tail = log_tail[-cap:]
|
||
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
|
||
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)
|
||
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()
|
||
agg: AggregateResourcesSummary
|
||
if res_err:
|
||
agg = AggregateResourcesSummary()
|
||
else:
|
||
agg = aggregate_kind_cluster_resources(res_blocks)
|
||
|
||
return StatsResponse(
|
||
container_cli=configured_container_cli(),
|
||
kind_clusters_count=len(kind_names),
|
||
local_cluster_dirs_count=len(subdirs),
|
||
total_workers_from_meta=total_workers,
|
||
jobs_total=len(jobs),
|
||
jobs_recent_failed=failed,
|
||
cluster_resources=res_blocks,
|
||
cluster_resources_error=res_err,
|
||
aggregate_cluster_resources=agg,
|
||
)
|
||
|
||
|
||
@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)
|
||
# В списке не тащим progress_log — экономим трафик; полный журнал только в GET /jobs/{id}.
|
||
return [_record_to_job_view(r, include_progress_log=False) for r in items]
|
||
|
||
|
||
@router.delete("/jobs", summary="Очистить завершённые задания из памяти")
|
||
async def delete_completed_jobs() -> dict[str, int]:
|
||
"""
|
||
Удаляет записи со статусом success, failed, cancelled. Задания в очереди и в работе не трогаются.
|
||
"""
|
||
removed = await job_store.purge_completed()
|
||
logger.info("Очистка заданий: удалено завершённых записей: %s", removed)
|
||
return {"removed": removed}
|
||
|
||
|
||
@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)
|
||
running_map = running_kind_clusters_mask(all_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"]),
|
||
kind_nodes_running=bool(running_map.get(name, False)),
|
||
has_local_kubeconfig=bool(summary["has_local_kubeconfig"]),
|
||
has_provision_log=bool(summary.get("has_provision_log")),
|
||
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:
|
||
"""
|
||
Kubeconfig для **kubectl на машине пользователя**: копия ``clusters/<имя>/kubeconfig``
|
||
с ``server=https://<KIND_K8S_KUBECONFIG_CLIENT_HOST или localhost>:<порт>`` (порт из ``docker port``).
|
||
|
||
Патч выполняется при каждом скачивании, чтобы учитывать переменные окружения и актуальный порт.
|
||
Если патч невозможен — отдаётся сохранённый ``kubeconfig.host`` или сырой ``kubeconfig``.
|
||
"""
|
||
if not validate_cluster_name(name):
|
||
raise HTTPException(status_code=400, detail="Некорректное имя кластера")
|
||
n = name.strip()
|
||
base = clusters_dir() / n
|
||
primary = base / "kubeconfig"
|
||
if not primary.is_file():
|
||
raise HTTPException(status_code=404, detail="kubeconfig не найден")
|
||
|
||
fd, tmp_name = tempfile.mkstemp(suffix=".yaml", prefix="kind-k8s-kube-")
|
||
os.close(fd)
|
||
tmp_path = Path(tmp_name)
|
||
try:
|
||
shutil.copyfile(primary, tmp_path, follow_symlinks=True)
|
||
except OSError as e:
|
||
tmp_path.unlink(missing_ok=True)
|
||
logger.warning("Копирование kubeconfig для скачивания: %s", e)
|
||
raise HTTPException(status_code=500, detail="Не удалось подготовить kubeconfig") from e
|
||
|
||
patched = await asyncio.to_thread(
|
||
partial(patch_kubeconfig_server_for_host, cluster_name=n, kube_path=tmp_path),
|
||
)
|
||
if patched:
|
||
logger.info("Отдача kubeconfig для кластера %s (сгенерировано с актуальным server)", n)
|
||
return FileResponse(
|
||
path=str(tmp_path),
|
||
filename=f"kubeconfig-{n}.yaml",
|
||
media_type="application/x-yaml",
|
||
background=BackgroundTask(_unlink_temp_kubeconfig, tmp_path),
|
||
)
|
||
|
||
tmp_path.unlink(missing_ok=True)
|
||
host_path = kubeconfig_host_file(base)
|
||
if host_path.is_file():
|
||
logger.info("Отдача kubeconfig для кластера %s (файл %s, патч при скачивании не удался)", n, host_path.name)
|
||
return FileResponse(
|
||
path=host_path,
|
||
filename=f"kubeconfig-{n}.yaml",
|
||
media_type="application/x-yaml",
|
||
)
|
||
logger.info("Отдача kubeconfig для кластера %s (сырой kubeconfig)", n)
|
||
return FileResponse(
|
||
path=primary,
|
||
filename=f"kubeconfig-{n}.yaml",
|
||
media_type="application/x-yaml",
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/clusters/{name}/provision-log",
|
||
summary="Журнал развёртывания (JSON)",
|
||
responses={404: {"description": "Файл не найден"}},
|
||
)
|
||
async def get_cluster_provision_log(name: str) -> JSONResponse:
|
||
"""Полный журнал последнего создания/старта кластера (``provision_log.json`` в каталоге кластера)."""
|
||
if not validate_cluster_name(name):
|
||
raise HTTPException(status_code=400, detail="Некорректное имя кластера")
|
||
path = provision_log_file_path(name)
|
||
if not path.is_file():
|
||
raise HTTPException(status_code=404, detail="Журнал развёртывания не найден")
|
||
try:
|
||
raw = await asyncio.to_thread(path.read_text, encoding="utf-8")
|
||
data = json.loads(raw)
|
||
except (OSError, json.JSONDecodeError) as e:
|
||
logger.warning("provision_log для %s не прочитан: %s", name, e)
|
||
raise HTTPException(status_code=500, detail="Не удалось прочитать журнал") from e
|
||
return JSONResponse(content=data)
|
||
|
||
|
||
@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,
|
||
cluster_name=name,
|
||
kubeconfig=kc,
|
||
)
|
||
pods_rc, pods_out = await asyncio.to_thread(
|
||
kubectl_pods_all_namespaces,
|
||
cluster_name=name,
|
||
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}/overview",
|
||
response_model=ClusterOverviewResponse,
|
||
summary="Сводка кластера (страница UI)",
|
||
)
|
||
async def cluster_overview(name: str) -> ClusterOverviewResponse:
|
||
"""
|
||
Мета, метрики узлов (docker/podman), агрегаты как на главной, вывод kubectl по **всему кластеру**:
|
||
узлы; для namespace-ресурсов везде **-A** (все пространства имён); отдельно namespaces, RS, jobs и т.д.
|
||
"""
|
||
if not validate_cluster_name(name):
|
||
raise HTTPException(status_code=400, detail="Некорректное имя кластера")
|
||
n = name.strip()
|
||
summary = cluster_summary_for_api(n)
|
||
run_mask = await asyncio.to_thread(running_kind_clusters_mask, [n])
|
||
kind_running = bool(run_mask.get(n, False))
|
||
|
||
block, res_err = await asyncio.to_thread(collect_single_cluster_resource_stats, n)
|
||
agg = aggregate_kind_cluster_resources([block])
|
||
|
||
kc = clusters_dir() / n / "kubeconfig"
|
||
kube_err: str | None = None
|
||
k8s_nodes = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_pods = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_deployments = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_statefulsets = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_daemonsets = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_services = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_ingresses = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_namespaces = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_replicasets = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_jobs = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_cronjobs = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
k8s_pvcs = K8sListJsonBlock(ok=False, items=[], message=None)
|
||
|
||
if not kc.is_file():
|
||
kube_err = "Нет сохранённого kubeconfig в clusters/<имя>/"
|
||
_kube_missing = K8sListJsonBlock(ok=False, items=[], message=kube_err)
|
||
k8s_nodes = _kube_missing
|
||
k8s_pods = k8s_deployments = k8s_statefulsets = k8s_daemonsets = _kube_missing
|
||
k8s_services = k8s_ingresses = k8s_namespaces = k8s_replicasets = _kube_missing
|
||
k8s_jobs = k8s_cronjobs = k8s_pvcs = _kube_missing
|
||
else:
|
||
(
|
||
n_res,
|
||
p_res,
|
||
d_res,
|
||
st_res,
|
||
dm_res,
|
||
svc_res,
|
||
ing_res,
|
||
ns_res,
|
||
rs_res,
|
||
job_res,
|
||
cj_res,
|
||
pvc_res,
|
||
) = await asyncio.gather(
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["nodes"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["pods", "-A"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["deployments", "-A"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["statefulsets", "-A"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["daemonsets", "-A"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["services", "-A"]),
|
||
asyncio.to_thread(
|
||
kubectl_get_json,
|
||
cluster_name=n,
|
||
kubeconfig=kc,
|
||
resource_args=["ingresses.networking.k8s.io", "-A"],
|
||
),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["namespaces"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["replicasets", "-A"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["jobs", "-A"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["cronjobs", "-A"]),
|
||
asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["pvc", "-A"]),
|
||
)
|
||
k8s_nodes = _k8s_list_from_kubectl(*n_res)
|
||
k8s_pods = _k8s_list_from_kubectl(*p_res)
|
||
k8s_deployments = _k8s_list_from_kubectl(*d_res)
|
||
k8s_statefulsets = _k8s_list_from_kubectl(*st_res)
|
||
k8s_daemonsets = _k8s_list_from_kubectl(*dm_res)
|
||
k8s_services = _k8s_list_from_kubectl(*svc_res)
|
||
k8s_ingresses = _k8s_list_from_kubectl(*ing_res)
|
||
k8s_namespaces = _k8s_list_from_kubectl(*ns_res)
|
||
k8s_replicasets = _k8s_list_from_kubectl(*rs_res)
|
||
k8s_jobs = _k8s_list_from_kubectl(*job_res)
|
||
k8s_cronjobs = _k8s_list_from_kubectl(*cj_res)
|
||
k8s_pvcs = _k8s_list_from_kubectl(*pvc_res)
|
||
|
||
meta = summary.get("meta")
|
||
return ClusterOverviewResponse(
|
||
cluster_name=n,
|
||
registered_in_kind=bool(summary["registered_in_kind"]),
|
||
kind_nodes_running=kind_running,
|
||
has_local_kubeconfig=bool(summary["has_local_kubeconfig"]),
|
||
has_provision_log=bool(summary.get("has_provision_log")),
|
||
meta=dict(meta) if isinstance(meta, dict) else {},
|
||
resources_error=res_err,
|
||
cluster_resources=block,
|
||
aggregate_resources=agg,
|
||
kubeconfig_error=kube_err,
|
||
nodes_rc=None,
|
||
nodes_output=None,
|
||
pods_rc=None,
|
||
pods_output=None,
|
||
deployments_rc=None,
|
||
deployments_output=None,
|
||
statefulsets_rc=None,
|
||
statefulsets_output=None,
|
||
daemonsets_rc=None,
|
||
daemonsets_output=None,
|
||
services_rc=None,
|
||
services_output=None,
|
||
ingresses_rc=None,
|
||
ingresses_output=None,
|
||
k8s_nodes=k8s_nodes,
|
||
k8s_pods=k8s_pods,
|
||
k8s_deployments=k8s_deployments,
|
||
k8s_statefulsets=k8s_statefulsets,
|
||
k8s_daemonsets=k8s_daemonsets,
|
||
k8s_services=k8s_services,
|
||
k8s_ingresses=k8s_ingresses,
|
||
k8s_namespaces=k8s_namespaces,
|
||
k8s_replicasets=k8s_replicasets,
|
||
k8s_jobs=k8s_jobs,
|
||
k8s_cronjobs=k8s_cronjobs,
|
||
k8s_pvcs=k8s_pvcs,
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/clusters/{name}/pods/restart",
|
||
response_model=PodRestartResponse,
|
||
summary="Рестарт пода (kubectl delete pod)",
|
||
)
|
||
async def restart_cluster_pod(name: str, body: PodRestartRequest) -> PodRestartResponse:
|
||
"""
|
||
Удаление пода: Deployment/RS/Job при необходимости создают новый Pod (типичный «рестарт»).
|
||
"""
|
||
if not validate_cluster_name(name):
|
||
raise HTTPException(status_code=400, detail="Некорректное имя кластера")
|
||
n = name.strip()
|
||
ns = body.namespace.strip()
|
||
pod = body.pod.strip()
|
||
if not _k8s_ns_pod_valid(ns) or not _k8s_ns_pod_valid(pod):
|
||
raise HTTPException(status_code=400, detail="Некорректное имя namespace или пода")
|
||
kc = clusters_dir() / n / "kubeconfig"
|
||
if not kc.is_file():
|
||
raise HTTPException(status_code=404, detail="kubeconfig не найден")
|
||
rc, msg = await asyncio.to_thread(
|
||
kubectl_delete_pod,
|
||
cluster_name=n,
|
||
kubeconfig=kc,
|
||
namespace=ns,
|
||
pod_name=pod,
|
||
grace_period=30,
|
||
)
|
||
if rc != 0:
|
||
logger.warning("restart pod %s/%s в кластере %s: rc=%s %s", ns, pod, n, rc, msg[:500])
|
||
raise HTTPException(status_code=502, detail=msg[:4000] if msg else "kubectl delete pod завершился с ошибкой")
|
||
logger.info("Рестарт пода %s/%s в кластере %s", ns, pod, n)
|
||
return PodRestartResponse(cluster_name=n, namespace=ns, pod=pod, rc=rc, message=msg or "ok")
|
||
|
||
|
||
@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,
|
||
cluster_name=name.strip(),
|
||
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:
|
||
register_uncapped_job_log(job_id)
|
||
cname = body.name.strip()
|
||
try:
|
||
async with kind_cluster_lock:
|
||
await job_store.set_running(job_id)
|
||
try:
|
||
result = await asyncio.to_thread(
|
||
create_cluster_non_interactive,
|
||
name=cname,
|
||
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:
|
||
lines = take_uncapped_log_finalize(job_id)
|
||
rec = await job_store.get(job_id)
|
||
if rec is not None:
|
||
await asyncio.to_thread(
|
||
lambda: write_cluster_provision_log(
|
||
cluster_name=cname,
|
||
job_id=job_id,
|
||
job_kind="create_cluster",
|
||
status=rec.status,
|
||
message=rec.message,
|
||
lines=lines,
|
||
result=rec.result,
|
||
),
|
||
)
|
||
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 в списке)."""
|
||
register_uncapped_job_log(job_id)
|
||
n = name.strip()
|
||
try:
|
||
async with kind_cluster_lock:
|
||
await job_store.set_running(job_id)
|
||
try:
|
||
result = await asyncio.to_thread(
|
||
create_cluster_non_interactive,
|
||
name=n,
|
||
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:
|
||
lines = take_uncapped_log_finalize(job_id)
|
||
rec = await job_store.get(job_id)
|
||
if rec is not None:
|
||
await asyncio.to_thread(
|
||
lambda: write_cluster_provision_log(
|
||
cluster_name=n,
|
||
job_id=job_id,
|
||
job_kind="start_cluster",
|
||
status=rec.status,
|
||
message=rec.message,
|
||
lines=lines,
|
||
result=rec.result,
|
||
),
|
||
)
|
||
end_job_tracking(job_id)
|
||
|
||
|
||
async def _run_stop_cluster_job(job_id: str, name: str) -> None:
|
||
"""Фоновая остановка узлов с журналом в GET /jobs/{id}."""
|
||
n = name.strip()
|
||
try:
|
||
async with kind_cluster_lock:
|
||
await job_store.set_running(job_id, stage="Остановка узлов", percent=6)
|
||
try:
|
||
ok, summary = await asyncio.to_thread(stop_kind_cluster_containers, name=n, 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("stop_containers 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("stop_containers job %s: непредвиденная ошибка", job_id)
|
||
return
|
||
|
||
await job_store.set_success(
|
||
job_id,
|
||
result={"name": n, "containers_stopped_ok": ok},
|
||
message=summary,
|
||
)
|
||
logger.info("stop_containers job %s: ok=%s", job_id, ok)
|
||
finally:
|
||
end_job_tracking(job_id)
|
||
|
||
|
||
async def _run_start_containers_job(job_id: str, name: str) -> None:
|
||
"""Фоновый запуск уже существующих узлов (кластер зарегистрирован в kind)."""
|
||
n = name.strip()
|
||
try:
|
||
async with kind_cluster_lock:
|
||
await job_store.set_running(job_id, stage="Запуск узлов", percent=6)
|
||
try:
|
||
ok, summary = await asyncio.to_thread(start_kind_cluster_containers, name=n, 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("start_containers 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_containers job %s: непредвиденная ошибка", job_id)
|
||
return
|
||
|
||
if not ok:
|
||
await job_store.set_failed(job_id, summary)
|
||
return
|
||
|
||
await job_store.set_success(
|
||
job_id,
|
||
result={"name": n, "containers_started_ok": ok},
|
||
message=summary,
|
||
)
|
||
logger.info("start_containers job %s: ok=%s", job_id, ok)
|
||
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="Остановить узлы кластера (фон, журнал в /jobs)",
|
||
responses={400: {"description": "Некорректное имя"}},
|
||
)
|
||
async def stop_cluster_nodes(name: str, background_tasks: BackgroundTasks) -> JSONResponse:
|
||
"""
|
||
Поставить остановку узлов в фон: тот же журнал в реальном времени, что и при создании.
|
||
|
||
Запись кластера в kind сохраняется; «Старт» снова поднимет узлы.
|
||
"""
|
||
if not validate_cluster_name(name):
|
||
raise HTTPException(status_code=400, detail="Некорректное имя кластера")
|
||
|
||
n = name.strip()
|
||
rec = await job_store.create_job("stop_containers", cluster_name=n)
|
||
background_tasks.add_task(_run_stop_cluster_job, rec.job_id, n)
|
||
logger.info("Остановка узлов %s в фоне, job_id=%s", n, rec.job_id)
|
||
return JSONResponse(
|
||
status_code=202,
|
||
content={
|
||
"job_id": rec.job_id,
|
||
"status": "queued",
|
||
"mode": "stop",
|
||
"message": "Остановка узлов; опросите GET /api/v1/jobs/{job_id}",
|
||
},
|
||
)
|
||
|
||
|
||
@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:
|
||
"""
|
||
Если кластер уже зарегистрирован — фоновый запуск сохранённых узлов (журнал в GET /jobs).
|
||
|
||
Иначе, при наличии сохранённого конфига в каталоге кластера — фоновый подъём как при создании.
|
||
"""
|
||
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:
|
||
rec = await job_store.create_job("start_containers", cluster_name=n)
|
||
background_tasks.add_task(_run_start_containers_job, rec.job_id, n)
|
||
logger.info("Запуск контейнеров кластера %s в фоне, job_id=%s", n, rec.job_id)
|
||
return JSONResponse(
|
||
status_code=202,
|
||
content={
|
||
"job_id": rec.job_id,
|
||
"status": "queued",
|
||
"mode": "containers",
|
||
"message": "Запуск узлов; опросите GET /api/v1/jobs/{job_id}",
|
||
},
|
||
)
|
||
|
||
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",
|
||
"mode": "kind_config",
|
||
"message": "Подъём кластера по сохранённому конфигу; опросите 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]:
|
||
"""
|
||
Прервать активное задание: скачивание образа, создание кластера, запуск/остановка узлов.
|
||
|
||
Для длительных команд дочерний процесс завершается принудительно; между узлами останов
|
||
также учитывает флаг отмены.
|
||
"""
|
||
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": "Запрошено прерывание; текущая команда будет остановлена, задание перейдёт в отменено",
|
||
}
|
||
|
||
|
||
@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)
|