Sergey Antropoff
8bd44adbb0
- Фоновые stop/start с job_id и poll; отмена с kill; docker pull plain + снятие ANSI - Лимиты журнала API/буфера; список jobs без progress_log; DELETE /jobs - UI: опрос чаще, подсказка при пустом логе, кнопка очистки завершённых
284 lines
11 KiB
Python
284 lines
11 KiB
Python
"""Хранилище фоновых заданий (создание кластера) в памяти процесса.
|
||
|
||
При перезапуске контейнера история заданий обнуляется — это ожидаемо для dev-среды.
|
||
|
||
Потокобезопасные флаги отмены и прогресс (для worker-thread) — через ``threading.Lock``.
|
||
|
||
Автор: Сергей Антропов
|
||
Сайт: https://devops.org.ru
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import os
|
||
import signal
|
||
import subprocess
|
||
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]] = {}
|
||
# Активный дочерний процесс (pull / kind create) — для принудительной отмены
|
||
_process_lock = threading.Lock()
|
||
_active_subprocess_by_job: dict[str, subprocess.Popen] = {}
|
||
|
||
|
||
def _max_job_log_lines() -> int:
|
||
# Буфер в памяти: старые строки вытесняются; по умолчанию 2500 — виден почти весь типичный лог.
|
||
raw = (os.environ.get("KIND_K8S_JOB_LOG_MAX_LINES") or "2500").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 _process_lock:
|
||
_active_subprocess_by_job.pop(job_id, 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 register_subprocess_sync(job_id: str, proc: subprocess.Popen) -> None:
|
||
"""Привязать текущий дочерний процесс к заданию (pull, kind create)."""
|
||
with _process_lock:
|
||
_active_subprocess_by_job[job_id] = proc
|
||
|
||
|
||
def unregister_subprocess_sync(job_id: str) -> None:
|
||
"""Снять регистрацию процесса (после завершения потока чтения)."""
|
||
with _process_lock:
|
||
_active_subprocess_by_job.pop(job_id, None)
|
||
|
||
|
||
def kill_subprocess_for_job_sync(job_id: str) -> None:
|
||
"""Принудительно завершить дочерний процесс задания (SIGKILL / группа процессов)."""
|
||
with _process_lock:
|
||
proc = _active_subprocess_by_job.get(job_id)
|
||
if proc is None:
|
||
return
|
||
if proc.poll() is not None:
|
||
return
|
||
pid = proc.pid
|
||
try:
|
||
try:
|
||
os.killpg(os.getpgid(pid), signal.SIGKILL)
|
||
logger.warning("Отмена: SIGKILL группы процесса PID %s (задание %s)", pid, job_id)
|
||
except (AttributeError, OSError, ProcessLookupError):
|
||
proc.kill()
|
||
logger.warning("Отмена: kill PID %s (задание %s)", pid, job_id)
|
||
except OSError as e:
|
||
logger.warning("Не удалось убить процесс задания %s: %s", job_id, e)
|
||
|
||
|
||
def request_cancel_sync(job_id: str) -> bool:
|
||
"""Запросить отмену и прервать активный подпроцесс (если есть)."""
|
||
with _thread_lock:
|
||
ev = _cancel_events.get(job_id)
|
||
if ev is None:
|
||
return False
|
||
ev.set()
|
||
logger.info("Запрошена отмена задания %s", job_id)
|
||
kill_subprocess_for_job_sync(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,
|
||
*,
|
||
stage: str = "Выполняется…",
|
||
percent: int = 5,
|
||
) -> 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, stage, percent)
|
||
|
||
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)]
|
||
|
||
async def purge_completed(self) -> int:
|
||
"""
|
||
Удалить из памяти только завершённые задания (success / failed / cancelled).
|
||
|
||
Активные (queued, running) не трогаем. Служебные структуры отслеживания подчищаются.
|
||
"""
|
||
async with self._lock:
|
||
to_remove = [
|
||
jid
|
||
for jid, rec in self._jobs.items()
|
||
if rec.status in ("success", "failed", "cancelled")
|
||
]
|
||
for jid in to_remove:
|
||
del self._jobs[jid]
|
||
for jid in to_remove:
|
||
end_job_tracking(jid)
|
||
return len(to_remove)
|
||
|
||
|
||
# Синглтон на процесс uvicorn
|
||
job_store = JobStore()
|