diff --git a/monitor/http_app.py b/monitor/http_app.py index b924b35..6a7aaa8 100644 --- a/monitor/http_app.py +++ b/monitor/http_app.py @@ -16,7 +16,7 @@ from .nacos import ( ) from .runtime_config import get_go_config_payload, save_go_config_payload, validate_go_config_payload from .service_release import build_update_targets_payload -from .service_update_jobs import get_update_job_payload, start_update_job +from .service_update_jobs import get_update_job_log_text, get_update_job_payload, start_update_job from .service_ops import build_restart_targets_payload, rolling_restart_payload @@ -189,6 +189,30 @@ class MonitorHandler(BaseHTTPRequestHandler): send_body=send_body, ) + # 返回单个服务更新任务的完整文本日志。 + if parsed.path == "/api/ops/update-log": + operation_id = self.query_value(query, "id") + if not operation_id: + return self.send_error_payload( + HTTPStatus.BAD_REQUEST, + "id is required", + send_body=send_body, + ) + + try: + content = get_update_job_log_text(operation_id) + except KeyError: + return self.send_error_payload(HTTPStatus.NOT_FOUND, "update log not found", send_body=send_body) + except Exception as exc: # noqa: BLE001 + return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=send_body) + + return self.send_payload( + HTTPStatus.OK, + content.encode("utf-8"), + "text/plain; charset=utf-8", + send_body=send_body, + ) + # favicon 直接返回空。 if parsed.path == "/favicon.ico": return self.send_payload(HTTPStatus.NO_CONTENT, b"", "image/x-icon", send_body=send_body) diff --git a/monitor/service_release.py b/monitor/service_release.py index 6f277ad..db0bcfc 100644 --- a/monitor/service_release.py +++ b/monitor/service_release.py @@ -27,6 +27,7 @@ from .config import ( HARBOR_REGISTRY, HARBOR_USERNAME, JAVA_REPO_ROOT, + ROOT, ROLLING_RESTART_STABILIZE_SECONDS, ROLLING_RESTART_TIMEOUT_SECONDS, SERVICE_BUILD_TIMEOUT_SECONDS, @@ -75,6 +76,8 @@ SERVICE_IMAGE_REPOSITORY_SUFFIXES = { # 后台任务的日志回调统一走这个签名。 ProgressLogger = Callable[[str, str, str], None] +# 私有 Maven 安装缓存统一放到 run 目录,避免混进 Git 工作区。 +PRIVATE_MAVEN_CACHE_DIR = ROOT / "run" / "maven_private_cache" def emit_progress(progress: ProgressLogger | None, stage: str, message: str, *, level: str = "info") -> None: @@ -272,6 +275,77 @@ def fetch_repo_refs(repo_root: Path, label: str) -> None: ) +def local_maven_repo_root() -> Path: + # 兼容外部显式指定 MAVEN_REPO_LOCAL,默认仍沿用 ~/.m2/repository。 + override = str(os.environ.get("MAVEN_REPO_LOCAL") or "").strip() + if override: + return Path(override).expanduser() + return Path.home() / ".m2" / "repository" + + +def private_maven_tree_id(repo_root: Path, commit: str) -> str: + # 用 git tree id 表示 maven_private 内容,只有私有依赖目录变化时才会变化。 + try: + return local_command_output( + ["git", "-C", str(repo_root), "rev-parse", "--verify", f"{commit}:maven_private"], + cwd=repo_root, + label="resolve private maven tree", + ) + except Exception: # noqa: BLE001 + return "" + + +def private_maven_cache_stamp_path(repo_root: Path) -> Path: + # 每个仓保留一份最近已安装的私有依赖 tree id。 + return PRIVATE_MAVEN_CACHE_DIR / f"{repo_root.name}.tree" + + +def private_maven_artifacts_present() -> bool: + # 本地 Maven 仓至少得有一份私有制品,缓存命中才有意义。 + artifacts_root = local_maven_repo_root() / "com" / "red" / "circle" + if not artifacts_root.exists(): + return False + for path in artifacts_root.rglob("*.pom"): + if path.is_file(): + return True + return False + + +def ensure_private_maven_installed( + repo_root: Path, + worktree_root: Path, + commit: str, + *, + progress: ProgressLogger | None = None, +) -> None: + # 私有依赖没变化时直接复用本机 ~/.m2 缓存,避免每次更新都重复 install-file。 + tree_id = private_maven_tree_id(repo_root, commit) + stamp_path = private_maven_cache_stamp_path(repo_root) + cached_tree_id = stamp_path.read_text(encoding="utf-8").strip() if stamp_path.exists() else "" + + if tree_id and cached_tree_id == tree_id and private_maven_artifacts_present(): + emit_progress(progress, "build.java.maven", f"私有 Maven 依赖未变化,跳过安装:{tree_id[:12]}") + return + + emit_progress(progress, "build.java.maven", "开始安装私有 Maven 依赖") + run_logged_command( + [str(worktree_root / "ci" / "install-private-maven.sh")], + cwd=worktree_root, + label="java install private maven", + timeout_seconds=min(SERVICE_BUILD_TIMEOUT_SECONDS, 1200), + progress_stage="build.java.maven.install", + progress=progress, + ) + + if tree_id: + PRIVATE_MAVEN_CACHE_DIR.mkdir(parents=True, exist_ok=True) + stamp_path.write_text(f"{tree_id}\n", encoding="utf-8") + emit_progress(progress, "build.java.maven", f"私有 Maven 依赖安装完成:{tree_id[:12]}") + return + + emit_progress(progress, "build.java.maven", "私有 Maven 依赖安装完成") + + def resolve_repo_commit(repo_root: Path, git_ref: str, label: str) -> str: # 优先把分支解析到 origin/,避免本地旧分支 HEAD 干扰构建。 candidates = [ @@ -440,13 +514,10 @@ def build_java_images( try: if not skip_maven_build: - emit_progress(progress, "build.java.maven", "开始安装私有 Maven 依赖") - run_logged_command( - [str(worktree_root / "ci" / "install-private-maven.sh")], - cwd=worktree_root, - label="java install private maven", - timeout_seconds=min(SERVICE_BUILD_TIMEOUT_SECONDS, 1200), - progress_stage="build.java.maven.install", + ensure_private_maven_installed( + repo_root, + worktree_root, + commit, progress=progress, ) emit_progress(progress, "build.java.maven", f"开始 Maven 打包:{maven_module_list(service_names)}") @@ -479,7 +550,8 @@ def build_java_images( label=f"java build {service_name}", env={ "DOCKER_PLATFORM": BUILD_DOCKER_PLATFORM, - "SKIP_MAVEN_BUILD": "1" if skip_maven_build else "0", + # 外层已经统一决定过是否 Maven 打包;镜像阶段一律复用现成产物。 + "SKIP_MAVEN_BUILD": "1", }, timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS, progress_stage=f"build.java.{service_name}.build", diff --git a/monitor/service_update_jobs.py b/monitor/service_update_jobs.py index f863efa..4c5cfff 100644 --- a/monitor/service_update_jobs.py +++ b/monitor/service_update_jobs.py @@ -2,13 +2,18 @@ from __future__ import annotations # 返回状态给 HTTP 层时直接用 deepcopy 最稳。 import copy +# 状态持久化到 JSON 文件需要标准库 json。 +import json +# 日志和状态文件落到 run 目录。 +from pathlib import Path # 后台更新任务需要用线程跑,避免 HTTP 请求一直阻塞。 import threading # operationId 用 uuid 生成最直接。 +import re import uuid from typing import Any -from .config import utc_now +from .config import ROOT, utc_now from .service_release import deploy_updated_services_payload @@ -16,6 +21,8 @@ from .service_release import deploy_updated_services_payload MAX_UPDATE_OPERATIONS = 20 # 单次更新最多保留最近这些日志行,超出后丢弃最旧的。 MAX_UPDATE_LOG_LINES = 1200 +# 任务状态和完整日志统一落到 run 目录。 +UPDATE_JOB_DIR = ROOT / "run" / "service_update_jobs" # 统一保护任务状态的锁。 UPDATE_JOB_LOCK = threading.Lock() # 当前进程内的更新任务表。 @@ -30,6 +37,126 @@ def normalize_services(service_names: list[str]) -> list[str]: return list(dict.fromkeys(normalized)) +def update_job_state_path(operation_id: str) -> Path: + # 每个任务一份状态 JSON。 + return UPDATE_JOB_DIR / f"{operation_id}.json" + + +def update_job_log_path(operation_id: str) -> Path: + # 每个任务一份完整文本日志。 + return UPDATE_JOB_DIR / f"{operation_id}.log" + + +def ensure_update_job_dir() -> None: + # 持久化目录不存在时先创建。 + UPDATE_JOB_DIR.mkdir(parents=True, exist_ok=True) + + +def valid_operation_id(operation_id: str) -> bool: + # 只接受简单安全字符,避免路径拼接被滥用。 + return re.fullmatch(r"[0-9A-Za-z_-]{6,64}", str(operation_id or "").strip()) is not None + + +def append_log_file(operation_id: str, line: str) -> None: + # 完整日志单独写入文本文件,供页面下载。 + ensure_update_job_dir() + with update_job_log_path(operation_id).open("a", encoding="utf-8") as handle: + handle.write(line) + + +def operation_payload(payload: dict[str, Any]) -> dict[str, Any]: + # 对外返回和落盘都统一走这个结构,避免前后不一致。 + return { + "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 {}), + "logUrl": f"/api/ops/update-log?id={payload['operationId']}", + } + + +def persist_job_unlocked(payload: dict[str, Any]) -> None: + # 每次状态变化都落盘,服务重启后还能恢复最近任务。 + ensure_update_job_dir() + state_path = update_job_state_path(payload["operationId"]) + state_path.write_text( + json.dumps({"ok": True, "operation": operation_payload(payload)}, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + + +def recover_lost_running_job_unlocked(payload: dict[str, Any]) -> None: + # 进程重启后,原后台线程已经不存在;把悬挂中的 running/queued 任务收口成失败态最直接。 + if payload.get("status") not in {"queued", "running"}: + return + + now = utc_now() + message = "monitor process restarted before update status could be finalized" + payload["status"] = "failed" + payload["stage"] = "failed" + payload["message"] = "service update interrupted" + payload["error"] = message + payload["updatedAt"] = now + payload["finishedAt"] = now + payload["logs"].append( + { + "time": now, + "stage": "failed", + "level": "error", + "message": message, + } + ) + 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 + + append_log_file(payload["operationId"], f"[{now}] [failed] [error] {message}\n") + persist_job_unlocked(payload) + + +def load_job_from_disk(operation_id: str) -> dict[str, Any] | None: + # 任务不在内存时,从持久化状态恢复。 + if not valid_operation_id(operation_id): + return None + + state_path = update_job_state_path(operation_id) + if not state_path.exists(): + return None + + payload = json.loads(state_path.read_text(encoding="utf-8")) + operation = dict(payload.get("operation") or {}) + if not operation: + return None + + restored = { + "operationId": str(operation.get("operationId") or operation_id).strip(), + "kind": str(operation.get("kind") or "service-update").strip() or "service-update", + "status": str(operation.get("status") or "").strip(), + "stage": str(operation.get("stage") or "").strip(), + "message": str(operation.get("message") or "").strip(), + "error": str(operation.get("error") or "").strip(), + "startedAt": str(operation.get("startedAt") or "").strip(), + "updatedAt": str(operation.get("updatedAt") or "").strip(), + "finishedAt": str(operation.get("finishedAt") or "").strip(), + "logDroppedCount": int(operation.get("logDroppedCount") or 0), + "logs": list(operation.get("logs") or []), + "result": operation.get("result"), + "request": dict(operation.get("request") or {}), + } + recover_lost_running_job_unlocked(restored) + return restored + + def create_job_payload( service_names: list[str], *, @@ -64,8 +191,11 @@ def create_job_payload( }, } with UPDATE_JOB_LOCK: + ensure_update_job_dir() UPDATE_JOBS[operation_id] = payload UPDATE_JOB_ORDER.append(operation_id) + update_job_log_path(operation_id).write_text("", encoding="utf-8") + persist_job_unlocked(payload) trim_jobs_unlocked() return snapshot_job(operation_id) @@ -102,6 +232,11 @@ def append_job_log(operation_id: str, stage: str, message: str, *, level: str = overflow = len(payload["logs"]) - MAX_UPDATE_LOG_LINES payload["logs"] = payload["logs"][overflow:] payload["logDroppedCount"] = int(payload.get("logDroppedCount") or 0) + overflow + append_log_file( + operation_id, + f"[{payload['updatedAt']}] [{str(stage or '').strip() or payload['stage']}] [{str(level or 'info').strip() or 'info'}] {text}\n", + ) + persist_job_unlocked(payload) def patch_job(operation_id: str, **fields: Any) -> None: @@ -112,6 +247,7 @@ def patch_job(operation_id: str, **fields: Any) -> None: return payload.update(fields) payload["updatedAt"] = utc_now() + persist_job_unlocked(payload) def finish_job(operation_id: str, *, result: dict[str, Any], status: str, error: str = "") -> None: @@ -131,6 +267,7 @@ def finish_job(operation_id: str, *, result: dict[str, Any], status: str, error: payload["finishedAt"] = now payload["updatedAt"] = now payload["result"] = result + persist_job_unlocked(payload) def snapshot_job(operation_id: str) -> dict[str, Any]: @@ -140,24 +277,7 @@ def snapshot_job(operation_id: str) -> dict[str, Any]: 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 {}), - }, - } + return {"ok": True, "operation": operation_payload(payload)} def job_progress_logger(operation_id: str): @@ -246,4 +366,27 @@ def start_update_job( def get_update_job_payload(operation_id: str) -> dict[str, Any]: # 供 HTTP 轮询接口直接读取任务状态。 - return snapshot_job(operation_id) + try: + return snapshot_job(operation_id) + except KeyError: + restored = load_job_from_disk(operation_id) + if restored is None: + raise + + with UPDATE_JOB_LOCK: + UPDATE_JOBS[operation_id] = restored + if operation_id not in UPDATE_JOB_ORDER: + UPDATE_JOB_ORDER.append(operation_id) + trim_jobs_unlocked() + return snapshot_job(operation_id) + + +def get_update_job_log_text(operation_id: str) -> str: + # 页面下载完整日志时直接返回文本文件内容。 + if not valid_operation_id(operation_id): + raise KeyError(operation_id) + + log_path = update_job_log_path(operation_id) + if not log_path.exists(): + raise KeyError(operation_id) + return log_path.read_text(encoding="utf-8", errors="replace") diff --git a/static/app.js b/static/app.js index 93bcf4c..1d8cfeb 100644 --- a/static/app.js +++ b/static/app.js @@ -333,6 +333,13 @@ createApp({ updateTargetLabel(item) { return `${item.service} · ${item.kind} · ${item.hosts.map((host) => host.host).join(" / ")}`; }, + updateLogDownloadUrl() { + if (!this.updateOps.operationId) { + return ""; + } + const query = new URLSearchParams({ id: this.updateOps.operationId }); + return `/api/ops/update-log?${query.toString()}`; + }, updateStatusBadgeClass() { if (this.updateOps.error || this.updateOps.status === "failed") { return "bad"; @@ -468,11 +475,30 @@ createApp({ } } } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + const notFound = errorMessage.includes("update operation not found") || errorMessage.includes("HTTP 404"); + + if (silent && !notFound) { + return; + } + this.stopUpdatePolling(); this.updateOps.running = false; - if (!silent) { - this.updateOps.error = error instanceof Error ? error.message : String(error); + + if (notFound) { + this.updateOps.status = "failed"; + this.updateOps.error = "更新任务状态不存在,可能因为服务重启或旧任务已清理"; + this.updateOps.message = ""; + this.updateOps.operationId = ""; + this.persistUpdateOperationId(); + return; } + + if (this.updateOps.status === "queued" || this.updateOps.status === "running") { + this.updateOps.status = "failed"; + this.updateOps.message = "更新状态查询失败"; + } + this.updateOps.error = errorMessage; } }, applyRefreshTimer() { diff --git a/static/index.html b/static/index.html index 36e5465..200ea09 100644 --- a/static/index.html +++ b/static/index.html @@ -501,16 +501,28 @@

Update Log

-

按阶段展示构建、推 Harbor、同步 compose、预拉镜像和滚动重建的实时输出

+

按阶段展示构建、推 Harbor、同步 compose、预拉镜像和滚动重建的实时输出,可下载完整日志

+
+
+ + + 下载完整日志 +
-
{{ updateLogText || '等待更新日志...' }}
-
+
@@ -522,7 +534,7 @@ - +
{{ build.kind }} {{ build.gitRef }} {{ build.commit }}