fix: persist update logs and status

This commit is contained in:
hy 2026-04-22 19:16:03 +08:00
parent a7d8b5ebfb
commit 0ed09f8e65
5 changed files with 314 additions and 37 deletions

View File

@ -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)

View File

@ -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/<branch>,避免本地旧分支 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",

View File

@ -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")

View File

@ -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() {

View File

@ -501,16 +501,28 @@
<div class="panel-head update-log-head">
<div>
<h3>Update Log</h3>
<p>按阶段展示构建、推 Harbor、同步 compose、预拉镜像和滚动重建的实时输出</p>
<p>按阶段展示构建、推 Harbor、同步 compose、预拉镜像和滚动重建的实时输出可下载完整日志</p>
</div>
<div class="action-group">
<button class="refresh ghost" @click="fetchUpdateOperationStatus(updateOps.operationId)" :disabled="!updateOps.operationId || updateOps.running">
刷新日志
</button>
<a
v-if="updateOps.operationId"
class="refresh ghost"
:href="updateLogDownloadUrl()"
target="_blank"
rel="noopener"
download
>
下载完整日志
</a>
</div>
<button class="refresh ghost" @click="fetchUpdateOperationStatus(updateOps.operationId)" :disabled="!updateOps.operationId || updateOps.running">
刷新日志
</button>
</div>
<pre id="service-update-log-output" class="update-log-output mono">{{ updateLogText || '等待更新日志...' }}</pre>
</section>
<div v-if="updateOps.result && updateOps.result.build && updateOps.result.build.length > 0" class="table-wrap">
<div v-if="updateOps.result && updateOps.result.builds && updateOps.result.builds.length > 0" class="table-wrap">
<table class="table compact">
<thead>
<tr>
@ -522,7 +534,7 @@
</tr>
</thead>
<tbody>
<tr v-for="build in updateOps.result.build" :key="build.kind + '-' + build.commit">
<tr v-for="build in updateOps.result.builds" :key="build.kind + '-' + build.commit">
<td>{{ build.kind }}</td>
<td class="mono">{{ build.gitRef }}</td>
<td class="mono">{{ build.commit }}</td>