KindClustersDashboard/app/core/job_journal.py

"""История завершённых заданий (create/start/stop и т.д.) в ``clusters/<имя>/journal/jobs_history.json``.

Дополняет глобальный ``kind_k8s_jobs.json``: записи дублируются в каталог кластера для архива на томе.

Автор: Сергей Антропов
Сайт: 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.job_journal")

JOURNAL_SUBDIR = "journal"
JOBS_HISTORY_FILENAME = "jobs_history.json"
JOURNAL_FILE_VERSION = 1

# Блокировки по имени кластера (один процесс uvicorn).
_locks_guard = threading.Lock()
_cluster_locks: dict[str, threading.Lock] = {}


def _cluster_lock(name: str) -> threading.Lock:
    with _locks_guard:
        if name not in _cluster_locks:
            _cluster_locks[name] = threading.Lock()
        return _cluster_locks[name]


def _max_entries() -> int:
    raw = (os.environ.get("KIND_K8S_CLUSTER_JOURNAL_MAX_ENTRIES") or "500").strip()
    try:
        return max(20, min(int(raw), 5000))
    except ValueError:
        return 500


def _max_log_lines() -> int:
    raw = (os.environ.get("KIND_K8S_CLUSTER_JOURNAL_MAX_LOG_LINES") or "2000").strip()
    try:
        return max(50, min(int(raw), 20000))
    except ValueError:
        return 2000


def journal_file_path(cluster_name: str) -> Path:
    """Путь к JSON с историей заданий кластера."""
    return clusters_dir() / cluster_name.strip() / JOURNAL_SUBDIR / JOBS_HISTORY_FILENAME


def append_job_from_record_sync(
    *,
    job_id: str,
    kind: str,
    cluster_name: str | None,
    status: str,
    message: str | None,
    created_at_utc: str,
    log_lines: list[str],
    result: dict[str, Any] | None,
) -> None:
    """
    Добавить запись о завершённом задании в журнал каталога кластера.

    Вызывается из потока после ``job_store`` (to_thread); без паролей в result.
    """
    if not cluster_name or not str(cluster_name).strip():
        return
    name = str(cluster_name).strip()
    if not validate_cluster_name(name):
        logger.debug("Журнал кластера: пропуск, некорректное имя «%s»", name)
        return

    cdir = clusters_dir() / name
    if not cdir.is_dir():
        logger.debug("Журнал кластера: каталог %s отсутствует", cdir)
        return

    jdir = cdir / JOURNAL_SUBDIR
    finished = datetime.now(timezone.utc).isoformat()
    cap_lines = _max_log_lines()
    lines = list(log_lines)[-cap_lines:] if log_lines else []

    entry: dict[str, Any] = {
        "version": JOURNAL_FILE_VERSION,
        "job_id": job_id,
        "kind": kind,
        "cluster_name": name,
        "status": status,
        "message": message,
        "created_at_utc": created_at_utc,
        "finished_at_utc": finished,
        "log_lines": lines,
        "result": result,
    }

    path = jdir / JOBS_HISTORY_FILENAME
    lock = _cluster_lock(name)
    with lock:
        jdir.mkdir(parents=True, exist_ok=True)
        entries: list[dict[str, Any]] = []
        if path.is_file():
            try:
                raw = path.read_text(encoding="utf-8")
                data = json.loads(raw)
                if isinstance(data, dict) and isinstance(data.get("entries"), list):
                    entries = [e for e in data["entries"] if isinstance(e, dict)]
            except (OSError, json.JSONDecodeError) as e:
                logger.warning("Журнал %s: не прочитан, создаём заново: %s", path, e)
                entries = []

        entries.insert(0, entry)
        max_e = _max_entries()
        if len(entries) > max_e:
            entries = entries[:max_e]

        payload = {"file_version": JOURNAL_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("Журнал кластера «%s»: запись %s (%s)", name, job_id, kind)
        except OSError as e:
            logger.warning("Журнал %s: запись не удалась: %s", path, e)
            try:
                tmp.unlink(missing_ok=True)
            except OSError:
                pass


def read_cluster_journal_sync(cluster_name: str) -> dict[str, Any] | None:
    """Прочитать файл журнала кластера или ``None``."""
    n = cluster_name.strip()
    if not validate_cluster_name(n):
        return None
    path = journal_file_path(n)
    if not path.is_file():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None


def _collect_all_journal_rows_sorted_sync() -> list[tuple[str, dict[str, Any]]]:
    """Все записи из журналов кластеров, отсортированные по времени (новые первыми)."""
    root = clusters_dir()
    if not root.is_dir():
        return []
    flat: list[tuple[str, dict[str, Any]]] = []
    for sub in sorted(root.iterdir()):
        if not sub.is_dir():
            continue
        name = sub.name
        if not validate_cluster_name(name):
            continue
        path = sub / JOURNAL_SUBDIR / JOBS_HISTORY_FILENAME
        if not path.is_file():
            continue
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        entries = data.get("entries") if isinstance(data, dict) else None
        if not isinstance(entries, list):
            continue
        for e in entries:
            if isinstance(e, dict):
                flat.append((name, e))

    def sort_key(item: tuple[str, dict[str, Any]]) -> str:
        return str(item[1].get("finished_at_utc") or item[1].get("created_at_utc") or "")

    flat.sort(key=sort_key, reverse=True)
    return flat


def collect_recent_journal_entries_page_sync(*, limit: int, offset: int) -> tuple[list[dict[str, Any]], int]:
    """
    Страница записей из всех ``clusters/*/journal/jobs_history.json``, новые первыми.

    В каждую запись добавляется поле ``source_cluster`` (имя каталога).
    Возвращает ``(страница записей, общее число записей)``.
    """
    lim = max(1, min(limit, 100))
    off = max(0, offset)
    flat = _collect_all_journal_rows_sorted_sync()
    total = len(flat)
    chunk = flat[off : off + lim]
    out: list[dict[str, Any]] = []
    for cluster_dir_name, e in chunk:
        row = dict(e)
        row["source_cluster"] = cluster_dir_name
        out.append(row)
    return out, total