Sergey Antropoff
eb063aec20
- Выделена страница списка кластеров, панель упрощена; nav_active и крошки ведут в раздел Кластеры; theme.js синхронизирует активную пилюлю по URL. - Доработки дашборда, аддонов, журнала, стилей и API-документации. - Поддержка Podman: docker-compose.podman.yml, скрипты сокета; Makefile и env.
264 lines
9.3 KiB
Python
264 lines
9.3 KiB
Python
"""Сохранение полного журнала развёртывания кластера в ``clusters/<имя>/provision_log.json``.
|
|
|
|
Журнал операций Helm-аддонов — ``helm_addon_log.json``: **история** операций (массив ``entries``),
|
|
старый формат «один объект в файле» при чтении мигрируется автоматически.
|
|
|
|
Автор: Сергей Антропов
|
|
Сайт: https://devops.org.ru
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from core.cluster_lifecycle import validate_cluster_name
|
|
from kind_k8s_paths import clusters_dir
|
|
|
|
logger = logging.getLogger("kind_k8s.provision_log")
|
|
|
|
PROVISION_LOG_FILENAME = "provision_log.json"
|
|
PROVISION_LOG_VERSION = 1
|
|
|
|
HELM_ADDON_LOG_FILENAME = "helm_addon_log.json"
|
|
"""Имя файла истории Helm; корневой объект с полем ``entries`` (версия файла 2)."""
|
|
HELM_ADDON_LOG_VERSION = 1
|
|
"""Версия схемы одной записи внутри ``entries``."""
|
|
HELM_ADDON_LOG_FILE_VERSION = 2
|
|
"""Версия обёртки файла (список ``entries``)."""
|
|
|
|
_locks_guard = threading.Lock()
|
|
_helm_log_locks: dict[str, threading.Lock] = {}
|
|
|
|
|
|
def _helm_cluster_lock(cluster_name: str) -> threading.Lock:
|
|
with _locks_guard:
|
|
if cluster_name not in _helm_log_locks:
|
|
_helm_log_locks[cluster_name] = threading.Lock()
|
|
return _helm_log_locks[cluster_name]
|
|
|
|
|
|
def _max_helm_addon_entries() -> int:
|
|
raw = (os.environ.get("KIND_K8S_HELM_ADDON_LOG_MAX_ENTRIES") or "500").strip()
|
|
try:
|
|
return max(20, min(int(raw), 5000))
|
|
except ValueError:
|
|
return 500
|
|
|
|
|
|
def provision_log_file_path(cluster_name: str) -> Path:
|
|
"""Путь к JSON с журналом операции для кластера ``cluster_name``."""
|
|
return clusters_dir() / cluster_name.strip() / PROVISION_LOG_FILENAME
|
|
|
|
|
|
def helm_addon_log_file_path(cluster_name: str) -> Path:
|
|
"""Путь к JSON с историей журналов установки/удаления Helm-аддонов."""
|
|
return clusters_dir() / cluster_name.strip() / HELM_ADDON_LOG_FILENAME
|
|
|
|
|
|
def write_cluster_provision_log(
|
|
*,
|
|
cluster_name: str,
|
|
job_id: str,
|
|
job_kind: str,
|
|
status: str,
|
|
message: str | None,
|
|
lines: list[str],
|
|
result: dict[str, Any] | None,
|
|
) -> Path | None:
|
|
"""
|
|
Атомарно записать полный журнал (все строки без обрезки буфера задания).
|
|
|
|
Возвращает путь к файлу или ``None``, если каталог кластера не существует.
|
|
"""
|
|
name = cluster_name.strip()
|
|
cdir = clusters_dir() / name
|
|
if not cdir.is_dir():
|
|
logger.debug("Каталог кластера отсутствует, provision_log не пишем: %s", cdir)
|
|
return None
|
|
|
|
path = cdir / PROVISION_LOG_FILENAME
|
|
payload: dict[str, Any] = {
|
|
"version": PROVISION_LOG_VERSION,
|
|
"job_id": job_id,
|
|
"kind": job_kind,
|
|
"cluster_name": name,
|
|
"finished_at_utc": datetime.now(timezone.utc).isoformat(),
|
|
"status": status,
|
|
"message": message,
|
|
"lines": list(lines),
|
|
"result": result,
|
|
}
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
try:
|
|
text = json.dumps(payload, ensure_ascii=False, indent=2, default=str)
|
|
tmp.write_text(text, encoding="utf-8")
|
|
tmp.replace(path)
|
|
logger.info("Сохранён журнал развёртывания: %s (%s строк)", path, len(lines))
|
|
return path
|
|
except OSError as e:
|
|
logger.warning("Не удалось записать %s: %s", path, e)
|
|
try:
|
|
tmp.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _helm_addon_rows_from_parsed_json(data: dict[str, Any], cname: str) -> list[dict[str, Any]]:
|
|
"""Разобрать содержимое ``helm_addon_log.json`` в плоский список строк для UI/API."""
|
|
raw_entries = data.get("entries")
|
|
if isinstance(raw_entries, list) and raw_entries:
|
|
out: list[dict[str, Any]] = []
|
|
for e in raw_entries:
|
|
if isinstance(e, dict):
|
|
row = dict(e)
|
|
row["source_cluster"] = cname
|
|
out.append(row)
|
|
return out
|
|
# Легаси: один объект операции в корне файла
|
|
if data.get("job_id") is not None or data.get("finished_at_utc") is not None:
|
|
row = dict(data)
|
|
row["source_cluster"] = cname
|
|
return [row]
|
|
return []
|
|
|
|
|
|
def write_cluster_helm_addon_log(
|
|
*,
|
|
cluster_name: str,
|
|
job_id: str,
|
|
job_kind: str,
|
|
status: str,
|
|
message: str | None,
|
|
lines: list[str],
|
|
result: dict[str, Any] | None,
|
|
) -> Path | None:
|
|
"""
|
|
Добавить запись в историю ``helm_addon_log.json`` (новые записи в начале списка).
|
|
|
|
Файл хранит ``{"file_version": 2, "entries": [ ... ]}``; лимит длины — ``KIND_K8S_HELM_ADDON_LOG_MAX_ENTRIES``.
|
|
"""
|
|
name = cluster_name.strip()
|
|
cdir = clusters_dir() / name
|
|
if not cdir.is_dir():
|
|
logger.debug("Каталог кластера отсутствует, helm_addon_log не пишем: %s", cdir)
|
|
return None
|
|
|
|
path = cdir / HELM_ADDON_LOG_FILENAME
|
|
new_entry: dict[str, Any] = {
|
|
"version": HELM_ADDON_LOG_VERSION,
|
|
"job_id": job_id,
|
|
"kind": job_kind,
|
|
"cluster_name": name,
|
|
"finished_at_utc": datetime.now(timezone.utc).isoformat(),
|
|
"status": status,
|
|
"message": message,
|
|
"lines": list(lines),
|
|
"result": result,
|
|
}
|
|
|
|
lock = _helm_cluster_lock(name)
|
|
with lock:
|
|
entries: list[dict[str, Any]] = []
|
|
if path.is_file():
|
|
try:
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
if isinstance(raw, dict):
|
|
ex = raw.get("entries")
|
|
if isinstance(ex, list):
|
|
entries = [e for e in ex if isinstance(e, dict)]
|
|
elif raw.get("job_id") is not None or raw.get("finished_at_utc") is not None:
|
|
legacy = {k: v for k, v in raw.items() if k != "source_cluster"}
|
|
entries = [legacy]
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
logger.warning("helm_addon_log %s не прочитан, создаём заново: %s", path, e)
|
|
entries = []
|
|
|
|
entries.insert(0, new_entry)
|
|
cap = _max_helm_addon_entries()
|
|
if len(entries) > cap:
|
|
entries = entries[:cap]
|
|
|
|
payload: dict[str, Any] = {
|
|
"file_version": HELM_ADDON_LOG_FILE_VERSION,
|
|
"entries": entries,
|
|
}
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
try:
|
|
tmp.write_text(
|
|
json.dumps(payload, ensure_ascii=False, indent=2, default=str),
|
|
encoding="utf-8",
|
|
)
|
|
tmp.replace(path)
|
|
logger.info(
|
|
"Сохранён журнал Helm-аддонов: %s (запись %s, всего записей в файле: %s)",
|
|
path,
|
|
job_id,
|
|
len(entries),
|
|
)
|
|
return path
|
|
except OSError as e:
|
|
logger.warning("Не удалось записать %s: %s", path, e)
|
|
try:
|
|
tmp.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def collect_cluster_dir_logs_page_sync(
|
|
*,
|
|
filename: str,
|
|
limit: int,
|
|
offset: int,
|
|
) -> tuple[list[dict[str, Any]], int]:
|
|
"""
|
|
Собрать JSON с диска по кластерам.
|
|
|
|
- ``provision_log.json`` — по одной строке на кластер (как раньше).
|
|
- ``helm_addon_log.json`` — по одной строке на **каждую** запись в ``entries`` (или одна для легаси-файла).
|
|
|
|
Сортировка по ``finished_at_utc`` (новые первыми), затем пагинация.
|
|
"""
|
|
lim = max(1, min(limit, 100))
|
|
off = max(0, offset)
|
|
root = clusters_dir()
|
|
if not root.is_dir():
|
|
return [], 0
|
|
rows: list[dict[str, Any]] = []
|
|
for sub in sorted(root.iterdir()):
|
|
if not sub.is_dir():
|
|
continue
|
|
cname = sub.name
|
|
if not validate_cluster_name(cname):
|
|
continue
|
|
path = sub / filename
|
|
if not path.is_file():
|
|
continue
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
if not isinstance(data, dict):
|
|
continue
|
|
if filename == HELM_ADDON_LOG_FILENAME:
|
|
rows.extend(_helm_addon_rows_from_parsed_json(data, cname))
|
|
else:
|
|
row = dict(data)
|
|
row["source_cluster"] = cname
|
|
rows.append(row)
|
|
|
|
def sort_key(r: dict[str, Any]) -> str:
|
|
return str(r.get("finished_at_utc") or "")
|
|
|
|
rows.sort(key=sort_key, reverse=True)
|
|
total = len(rows)
|
|
chunk = rows[off : off + lim]
|
|
return chunk, total
|