"""Хранилище фоновых заданий (создание кластера) в памяти процесса. При перезапуске контейнера история заданий обнуляется — это ожидаемо для dev-среды. Потокобезопасные флаги отмены и прогресс (для worker-thread) — через ``threading.Lock``. Автор: Сергей Антропов Сайт: https://devops.org.ru """ from __future__ import annotations import asyncio import logging import os import threading import uuid from collections import deque from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Literal logger = logging.getLogger("kind_k8s.job_store") # Лимит записей в памяти (dev-инструмент; старые задания вытесняются) _MAX_JOBS = 200 JobStatus = Literal["queued", "running", "success", "failed", "cancelled"] # --- Синхронное сопровождение задания (worker-thread и HTTP отмена) --- _thread_lock = threading.Lock() _cancel_events: dict[str, threading.Event] = {} _progress: dict[str, tuple[str, int]] = {} # Хвост логов для активных заданий (kind create и т.д.); после завершения копируется в JobRecord.log_lines _job_log_deques: dict[str, deque[str]] = {} def _max_job_log_lines() -> int: raw = (os.environ.get("KIND_K8S_JOB_LOG_MAX_LINES") or "500").strip() try: return max(50, min(int(raw), 5000)) except ValueError: return 500 def append_log_sync(job_id: str, line: str) -> None: """Добавить строку в журнал задания (вызывается из worker-thread во время долгих команд).""" text = (line or "").rstrip() if not text: return cap = _max_job_log_lines() with _thread_lock: if job_id not in _job_log_deques: _job_log_deques[job_id] = deque(maxlen=cap) _job_log_deques[job_id].append(text) def get_logs_snapshot_sync(job_id: str) -> list[str]: """Снимок текущего журнала (для API во время running/queued).""" with _thread_lock: d = _job_log_deques.get(job_id) return list(d) if d else [] def take_logs_finalize_sync(job_id: str) -> list[str]: """ Забрать журнал в список и удалить deque (после успеха/ошибки/отмены). Вызывать перед или внутри обновления JobRecord. """ with _thread_lock: d = _job_log_deques.pop(job_id, None) return list(d) if d else [] def begin_job_tracking(job_id: str) -> None: """Зарегистрировать отмену/прогресс для нового job_id (вызывать при создании задания).""" with _thread_lock: _cancel_events[job_id] = threading.Event() _progress[job_id] = ("В очереди", 0) def end_job_tracking(job_id: str) -> None: """Очистить служебные структуры после завершения задания.""" with _thread_lock: _cancel_events.pop(job_id, None) _progress.pop(job_id, None) _job_log_deques.pop(job_id, None) def set_progress_sync(job_id: str, stage: str, percent: int) -> None: """Обновить текст этапа и процент (0–100) из worker-thread.""" pct = max(0, min(100, int(percent))) with _thread_lock: if job_id in _progress: _progress[job_id] = (stage, pct) def get_progress_sync(job_id: str) -> tuple[str, int] | None: """Снимок прогресса для ответа API.""" with _thread_lock: return _progress.get(job_id) def request_cancel_sync(job_id: str) -> bool: """Запросить отмену. Вернуть False, если job_id не отслеживается.""" with _thread_lock: ev = _cancel_events.get(job_id) if ev is None: return False ev.set() logger.info("Запрошена отмена задания %s", job_id) return True def is_cancelled_sync(job_id: str) -> bool: """Проверка из worker-thread между этапами создания кластера.""" with _thread_lock: ev = _cancel_events.get(job_id) return bool(ev and ev.is_set()) @dataclass class JobRecord: """Описание одного задания.""" job_id: str kind: str status: JobStatus cluster_name: str | None created_at_utc: str message: str | None = None result: dict[str, Any] | None = None # Журнал после завершения (stdout/stderr kind create и этапы); пока задание активно — см. deque log_lines: list[str] = field(default_factory=list) class JobStore: """Потокобезопасное (asyncio) хранилище заданий.""" def __init__(self) -> None: self._jobs: dict[str, JobRecord] = {} self._lock = asyncio.Lock() async def create_job(self, kind: str, *, cluster_name: str | None) -> JobRecord: """Зарегистрировать задание в статусе ``queued``.""" jid = uuid.uuid4().hex now = datetime.now(timezone.utc).isoformat() rec = JobRecord( job_id=jid, kind=kind, status="queued", cluster_name=cluster_name, created_at_utc=now, ) async with self._lock: self._jobs[jid] = rec while len(self._jobs) > _MAX_JOBS: oldest_id = min(self._jobs, key=lambda k: self._jobs[k].created_at_utc) end_job_tracking(oldest_id) del self._jobs[oldest_id] logger.debug("Вытеснено старое задание из хранилища: %s", oldest_id) begin_job_tracking(jid) logger.info("Создано задание %s kind=%s cluster=%s", jid, kind, cluster_name) return rec async def set_running(self, job_id: str) -> None: async with self._lock: if job_id in self._jobs: self._jobs[job_id].status = "running" self._jobs[job_id].message = None set_progress_sync(job_id, "Запуск создания кластера…", 5) async def set_success(self, job_id: str, *, result: dict[str, Any] | None = None, message: str | None = None) -> None: logs = take_logs_finalize_sync(job_id) async with self._lock: if job_id in self._jobs: self._jobs[job_id].status = "success" self._jobs[job_id].result = result self._jobs[job_id].message = message self._jobs[job_id].log_lines = logs set_progress_sync(job_id, "Готово", 100) async def set_failed(self, job_id: str, message: str) -> None: logs = take_logs_finalize_sync(job_id) async with self._lock: if job_id in self._jobs: self._jobs[job_id].status = "failed" self._jobs[job_id].message = message self._jobs[job_id].log_lines = logs logger.warning("Задание %s завершилось ошибкой: %s", job_id, message) async def set_cancelled(self, job_id: str, message: str = "Создание отменено пользователем") -> None: logs = take_logs_finalize_sync(job_id) async with self._lock: if job_id in self._jobs: self._jobs[job_id].status = "cancelled" self._jobs[job_id].message = message self._jobs[job_id].log_lines = logs logger.info("Задание %s отменено: %s", job_id, message) async def get(self, job_id: str) -> JobRecord | None: async with self._lock: return self._jobs.get(job_id) def snapshot_all(self) -> list[JobRecord]: """Снимок всех заданий (для отладки; без блокировки — eventual consistency).""" return list(self._jobs.values()) def snapshot_recent_sorted(self, *, limit: int) -> list[JobRecord]: """Задания от новых к старым, не более ``limit``.""" items = self.snapshot_all() items.sort(key=lambda r: r.created_at_utc, reverse=True) return items[: max(1, limit)] # Синглтон на процесс uvicorn job_store = JobStore()