diff --git a/config/targets.json b/config/targets.json index fc2c71f..8ddfe25 100644 --- a/config/targets.json +++ b/config/targets.json @@ -90,6 +90,12 @@ "kind": "golang", "port": 2900, "path": "/health" + }, + { + "name": "dashboard-cdc-worker", + "kind": "golang", + "port": 2910, + "path": "/health" } ] }, diff --git a/core/config.py b/core/config.py index 0cc73be..cceda76 100644 --- a/core/config.py +++ b/core/config.py @@ -222,6 +222,21 @@ GOLANG_REPO_ROOT = env_or_candidate_path( Path("/opt/deploy/sources/chatapp3-golang"), ], ) +# Dashboard CDC Worker 仓库地址；仓库缺失时允许按这里自动 clone。 +DASHBOARD_CDC_REPO_URL = env_str("DASHBOARD_CDC_REPO_URL", "git@gitea.haiyihy.com:admin/dashboard-cdc-worker.git") +# Dashboard CDC Worker 仓库路径，默认兼容本地工作区和远端 deploy 机。 +DASHBOARD_CDC_REPO_ROOT = env_or_candidate_path( + "DASHBOARD_CDC_REPO_ROOT", + [ + WORKSPACE_ROOT / "chatapp3" / "dashboard-cdc-worker", + WORKSPACE_ROOT / "dashboard-cdc-worker", + Path("/opt/deploy/sources/dashboard-cdc-worker"), + ], +) +# Dashboard CDC Worker 在发布页和 compose 里的服务名。 +DASHBOARD_CDC_SERVICE_NAME = env_str("DASHBOARD_CDC_SERVICE_NAME", "dashboard-cdc-worker").strip() or "dashboard-cdc-worker" +# Dashboard CDC Worker 是否接入发布；默认启用。 +DASHBOARD_CDC_ENABLED = truthy(os.environ.get("DASHBOARD_CDC_ENABLED"), True) # Admin 前端仓库地址；仓库缺失时允许按这里自动 clone。 ADMIN_FRONTEND_REPO_URL = env_str("ADMIN_FRONTEND_REPO_URL", "") # Admin 前端仓库路径，默认兼容本地工作区和远端 deploy 机。 diff --git a/monitors/yumi/compose_templates.py b/monitors/yumi/compose_templates.py index 93731fd..2303ab3 100644 --- a/monitors/yumi/compose_templates.py +++ b/monitors/yumi/compose_templates.py @@ -10,7 +10,7 @@ import shlex import re from typing import Any -from core.config import ROOT, RUNTIME_BASE_DIR, load_hosts, utc_now +from core.config import DASHBOARD_CDC_SERVICE_NAME, HARBOR_PROJECT, HARBOR_REGISTRY, ROOT, RUNTIME_BASE_DIR, load_hosts, utc_now from .runtime_config import read_remote_text, write_remote_text from .ssh_remote import copy_remote_file, run_host_script_checked @@ -350,6 +350,32 @@ def build_service_definition( return service_definition +def build_dashboard_cdc_service_definition(host_name: str) -> dict[str, Any]: + # dashboard-cdc-worker 首次上线时还没有容器可 inspect，只能从约定生成模板。 + return { + "image": f"{HARBOR_REGISTRY}/{HARBOR_PROJECT}/dashboard-cdc-worker:bootstrap", + "container_name": f"likei-{host_name}-{DASHBOARD_CDC_SERVICE_NAME}", + "network_mode": "host", + "restart": "unless-stopped", + "stop_signal": "SIGTERM", + "env_file": ["./dashboard-cdc-worker.env"], + "pull_policy": "always", + "working_dir": "/application", + "entrypoint": [], + "command": ["/application/service"], + "healthcheck": { + "test": ["CMD-SHELL", "wget -qO- http://127.0.0.1:2910/health | grep -q '\"code\":\"ok\"'"], + "interval": "10s", + "timeout": "10s", + "retries": 30, + }, + "logging": { + "driver": "json-file", + "options": {"max-size": "200m", "max-file": "5"}, + }, + } + + def generate_host_compose_template(host_name: str) -> dict[str, Any]: # 生成模板时严格按 targets.json 里的服务顺序输出，后续 diff 更稳定。 host_map = managed_host_map() @@ -374,6 +400,9 @@ def generate_host_compose_template(host_name: str) -> dict[str, Any]: expected_container = container_name(host_name, service_name) inspect_item = inspect_by_name.get(expected_container) if inspect_item is None: + if service_name == DASHBOARD_CDC_SERVICE_NAME: + service_map[service_name] = build_dashboard_cdc_service_definition(host_name) + continue missing_services.append(service_name) continue service_map[service_name] = build_service_definition( diff --git a/monitors/yumi/service_release.py b/monitors/yumi/service_release.py index c5ddd4c..379d8cb 100644 --- a/monitors/yumi/service_release.py +++ b/monitors/yumi/service_release.py @@ -15,9 +15,11 @@ import tempfile import threading import time from typing import Any, Callable +from urllib.parse import quote, urlparse from .compose_templates import ( load_local_host_compose_model, + save_local_host_compose_model, sync_remote_host_compose_template, update_local_service_image, ) @@ -42,6 +44,10 @@ from core.config import ( ADMIN_FRONTEND_SSH_USE_SUDO, ADMIN_FRONTEND_SSH_USER, BUILD_DOCKER_PLATFORM, + DASHBOARD_CDC_ENABLED, + DASHBOARD_CDC_REPO_ROOT, + DASHBOARD_CDC_REPO_URL, + DASHBOARD_CDC_SERVICE_NAME, GATEWAY_CLB_DRAIN_SECONDS, GOLANG_REPO_ROOT, HARBOR_PASSWORD, @@ -54,6 +60,7 @@ from core.config import ( ROOT, ROLLING_RESTART_STABILIZE_SECONDS, ROLLING_RESTART_TIMEOUT_SECONDS, + RUNTIME_BASE_DIR, SERVICE_BUILD_CACHE_ROOT, SERVICE_BUILD_TIMEOUT_SECONDS, SERVICE_PULL_TIMEOUT_SECONDS, @@ -86,6 +93,8 @@ from .tencent_clb import ( JAVA_BUILDABLE_SERVICES = ("auth", "gateway", "external", "console", "other", "live", "wallet", "order") # Go 仓目前只有主业务服务。 GOLANG_BUILDABLE_SERVICES = ("golang",) +# Dashboard CDC Worker 是独立 Go 仓库，但部署形态仍是普通镜像服务。 +DASHBOARD_CDC_BUILDABLE_SERVICES = (DASHBOARD_CDC_SERVICE_NAME,) if DASHBOARD_CDC_ENABLED else () # Admin 前端当前只接入一个发布目标。 FRONTEND_BUILDABLE_SERVICES = (ADMIN_FRONTEND_SERVICE_NAME,) # Java 批量 Maven 构建时，需要映射回对应的启动模块。 @@ -110,6 +119,7 @@ SERVICE_IMAGE_REPOSITORY_SUFFIXES = { "wallet": "java/chatapp3-wallet", "order": "order", "golang": "golang", + DASHBOARD_CDC_SERVICE_NAME: "dashboard-cdc-worker", } # 后台任务的日志回调统一走这个签名。 @@ -173,7 +183,7 @@ def stream_log_file_lines( def buildable_service_names() -> list[str]: # 页面和接口统一按这里判定哪些服务允许“构建并更新”。 - names = [*JAVA_BUILDABLE_SERVICES, *GOLANG_BUILDABLE_SERVICES] + names = [*JAVA_BUILDABLE_SERVICES, *GOLANG_BUILDABLE_SERVICES, *DASHBOARD_CDC_BUILDABLE_SERVICES] if frontend_release_enabled(): names.extend(service_name for service_name in FRONTEND_BUILDABLE_SERVICES if str(service_name or "").strip()) return names @@ -191,6 +201,8 @@ def service_kind(service_name: str) -> str: return "java" if service_name in GOLANG_BUILDABLE_SERVICES: return "golang" + if service_name in DASHBOARD_CDC_BUILDABLE_SERVICES: + return "golang" if frontend_release_enabled() and service_name in FRONTEND_BUILDABLE_SERVICES: return "frontend" raise RuntimeError(f"unsupported update service: {service_name}") @@ -227,6 +239,27 @@ def ensure_frontend_repo_root(*, progress: ProgressLogger | None = None) -> Path return ensure_repo_root(repo_root, "frontend") +def ensure_dashboard_cdc_repo_root(*, progress: ProgressLogger | None = None) -> Path: + # Dashboard CDC Worker 是新独立仓库，deploy 机第一次使用时自动 clone。 + repo_root = Path(DASHBOARD_CDC_REPO_ROOT).expanduser().resolve() + if (repo_root / ".git").exists(): + return repo_root + + repo_url = str(DASHBOARD_CDC_REPO_URL or "").strip() + if not repo_url: + raise RuntimeError("missing DASHBOARD_CDC_REPO_URL in .env") + + repo_root.parent.mkdir(parents=True, exist_ok=True) + emit_progress(progress, "build.dashboard-cdc.repo", f"初始化 Dashboard CDC 仓库：{repo_url} -> {repo_root}") + local_command_output( + ["git", "clone", repo_url, str(repo_root)], + cwd=repo_root.parent, + label="dashboard cdc clone repo", + timeout_seconds=900, + ) + return ensure_repo_root(repo_root, "dashboard-cdc") + + def local_command_output( command: list[str], *, @@ -1043,6 +1076,80 @@ def build_golang_images( } +def build_dashboard_cdc_images( + service_names: list[str], + git_ref: str, + image_tag: str, + *, + progress: ProgressLogger | None = None, +) -> dict[str, Any]: + # Dashboard CDC Worker 使用独立 Go 仓库和同名 ci/build-image.sh。 + repo_root = ensure_dashboard_cdc_repo_root(progress=progress) + resolved_ref = str(git_ref or "").strip() or UPDATE_DEFAULT_GO_GIT_REF + emit_progress(progress, "build.dashboard-cdc", f"刷新 Dashboard CDC 仓库 refs：{resolved_ref}") + fetch_repo_refs(repo_root, "dashboard-cdc") + commit = resolve_repo_commit(repo_root, resolved_ref, "dashboard-cdc") + emit_progress(progress, "build.dashboard-cdc", f"Dashboard CDC ref 已解析：{resolved_ref} -> {commit[:12]}") + worktree_root = create_temp_worktree(repo_root, commit, "dashboard-cdc") + emit_progress(progress, "build.dashboard-cdc", f"已创建 Dashboard CDC 临时 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.dashboard-cdc.{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"dashboard cdc build {service_name}", + env={"DOCKER_PLATFORM": BUILD_DOCKER_PLATFORM}, + timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS, + progress_stage=f"build.dashboard-cdc.{service_name}.build", + progress=progress, + ) + emit_progress(progress, f"build.dashboard-cdc.{service_name}", f"开始推送 Harbor：{target_image_ref}") + push_tail = run_logged_command( + ["docker", "push", target_image_ref], + cwd=worktree_root, + label=f"dashboard cdc push {service_name}", + timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS, + progress_stage=f"build.dashboard-cdc.{service_name}.push", + progress=progress, + ) + digest_ref = docker_digest(target_image_ref, cwd=worktree_root) + emit_progress(progress, f"build.dashboard-cdc.{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.dashboard-cdc", f"已清理 Dashboard CDC worktree：{worktree_root}") + + return { + "kind": "dashboard-cdc", + "repoRoot": str(repo_root), + "repoUrl": str(DASHBOARD_CDC_REPO_URL or "").strip(), + "gitRef": resolved_ref, + "commit": commit, + "imageTag": resolved_image_tag, + "services": list(service_names), + "items": outputs, + "serviceImages": service_images, + } + + def build_frontend_bundle( service_name: str, git_ref: str, @@ -1130,15 +1237,23 @@ def build_service_images( # 一个请求里可能同时选了 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"] + dashboard_cdc_services = [service_name for service_name in normalized if service_name in DASHBOARD_CDC_BUILDABLE_SERVICES] + golang_services = [ + service_name + for service_name in normalized + if service_kind(service_name) == "golang" and service_name not in DASHBOARD_CDC_BUILDABLE_SERVICES + ] 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, - ) + if java_services: + login_repo = ensure_repo_root(JAVA_REPO_ROOT, "build") + elif golang_services: + login_repo = ensure_repo_root(GOLANG_REPO_ROOT, "build") + else: + login_repo = ensure_dashboard_cdc_repo_root(progress=progress) + docker_login_local(login_repo, progress=progress) builds: list[dict[str, Any]] = [] service_images: dict[str, str] = {} @@ -1156,6 +1271,15 @@ def build_service_images( golang_build = build_golang_images(golang_services, go_git_ref, image_tag, progress=progress) builds.append(golang_build) service_images.update(golang_build["serviceImages"]) + if dashboard_cdc_services: + dashboard_cdc_build = build_dashboard_cdc_images( + dashboard_cdc_services, + go_git_ref, + image_tag, + progress=progress, + ) + builds.append(dashboard_cdc_build) + service_images.update(dashboard_cdc_build["serviceImages"]) return { "builds": builds, @@ -1231,6 +1355,101 @@ def prepull_host_images(host_entry: dict[str, Any]) -> dict[str, Any]: } +def dashboard_cdc_env_path(host_name: str) -> str: + return f"{RUNTIME_BASE_DIR.rstrip('/')}/{host_name}/dashboard-cdc-worker.env" + + +def render_dashboard_cdc_env() -> str: + # 运行时配置从 Nacos 读取并写到目标机 env 文件，不把数据库密钥提交进 Git。 + from .nacos import current_namespace, get_config_content + + namespace_id, _ = current_namespace() + content = get_config_content(namespace_id, "common", "rds-config.yml") + + def yaml_value(key: str) -> str: + match = re.search(rf"(?im)^\s*{re.escape(key)}\s*:\s*(.*?)\s*$", content) + if not match: + raise RuntimeError(f"missing {key} in common/rds-config.yml") + value = match.group(1).strip() + if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")): + value = value[1:-1] + return value + + jdbc_url = ( + yaml_value("url") + .replace("jdbc:p6spy:mysql://", "mysql://", 1) + .replace("jdbc:mysql://", "mysql://", 1) + ) + parsed = urlparse(jdbc_url) + if not parsed.hostname: + raise RuntimeError("invalid rds-config url for dashboard cdc worker") + + username = yaml_value("username") + password = yaml_value("password") + database = parsed.path.lstrip("/") or "likei" + addr = f"{parsed.hostname}:{parsed.port or 3306}" + dsn = ( + f"{quote(username, safe='')}:{quote(password, safe='')}@tcp({addr})/{database}" + "?charset=utf8mb4&parseTime=true&loc=Asia%2FRiyadh" + ) + return "\n".join( + [ + f"CDC_MYSQL_ADDR={addr}", + f"CDC_MYSQL_USER={username}", + f"CDC_MYSQL_PASSWORD={password}", + f"DASHBOARD_MYSQL_DSN={dsn}", + "CDC_START_MODE=checkpoint", + "CDC_SERVER_ID=29301", + "CDC_DRY_RUN=false", + "DASHBOARD_SYS_ORIGINS=LIKEI", + "DASHBOARD_STAT_TIMEZONES=Asia/Riyadh", + "DASHBOARD_STORAGE_TIMEZONE=Asia/Riyadh", + "REDIS_ENABLED=false", + "HEALTH_ADDR=:2910", + "", + ] + ) + + +def ensure_dashboard_cdc_remote_env(host_entry: dict[str, Any]) -> dict[str, Any]: + env_text = render_dashboard_cdc_env() + env_path = dashboard_cdc_env_path(host_entry["host"]) + write_remote_text(host_entry, env_path, env_text, timeout_seconds=30) + return {"path": env_path, "changed": True} + + +def ensure_dashboard_cdc_compose_service(host_name: str) -> dict[str, Any] | None: + # 新服务首次上线前远端还没有运行容器，不能依赖 docker inspect 自动生成模板。 + model = load_local_host_compose_model(host_name) + service_map = dict(model.get("services") or {}) + if DASHBOARD_CDC_SERVICE_NAME in service_map: + return None + + service_map[DASHBOARD_CDC_SERVICE_NAME] = { + "image": f"{HARBOR_REGISTRY}/{HARBOR_PROJECT}/dashboard-cdc-worker:bootstrap", + "container_name": f"likei-{host_name}-{DASHBOARD_CDC_SERVICE_NAME}", + "network_mode": "host", + "restart": "unless-stopped", + "stop_signal": "SIGTERM", + "env_file": ["./dashboard-cdc-worker.env"], + "pull_policy": "always", + "working_dir": "/application", + "entrypoint": [], + "command": ["/application/service"], + "healthcheck": { + "test": ["CMD-SHELL", "wget -qO- http://127.0.0.1:2910/health | grep -q '\"code\":\"ok\"'"], + "interval": "10s", + "timeout": "10s", + "retries": 30, + }, + "logging": { + "driver": "json-file", + "options": {"max-size": "200m", "max-file": "5"}, + }, + } + return save_local_host_compose_model(host_name, service_map) + + def prepare_frontend_remote_tmp(host: dict[str, Any]) -> str: # 先在远端站点目录旁边建一份临时上传目录，避免直接覆盖线上内容。 site_dir = str(ADMIN_FRONTEND_SITE_DIR or "").rstrip("/") @@ -1526,6 +1745,8 @@ def build_update_targets_payload() -> dict[str, Any]: "harborProject": HARBOR_PROJECT, "javaRepoRoot": str(Path(JAVA_REPO_ROOT).expanduser()), "golangRepoRoot": str(Path(GOLANG_REPO_ROOT).expanduser()), + "dashboardCdcRepoRoot": str(Path(DASHBOARD_CDC_REPO_ROOT).expanduser()), + "dashboardCdcRepoUrl": str(DASHBOARD_CDC_REPO_URL or "").strip(), "frontendRepoRoot": str(Path(ADMIN_FRONTEND_REPO_ROOT).expanduser()), "frontendRepoUrl": str(ADMIN_FRONTEND_REPO_URL or "").strip(), "frontendPublicUrl": ADMIN_FRONTEND_PUBLIC_URL, @@ -1572,6 +1793,10 @@ def deploy_runtime_services_payload( for service_name in normalized: for record in records_map[service_name]: image_ref = service_images[service_name] + if service_name in DASHBOARD_CDC_BUILDABLE_SERVICES: + ensure_result = ensure_dashboard_cdc_compose_service(record["host"]) + if ensure_result is not None: + template_updates.append(ensure_result) 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)) @@ -1581,6 +1806,10 @@ def deploy_runtime_services_payload( try: for host_entry in host_plan: + if any(service_name in DASHBOARD_CDC_BUILDABLE_SERVICES for service_name in host_entry.get("services") or []): + emit_progress(progress, f"compose.env.{host_entry['host']}", "写入 Dashboard CDC Worker 运行配置") + env_result = ensure_dashboard_cdc_remote_env(host_entry) + emit_progress(progress, f"compose.env.{host_entry['host']}", f"运行配置已写入：{env_result['path']}") 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(