from __future__ import annotations # 返回状态给 HTTP 层时直接用 deepcopy 最稳。 import copy # 后台更新任务需要用线程跑，避免 HTTP 请求一直阻塞。 import threading # operationId 用 uuid 生成最直接。 import uuid from typing import Any from .config import utc_now from .service_release import deploy_updated_services_payload # 内存里最多保留最近若干次更新记录，避免 deploy 机常驻进程越跑越大。 MAX_UPDATE_OPERATIONS = 20 # 单次更新最多保留最近这些日志行，超出后丢弃最旧的。 MAX_UPDATE_LOG_LINES = 1200 # 统一保护任务状态的锁。 UPDATE_JOB_LOCK = threading.Lock() # 当前进程内的更新任务表。 UPDATE_JOBS: dict[str, dict[str, Any]] = {} # 记录任务插入顺序，便于淘汰最旧项。 UPDATE_JOB_ORDER: list[str] = [] def normalize_services(service_names: list[str]) -> list[str]: # 请求里允许重复服务名，这里统一去重并保序。 normalized = [str(item or "").strip() for item in service_names if str(item or "").strip()] return list(dict.fromkeys(normalized)) def create_job_payload( service_names: list[str], *, java_git_ref: str, go_git_ref: str, image_tag: str, skip_maven_build: bool, ) -> dict[str, Any]: # 后台任务一创建就落一份初始状态，前端可以马上开始轮询。 normalized = normalize_services(service_names) operation_id = uuid.uuid4().hex[:12] now = utc_now() payload = { "operationId": operation_id, "kind": "service-update", "status": "queued", "stage": "queued", "message": "service update queued", "error": "", "startedAt": now, "updatedAt": now, "finishedAt": "", "logDroppedCount": 0, "logs": [], "result": None, "request": { "services": normalized, "javaGitRef": str(java_git_ref or "").strip(), "goGitRef": str(go_git_ref or "").strip(), "imageTag": str(image_tag or "").strip(), "skipMavenBuild": bool(skip_maven_build), }, } with UPDATE_JOB_LOCK: UPDATE_JOBS[operation_id] = payload UPDATE_JOB_ORDER.append(operation_id) trim_jobs_unlocked() return snapshot_job(operation_id) def trim_jobs_unlocked() -> None: # 只保留最近的若干次任务，老数据直接淘汰。 while len(UPDATE_JOB_ORDER) > MAX_UPDATE_OPERATIONS: operation_id = UPDATE_JOB_ORDER.pop(0) UPDATE_JOBS.pop(operation_id, None) def append_job_log(operation_id: str, stage: str, message: str, *, level: str = "info") -> None: # 每条日志都带时间和阶段，页面可以直接展示。 text = str(message or "").rstrip() if not text: return with UPDATE_JOB_LOCK: payload = UPDATE_JOBS.get(operation_id) if payload is None: return payload["updatedAt"] = utc_now() payload["stage"] = str(stage or payload.get("stage") or "running") payload["logs"].append( { "time": payload["updatedAt"], "stage": str(stage or "").strip(), "level": str(level or "info").strip() or "info", "message": text, } ) if len(payload["logs"]) > MAX_UPDATE_LOG_LINES: overflow = len(payload["logs"]) - MAX_UPDATE_LOG_LINES payload["logs"] = payload["logs"][overflow:] payload["logDroppedCount"] = int(payload.get("logDroppedCount") or 0) + overflow def patch_job(operation_id: str, **fields: Any) -> None: # 任务元信息更新统一走这里，避免漏掉 updatedAt。 with UPDATE_JOB_LOCK: payload = UPDATE_JOBS.get(operation_id) if payload is None: return payload.update(fields) payload["updatedAt"] = utc_now() def finish_job(operation_id: str, *, result: dict[str, Any], status: str, error: str = "") -> None: # 结束态统一收口，前端轮询时不用自己猜。 now = utc_now() with UPDATE_JOB_LOCK: payload = UPDATE_JOBS.get(operation_id) if payload is None: return payload["status"] = status payload["stage"] = status payload["message"] = ( str(result.get("message") or "").strip() or ("service update complete" if status == "success" else "service update failed") ) payload["error"] = str(error or "").strip() payload["finishedAt"] = now payload["updatedAt"] = now payload["result"] = result def snapshot_job(operation_id: str) -> dict[str, Any]: # HTTP 返回前复制一份，避免把锁外的可变对象直接暴露出去。 with UPDATE_JOB_LOCK: payload = UPDATE_JOBS.get(operation_id) if payload is None: raise KeyError(operation_id) return { "ok": True, "operation": { "operationId": payload["operationId"], "kind": payload["kind"], "status": payload["status"], "stage": payload["stage"], "message": payload["message"], "error": payload["error"], "startedAt": payload["startedAt"], "updatedAt": payload["updatedAt"], "finishedAt": payload["finishedAt"], "logDroppedCount": payload["logDroppedCount"], "logs": copy.deepcopy(payload.get("logs") or []), "result": copy.deepcopy(payload.get("result")), "request": copy.deepcopy(payload.get("request") or {}), }, } def job_progress_logger(operation_id: str): # service_release 侧只关心 stage / level / message，落盘细节都交给这里。 def _logger(stage: str, level: str, message: str) -> None: append_job_log(operation_id, stage, message, level=level) return _logger def run_update_job( operation_id: str, service_names: list[str], *, java_git_ref: str, go_git_ref: str, image_tag: str, skip_maven_build: bool, ) -> None: # 真正的更新逻辑放后台线程里执行。 patch_job(operation_id, status="running", stage="prepare", message="service update running") append_job_log(operation_id, "prepare", f"开始更新服务：{', '.join(normalize_services(service_names))}") try: result = deploy_updated_services_payload( service_names, java_git_ref=java_git_ref, go_git_ref=go_git_ref, image_tag=image_tag, skip_maven_build=skip_maven_build, progress=job_progress_logger(operation_id), ) if result.get("ok"): append_job_log(operation_id, "success", "服务更新完成") finish_job(operation_id, result=result, status="success") return error = str(result.get("error") or "service update failed").strip() append_job_log(operation_id, "failed", error, level="error") finish_job(operation_id, result=result, status="failed", error=error) except Exception as exc: # noqa: BLE001 error = str(exc) append_job_log(operation_id, "failed", error, level="error") finish_job( operation_id, result={"ok": False, "updatedAt": utc_now(), "error": error}, status="failed", error=error, ) def start_update_job( service_names: list[str], *, java_git_ref: str, go_git_ref: str, image_tag: str, skip_maven_build: bool, ) -> dict[str, Any]: # 创建任务后立刻拉起后台线程，并把 operationId 返回给前端。 payload = create_job_payload( service_names, java_git_ref=java_git_ref, go_git_ref=go_git_ref, image_tag=image_tag, skip_maven_build=skip_maven_build, ) operation_id = payload["operation"]["operationId"] worker = threading.Thread( target=run_update_job, kwargs={ "operation_id": operation_id, "service_names": list(service_names), "java_git_ref": java_git_ref, "go_git_ref": go_git_ref, "image_tag": image_tag, "skip_maven_build": skip_maven_build, }, name=f"service-update-{operation_id}", daemon=True, ) worker.start() return snapshot_job(operation_id) def get_update_job_payload(operation_id: str) -> dict[str, Any]: # 供 HTTP 轮询接口直接读取任务状态。 return snapshot_job(operation_id)