diff --git a/.env.example b/.env.example index 3c08947..ae098e8 100644 --- a/.env.example +++ b/.env.example @@ -8,12 +8,26 @@ PROBE_MAX_WORKERS=12 HOST_METRIC_CACHE_TTL_SECONDS=25 TAT_METRIC_TIMEOUT_SECONDS=30 TAT_CPU_SAMPLE_SECONDS=1 +ROLLING_RESTART_TIMEOUT_SECONDS=180 +ROLLING_RESTART_POLL_SECONDS=2 +ROLLING_RESTART_STABILIZE_SECONDS=2 +SERVICE_BUILD_TIMEOUT_SECONDS=3600 +SERVICE_PULL_TIMEOUT_SECONDS=600 NACOS_BASE_URL=http://10.2.1.17:8848 NACOS_NAMESPACE_ID=f7a8594e-090c-4830-a3ad-b5c0f9afcc07 NACOS_NAMESPACE_NAME=RedCircleServiceProd NACOS_DISCOVERY_GROUP=DEFAULT_GROUP NACOS_DISCOVERY_CLUSTER_NAME=DEFAULT NACOS_CACHE_TTL_SECONDS=15 +HARBOR_REGISTRY=10.2.1.3:18082 +HARBOR_PROJECT=likei +HARBOR_USERNAME= +HARBOR_PASSWORD= +BUILD_DOCKER_PLATFORM=linux/amd64 +UPDATE_DEFAULT_JAVA_GIT_REF=main +UPDATE_DEFAULT_GO_GIT_REF=main +JAVA_REPO_ROOT=/opt/deploy/sources/chatapp3-java +GOLANG_REPO_ROOT=/opt/deploy/sources/chatapp3-golang MONGO_URI= MONGO_DB_NAME=tarab_all MONGO_INSTANCE_ID=ins-o9mmob66 diff --git a/README.md b/README.md index 4fbf11c..7ef0e7a 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,19 @@ TENCENT_SECRET_ID=*** TENCENT_SECRET_KEY=*** ``` +如果你要直接在页面里做“构建镜像 -> 推 Harbor -> 更新服务”，还需要补这几项： + +```bash +HARBOR_REGISTRY=10.2.1.3:18082 +HARBOR_PROJECT=likei +HARBOR_USERNAME=*** +HARBOR_PASSWORD=*** +JAVA_REPO_ROOT=/opt/deploy/sources/chatapp3-java +GOLANG_REPO_ROOT=/opt/deploy/sources/chatapp3-golang +UPDATE_DEFAULT_JAVA_GIT_REF=main +UPDATE_DEFAULT_GO_GIT_REF=main +``` + 可以直接复制模板： ```bash @@ -101,6 +114,41 @@ python3 ops/deploy_via_tat.py 5. 安装并启用 `systemd` 6. 重启服务并检查 `2026` 监听 +## 维护完整 Compose + +默认只刷新本地完整模板： + +```bash +python3 ops/sync_runtime_compose.py +``` + +如果要把完整模板同步到远端业务主机： + +```bash +python3 ops/sync_runtime_compose.py --sync-remote +``` + +模板文件会写到本地 `run/compose_templates/`。 +滚动重启在真正执行 `docker compose up -d --force-recreate --no-deps ` 之前，也会先把对应宿主机的完整 `docker-compose.yml` 同步到远端，避免线上 compose 被裁剪成子集。 + +## 服务更新 + +页面里的“服务更新”会直接在当前机器上执行这条链路： + +1. 从 `JAVA_REPO_ROOT` / `GOLANG_REPO_ROOT` 指向的业务仓按 `git ref` 创建临时 `git worktree` +2. 本地构建镜像并推送到 Harbor +3. 取回 Harbor `digest` +4. 用新 `digest` 更新本地完整 compose 模板 +5. 把完整 compose 下发到目标宿主机 +6. 先在目标机 `docker pull` 新镜像 +7. 最后只对你选中的服务执行 `docker compose up -d --force-recreate --no-deps ` + +说明： + +- `other` 现在也按 Harbor 业务镜像更新，不再继续依赖本地 `./other/service.jar` 挂载 +- 页面里的“滚动重启”仍然保留，适合只想让配置生效、不涉及代码构建的场景 +- 这条链路默认假设 `hy-app-monitor` 就运行在打包机 / deploy 机上，并且本机已经能访问业务仓、Docker 和 Harbor + ## 页面内容 - 主机：`CPU / 内存 / 磁盘 / 进程数`，带阈值颜色分级 @@ -108,6 +156,7 @@ python3 ops/deploy_via_tat.py - Nacos：命名空间、服务、注册实例、健康状态 - Nacos 配置：读取当前 namespace 下的配置列表，支持格式校验、在线编辑保存 - Golang 配置：读取 app 节点当前运行中的 `service.env`，展示 `chatapp3-golang` 实际消费的环境变量并支持保存 +- 服务更新：本地构建选中服务、推送 Harbor、更新完整 compose，并只重建选中的服务 - 滚动重启：可按服务选择 Java / Golang 节点，逐台重启并等待健康恢复 - MongoDB：连接、内存、网络、操作计数、复制集、数据库统计 - 自动刷新：前端可切换刷新间隔，并保存在浏览器本地 diff --git a/monitor/compose_templates.py b/monitor/compose_templates.py new file mode 100644 index 0000000..2d91443 --- /dev/null +++ b/monitor/compose_templates.py @@ -0,0 +1,547 @@ +from __future__ import annotations + +# compose 模板生成需要解析 docker inspect 的 JSON。 +import json +# 本地模板目录和远端运行目录都走现有配置。 +from pathlib import Path +# 远端命令和文件路径都需要安全转义。 +import shlex +# 一些字段要做正则判断。 +import re +from typing import Any + +from .config import ROOT, RUNTIME_BASE_DIR, load_hosts, utc_now +from .runtime_config import read_remote_text, write_remote_text +from .tencent_tat import run_shell_command + + +# 本地完整 compose 模板统一放到 run 目录，避免把运行期产物写进 Git。 +COMPOSE_TEMPLATE_DIR = ROOT / "run" / "compose_templates" + +def managed_compose_hosts() -> list[dict[str, Any]]: + # 只维护当前 targets.json 里真正承载业务服务的宿主机。 + return [host for host in load_hosts() if host.get("services")] + + +def managed_compose_host_names() -> list[str]: + # 页面和运维脚本都更适合拿主机名列表。 + return [host["host"] for host in managed_compose_hosts()] + + +def managed_host_map() -> dict[str, dict[str, Any]]: + # 方便按主机名快速定位实例 ID 和服务清单。 + return {host["host"]: host for host in managed_compose_hosts()} + + +def compose_template_path(host_name: str) -> Path: + # 每台机器一份本地模板文件。 + return COMPOSE_TEMPLATE_DIR / f"{host_name}.yml" + + +def compose_remote_path(host_name: str) -> str: + # 远端 compose 路径沿用现网目录结构。 + return f"{RUNTIME_BASE_DIR.rstrip('/')}/{host_name}/docker-compose.yml" + + +def container_name(host_name: str, service_name: str) -> str: + # 当前线上容器命名规则固定。 + return f"likei-{host_name}-{service_name}" + + +def docker_inspect_payload(host: dict[str, Any], service_names: list[str]) -> list[dict[str, Any]]: + # 这里只抓 compose 需要的关键字段，避免完整 inspect JSON 被 TAT 输出截断。 + container_names = [container_name(host["host"], service_name) for service_name in service_names] + if not container_names: + raise RuntimeError(f"no compose services configured for host: {host['host']}") + + quoted_names = " ".join(shlex.quote(name) for name in container_names) + python_code = """ +import json +import sys + +payload = json.load(sys.stdin) +for item in payload: + config = item.get("Config") or {} + host_config = item.get("HostConfig") or {} + print( + json.dumps( + { + "Name": item.get("Name"), + "Config": { + "Image": config.get("Image"), + "StopSignal": config.get("StopSignal"), + "WorkingDir": config.get("WorkingDir"), + "Entrypoint": config.get("Entrypoint"), + "Cmd": config.get("Cmd"), + "Healthcheck": config.get("Healthcheck"), + }, + "HostConfig": { + "NetworkMode": host_config.get("NetworkMode"), + "RestartPolicy": {"Name": (host_config.get("RestartPolicy") or {}).get("Name")}, + "StopTimeout": host_config.get("StopTimeout"), + }, + "Mounts": item.get("Mounts") or [], + }, + ensure_ascii=False, + ) + ) +""".strip() + script = f"""set -eu +docker inspect {quoted_names} | python3 -c {shlex.quote(python_code)} +""" + result_map = run_shell_command( + [host["instanceId"]], + script, + command_name=f"inspect-compose-{host['host']}", + timeout_seconds=90, + ) + result = result_map[host["instanceId"]] + if result.get("status") != "SUCCESS": + raise RuntimeError(f"docker inspect failed on {host['host']}: {result.get('status')}") + + payload: list[dict[str, Any]] = [] + for raw_line in str(result.get("output") or "").splitlines(): + line = raw_line.strip() + if not line: + continue + try: + payload.append(json.loads(line)) + except json.JSONDecodeError as exc: + raise RuntimeError(f"invalid inspect field payload on {host['host']}") from exc + + if not payload: + raise RuntimeError(f"empty docker inspect payload on {host['host']}") + return payload + + +def ns_to_duration_text(raw_value: int | None) -> str | None: + # docker inspect 的健康检查时间单位是纳秒，compose 更适合秒级文本。 + if not raw_value: + return None + if raw_value % 1_000_000_000 == 0: + return f"{raw_value // 1_000_000_000}s" + if raw_value % 1_000_000 == 0: + return f"{raw_value // 1_000_000}ms" + return str(raw_value) + + +def stop_grace_period_text(raw_value: int | None) -> str | None: + # docker inspect 的 StopTimeout 单位本来就是秒。 + if raw_value is None or raw_value <= 0: + return None + return f"{int(raw_value)}s" + + +def image_is_pullable(image_ref: str) -> bool: + # 纯十六进制镜像 ID 不能当成远端可拉取镜像。 + text = str(image_ref or "").strip() + if not text: + return False + return re.fullmatch(r"[0-9a-f]{12,64}", text) is None + + +def normalize_command(raw_value: Any) -> list[str] | str | None: + # docker inspect 的 Cmd / Entrypoint 可能是数组、空值或字符串。 + if raw_value is None: + return None + if isinstance(raw_value, list): + return [str(item) for item in raw_value] + if isinstance(raw_value, str): + return raw_value + return [str(item) for item in raw_value] + + +def healthcheck_block(config: dict[str, Any]) -> dict[str, Any] | None: + # 把 inspect 里的健康检查结构翻回 compose 可接受格式。 + healthcheck = config.get("Healthcheck") or {} + test = healthcheck.get("Test") + if not test: + return None + + block: dict[str, Any] = {"test": [str(item) for item in test]} + interval = ns_to_duration_text(int(healthcheck.get("Interval") or 0)) + timeout = ns_to_duration_text(int(healthcheck.get("Timeout") or 0)) + start_period = ns_to_duration_text(int(healthcheck.get("StartPeriod") or 0)) + retries = int(healthcheck.get("Retries") or 0) + if interval: + block["interval"] = interval + if timeout: + block["timeout"] = timeout + if start_period: + block["start_period"] = start_period + if retries > 0: + block["retries"] = retries + return block + + +def render_compose_yaml(service_map: dict[str, dict[str, Any]]) -> str: + # 这里按需导入 PyYAML，和现有项目依赖策略保持一致。 + try: + import yaml + except Exception as exc: # noqa: BLE001 + raise RuntimeError(f"yaml renderer unavailable: {exc}") from exc + + payload = {"services": service_map} + return yaml.safe_dump(payload, allow_unicode=True, sort_keys=False).rstrip() + "\n" + + +def parse_compose_yaml(content: str) -> dict[str, Any]: + # 服务更新需要读取本地模板并直接改 image 字段。 + try: + import yaml + except Exception as exc: # noqa: BLE001 + raise RuntimeError(f"yaml parser unavailable: {exc}") from exc + + payload = yaml.safe_load(content) or {} + if not isinstance(payload, dict): + raise RuntimeError("invalid compose yaml payload") + + services = payload.get("services") or {} + if not isinstance(services, dict): + raise RuntimeError("invalid compose yaml services block") + return {"services": {str(name): dict(value or {}) for name, value in services.items()}} + + +def render_volume_specs(host_name: str, mounts: list[dict[str, Any]]) -> list[str]: + # 把 docker inspect 的挂载信息翻回 compose volumes。 + host_dir = f"{RUNTIME_BASE_DIR.rstrip('/')}/{host_name}/" + specs: list[str] = [] + for mount in mounts: + mount_type = str(mount.get("Type") or "").strip() + source = str(mount.get("Source") or "").strip() + destination = str(mount.get("Destination") or "").strip() + if mount_type != "bind" or not source or not destination: + continue + + rendered_source = source + if source.startswith(host_dir): + rendered_source = f"./{source[len(host_dir):]}" + + volume_spec = f"{rendered_source}:{destination}" + if not bool(mount.get("RW")): + volume_spec += ":ro" + specs.append(volume_spec) + return specs + + +def uses_local_runtime_artifact(spec: str) -> bool: + # 这些挂载代表服务依赖宿主机上的 jar / 二进制，不再适合镜像化发布。 + text = str(spec or "").strip() + return ( + text.endswith(":/application/service.jar:ro") + or text.endswith(":/application/service.jar") + or text.endswith(":/application/service:ro") + or text.endswith(":/application/service") + ) + + +def strip_local_runtime_artifacts(volume_specs: list[str]) -> list[str]: + # 发布到 Harbor 后，需要把本地运行时制品挂载从模板里移掉。 + return [spec for spec in volume_specs if not uses_local_runtime_artifact(spec)] + + +def build_service_definition( + host_name: str, + service_name: str, + inspect_payload: dict[str, Any], + *, + include_env_file: bool, +) -> dict[str, Any]: + # 只抽 compose 真正需要的字段，避免把 docker inspect 的噪声全部带回来。 + config = inspect_payload.get("Config") or {} + host_config = inspect_payload.get("HostConfig") or {} + service_definition: dict[str, Any] = {} + + image_ref = str(config.get("Image") or "").strip() + if not image_ref: + raise RuntimeError(f"missing image ref for {host_name}/{service_name}") + + service_definition["image"] = image_ref + + service_definition["container_name"] = str(inspect_payload.get("Name") or f"/{container_name(host_name, service_name)}").lstrip("/") + + network_mode = str(host_config.get("NetworkMode") or "").strip() + if network_mode: + service_definition["network_mode"] = network_mode + + restart_policy = str((host_config.get("RestartPolicy") or {}).get("Name") or "").strip() + if restart_policy: + service_definition["restart"] = restart_policy + + stop_signal = str(config.get("StopSignal") or "").strip() + if stop_signal: + service_definition["stop_signal"] = stop_signal + + stop_grace_period = stop_grace_period_text(host_config.get("StopTimeout")) + if stop_grace_period: + service_definition["stop_grace_period"] = stop_grace_period + + if include_env_file: + service_definition["env_file"] = ["./service.env"] + + volume_specs = render_volume_specs(host_name, list(inspect_payload.get("Mounts") or [])) + if volume_specs: + service_definition["volumes"] = volume_specs + + # 如果运行时依赖本地挂载的 jar / 二进制，就不要再强制拉镜像。 + uses_local_runtime_artifact = any( + spec.endswith(":/application/service.jar:ro") + or spec.endswith(":/application/service") + or spec.endswith(":/application/service:ro") + for spec in volume_specs + ) + if uses_local_runtime_artifact: + service_definition["pull_policy"] = "never" + elif image_is_pullable(image_ref): + service_definition["pull_policy"] = "always" + + working_dir = str(config.get("WorkingDir") or "").strip() + if working_dir: + service_definition["working_dir"] = working_dir + + entrypoint = normalize_command(config.get("Entrypoint")) + if entrypoint is not None: + service_definition["entrypoint"] = entrypoint + + command = normalize_command(config.get("Cmd")) + if command is not None: + service_definition["command"] = command + + healthcheck = healthcheck_block(config) + if healthcheck: + service_definition["healthcheck"] = healthcheck + + return service_definition + + +def generate_host_compose_template(host_name: str) -> dict[str, Any]: + # 生成模板时严格按 targets.json 里的服务顺序输出，后续 diff 更稳定。 + host_map = managed_host_map() + host = host_map.get(host_name) + if host is None: + raise RuntimeError(f"unsupported compose host: {host_name}") + + service_names = [str(service.get("name") or "").strip() for service in host.get("services") or []] + service_names = [service_name for service_name in service_names if service_name] + if not service_names: + raise RuntimeError(f"host has no managed services: {host_name}") + + inspect_items = docker_inspect_payload(host, service_names) + inspect_by_name = { + str((item.get("Name") or "").lstrip("/")): item + for item in inspect_items + } + + service_map: dict[str, dict[str, Any]] = {} + missing_services: list[str] = [] + for service_name in service_names: + expected_container = container_name(host_name, service_name) + inspect_item = inspect_by_name.get(expected_container) + if inspect_item is None: + missing_services.append(service_name) + continue + service_map[service_name] = build_service_definition( + host_name, + service_name, + inspect_item, + include_env_file=True, + ) + + if missing_services: + raise RuntimeError(f"missing running containers on {host_name}: {', '.join(missing_services)}") + + content = render_compose_yaml(service_map) + return { + "ok": True, + "host": host_name, + "services": list(service_map.keys()), + "updatedAt": utc_now(), + "content": content, + "path": str(compose_template_path(host_name)), + } + + +def save_local_host_compose_template(host_name: str) -> dict[str, Any]: + # 从远端运行态重新生成本地完整模板。 + payload = generate_host_compose_template(host_name) + path = compose_template_path(host_name) + path.parent.mkdir(parents=True, exist_ok=True) + previous = path.read_text(encoding="utf-8") if path.exists() else "" + path.write_text(payload["content"], encoding="utf-8") + return { + **payload, + "changed": previous != payload["content"], + "source": "remote-runtime", + } + + +def load_local_host_compose_template(host_name: str) -> dict[str, Any]: + # 本地模板优先作为远端同步的稳定来源。 + path = compose_template_path(host_name) + if not path.exists(): + raise RuntimeError(f"local compose template missing: {host_name}") + return { + "ok": True, + "host": host_name, + "services": [str(service.get("name") or "").strip() for service in managed_host_map()[host_name].get("services") or []], + "updatedAt": utc_now(), + "content": path.read_text(encoding="utf-8"), + "path": str(path), + "source": "local-cache", + } + + +def ensure_local_host_compose_template(host_name: str) -> dict[str, Any]: + # 第一次进入服务更新时，模板可能还没在本地生成。 + path = compose_template_path(host_name) + if path.exists(): + return load_local_host_compose_template(host_name) + return save_local_host_compose_template(host_name) + + +def load_local_host_compose_model(host_name: str) -> dict[str, Any]: + # 同时返回解析后的 services，方便直接修改模板。 + template_payload = ensure_local_host_compose_template(host_name) + parsed = parse_compose_yaml(str(template_payload.get("content") or "")) + return { + **template_payload, + "services": parsed["services"], + } + + +def save_local_host_compose_model(host_name: str, service_map: dict[str, dict[str, Any]]) -> dict[str, Any]: + # 模板编辑完成后，统一走这里回写磁盘并返回 diff 信息。 + path = compose_template_path(host_name) + path.parent.mkdir(parents=True, exist_ok=True) + content = render_compose_yaml(service_map) + previous = path.read_text(encoding="utf-8") if path.exists() else "" + path.write_text(content, encoding="utf-8") + return { + "ok": True, + "host": host_name, + "updatedAt": utc_now(), + "path": str(path), + "content": content, + "changed": previous != content, + "source": "local-cache", + "services": list(service_map.keys()), + } + + +def update_local_service_image(host_name: str, service_name: str, image_ref: str) -> dict[str, Any]: + # 服务更新链路会把新镜像 digest 写回本地完整模板。 + if not str(image_ref or "").strip(): + raise RuntimeError("image ref is required") + + model = load_local_host_compose_model(host_name) + service_map = dict(model.get("services") or {}) + service_definition = dict(service_map.get(service_name) or {}) + if not service_definition: + raise RuntimeError(f"missing service in local compose template: {host_name}/{service_name}") + + service_definition["image"] = str(image_ref).strip() + + volume_specs = [str(item) for item in service_definition.get("volumes") or []] + volume_specs = strip_local_runtime_artifacts(volume_specs) + if volume_specs: + service_definition["volumes"] = volume_specs + else: + service_definition.pop("volumes", None) + + if image_is_pullable(image_ref): + service_definition["pull_policy"] = "always" + else: + service_definition.pop("pull_policy", None) + + service_map[service_name] = service_definition + saved = save_local_host_compose_model(host_name, service_map) + saved["service"] = service_name + saved["image"] = str(image_ref).strip() + return saved + + +def verify_remote_compose(host: dict[str, Any], remote_path: str) -> list[str]: + # 下发后立即让 docker compose 解析一遍，避免把坏文件写到线上却没人发现。 + script = f"""set -eu +FILE={shlex.quote(remote_path)} +docker compose -f "$FILE" config --services +""" + result_map = run_shell_command( + [host["instanceId"]], + script, + command_name=f"verify-compose-{host['host']}", + timeout_seconds=60, + ) + result = result_map[host["instanceId"]] + if result.get("status") != "SUCCESS": + raise RuntimeError(f"compose verify failed on {host['host']}: {result.get('status')}") + return [line.strip() for line in str(result.get("output") or "").splitlines() if line.strip()] + + +def sync_remote_host_compose_template(host_name: str, *, refresh_from_remote: bool = True) -> dict[str, Any]: + # 对外暴露的主入口：优先刷新本地模板，再把完整 compose 同步到远端。 + host_map = managed_host_map() + host = host_map.get(host_name) + if host is None: + raise RuntimeError(f"unsupported compose host: {host_name}") + + template_payload: dict[str, Any] | None = None + refresh_error = "" + if refresh_from_remote: + try: + template_payload = save_local_host_compose_template(host_name) + except Exception as exc: # noqa: BLE001 + refresh_error = str(exc) + + if template_payload is None: + template_payload = load_local_host_compose_template(host_name) + template_payload["refreshError"] = refresh_error + + remote_path = compose_remote_path(host_name) + try: + remote_current = read_remote_text(host, remote_path) + except Exception: + remote_current = "" + + backup_path = "" + if remote_current and remote_current != template_payload["content"]: + backup_path = f"{remote_path}.hyapp.{utc_now().replace(':', '').replace('+', '_')}.bak" + write_remote_text(host, backup_path, remote_current) + + changed = remote_current != template_payload["content"] + if changed: + write_remote_text(host, remote_path, template_payload["content"]) + + remote_after = read_remote_text(host, remote_path) + if remote_after != template_payload["content"]: + raise RuntimeError(f"compose verify content mismatch on {host_name}") + + remote_services = verify_remote_compose(host, remote_path) + return { + "ok": True, + "host": host_name, + "services": template_payload["services"], + "updatedAt": utc_now(), + "path": remote_path, + "templatePath": template_payload["path"], + "changed": changed, + "backupPath": backup_path, + "remoteServices": remote_services, + "source": template_payload.get("source") or "local-cache", + "refreshError": template_payload.get("refreshError") or "", + } + + +def sync_remote_compose_templates(host_names: list[str], *, refresh_from_remote: bool = True) -> dict[str, Any]: + # 允许对一组主机做完整模板同步。 + normalized_names = [str(name or "").strip() for name in host_names if str(name or "").strip()] + normalized_names = list(dict.fromkeys(normalized_names)) + if not normalized_names: + normalized_names = managed_compose_host_names() + + items = [sync_remote_host_compose_template(host_name, refresh_from_remote=refresh_from_remote) for host_name in normalized_names] + return { + "ok": True, + "updatedAt": utc_now(), + "hosts": items, + } diff --git a/monitor/config.py b/monitor/config.py index c835e32..3f4088e 100644 --- a/monitor/config.py +++ b/monitor/config.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os from pathlib import Path from typing import Any @@ -9,6 +10,8 @@ from .env import env_float, env_int, env_int_list, env_str, load_env_file # 项目根目录。 ROOT = Path(__file__).resolve().parents[1] +# 常见工作区根目录，便于本地开发时自动发现业务仓。 +WORKSPACE_ROOT = ROOT.parent # 静态资源目录。 STATIC_DIR = ROOT / "static" # 默认 .env 路径。 @@ -45,6 +48,10 @@ ROLLING_RESTART_TIMEOUT_SECONDS = env_int("ROLLING_RESTART_TIMEOUT_SECONDS", 180 ROLLING_RESTART_POLL_SECONDS = env_float("ROLLING_RESTART_POLL_SECONDS", 2.0) # 单实例恢复后额外稳定等待秒数。 ROLLING_RESTART_STABILIZE_SECONDS = env_float("ROLLING_RESTART_STABILIZE_SECONDS", 2.0) +# 服务更新时本地构建最长等待秒数。 +SERVICE_BUILD_TIMEOUT_SECONDS = env_int("SERVICE_BUILD_TIMEOUT_SECONDS", 3600) +# 服务更新时目标机预拉镜像最长等待秒数。 +SERVICE_PULL_TIMEOUT_SECONDS = env_int("SERVICE_PULL_TIMEOUT_SECONDS", 600) # TAT 所在区域。 DEPLOY_REGION = env_str("DEPLOY_REGION", "me-saudi-arabia") @@ -53,6 +60,22 @@ TENCENT_SECRET_ID = env_str("TENCENT_SECRET_ID", "") # 腾讯云 SecretKey。 TENCENT_SECRET_KEY = env_str("TENCENT_SECRET_KEY", "") + +def pick_existing_path(candidates: list[Path]) -> Path: + # 配置未显式指定时，优先挑当前机器上真正存在的路径。 + for candidate in candidates: + if candidate.exists(): + return candidate + return candidates[0] + + +def env_or_candidate_path(env_key: str, candidates: list[Path]) -> Path: + # 允许 .env 显式覆盖，也兼容本地开发和远端 deploy 两种目录布局。 + override = os.environ.get(env_key, "").strip() + if override: + return Path(override).expanduser() + return pick_existing_path(candidates) + # 页面默认刷新间隔。 DEFAULT_REFRESH_INTERVAL_SECONDS = env_int("DEFAULT_REFRESH_INTERVAL_SECONDS", 10) # 页面可选刷新间隔。 @@ -79,6 +102,37 @@ NACOS_USERNAME = env_str("NACOS_USERNAME", "") # Nacos 可选密码。 NACOS_PASSWORD = env_str("NACOS_PASSWORD", "") +# Harbor 镜像仓地址，服务更新会推送镜像并让目标机主动拉取。 +HARBOR_REGISTRY = env_str("HARBOR_REGISTRY", "10.2.1.3:18082") +# Harbor 项目名。 +HARBOR_PROJECT = env_str("HARBOR_PROJECT", "likei") +# Harbor 登录用户名。 +HARBOR_USERNAME = env_str("HARBOR_USERNAME", "") +# Harbor 登录密码。 +HARBOR_PASSWORD = env_str("HARBOR_PASSWORD", "") +# Docker 构建平台，和现有 Jenkins/构建脚本口径保持一致。 +BUILD_DOCKER_PLATFORM = env_str("BUILD_DOCKER_PLATFORM", "linux/amd64") +# Java 服务默认拉取的 Git ref。 +UPDATE_DEFAULT_JAVA_GIT_REF = env_str("UPDATE_DEFAULT_JAVA_GIT_REF", "main") +# Go 服务默认拉取的 Git ref。 +UPDATE_DEFAULT_GO_GIT_REF = env_str("UPDATE_DEFAULT_GO_GIT_REF", "main") +# Java 仓库路径，默认兼容本地工作区和远端 deploy 机。 +JAVA_REPO_ROOT = env_or_candidate_path( + "JAVA_REPO_ROOT", + [ + WORKSPACE_ROOT / "chatapp3" / "chatapp3-java", + Path("/opt/deploy/sources/chatapp3-java"), + ], +) +# Go 仓库路径，默认兼容本地工作区和远端 deploy 机。 +GOLANG_REPO_ROOT = env_or_candidate_path( + "GOLANG_REPO_ROOT", + [ + WORKSPACE_ROOT / "chatapp3" / "chatapp3-golang", + Path("/opt/deploy/sources/chatapp3-golang"), + ], +) + # Mongo 连接串。 MONGO_URI = env_str("MONGO_URI", "") # Mongo 默认关注的业务库名。 diff --git a/monitor/http_app.py b/monitor/http_app.py index d6e4fee..2472e4c 100644 --- a/monitor/http_app.py +++ b/monitor/http_app.py @@ -15,6 +15,7 @@ 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_ops import build_restart_targets_payload, rolling_restart_payload @@ -149,6 +150,20 @@ class MonitorHandler(BaseHTTPRequestHandler): send_body=send_body, ) + # 返回可构建并发布的业务服务。 + if parsed.path == "/api/ops/update-targets": + try: + payload = build_update_targets_payload() + 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) @@ -180,6 +195,9 @@ class MonitorHandler(BaseHTTPRequestHandler): if parsed.path == "/api/ops/rolling-restart": return self.handle_rolling_restart(payload) + if parsed.path == "/api/ops/update-services": + return self.handle_update_services(payload) + return self.send_error_payload(HTTPStatus.NOT_FOUND, "not found", send_body=True) def handle_nacos_validate(self, payload: dict[str, Any]) -> None: @@ -284,6 +302,34 @@ class MonitorHandler(BaseHTTPRequestHandler): send_body=True, ) + def handle_update_services(self, payload: dict[str, Any]) -> None: + services = payload.get("services") + if not isinstance(services, list): + return self.send_error_payload(HTTPStatus.BAD_REQUEST, "services must be a list", send_body=True) + + java_git_ref = str(payload.get("javaGitRef") or "").strip() + go_git_ref = str(payload.get("goGitRef") or "").strip() + image_tag = str(payload.get("imageTag") or "").strip() + skip_maven_build = bool(payload.get("skipMavenBuild")) + + try: + result = deploy_updated_services_payload( + services, + java_git_ref=java_git_ref, + go_git_ref=go_git_ref, + image_tag=image_tag, + skip_maven_build=skip_maven_build, + ) + except Exception as exc: # noqa: BLE001 + return self.send_error_payload(HTTPStatus.BAD_GATEWAY, str(exc), send_body=True) + + return self.send_payload( + HTTPStatus.OK, + response_json(result), + "application/json; charset=utf-8", + send_body=True, + ) + def query_value(self, query: dict[str, list[str]], key: str) -> str: # 统一读取单值 query 参数。 return str((query.get(key) or [""])[0]).strip() diff --git a/monitor/service_ops.py b/monitor/service_ops.py index a861bd8..8875845 100644 --- a/monitor/service_ops.py +++ b/monitor/service_ops.py @@ -6,6 +6,7 @@ import shlex import time from typing import Any +from .compose_templates import sync_remote_host_compose_template from .config import ( ROLLING_RESTART_POLL_SECONDS, ROLLING_RESTART_STABILIZE_SECONDS, @@ -194,6 +195,7 @@ def rolling_restart_payload(service_names: list[str]) -> dict[str, Any]: records_map = service_records_by_name() steps: list[dict[str, Any]] = [] + prepared_hosts: dict[str, dict[str, Any]] = {} for service_name in normalized_names: if service_name not in records_map: @@ -210,6 +212,18 @@ def rolling_restart_payload(service_names: list[str]) -> dict[str, Any]: } try: + # 单机第一次进入重启前，先把完整 compose 模板同步到远端。 + if service["host"] not in prepared_hosts: + prepared_hosts[service["host"]] = sync_remote_host_compose_template( + service["host"], + refresh_from_remote=True, + ) + step["compose"] = { + "path": prepared_hosts[service["host"]]["path"], + "changed": bool(prepared_hosts[service["host"]].get("changed")), + "source": str(prepared_hosts[service["host"]].get("source") or ""), + } + restart_result = restart_service_instance(service) step["tatStatus"] = restart_result.get("status") step["tatOutput"] = str(restart_result.get("output") or "").strip()[:500] diff --git a/monitor/service_release.py b/monitor/service_release.py new file mode 100644 index 0000000..94226ff --- /dev/null +++ b/monitor/service_release.py @@ -0,0 +1,766 @@ +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 time +from typing import Any + +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, + 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", +} + + +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, +) -> 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: + 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, + ) + try: + return_code = process.wait(timeout=timeout_seconds) + except subprocess.TimeoutExpired as exc: + process.kill() + process.wait() + raise RuntimeError(f"{label} timed out: {tail_text(log_path)}") from exc + + 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 resolve_repo_commit(repo_root: Path, git_ref: str, label: str) -> str: + # 优先把分支解析到 origin/，避免本地旧分支 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) -> None: + # 本机构建前先登录 Harbor，避免 push 时才失败。 + require_harbor_credentials() + 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, + ) + + +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) -> 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 + fetch_repo_refs(repo_root, "java") + commit = resolve_repo_commit(repo_root, resolved_ref, "java") + worktree_root = create_temp_worktree(repo_root, commit, "java") + 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: + 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), + ) + 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, + ) + + for service_name in service_names: + repository_ref = service_repository_ref(service_name) + target_image_ref = f"{repository_ref}:{resolved_image_tag}" + 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", + }, + timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS, + ) + 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, + ) + digest_ref = docker_digest(target_image_ref, cwd=worktree_root) + 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) + + 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) -> 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 + fetch_repo_refs(repo_root, "golang") + commit = resolve_repo_commit(repo_root, resolved_ref, "golang") + worktree_root = create_temp_worktree(repo_root, commit, "golang") + 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}" + 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, + ) + 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, + ) + digest_ref = docker_digest(target_image_ref, cwd=worktree_root) + 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) + + 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, +) -> 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")) + + 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) + 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) + 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, +) -> 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}") + + 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, + ) + service_images = dict(build_result.get("serviceImages") or {}) + + 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])) + + 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: + compose_result = sync_remote_host_compose_template(host_entry["host"], refresh_from_remote=False) + prepull_result = prepull_host_images(host_entry) + 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: dict[str, Any] = { + "service": service_name, + "host": service["host"], + "ip": service["ip"], + "kind": service["kind"], + "image": service_images[service_name], + "status": "running", + } + try: + restart_result = restart_service_instance(service) + step["tatStatus"] = restart_result.get("status") + step["tatOutput"] = str(restart_result.get("output") or "").strip()[:500] + + 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 ""), + } + + if service["kind"] == "java": + step["nacos"] = wait_java_registration_ready( + service, + timeout_seconds=ROLLING_RESTART_TIMEOUT_SECONDS, + ) + + if ROLLING_RESTART_STABILIZE_SECONDS > 0: + time.sleep(ROLLING_RESTART_STABILIZE_SECONDS) + + step["status"] = "success" + step["durationSeconds"] = round(time.time() - started_at, 2) + steps.append(step) + except Exception as exc: # noqa: BLE001 + step["status"] = "failed" + step["error"] = str(exc) + step["durationSeconds"] = round(time.time() - started_at, 2) + steps.append(step) + 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() + + 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", + } diff --git a/ops/sync_runtime_compose.py b/ops/sync_runtime_compose.py new file mode 100644 index 0000000..69014ca --- /dev/null +++ b/ops/sync_runtime_compose.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + + +# 直接把项目根目录加到 import path，脚本可以从任意 cwd 调起。 +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from monitor.compose_templates import managed_compose_host_names, save_local_host_compose_template, sync_remote_compose_templates + + +def parse_args() -> argparse.Namespace: + # 默认只刷新本地完整模板，不主动改线上 compose。 + parser = argparse.ArgumentParser(description="refresh full docker compose templates for runtime hosts") + parser.add_argument("--hosts", nargs="*", default=[], help="target host names, default all managed service hosts") + parser.add_argument( + "--sync-remote", + action="store_true", + help="after refreshing local templates, also sync full compose files to remote hosts", + ) + return parser.parse_args() + + +def main() -> None: + args = parse_args() + host_names = [host_name for host_name in (args.hosts or managed_compose_host_names()) if host_name] + + if args.sync_remote: + payload = sync_remote_compose_templates(host_names, refresh_from_remote=True) + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return + + items = [save_local_host_compose_template(host_name) for host_name in host_names] + print( + json.dumps( + { + "ok": True, + "updatedAt": items[-1]["updatedAt"] if items else "", + "hosts": items, + }, + ensure_ascii=False, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/static/app.css b/static/app.css index 2daa211..1a3ac49 100644 --- a/static/app.css +++ b/static/app.css @@ -527,6 +527,39 @@ body { min-width: 220px; } +.update-form-grid { + display: grid; + grid-template-columns: repeat(4, minmax(0, 1fr)); + gap: 12px; + margin-top: 14px; +} + +.update-field, +.update-check { + display: flex; + flex-direction: column; + gap: 8px; +} + +.update-field span, +.update-check span { + color: var(--muted); + font-size: 12px; +} + +.update-check { + justify-content: center; + padding: 12px 14px; + border: 1px solid var(--line); + border-radius: 18px; + background: color-mix(in oklab, var(--panel-strong) 92%, transparent); +} + +.update-check input { + width: 18px; + height: 18px; +} + .config-layout { display: grid; grid-template-columns: minmax(260px, 340px) minmax(0, 1fr); @@ -664,6 +697,10 @@ body { .config-meta-grid { grid-template-columns: repeat(2, minmax(0, 1fr)); } + + .update-form-grid { + grid-template-columns: repeat(2, minmax(0, 1fr)); + } } @media (max-width: 720px) { @@ -683,7 +720,8 @@ body { .host-metrics, .summary-grid, .metric-grid, - .config-meta-grid { + .config-meta-grid, + .update-form-grid { grid-template-columns: 1fr; } diff --git a/static/app.js b/static/app.js index 805022b..d1775a3 100644 --- a/static/app.js +++ b/static/app.js @@ -85,6 +85,30 @@ createApp({ diffKeys: [], hosts: [], }, + updateOps: { + loading: false, + error: "", + running: false, + message: "", + items: [], + selectedServices: [], + javaGitRef: "main", + goGitRef: "main", + imageTag: "", + skipMavenBuild: false, + defaults: { + javaGitRef: "main", + goGitRef: "main", + dockerPlatform: "linux/amd64", + }, + builder: { + harborRegistry: "", + harborProject: "", + javaRepoRoot: "", + golangRepoRoot: "", + }, + result: null, + }, restartOps: { loading: false, error: "", @@ -194,6 +218,9 @@ createApp({ restartTargetMap() { return Object.fromEntries((this.restartOps.items || []).map((item) => [item.service, item])); }, + updateTargetMap() { + return Object.fromEntries((this.updateOps.items || []).map((item) => [item.service, item])); + }, }, watch: { refreshIntervalSeconds() { @@ -284,6 +311,9 @@ createApp({ restartTargetLabel(item) { return `${item.service} · ${item.kind} · ${item.hosts.map((host) => host.host).join(" / ")}`; }, + updateTargetLabel(item) { + return `${item.service} · ${item.kind} · ${item.hosts.map((host) => host.host).join(" / ")}`; + }, syncRefreshSettings() { const settings = this.payload.settings || {}; const options = Array.isArray(settings.refreshIntervalOptions) @@ -606,6 +636,92 @@ createApp({ this.restartOps.loading = false; } }, + async refreshUpdateTargets() { + this.updateOps.loading = true; + this.updateOps.error = ""; + + try { + const response = await fetch("/api/ops/update-targets", { cache: "no-store" }); + const payload = await this.readJsonResponse(response); + if (!response.ok) { + throw new Error(payload.error || `HTTP ${response.status}`); + } + + this.updateOps.items = Array.isArray(payload.items) ? payload.items : []; + this.updateOps.defaults = payload.defaults || this.updateOps.defaults; + this.updateOps.builder = payload.builder || this.updateOps.builder; + + if (!this.updateOps.javaGitRef) { + this.updateOps.javaGitRef = this.updateOps.defaults.javaGitRef || "main"; + } else if (this.updateOps.javaGitRef === "main" && this.updateOps.defaults.javaGitRef) { + this.updateOps.javaGitRef = this.updateOps.defaults.javaGitRef; + } + + if (!this.updateOps.goGitRef) { + this.updateOps.goGitRef = this.updateOps.defaults.goGitRef || "main"; + } else if (this.updateOps.goGitRef === "main" && this.updateOps.defaults.goGitRef) { + this.updateOps.goGitRef = this.updateOps.defaults.goGitRef; + } + + const available = new Set(this.updateOps.items.map((item) => item.service)); + this.updateOps.selectedServices = this.updateOps.selectedServices.filter((service) => available.has(service)); + } catch (error) { + this.updateOps.error = error instanceof Error ? error.message : String(error); + } finally { + this.updateOps.loading = false; + } + }, + toggleUpdateService(serviceName) { + const next = new Set(this.updateOps.selectedServices); + if (next.has(serviceName)) { + next.delete(serviceName); + } else { + next.add(serviceName); + } + this.updateOps.selectedServices = [...next]; + }, + async runServiceUpdate() { + if (this.updateOps.selectedServices.length === 0) { + return; + } + + this.updateOps.running = true; + this.updateOps.error = ""; + this.updateOps.message = ""; + + try { + const response = await fetch("/api/ops/update-services", { + method: "POST", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify({ + services: this.updateOps.selectedServices, + javaGitRef: this.updateOps.javaGitRef, + goGitRef: this.updateOps.goGitRef, + imageTag: this.updateOps.imageTag, + skipMavenBuild: this.updateOps.skipMavenBuild, + }), + }); + const payload = await this.readJsonResponse(response); + if (!response.ok) { + 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(); + } catch (error) { + this.updateOps.error = error instanceof Error ? error.message : String(error); + } finally { + this.updateOps.running = false; + } + }, toggleRestartService(serviceName) { const next = new Set(this.restartOps.selectedServices); if (next.has(serviceName)) { @@ -663,6 +779,7 @@ createApp({ this.refresh(); this.refreshNacosConfigs({ preserveSelection: false, reloadSelected: true }); this.refreshGoConfig(); + this.refreshUpdateTargets(); this.refreshRestartTargets(); }, beforeUnmount() { diff --git a/static/index.html b/static/index.html index 0d86ceb..1f0b6b3 100644 --- a/static/index.html +++ b/static/index.html @@ -418,6 +418,156 @@ +
+

