1017 lines
39 KiB
Python
1017 lines
39 KiB
Python
from __future__ import annotations
|
||
|
||
# 镜像 tag 需要拼时间戳。
|
||
from datetime import datetime
|
||
# 本地构建和 git worktree 都依赖标准库 subprocess / tempfile。
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import shlex
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from typing import Any, Callable
|
||
|
||
from .compose_templates import (
|
||
load_local_host_compose_model,
|
||
sync_remote_host_compose_template,
|
||
update_local_service_image,
|
||
)
|
||
from .config import (
|
||
BUILD_DOCKER_PLATFORM,
|
||
GOLANG_REPO_ROOT,
|
||
HARBOR_PASSWORD,
|
||
HARBOR_PROJECT,
|
||
HARBOR_REGISTRY,
|
||
HARBOR_USERNAME,
|
||
JAVA_REPO_ROOT,
|
||
ROOT,
|
||
ROLLING_RESTART_STABILIZE_SECONDS,
|
||
ROLLING_RESTART_TIMEOUT_SECONDS,
|
||
SERVICE_BUILD_TIMEOUT_SECONDS,
|
||
SERVICE_PULL_TIMEOUT_SECONDS,
|
||
UPDATE_DEFAULT_GO_GIT_REF,
|
||
UPDATE_DEFAULT_JAVA_GIT_REF,
|
||
utc_now,
|
||
)
|
||
from .service_ops import (
|
||
clear_operation_caches,
|
||
restart_service_instance,
|
||
service_records_by_name,
|
||
wait_http_ready,
|
||
wait_java_registration_ready,
|
||
)
|
||
from .tencent_tat import run_shell_command
|
||
|
||
|
||
# Java 仓里当前支持 Harbor 更新的服务集合。
|
||
JAVA_BUILDABLE_SERVICES = ("auth", "gateway", "external", "console", "other", "live", "wallet", "order")
|
||
# Go 仓目前只有主业务服务。
|
||
GOLANG_BUILDABLE_SERVICES = ("golang",)
|
||
# Java 批量 Maven 构建时，需要映射回对应的启动模块。
|
||
JAVA_SERVICE_MODULES = {
|
||
"auth": "rc-auth",
|
||
"gateway": "rc-gateway",
|
||
"external": "rc-service/rc-service-external/external-start",
|
||
"console": "rc-service/rc-service-console/console-start",
|
||
"other": "rc-service/rc-service-other/other-start",
|
||
"live": "rc-service/rc-service-live/live-start",
|
||
"wallet": "rc-service/rc-service-wallet/wallet-start",
|
||
"order": "rc-service/rc-service-order/order-start",
|
||
}
|
||
# 当前线上镜像仓路径并不完全统一，这里只给缺失模板信息时兜底。
|
||
SERVICE_IMAGE_REPOSITORY_SUFFIXES = {
|
||
"auth": "java/chatapp3-auth",
|
||
"gateway": "java/chatapp3-gateway",
|
||
"external": "java/chatapp3-external",
|
||
"console": "console",
|
||
"other": "java/chatapp3-other",
|
||
"live": "java/chatapp3-live",
|
||
"wallet": "java/chatapp3-wallet",
|
||
"order": "order",
|
||
"golang": "golang",
|
||
}
|
||
|
||
# 后台任务的日志回调统一走这个签名。
|
||
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:
|
||
# 所有过程日志都经这里发出，调用方传空时自动静默。
|
||
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]:
|
||
# 页面和接口统一按这里判定哪些服务允许“构建并更新”。
|
||
return [*JAVA_BUILDABLE_SERVICES, *GOLANG_BUILDABLE_SERVICES]
|
||
|
||
|
||
def normalize_service_names(service_names: list[str]) -> list[str]:
|
||
# 请求里可能重复勾选同一个服务，这里先去重并保序。
|
||
normalized = [str(name or "").strip() for name in service_names if str(name or "").strip()]
|
||
return list(dict.fromkeys(normalized))
|
||
|
||
|
||
def service_kind(service_name: str) -> str:
|
||
# 更新链路目前只分 Java / Golang 两类仓库。
|
||
if service_name in JAVA_BUILDABLE_SERVICES:
|
||
return "java"
|
||
if service_name in GOLANG_BUILDABLE_SERVICES:
|
||
return "golang"
|
||
raise RuntimeError(f"unsupported update service: {service_name}")
|
||
|
||
|
||
def ensure_repo_root(repo_root: Path, label: str) -> Path:
|
||
# 构建前先确认业务仓真的存在，避免跑到一半才发现路径错了。
|
||
resolved = Path(repo_root).expanduser().resolve()
|
||
if not resolved.exists():
|
||
raise RuntimeError(f"{label} repo root missing: {resolved}")
|
||
if not (resolved / ".git").exists():
|
||
raise RuntimeError(f"{label} repo root is not a git repo: {resolved}")
|
||
return resolved
|
||
|
||
|
||
def local_command_output(
|
||
command: list[str],
|
||
*,
|
||
cwd: Path,
|
||
label: str,
|
||
env: dict[str, str] | None = None,
|
||
input_text: str = "",
|
||
timeout_seconds: int = 120,
|
||
) -> str:
|
||
# 小输出命令直接走 capture_output，适合 git rev-parse / docker inspect 这类场景。
|
||
merged_env = os.environ.copy()
|
||
if env:
|
||
merged_env.update(env)
|
||
|
||
result = subprocess.run(
|
||
command,
|
||
cwd=str(cwd),
|
||
text=True,
|
||
input=input_text if input_text else None,
|
||
capture_output=True,
|
||
env=merged_env,
|
||
timeout=timeout_seconds,
|
||
check=False,
|
||
)
|
||
if result.returncode != 0:
|
||
detail = (result.stderr or result.stdout or "").strip()
|
||
raise RuntimeError(f"{label} failed: {detail or f'exit {result.returncode}'}")
|
||
return (result.stdout or "").strip()
|
||
|
||
|
||
def tail_text(path: Path, max_bytes: int = 8192) -> str:
|
||
# 大构建日志只回最后一小段，避免把整份 Maven 输出塞进响应体。
|
||
if not path.exists():
|
||
return ""
|
||
|
||
with path.open("rb") as handle:
|
||
handle.seek(0, os.SEEK_END)
|
||
size = handle.tell()
|
||
handle.seek(max(0, size - max_bytes))
|
||
return handle.read().decode("utf-8", errors="replace").strip()
|
||
|
||
|
||
def run_logged_command(
|
||
command: list[str],
|
||
*,
|
||
cwd: Path,
|
||
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()
|
||
if env:
|
||
merged_env.update(env)
|
||
|
||
log_file = tempfile.NamedTemporaryFile(prefix="hy-app-monitor-build-", suffix=".log", delete=False)
|
||
log_path = Path(log_file.name)
|
||
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,
|
||
cwd=str(cwd),
|
||
stdout=handle,
|
||
stderr=subprocess.STDOUT,
|
||
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:
|
||
raise RuntimeError(f"{label} failed: {output_tail or f'exit {return_code}'}")
|
||
return output_tail
|
||
finally:
|
||
log_path.unlink(missing_ok=True)
|
||
|
||
|
||
def fetch_repo_refs(repo_root: Path, label: str) -> None:
|
||
# 统一先刷新远端分支和 tag，后面的 ref 解析才可信。
|
||
local_command_output(
|
||
["git", "-C", str(repo_root), "fetch", "--all", "--tags", "--prune"],
|
||
cwd=repo_root,
|
||
label=f"{label} fetch refs",
|
||
timeout_seconds=300,
|
||
)
|
||
|
||
|
||
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 = [
|
||
f"refs/remotes/origin/{git_ref}^{{commit}}",
|
||
f"refs/tags/{git_ref}^{{commit}}",
|
||
f"{git_ref}^{{commit}}",
|
||
]
|
||
for candidate in candidates:
|
||
try:
|
||
return local_command_output(
|
||
["git", "-C", str(repo_root), "rev-parse", "--verify", candidate],
|
||
cwd=repo_root,
|
||
label=f"{label} resolve ref",
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
raise RuntimeError(f"{label} git ref not found: {git_ref}")
|
||
|
||
|
||
def create_temp_worktree(repo_root: Path, commit: str, label: str) -> Path:
|
||
# 用临时 worktree 构建，避免直接改业务仓当前工作区。
|
||
worktree_root = Path(tempfile.mkdtemp(prefix=f"hy-app-monitor-{repo_root.name}-"))
|
||
try:
|
||
local_command_output(
|
||
["git", "-C", str(repo_root), "worktree", "add", "--detach", str(worktree_root), commit],
|
||
cwd=repo_root,
|
||
label=f"{label} create worktree",
|
||
timeout_seconds=300,
|
||
)
|
||
return worktree_root
|
||
except Exception:
|
||
shutil.rmtree(worktree_root, ignore_errors=True)
|
||
raise
|
||
|
||
|
||
def remove_temp_worktree(repo_root: Path, worktree_root: Path) -> None:
|
||
# 构建结束后清理 worktree，避免 deploy 机累积一堆临时目录。
|
||
try:
|
||
local_command_output(
|
||
["git", "-C", str(repo_root), "worktree", "remove", "--force", str(worktree_root)],
|
||
cwd=repo_root,
|
||
label="remove worktree",
|
||
timeout_seconds=120,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
shutil.rmtree(worktree_root, ignore_errors=True)
|
||
|
||
|
||
def require_harbor_credentials() -> None:
|
||
# 构建推送和目标机预拉都依赖 Harbor 账号。
|
||
if not HARBOR_REGISTRY:
|
||
raise RuntimeError("missing HARBOR_REGISTRY in .env")
|
||
if not HARBOR_USERNAME or not HARBOR_PASSWORD:
|
||
raise RuntimeError("missing HARBOR_USERNAME / HARBOR_PASSWORD in .env")
|
||
|
||
|
||
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,
|
||
label="docker login harbor",
|
||
input_text=HARBOR_PASSWORD,
|
||
timeout_seconds=120,
|
||
)
|
||
emit_progress(progress, "prepare", "Harbor 登录成功")
|
||
|
||
|
||
def image_repository_ref(image_ref: str) -> str:
|
||
# 把 tag / digest 去掉，保留完整仓库地址。
|
||
text = str(image_ref or "").strip()
|
||
if "@" in text:
|
||
return text.split("@", 1)[0]
|
||
|
||
last_slash = text.rfind("/")
|
||
last_colon = text.rfind(":")
|
||
if last_colon > last_slash:
|
||
return text[:last_colon]
|
||
return text
|
||
|
||
|
||
def sanitized_ref_fragment(git_ref: str) -> str:
|
||
# Harbor tag 只保留相对安全的字符。
|
||
cleaned = re.sub(r"[^0-9A-Za-z._-]+", "-", str(git_ref or "").strip()) or "manual"
|
||
return cleaned.strip("-") or "manual"
|
||
|
||
|
||
def derive_image_tag(git_ref: str, commit: str, image_tag: str) -> str:
|
||
# 用户没指定 tag 时，默认用 ref + commit12 + 时间戳。
|
||
explicit = str(image_tag or "").strip()
|
||
if explicit:
|
||
return explicit
|
||
timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
|
||
return f"{sanitized_ref_fragment(git_ref)}-{commit[:12]}-{timestamp}"
|
||
|
||
|
||
def docker_digest(image_ref: str, *, cwd: Path) -> str:
|
||
# push 完要拿 digest，后续 compose 一律按 digest 部署。
|
||
return local_command_output(
|
||
["docker", "image", "inspect", image_ref, "--format", "{{index .RepoDigests 0}}"],
|
||
cwd=cwd,
|
||
label=f"inspect digest {image_ref}",
|
||
timeout_seconds=120,
|
||
)
|
||
|
||
|
||
def current_template_image(service_name: str) -> str:
|
||
# 优先从本地完整模板里拿当前运行时镜像仓路径。
|
||
records = service_records_by_name().get(service_name) or []
|
||
for record in records:
|
||
model = load_local_host_compose_model(record["host"])
|
||
image_ref = str(((model.get("services") or {}).get(service_name) or {}).get("image") or "").strip()
|
||
if image_ref.startswith(f"{HARBOR_REGISTRY}/{HARBOR_PROJECT}/"):
|
||
return image_ref
|
||
return ""
|
||
|
||
|
||
def service_repository_ref(service_name: str) -> str:
|
||
# 线上已有镜像时沿用现有仓路径；没有时再走保守兜底。
|
||
current_image = current_template_image(service_name)
|
||
if current_image:
|
||
return image_repository_ref(current_image)
|
||
|
||
suffix = SERVICE_IMAGE_REPOSITORY_SUFFIXES.get(service_name, "").strip()
|
||
if not suffix:
|
||
raise RuntimeError(f"missing Harbor repository mapping for service: {service_name}")
|
||
return f"{HARBOR_REGISTRY}/{HARBOR_PROJECT}/{suffix}"
|
||
|
||
|
||
def maven_module_list(service_names: list[str]) -> str:
|
||
# Java 多服务更新时，先一次性 Maven 打包对应模块，避免重复 clean package。
|
||
modules: list[str] = []
|
||
for service_name in service_names:
|
||
module = JAVA_SERVICE_MODULES.get(service_name)
|
||
if not module:
|
||
raise RuntimeError(f"missing java maven module mapping: {service_name}")
|
||
if module not in modules:
|
||
modules.append(module)
|
||
return ",".join(modules)
|
||
|
||
|
||
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)
|
||
|
||
outputs: list[dict[str, Any]] = []
|
||
service_images: dict[str, str] = {}
|
||
|
||
try:
|
||
if not skip_maven_build:
|
||
ensure_private_maven_installed(
|
||
repo_root,
|
||
worktree_root,
|
||
commit,
|
||
progress=progress,
|
||
)
|
||
emit_progress(progress, "build.java.maven", f"开始 Maven 打包：{maven_module_list(service_names)}")
|
||
run_logged_command(
|
||
[
|
||
"mvn",
|
||
"-pl",
|
||
maven_module_list(service_names),
|
||
"-am",
|
||
"-Dmaven.test.skip=true",
|
||
"clean",
|
||
"package",
|
||
],
|
||
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,
|
||
# 外层已经统一决定过是否 Maven 打包；镜像阶段一律复用现成产物。
|
||
"SKIP_MAVEN_BUILD": "1",
|
||
},
|
||
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(
|
||
{
|
||
"service": service_name,
|
||
"repository": repository_ref,
|
||
"tag": target_image_ref,
|
||
"image": digest_ref,
|
||
"buildTail": build_tail,
|
||
"pushTail": push_tail,
|
||
}
|
||
)
|
||
finally:
|
||
remove_temp_worktree(repo_root, worktree_root)
|
||
emit_progress(progress, "build.java", f"已清理 Java worktree：{worktree_root}")
|
||
|
||
return {
|
||
"kind": "java",
|
||
"repoRoot": str(repo_root),
|
||
"gitRef": resolved_ref,
|
||
"commit": commit,
|
||
"imageTag": resolved_image_tag,
|
||
"services": list(service_names),
|
||
"items": outputs,
|
||
"serviceImages": service_images,
|
||
}
|
||
|
||
|
||
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)
|
||
|
||
outputs: list[dict[str, Any]] = []
|
||
service_images: dict[str, str] = {}
|
||
|
||
try:
|
||
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(
|
||
{
|
||
"service": service_name,
|
||
"repository": repository_ref,
|
||
"tag": target_image_ref,
|
||
"image": digest_ref,
|
||
"buildTail": build_tail,
|
||
"pushTail": push_tail,
|
||
}
|
||
)
|
||
finally:
|
||
remove_temp_worktree(repo_root, worktree_root)
|
||
emit_progress(progress, "build.golang", f"已清理 Go worktree：{worktree_root}")
|
||
|
||
return {
|
||
"kind": "golang",
|
||
"repoRoot": str(repo_root),
|
||
"gitRef": resolved_ref,
|
||
"commit": commit,
|
||
"imageTag": resolved_image_tag,
|
||
"services": list(service_names),
|
||
"items": outputs,
|
||
"serviceImages": service_images,
|
||
}
|
||
|
||
|
||
def build_service_images(
|
||
service_names: list[str],
|
||
*,
|
||
java_git_ref: str,
|
||
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)
|
||
java_services = [service_name for service_name in normalized if service_kind(service_name) == "java"]
|
||
golang_services = [service_name for service_name in normalized if service_kind(service_name) == "golang"]
|
||
|
||
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"),
|
||
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,
|
||
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, progress=progress)
|
||
builds.append(golang_build)
|
||
service_images.update(golang_build["serviceImages"])
|
||
|
||
return {
|
||
"builds": builds,
|
||
"serviceImages": service_images,
|
||
}
|
||
|
||
|
||
def host_image_plan(service_names: list[str], service_images: dict[str, str]) -> list[dict[str, Any]]:
|
||
# 先按宿主机聚合 image，后面可以先预拉再做滚动重建。
|
||
records_map = service_records_by_name()
|
||
items: list[dict[str, Any]] = []
|
||
host_index: dict[str, dict[str, Any]] = {}
|
||
|
||
for service_name in service_names:
|
||
image_ref = str(service_images.get(service_name) or "").strip()
|
||
if not image_ref:
|
||
raise RuntimeError(f"missing built image for service: {service_name}")
|
||
|
||
for record in records_map.get(service_name) or []:
|
||
host_entry = host_index.get(record["host"])
|
||
if host_entry is None:
|
||
host_entry = {
|
||
"host": record["host"],
|
||
"ip": record["ip"],
|
||
"instanceId": record["instanceId"],
|
||
"services": [],
|
||
"images": [],
|
||
}
|
||
host_index[record["host"]] = host_entry
|
||
items.append(host_entry)
|
||
|
||
if service_name not in host_entry["services"]:
|
||
host_entry["services"].append(service_name)
|
||
if image_ref not in host_entry["images"]:
|
||
host_entry["images"].append(image_ref)
|
||
|
||
return items
|
||
|
||
|
||
def render_remote_login_lines() -> list[str]:
|
||
# 目标机拉 Harbor 镜像前同样需要显式登录。
|
||
require_harbor_credentials()
|
||
return [
|
||
f"printf '%s' {shlex.quote(HARBOR_PASSWORD)} | docker login {shlex.quote(HARBOR_REGISTRY)} -u {shlex.quote(HARBOR_USERNAME)} --password-stdin",
|
||
]
|
||
|
||
|
||
def remote_host_result(host_entry: dict[str, Any], script: str, command_name: str, timeout_seconds: int) -> dict[str, Any]:
|
||
# 统一执行单宿主机命令，便于预拉镜像和其他操作复用。
|
||
result_map = run_shell_command(
|
||
[host_entry["instanceId"]],
|
||
script,
|
||
command_name=command_name,
|
||
timeout_seconds=timeout_seconds,
|
||
)
|
||
result = result_map[host_entry["instanceId"]]
|
||
if result.get("status") != "SUCCESS":
|
||
raise RuntimeError(f"{host_entry['host']} {command_name} failed: {result.get('status')}")
|
||
return result
|
||
|
||
|
||
def prepull_host_images(host_entry: dict[str, Any]) -> dict[str, Any]:
|
||
# 先把该宿主机上要更新的镜像拉好，减少单实例重建时的等待。
|
||
lines = ["set -eu", *render_remote_login_lines()]
|
||
for image_ref in host_entry.get("images") or []:
|
||
lines.append(f"docker pull {shlex.quote(str(image_ref))}")
|
||
script = "\n".join(lines) + "\n"
|
||
result = remote_host_result(
|
||
host_entry,
|
||
script,
|
||
f"prepull-{host_entry['host']}",
|
||
timeout_seconds=max(SERVICE_PULL_TIMEOUT_SECONDS, 120),
|
||
)
|
||
return {
|
||
"status": result.get("status"),
|
||
"output": str(result.get("output") or "").strip()[-1200:],
|
||
}
|
||
|
||
|
||
def build_update_targets_payload() -> dict[str, Any]:
|
||
# 页面需要展示当前可更新服务、宿主机，以及本地构建所依赖的仓路径。
|
||
records_map = service_records_by_name()
|
||
items: list[dict[str, Any]] = []
|
||
|
||
for service_name in buildable_service_names():
|
||
records = records_map.get(service_name) or []
|
||
if not records:
|
||
continue
|
||
current_images: list[str] = []
|
||
for record in records:
|
||
model = load_local_host_compose_model(record["host"])
|
||
image_ref = str(((model.get("services") or {}).get(service_name) or {}).get("image") or "").strip()
|
||
if image_ref and image_ref not in current_images:
|
||
current_images.append(image_ref)
|
||
|
||
items.append(
|
||
{
|
||
"service": service_name,
|
||
"kind": service_kind(service_name),
|
||
"repository": service_repository_ref(service_name),
|
||
"currentImages": current_images,
|
||
"hosts": [
|
||
{
|
||
"host": record["host"],
|
||
"ip": record["ip"],
|
||
"instanceId": record["instanceId"],
|
||
"port": record["port"],
|
||
"path": record["path"],
|
||
}
|
||
for record in records
|
||
],
|
||
}
|
||
)
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"defaults": {
|
||
"javaGitRef": UPDATE_DEFAULT_JAVA_GIT_REF,
|
||
"goGitRef": UPDATE_DEFAULT_GO_GIT_REF,
|
||
"dockerPlatform": BUILD_DOCKER_PLATFORM,
|
||
},
|
||
"builder": {
|
||
"harborRegistry": HARBOR_REGISTRY,
|
||
"harborProject": HARBOR_PROJECT,
|
||
"javaRepoRoot": str(Path(JAVA_REPO_ROOT).expanduser()),
|
||
"golangRepoRoot": str(Path(GOLANG_REPO_ROOT).expanduser()),
|
||
},
|
||
"items": items,
|
||
}
|
||
|
||
|
||
def deploy_updated_services_payload(
|
||
service_names: list[str],
|
||
*,
|
||
java_git_ref: str,
|
||
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)
|
||
if not normalized:
|
||
raise RuntimeError("services are required")
|
||
|
||
records_map = service_records_by_name()
|
||
unsupported = [service_name for service_name in normalized if service_name not in buildable_service_names()]
|
||
if unsupported:
|
||
raise RuntimeError(f"unsupported update services: {', '.join(unsupported)}")
|
||
for service_name in normalized:
|
||
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]:
|
||
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]] = []
|
||
steps: list[dict[str, Any]] = []
|
||
|
||
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"],
|
||
"ip": host_entry["ip"],
|
||
"services": list(host_entry["services"]),
|
||
"images": list(host_entry["images"]),
|
||
"compose": {
|
||
"path": compose_result["path"],
|
||
"changed": bool(compose_result.get("changed")),
|
||
"source": str(compose_result.get("source") or ""),
|
||
"backupPath": str(compose_result.get("backupPath") or ""),
|
||
},
|
||
"prepull": {
|
||
"status": prepull_result["status"],
|
||
"output": prepull_result["output"],
|
||
},
|
||
}
|
||
)
|
||
|
||
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"],
|
||
"ip": service["ip"],
|
||
"kind": service["kind"],
|
||
"image": service_images[service_name],
|
||
"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"] = {
|
||
"ok": bool(http_result.get("ok")),
|
||
"statusCode": int(http_result.get("statusCode") or 0),
|
||
"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,
|
||
"updatedAt": utc_now(),
|
||
"services": normalized,
|
||
"build": build_result.get("builds") or [],
|
||
"templates": template_updates,
|
||
"preparedHosts": prepared_hosts,
|
||
"steps": steps,
|
||
"error": step["error"],
|
||
}
|
||
finally:
|
||
clear_operation_caches()
|
||
emit_progress(progress, "finish", "已清理监控缓存")
|
||
|
||
emit_progress(progress, "finish", "全部服务更新完成")
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"services": normalized,
|
||
"build": build_result.get("builds") or [],
|
||
"templates": template_updates,
|
||
"preparedHosts": prepared_hosts,
|
||
"steps": steps,
|
||
"message": "service update complete",
|
||
}
|