add dashboard cdc worker deployment

config/targets.json

@@ -90,6 +90,12 @@
                     "kind": "golang",
                     "port": 2900,
                     "path": "/health"
+                },
+                {
+                    "name": "dashboard-cdc-worker",
+                    "kind": "golang",
+                    "port": 2910,
+                    "path": "/health"
                 }
             ]
         },

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 机。

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(

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(
+    if java_services:
-        ensure_repo_root(JAVA_REPO_ROOT if java_services else GOLANG_REPO_ROOT, "build"),
+        login_repo = ensure_repo_root(JAVA_REPO_ROOT, "build")
-        progress=progress,
+    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(