diff --git a/monitor/http_app.py b/monitor/http_app.py
index 2472e4c..b924b35 100644
--- a/monitor/http_app.py
+++ b/monitor/http_app.py
@@ -15,7 +15,8 @@ from .nacos import (
validate_nacos_config_payload,
)
from .runtime_config import get_go_config_payload, save_go_config_payload, validate_go_config_payload
-from .service_release import build_update_targets_payload, deploy_updated_services_payload
+from .service_release import build_update_targets_payload
+from .service_update_jobs import get_update_job_payload, start_update_job
from .service_ops import build_restart_targets_payload, rolling_restart_payload
@@ -164,6 +165,30 @@ class MonitorHandler(BaseHTTPRequestHandler):
send_body=send_body,
)
+ # 返回当前服务更新任务的阶段和日志。
+ if parsed.path == "/api/ops/update-status":
+ 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:
+ payload = get_update_job_payload(operation_id)
+ except KeyError:
+ return self.send_error_payload(HTTPStatus.NOT_FOUND, "update operation 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,
+ response_json(payload),
+ "application/json; 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)
@@ -313,7 +338,7 @@ class MonitorHandler(BaseHTTPRequestHandler):
skip_maven_build = bool(payload.get("skipMavenBuild"))
try:
- result = deploy_updated_services_payload(
+ result = start_update_job(
services,
java_git_ref=java_git_ref,
go_git_ref=go_git_ref,
@@ -324,7 +349,7 @@ class MonitorHandler(BaseHTTPRequestHandler):
return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True)
return self.send_payload(
- HTTPStatus.OK,
+ HTTPStatus.ACCEPTED,
response_json(result),
"application/json; charset=utf-8",
send_body=True,
diff --git a/monitor/nacos.py b/monitor/nacos.py
index 5034f7a..a4d3d91 100644
--- a/monitor/nacos.py
+++ b/monitor/nacos.py
@@ -665,14 +665,31 @@ def build_nacos_payload() -> dict[str, Any]:
# 统计全健康服务数量。
healthy_service_total = 0
- # 逐个 namespace 读取服务和实例。
- for namespace in normalized_namespaces:
+ # 只对当前目标 namespace 构建服务和实例视图，避免把 public 下的空查询混进来。
+ target_namespaces = [
+ namespace
+ for namespace in normalized_namespaces
+ if namespace["namespaceId"] == namespace_id or namespace["name"] == namespace_name
+ ]
+ if not target_namespaces:
+ target_namespaces = [
+ {
+ "namespaceId": namespace_id,
+ "name": namespace_name,
+ "configCount": 0,
+ "type": 0,
+ "visible": False,
+ }
+ ]
+
+ # 逐个目标 namespace 读取服务和实例。
+ for namespace in target_namespaces:
# 当前 namespace 的自动发现结果。
raw_services = list_services(namespace["namespaceId"])
# 这里保存当前 namespace 下所有待查询服务。
queries_by_key: dict[tuple[str, str], dict[str, str]] = {}
- # 只在可见 namespace，或者根本没有可见 namespace 时，才补 fallback 查询。
- allow_fallback_queries = namespace["visible"] or not any(item["visible"] for item in normalized_namespaces)
+ # 当前目标 namespace 即使 service/list 为空，也要把已知注册服务补查一遍。
+ allow_fallback_queries = True
# 先收 service/list 返回的服务。
for raw_name in raw_services:
diff --git a/monitor/service_release.py b/monitor/service_release.py
index 94226ff..6f277ad 100644
--- a/monitor/service_release.py
+++ b/monitor/service_release.py
@@ -10,8 +10,9 @@ import shlex
import shutil
import subprocess
import tempfile
+import threading
import time
-from typing import Any
+from typing import Any, Callable
from .compose_templates import (
load_local_host_compose_model,
@@ -72,6 +73,60 @@ SERVICE_IMAGE_REPOSITORY_SUFFIXES = {
"golang": "golang",
}
+# 后台任务的日志回调统一走这个签名。
+ProgressLogger = Callable[[str, str, str], None]
+
+
+def emit_progress(progress: ProgressLogger | None, stage: str, message: str, *, level: str = "info") -> None:
+ # 所有过程日志都经这里发出，调用方传空时自动静默。
+ if progress is None:
+ return
+ progress(str(stage or "").strip(), str(level or "info").strip() or "info", str(message or "").rstrip())
+
+
+def short_text(text: str, limit: int = 600) -> str:
+ # 远端输出和异常信息只保留一小段，避免页面日志爆掉。
+ normalized = " ".join(str(text or "").split())
+ if len(normalized) <= limit:
+ return normalized
+ return normalized[:limit].rstrip() + " ..."
+
+
+def stream_log_file_lines(
+ log_path: Path,
+ stop_event: threading.Event,
+ *,
+ stage: str,
+ progress: ProgressLogger | None,
+) -> None:
+ # 构建日志落本地文件后，这里持续追尾并把新增行推给前端。
+ if progress is None:
+ return
+
+ position = 0
+ remainder = ""
+
+ while not stop_event.is_set():
+ if log_path.exists():
+ with log_path.open("r", encoding="utf-8", errors="replace") as handle:
+ handle.seek(position)
+ chunk = handle.read()
+ position = handle.tell()
+ if chunk:
+ remainder += chunk
+ lines = remainder.splitlines(keepends=True)
+ remainder = ""
+ if lines and not lines[-1].endswith(("\n", "\r")):
+ remainder = lines.pop()
+ for line in lines:
+ text = line.rstrip()
+ if text:
+ emit_progress(progress, stage, text)
+ stop_event.wait(0.4)
+
+ if remainder.strip():
+ emit_progress(progress, stage, remainder.rstrip())
+
def buildable_service_names() -> list[str]:
# 页面和接口统一按这里判定哪些服务允许“构建并更新”。
@@ -152,6 +207,8 @@ def run_logged_command(
label: str,
env: dict[str, str] | None = None,
timeout_seconds: int = SERVICE_BUILD_TIMEOUT_SECONDS,
+ progress_stage: str = "",
+ progress: ProgressLogger | None = None,
) -> str:
# Maven / docker build 输出很大，单独落临时日志再取尾部更稳。
merged_env = os.environ.copy()
@@ -163,6 +220,9 @@ def run_logged_command(
log_file.close()
try:
+ stop_event = threading.Event()
+ follower: threading.Thread | None = None
+
with log_path.open("w", encoding="utf-8") as handle:
process = subprocess.Popen(
command,
@@ -172,12 +232,27 @@ def run_logged_command(
text=True,
env=merged_env,
)
+ if progress is not None:
+ follower = threading.Thread(
+ target=stream_log_file_lines,
+ args=(log_path, stop_event),
+ kwargs={"stage": progress_stage or label, "progress": progress},
+ daemon=True,
+ )
+ follower.start()
try:
return_code = process.wait(timeout=timeout_seconds)
except subprocess.TimeoutExpired as exc:
process.kill()
process.wait()
+ stop_event.set()
+ if follower is not None:
+ follower.join(timeout=1)
raise RuntimeError(f"{label} timed out: {tail_text(log_path)}") from exc
+ finally:
+ stop_event.set()
+ if follower is not None:
+ follower.join(timeout=1)
output_tail = tail_text(log_path)
if return_code != 0:
@@ -253,9 +328,10 @@ def require_harbor_credentials() -> None:
raise RuntimeError("missing HARBOR_USERNAME / HARBOR_PASSWORD in .env")
-def docker_login_local(repo_root: Path) -> None:
+def docker_login_local(repo_root: Path, *, progress: ProgressLogger | None = None) -> None:
# 本机构建前先登录 Harbor，避免 push 时才失败。
require_harbor_credentials()
+ emit_progress(progress, "prepare", f"登录 Harbor：{HARBOR_REGISTRY}")
local_command_output(
["docker", "login", HARBOR_REGISTRY, "-u", HARBOR_USERNAME, "--password-stdin"],
cwd=repo_root,
@@ -263,6 +339,7 @@ def docker_login_local(repo_root: Path) -> None:
input_text=HARBOR_PASSWORD,
timeout_seconds=120,
)
+ emit_progress(progress, "prepare", "Harbor 登录成功")
def image_repository_ref(image_ref: str) -> str:
@@ -338,13 +415,23 @@ def maven_module_list(service_names: list[str]) -> str:
return ",".join(modules)
-def build_java_images(service_names: list[str], git_ref: str, image_tag: str, *, skip_maven_build: bool) -> dict[str, Any]:
+def build_java_images(
+ service_names: list[str],
+ git_ref: str,
+ image_tag: str,
+ *,
+ skip_maven_build: bool,
+ progress: ProgressLogger | None = None,
+) -> dict[str, Any]:
# Java 服务共享一个仓库和 Maven 本地缓存，这里按仓统一构建。
repo_root = ensure_repo_root(JAVA_REPO_ROOT, "java")
resolved_ref = str(git_ref or "").strip() or UPDATE_DEFAULT_JAVA_GIT_REF
+ emit_progress(progress, "build.java", f"刷新 Java 仓库 refs：{resolved_ref}")
fetch_repo_refs(repo_root, "java")
commit = resolve_repo_commit(repo_root, resolved_ref, "java")
+ emit_progress(progress, "build.java", f"Java ref 已解析：{resolved_ref} -> {commit[:12]}")
worktree_root = create_temp_worktree(repo_root, commit, "java")
+ emit_progress(progress, "build.java", f"已创建 Java 临时 worktree：{worktree_root}")
short_commit = commit[:12]
resolved_image_tag = derive_image_tag(resolved_ref, commit, image_tag)
@@ -353,12 +440,16 @@ def build_java_images(service_names: list[str], git_ref: str, image_tag: str, *,
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",
+ progress=progress,
)
+ emit_progress(progress, "build.java.maven", f"开始 Maven 打包：{maven_module_list(service_names)}")
run_logged_command(
[
"mvn",
@@ -372,28 +463,39 @@ def build_java_images(service_names: list[str], git_ref: str, image_tag: str, *,
cwd=worktree_root,
label="java maven package",
timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
+ progress_stage="build.java.maven.package",
+ progress=progress,
)
+ else:
+ emit_progress(progress, "build.java.maven", "跳过 Maven，直接复用已有 Jar")
for service_name in service_names:
repository_ref = service_repository_ref(service_name)
target_image_ref = f"{repository_ref}:{resolved_image_tag}"
+ emit_progress(progress, f"build.java.{service_name}", f"开始构建镜像：{target_image_ref}")
build_tail = run_logged_command(
[str(worktree_root / "ci" / "build-service-image.sh"), service_name, target_image_ref, short_commit],
cwd=worktree_root,
label=f"java build {service_name}",
env={
"DOCKER_PLATFORM": BUILD_DOCKER_PLATFORM,
- "SKIP_MAVEN_BUILD": "1" if not skip_maven_build else "1",
+ "SKIP_MAVEN_BUILD": "1" if skip_maven_build else "0",
},
timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
+ progress_stage=f"build.java.{service_name}.build",
+ progress=progress,
)
+ emit_progress(progress, f"build.java.{service_name}", f"开始推送 Harbor：{target_image_ref}")
push_tail = run_logged_command(
["docker", "push", target_image_ref],
cwd=worktree_root,
label=f"java push {service_name}",
timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
+ progress_stage=f"build.java.{service_name}.push",
+ progress=progress,
)
digest_ref = docker_digest(target_image_ref, cwd=worktree_root)
+ emit_progress(progress, f"build.java.{service_name}", f"镜像已推送完成：{digest_ref}")
service_images[service_name] = digest_ref
outputs.append(
{
@@ -407,6 +509,7 @@ def build_java_images(service_names: list[str], git_ref: str, image_tag: str, *,
)
finally:
remove_temp_worktree(repo_root, worktree_root)
+ emit_progress(progress, "build.java", f"已清理 Java worktree：{worktree_root}")
return {
"kind": "java",
@@ -420,13 +523,22 @@ def build_java_images(service_names: list[str], git_ref: str, image_tag: str, *,
}
-def build_golang_images(service_names: list[str], git_ref: str, image_tag: str) -> dict[str, Any]:
+def build_golang_images(
+ service_names: list[str],
+ git_ref: str,
+ image_tag: str,
+ *,
+ progress: ProgressLogger | None = None,
+) -> dict[str, Any]:
# Go 服务当前是单仓构建，逻辑更简单。
repo_root = ensure_repo_root(GOLANG_REPO_ROOT, "golang")
resolved_ref = str(git_ref or "").strip() or UPDATE_DEFAULT_GO_GIT_REF
+ emit_progress(progress, "build.golang", f"刷新 Go 仓库 refs：{resolved_ref}")
fetch_repo_refs(repo_root, "golang")
commit = resolve_repo_commit(repo_root, resolved_ref, "golang")
+ emit_progress(progress, "build.golang", f"Go ref 已解析：{resolved_ref} -> {commit[:12]}")
worktree_root = create_temp_worktree(repo_root, commit, "golang")
+ emit_progress(progress, "build.golang", f"已创建 Go 临时 worktree：{worktree_root}")
short_commit = commit[:12]
resolved_image_tag = derive_image_tag(resolved_ref, commit, image_tag)
@@ -437,20 +549,27 @@ def build_golang_images(service_names: list[str], git_ref: str, image_tag: str)
for service_name in service_names:
repository_ref = service_repository_ref(service_name)
target_image_ref = f"{repository_ref}:{resolved_image_tag}"
+ emit_progress(progress, f"build.golang.{service_name}", f"开始构建镜像：{target_image_ref}")
build_tail = run_logged_command(
[str(worktree_root / "ci" / "build-image.sh"), target_image_ref, short_commit],
cwd=worktree_root,
label=f"golang build {service_name}",
env={"DOCKER_PLATFORM": BUILD_DOCKER_PLATFORM},
timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
+ progress_stage=f"build.golang.{service_name}.build",
+ progress=progress,
)
+ emit_progress(progress, f"build.golang.{service_name}", f"开始推送 Harbor：{target_image_ref}")
push_tail = run_logged_command(
["docker", "push", target_image_ref],
cwd=worktree_root,
label=f"golang push {service_name}",
timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
+ progress_stage=f"build.golang.{service_name}.push",
+ progress=progress,
)
digest_ref = docker_digest(target_image_ref, cwd=worktree_root)
+ emit_progress(progress, f"build.golang.{service_name}", f"镜像已推送完成：{digest_ref}")
service_images[service_name] = digest_ref
outputs.append(
{
@@ -464,6 +583,7 @@ def build_golang_images(service_names: list[str], git_ref: str, image_tag: str)
)
finally:
remove_temp_worktree(repo_root, worktree_root)
+ emit_progress(progress, "build.golang", f"已清理 Go worktree：{worktree_root}")
return {
"kind": "golang",
@@ -484,6 +604,7 @@ def build_service_images(
go_git_ref: str,
image_tag: str,
skip_maven_build: bool,
+ progress: ProgressLogger | None = None,
) -> dict[str, Any]:
# 一个请求里可能同时选了 Java 和 Go，需要分仓构建后再合并 digest。
normalized = normalize_service_names(service_names)
@@ -493,16 +614,25 @@ def build_service_images(
if not normalized:
raise RuntimeError("services are required")
- docker_login_local(ensure_repo_root(JAVA_REPO_ROOT if java_services else GOLANG_REPO_ROOT, "build"))
+ docker_login_local(
+ ensure_repo_root(JAVA_REPO_ROOT if java_services else GOLANG_REPO_ROOT, "build"),
+ progress=progress,
+ )
builds: list[dict[str, Any]] = []
service_images: dict[str, str] = {}
if java_services:
- java_build = build_java_images(java_services, java_git_ref, image_tag, skip_maven_build=skip_maven_build)
+ java_build = build_java_images(
+ java_services,
+ java_git_ref,
+ image_tag,
+ skip_maven_build=skip_maven_build,
+ progress=progress,
+ )
builds.append(java_build)
service_images.update(java_build["serviceImages"])
if golang_services:
- golang_build = build_golang_images(golang_services, go_git_ref, image_tag)
+ golang_build = build_golang_images(golang_services, go_git_ref, image_tag, progress=progress)
builds.append(golang_build)
service_images.update(golang_build["serviceImages"])
@@ -644,6 +774,7 @@ def deploy_updated_services_payload(
go_git_ref: str,
image_tag: str,
skip_maven_build: bool,
+ progress: ProgressLogger | None = None,
) -> dict[str, Any]:
# 这是“服务更新”主链路：本地 build -> Harbor digest -> 完整 compose -> 预拉 -> 滚动重建。
normalized = normalize_service_names(service_names)
@@ -658,19 +789,24 @@ def deploy_updated_services_payload(
if service_name not in records_map:
raise RuntimeError(f"service not found in runtime topology: {service_name}")
+ emit_progress(progress, "prepare", f"准备更新服务：{', '.join(normalized)}")
build_result = build_service_images(
normalized,
java_git_ref=java_git_ref,
go_git_ref=go_git_ref,
image_tag=image_tag,
skip_maven_build=skip_maven_build,
+ progress=progress,
)
service_images = dict(build_result.get("serviceImages") or {})
+ emit_progress(progress, "prepare", "镜像构建完成，开始更新本地完整 compose 模板")
template_updates: list[dict[str, Any]] = []
for service_name in normalized:
for record in records_map[service_name]:
- template_updates.append(update_local_service_image(record["host"], service_name, service_images[service_name]))
+ image_ref = service_images[service_name]
+ emit_progress(progress, "compose.template", f"更新模板：{record['host']} / {service_name} -> {image_ref}")
+ template_updates.append(update_local_service_image(record["host"], service_name, image_ref))
host_plan = host_image_plan(normalized, service_images)
prepared_hosts: list[dict[str, Any]] = []
@@ -678,8 +814,20 @@ def deploy_updated_services_payload(
try:
for host_entry in host_plan:
+ emit_progress(progress, f"compose.sync.{host_entry['host']}", f"同步完整 compose 到远端：{host_entry['host']}")
compose_result = sync_remote_host_compose_template(host_entry["host"], refresh_from_remote=False)
+ emit_progress(
+ progress,
+ f"compose.sync.{host_entry['host']}",
+ f"compose 同步完成：changed={bool(compose_result.get('changed'))} path={compose_result.get('path')}",
+ )
+ emit_progress(progress, f"image.prepull.{host_entry['host']}", f"预拉镜像：{', '.join(host_entry.get('images') or [])}")
prepull_result = prepull_host_images(host_entry)
+ emit_progress(
+ progress,
+ f"image.prepull.{host_entry['host']}",
+ f"预拉完成：{prepull_result['status']} {short_text(prepull_result['output'])}",
+ )
prepared_hosts.append(
{
"host": host_entry["host"],
@@ -702,6 +850,7 @@ def deploy_updated_services_payload(
for service_name in normalized:
for service in records_map[service_name]:
started_at = time.time()
+ step_stage = f"restart.{service['host']}.{service_name}"
step: dict[str, Any] = {
"service": service_name,
"host": service["host"],
@@ -711,9 +860,15 @@ def deploy_updated_services_payload(
"status": "running",
}
try:
+ emit_progress(progress, step_stage, f"开始滚动更新：{service['host']} / {service_name}")
restart_result = restart_service_instance(service)
step["tatStatus"] = restart_result.get("status")
step["tatOutput"] = str(restart_result.get("output") or "").strip()[:500]
+ emit_progress(
+ progress,
+ step_stage,
+ f"TAT 重建完成：{restart_result.get('status')} {short_text(step['tatOutput'])}",
+ )
http_result = wait_http_ready(service, timeout_seconds=ROLLING_RESTART_TIMEOUT_SECONDS)
step["http"] = {
@@ -722,24 +877,45 @@ def deploy_updated_services_payload(
"latencyMs": int(http_result.get("latencyMs") or 0),
"detail": str(http_result.get("detail") or ""),
}
+ emit_progress(
+ progress,
+ step_stage,
+ f"HTTP 恢复：{step['http']['statusCode']} / {step['http']['latencyMs']}ms",
+ )
if service["kind"] == "java":
step["nacos"] = wait_java_registration_ready(
service,
timeout_seconds=ROLLING_RESTART_TIMEOUT_SECONDS,
)
+ emit_progress(
+ progress,
+ step_stage,
+ f"Nacos 注册恢复：{step['nacos'].get('source') or 'ok'}",
+ )
if ROLLING_RESTART_STABILIZE_SECONDS > 0:
+ emit_progress(
+ progress,
+ step_stage,
+ f"稳定观察 {ROLLING_RESTART_STABILIZE_SECONDS}s",
+ )
time.sleep(ROLLING_RESTART_STABILIZE_SECONDS)
step["status"] = "success"
step["durationSeconds"] = round(time.time() - started_at, 2)
steps.append(step)
+ emit_progress(
+ progress,
+ step_stage,
+ f"实例更新成功，耗时 {step['durationSeconds']}s",
+ )
except Exception as exc: # noqa: BLE001
step["status"] = "failed"
step["error"] = str(exc)
step["durationSeconds"] = round(time.time() - started_at, 2)
steps.append(step)
+ emit_progress(progress, step_stage, f"实例更新失败：{step['error']}", level="error")
clear_operation_caches()
return {
"ok": False,
@@ -753,7 +929,9 @@ def deploy_updated_services_payload(
}
finally:
clear_operation_caches()
+ emit_progress(progress, "finish", "已清理监控缓存")
+ emit_progress(progress, "finish", "全部服务更新完成")
return {
"ok": True,
"updatedAt": utc_now(),
diff --git a/monitor/service_update_jobs.py b/monitor/service_update_jobs.py
new file mode 100644
index 0000000..f863efa
--- /dev/null
+++ b/monitor/service_update_jobs.py
@@ -0,0 +1,249 @@
+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)
diff --git a/static/app.css b/static/app.css
index 1a3ac49..7d41c78 100644
--- a/static/app.css
+++ b/static/app.css
@@ -534,6 +534,38 @@ body {
margin-top: 14px;
}
+.operation-meta {
+ margin-top: 14px;
+}
+
+.update-log-panel {
+ margin-top: 14px;
+ padding: 16px;
+ border: 1px solid var(--line);
+ border-radius: 22px;
+ background: color-mix(in oklab, var(--panel-strong) 94%, transparent);
+}
+
+.update-log-head {
+ margin-bottom: 12px;
+}
+
+.update-log-output {
+ margin: 0;
+ min-height: 180px;
+ max-height: 420px;
+ overflow: auto;
+ padding: 16px;
+ border: 1px solid color-mix(in oklab, var(--line) 88%, transparent);
+ border-radius: 18px;
+ background: oklch(19% 0.02 248 / 0.96);
+ color: oklch(92% 0.01 250);
+ font-size: 12px;
+ line-height: 1.55;
+ white-space: pre-wrap;
+ word-break: break-word;
+}
+
.update-field,
.update-check {
display: flex;
diff --git a/static/app.js b/static/app.js
index d1775a3..93bcf4c 100644
--- a/static/app.js
+++ b/static/app.js
@@ -1,6 +1,7 @@
const { createApp } = Vue;
const STORAGE_KEY = "hy-app-monitor-refresh-interval";
+const UPDATE_OPERATION_STORAGE_KEY = "hy-app-monitor-service-update-operation";
createApp({
data() {
@@ -89,6 +90,13 @@ createApp({
loading: false,
error: "",
running: false,
+ operationId: "",
+ status: "",
+ stage: "",
+ startedAt: "",
+ finishedAt: "",
+ logDroppedCount: 0,
+ logs: [],
message: "",
items: [],
selectedServices: [],
@@ -108,6 +116,7 @@ createApp({
golangRepoRoot: "",
},
result: null,
+ pollTimer: null,
},
restartOps: {
loading: false,
@@ -221,6 +230,16 @@ createApp({
updateTargetMap() {
return Object.fromEntries((this.updateOps.items || []).map((item) => [item.service, item]));
},
+ updateLogText() {
+ return (this.updateOps.logs || [])
+ .map((entry) => {
+ const time = this.formatDateTime(entry.time);
+ const stage = entry.stage ? ` [${entry.stage}]` : "";
+ const level = entry.level && entry.level !== "info" ? ` (${entry.level})` : "";
+ return `${time}${stage}${level} ${entry.message || ""}`.trim();
+ })
+ .join("\n");
+ },
},
watch: {
refreshIntervalSeconds() {
@@ -314,6 +333,33 @@ createApp({
updateTargetLabel(item) {
return `${item.service} · ${item.kind} · ${item.hosts.map((host) => host.host).join(" / ")}`;
},
+ updateStatusBadgeClass() {
+ if (this.updateOps.error || this.updateOps.status === "failed") {
+ return "bad";
+ }
+ if (this.updateOps.running || this.updateOps.status === "running" || this.updateOps.status === "queued") {
+ return "warn";
+ }
+ if (this.updateOps.status === "success") {
+ return "ok";
+ }
+ return "neutral";
+ },
+ updateStatusLabel() {
+ if (this.updateOps.error || this.updateOps.status === "failed") {
+ return "失败";
+ }
+ if (this.updateOps.status === "queued") {
+ return "排队中";
+ }
+ if (this.updateOps.running || this.updateOps.status === "running") {
+ return "执行中";
+ }
+ if (this.updateOps.status === "success") {
+ return "已完成";
+ }
+ return "未开始";
+ },
syncRefreshSettings() {
const settings = this.payload.settings || {};
const options = Array.isArray(settings.refreshIntervalOptions)
@@ -339,6 +385,96 @@ createApp({
this.applyRefreshTimer();
}
},
+ persistUpdateOperationId() {
+ if (this.updateOps.operationId) {
+ window.sessionStorage.setItem(UPDATE_OPERATION_STORAGE_KEY, this.updateOps.operationId);
+ return;
+ }
+ window.sessionStorage.removeItem(UPDATE_OPERATION_STORAGE_KEY);
+ },
+ stopUpdatePolling() {
+ if (this.updateOps.pollTimer) {
+ window.clearInterval(this.updateOps.pollTimer);
+ this.updateOps.pollTimer = null;
+ }
+ },
+ scrollUpdateLogToBottom() {
+ this.$nextTick(() => {
+ const node = document.getElementById("service-update-log-output");
+ if (node) {
+ node.scrollTop = node.scrollHeight;
+ }
+ });
+ },
+ applyUpdateOperation(operation) {
+ if (!operation) {
+ return;
+ }
+
+ this.updateOps.operationId = operation.operationId || "";
+ this.updateOps.status = operation.status || "";
+ this.updateOps.stage = operation.stage || "";
+ this.updateOps.startedAt = operation.startedAt || "";
+ this.updateOps.finishedAt = operation.finishedAt || "";
+ this.updateOps.logDroppedCount = Number(operation.logDroppedCount || 0);
+ this.updateOps.logs = Array.isArray(operation.logs) ? operation.logs : [];
+ this.updateOps.result = operation.result || null;
+ this.updateOps.running = operation.status === "queued" || operation.status === "running";
+
+ if (operation.status === "success") {
+ this.updateOps.message = operation.message || operation.result?.message || "服务更新完成";
+ this.updateOps.error = "";
+ } else if (operation.status === "failed") {
+ this.updateOps.error = operation.error || operation.result?.error || "服务更新失败";
+ this.updateOps.message = operation.message || "";
+ } else {
+ this.updateOps.message = operation.message || "";
+ }
+
+ this.persistUpdateOperationId();
+ this.scrollUpdateLogToBottom();
+ },
+ startUpdatePolling() {
+ this.stopUpdatePolling();
+ if (!this.updateOps.operationId) {
+ return;
+ }
+ this.updateOps.pollTimer = window.setInterval(() => {
+ this.fetchUpdateOperationStatus(this.updateOps.operationId, { silent: true });
+ }, 2000);
+ },
+ async fetchUpdateOperationStatus(operationId, { silent = false } = {}) {
+ if (!operationId) {
+ return;
+ }
+
+ const wasRunning = this.updateOps.running;
+ try {
+ const query = new URLSearchParams({ id: operationId });
+ const response = await fetch(`/api/ops/update-status?${query.toString()}`, { cache: "no-store" });
+ const payload = await this.readJsonResponse(response);
+ if (!response.ok) {
+ throw new Error(payload.error || `HTTP ${response.status}`);
+ }
+
+ this.applyUpdateOperation(payload.operation || {});
+
+ if (!this.updateOps.running) {
+ this.stopUpdatePolling();
+ if (wasRunning) {
+ await this.refresh();
+ await this.refreshUpdateTargets();
+ await this.refreshRestartTargets();
+ }
+ }
+ } catch (error) {
+ this.stopUpdatePolling();
+ this.updateOps.running = false;
+ if (!silent) {
+ this.updateOps.error = error instanceof Error ? error.message : String(error);
+ }
+ }
+ },
applyRefreshTimer() {
if (this.timer) {
window.clearInterval(this.timer);
@@ -685,9 +821,19 @@ createApp({
return;
}
+ this.stopUpdatePolling();
this.updateOps.running = true;
this.updateOps.error = "";
- this.updateOps.message = "";
+ this.updateOps.message = "已提交更新任务，等待开始执行";
+ this.updateOps.operationId = "";
+ this.updateOps.status = "queued";
+ this.updateOps.stage = "queued";
+ this.updateOps.startedAt = "";
+ this.updateOps.finishedAt = "";
+ this.updateOps.logDroppedCount = 0;
+ this.updateOps.logs = [];
+ this.updateOps.result = null;
+ this.persistUpdateOperationId();
try {
const response = await fetch("/api/ops/update-services", {
@@ -708,18 +854,14 @@ createApp({
throw new Error(payload.error || `HTTP ${response.status}`);
}
- this.updateOps.result = payload;
- this.updateOps.message = payload.ok
- ? (payload.message || "服务更新完成")
- : (payload.error || "服务更新失败");
-
- await this.refresh();
- await this.refreshUpdateTargets();
- await this.refreshRestartTargets();
+ this.applyUpdateOperation(payload.operation || {});
+ this.startUpdatePolling();
+ await this.fetchUpdateOperationStatus(this.updateOps.operationId, { silent: true });
} catch (error) {
this.updateOps.error = error instanceof Error ? error.message : String(error);
- } finally {
this.updateOps.running = false;
+ } finally {
+ this.scrollUpdateLogToBottom();
}
},
toggleRestartService(serviceName) {
@@ -781,10 +923,16 @@ createApp({
this.refreshGoConfig();
this.refreshUpdateTargets();
this.refreshRestartTargets();
+ const lastOperationId = window.sessionStorage.getItem(UPDATE_OPERATION_STORAGE_KEY) || "";
+ if (lastOperationId) {
+ this.fetchUpdateOperationStatus(lastOperationId, { silent: true });
+ this.startUpdatePolling();
+ }
},
beforeUnmount() {
if (this.timer) {
window.clearInterval(this.timer);
}
+ this.stopUpdatePolling();
},
}).mount("#app");
diff --git a/static/index.html b/static/index.html
index 1f0b6b3..36e5465 100644
--- a/static/index.html
+++ b/static/index.html
@@ -427,8 +427,8 @@
Service Update
本机构建选中的 Java / Golang 服务，推到 Harbor，更新完整 compose，并只重建你选中的服务
-
- {{ updateOps.error ? '存在错误' : updateOps.running ? '执行中' : '可操作' }}
+
+ {{ updateStatusLabel() }}
@@ -488,6 +488,28 @@
{{ updateOps.error }}
{{ updateOps.message }}
+
+ 任务：{{ updateOps.operationId }}
+ 状态：{{ updateStatusLabel() }}
+ 阶段：{{ updateOps.stage || '-' }}
+ 开始：{{ formatDateTime(updateOps.startedAt) }}
+ 结束：{{ formatDateTime(updateOps.finishedAt) }}
+ 已截断 {{ updateOps.logDroppedCount }} 行旧日志
+
+
+
+
+
+
Update Log
+
按阶段展示构建、推 Harbor、同步 compose、预拉镜像和滚动重建的实时输出
+
+
+
{{ updateLogText || '等待更新日志...' }}
+
+