服务更新

+
+
+
+
+

Service Update

+

本机构建选中的 Java / Golang 服务，推到 Harbor，更新完整 compose，并只重建你选中的服务

+
+ + {{ updateOps.error ? '存在错误' : updateOps.running ? '执行中' : '可操作' }} + +
+ +
+
+ +
+
+ + +
+
+ +
+ + Harbor：{{ updateOps.builder.harborRegistry }} / {{ updateOps.builder.harborProject }} + + Java 仓：{{ updateOps.builder.javaRepoRoot }} + Go 仓：{{ updateOps.builder.golangRepoRoot }} + 平台：{{ updateOps.defaults.dockerPlatform }} +
+ +
+ + + + +
+ +

{{ updateOps.error }}

+

{{ updateOps.message }}

+ +
+ + + + + + + + + + + + + + + + + + + +
仓库类型Git RefCommitImage Tag服务
{{ build.kind }}{{ build.gitRef }}{{ build.commit }}{{ build.imageTag }}{{ (build.services || []).join(', ') }}
+
+ +
+ + + + + + + + + + + + + + + + + +
主机服务Compose预拉镜像
{{ item.host }}{{ (item.services || []).join(', ') }}{{ item.compose && item.compose.changed ? '已更新' : '未变化' }} + {{ item.prepull ? item.prepull.status : '-' }} +
+
+ +
+ + + + + + + + + + + + + + + + + + + + + + + +
服务主机镜像状态HTTPNacos耗时
{{ step.service }}{{ step.host }}{{ step.image }}{{ step.status }} + {{ step.http.statusCode }} / {{ step.http.latencyMs }}ms + - + + {{ step.nacos.source || 'ok' }} + - + {{ step.durationSeconds || '-' }}s
+
+
+

滚动重启