diff --git a/docs/19_domainCheck_国内外联调交接说明.md b/docs/19_domainCheck_国内外联调交接说明.md new file mode 100644 index 0000000..6dafa88 --- /dev/null +++ b/docs/19_domainCheck_国内外联调交接说明.md @@ -0,0 +1,404 @@ +# domainCheck 国内外联调交接说明 + +## 1. 当前目标 + +本次联调的目标是把现有架构整理成下面这条可运行链路: + +- 海外机保留主数据与后台展示 +- 海外机把待检测任务批次下发到国内 controller +- 国内 controller 落本地任务与本地 `domains` +- 国内 worker 执行检测 +- 国内运行时状态、检测结果再同步回海外 +- 海外后台统一查看总体进度与结果 + +当前主链路已经基本跑通,剩下的是: + +- 海外 cluster/readiness 视图的最后收尾 +- 文档固化 +- 最好增加“国内日志回调到海外”的调试手段,减少人工复制日志 + +--- + +## 2. 当前已确认跑通的链路 + +### 2.1 海外 API 正常 + +海外 API 已经具备: + +- `/health` +- `/api/v1/runtime/cluster` +- `/api/v1/runtime/readiness` +- `/api/v1/runtime/preflight` +- `/api/v1/runtime/sync-summary` + +并且 nginx 对 `/api/v1/runtime/*` 的转发已经修好。 + +### 2.2 国内 controller 正常 + +国内 controller 这边已经确认: + +- `domaincheck-api` 正常运行 +- `domaincheck-sync-agent` 正常运行 +- 本地 PostgreSQL 可用 +- 本地 Redis 可用 +- `detect/start` 不再受 `sudo: a password is required` 影响 + +### 2.3 国内 worker 正常 + +国内 worker 这边已经确认: + +- `domaincheck-worker` 正常运行 +- 可以连接国内 controller 的 PostgreSQL +- 可以连接国内 controller 的 Redis +- Redis 密码配置后,Worker 控制指令已能正确消费 + +### 2.4 海外任务下发到国内已跑通 + +已实现并验证: + +- 海外导出待检测任务批次 +- 国内 controller 主动拉取任务批次 +- 国内本地写入 `domains` +- 国内本地生成 `detect_jobs` / `detect_job_items` +- 国内 worker 已开始执行实际检测任务 + +实际验证过的现象: + +- 国内 `domains` 从 `0` 变成 `200` +- `detect/start` 能创建任务 +- `job_id=1` 已进入 `running` +- `items_completed`、`items_running`、`items_claimed` 已持续变化 + +### 2.5 结果与运行时同步部分已跑通 + +已确认: + +- 国内 `runtime_projection` 已成功推送到海外 +- 国内 `detect_result_projection` 已开始生成 +- 海外已能看到新的大陆 controller 节点在线 + +--- + +## 3. 本次实际修改过的核心代码 + +以下文件已经被修改,是后续接手时最需要优先查看的: + +### 3.1 任务下发/同步相关 + +- [domain-api/app/services/sync_push_service.py](/www/wwwroot/getDomain/domain-api/app/services/sync_push_service.py:1) +- [domain-api/app/services/sync_record_service.py](/www/wwwroot/getDomain/domain-api/app/services/sync_record_service.py:1) +- [domain-api/app/api/routes/runtime.py](/www/wwwroot/getDomain/domain-api/app/api/routes/runtime.py:1) +- [domain-api/app/sync_agent.py](/www/wwwroot/getDomain/domain-api/app/sync_agent.py:1) + +新增/补充的能力包括: + +- `runtime_projection` 推送与接收 +- `detect_result_projection` 推送与接收 +- `detect_task_projection` 任务批次导出 +- 国内 `pull_detect_task_batch_now()` +- 海外 `task-export` +- 海外 `task-ack` +- 国内任务批次入库逻辑 + +### 3.2 任务创建与控制 + +- [domain-api/app/services/detect_job_service.py](/www/wwwroot/getDomain/domain-api/app/services/detect_job_service.py:1) +- [domain-api/app/services/worker_control_service.py](/www/wwwroot/getDomain/domain-api/app/services/worker_control_service.py:1) +- [domain-api/app/services/runtime_control_service.py](/www/wwwroot/getDomain/domain-api/app/services/runtime_control_service.py:1) +- [domain-api/app/api/routes/detect.py](/www/wwwroot/getDomain/domain-api/app/api/routes/detect.py:1) + +关键变化: + +- 本地没有任务时,`create_detect_job_if_needed()` 会先尝试从海外拉一批任务 +- Linux `systemd` 模式下不再强依赖 `sudo -n` +- 增加了 `pull_tasks` 运行时动作 + +### 3.3 运行时节点与 cluster 视图 + +- [domain-api/app/services/cluster_runtime_service.py](/www/wwwroot/getDomain/domain-api/app/services/cluster_runtime_service.py:1) + +关键变化: + +- 支持导入远端 runtime 节点 +- 支持清理历史 `*-imported` 脏节点 +- 正在补齐“海外显示大陆 worker 节点”的投影逻辑 + +### 3.4 前面已经做过的部署类改动 + +这些文件之前也已经做过较多修复,交接时可一起复核: + +- `domain-api/deploy/multi-region/install_overseas_quick.sh` +- `domain-api/deploy/multi-region/install_mainland_controller_quick.sh` +- `domain-api/deploy/multi-region/install_mainland_worker_quick.sh` +- `docs/17_domainCheck_全流程部署实操手册.md` +- `docs/18_domainCheck_CentOS9一键复制部署与更新文档.md` + +--- + +## 4. 当前仍未完全收尾的问题 + +### 4.1 海外 cluster 里大陆 worker 显示不完整 + +当前已经能在海外看到: + +- `mainland-controller-01` + +但一度没有稳定看到: + +- `mainland-worker-01` + +原因: + +- 海外主要是根据大陆 controller 回传的 `runtime_projection` 做远端节点物化 +- 之前 projection 里缺少 `active_job.node_stats` +- 后续已补代码,让海外可根据 `node_stats` 额外注册 worker 节点 + +接手人需要重点验证: + +- 海外 `/api/v1/runtime/cluster` 中是否出现 `mainland-worker-01` +- `online_worker_nodes` 是否从 `0` 变为 `1` + +### 4.2 历史脏节点 `mainland-control-imported` + +这是早期调试阶段遗留的 imported 节点。 + +现象: + +- 海外 `readiness` / `cluster` 曾显示: + - `mainland-control-imported` + - `status: offline` + +影响: + +- 不影响主流程执行 +- 但会污染 readiness / cluster 展示 + +已经补了清理逻辑,但需要继续验证是否彻底清干净。 + +### 4.3 结果回传展示还需要继续观察 + +当前已经看到: + +- `detect_result_projection` 在国内生成 +- 海外 `sync-summary` 中出现结果相关记录 + +还需要继续确认: + +- 海外是否稳定生成 `detect_result_ingest` +- 海外业务展示页/结果页是否正确反映国内执行结果 +- 海外主库中的业务表是否已被正确更新 + +--- + +## 5. 两边部署时容易踩的坑 + +### 5.1 root 和 www 的职责不要混 + +建议约定: + +- `git pull` / 仓库代码操作:用 `www` +- `systemctl restart ...`:用 `root` + +原因: + +- `root` 在当前环境下经常遇到 git safe.directory / ssh / dubious ownership 问题 +- `www` 是仓库实际维护用户 + +### 5.2 国内 controller 与 worker 的 Redis 密码必须一致 + +至少这些位置要确认: + +- `/etc/default/domaincheck-api` +- `/etc/default/domaincheck-worker` +- `/www/wwwroot/getDomain/domainCheck/.env` + +如果 Redis 开了密码,但上述文件里 `REDIS_PASSWORD=` 为空,会出现: + +- `Authentication required.` +- API 无法发 Worker 控制指令 +- Worker 无法订阅控制消息 + +### 5.3 国内 controller 本地数据库并不是海外主数据 + +之前最大的误区就是: + +- 海外有 `16w` 数据 +- 但国内本地 `domains` 一开始是 `0` + +当前设计已经调整为: + +- 海外通过 `task-export` 导出待检测批次 +- 国内主动拉批次,落本地最小 `domains` +- 国内 worker 只跑本地任务 + +不要再假设“国内本地库天然就有海外那 16w 主数据”。 + +### 5.4 代码更新要确认文件内容,不要只看 restart + +如果 `systemctl restart domaincheck-api` 之后行为没变,不要只怀疑服务。 + +先直接检查源码关键片段是否真的更新到位。 + +例如本次调试里最典型的就是: + +- `domain-api/app/services/sync_push_service.py` + +通过 `grep` 关键代码判断是否已是新版,比只看服务状态更可靠。 + +--- + +## 6. 建议给海外 Codex 的接手任务 + +建议在海外机新开 Codex 后,直接按下面顺序接手: + +### 6.1 第一优先级 + +- 确认海外 cluster 能稳定显示: + - `overseas-control-01` + - `mainland-controller-01` + - `mainland-worker-01` + +- 确认海外 `readiness` 中: + - `online_worker_nodes >= 1` + - 不再被 `mainland-control-imported` 影响 + +### 6.2 第二优先级 + +- 确认 `detect_result_projection -> detect_result_ingest` 的闭环 +- 确认海外业务查询接口/后台页面能看到国内执行回流结果 + +### 6.3 第三优先级 + +- 整理 `docs/18_domainCheck_CentOS9一键复制部署与更新文档.md` +- 把国内 controller / worker 的最终正确部署流程写成“复制即可执行”的版本 + +--- + +## 7. 建议的优秀调试方式:国内日志主动回调到海外 + +当前最大问题不是代码本身,而是: + +- 每次都要在国内机执行命令 +- 再手工复制长日志给海外 +- 调试效率太低 + +最推荐的改法不是继续人工贴日志,而是做“日志回调到海外调试面板”。 + +### 7.1 推荐目标 + +增加一个轻量调试链路: + +- 国内 controller +- 国内 worker +- 国内 sync-agent + +把最近 N 条关键日志、关键事件、关键指标,主动推送到海外一个调试接口,或写入海外一张调试表。 + +### 7.2 最小可行方案 + +推荐只同步“结构化调试事件”,不要直接全量推原始日志文件。 + +建议结构: + +- `source_region` +- `node_code` +- `service` +- `event_type` +- `level` +- `message` +- `payload_json` +- `created_at` + +来源可以先选这三类: + +- `domaincheck-api` +- `domaincheck-worker` +- `domaincheck-sync-agent` + +### 7.3 最值得先回调的事件 + +不是所有日志都要回传,优先回传高价值事件: + +- Worker 控制指令成功/失败 +- 任务批次拉取成功/失败 +- 任务创建成功/失败 +- 结果投影生成成功/失败 +- 结果同步推送成功/失败 +- Redis/PostgreSQL 连接失败 +- 代理池刷新结果摘要 +- `domain_started` / `domain_completed` / `domain_failed` + +### 7.4 为什么推荐结构化事件而不是原始日志 + +因为结构化事件更适合: + +- 海外 Codex 直接查询分析 +- 后台页面直接展示 +- 做筛选、聚合、时间线回放 +- 后续做自动告警 + +### 7.5 建议实现方式 + +建议海外增加一个轻量接口,例如: + +- `POST /api/v1/runtime/debug-ingest` + +国内三类服务把关键事件 POST 到海外。 + +海外落库到新表,例如: + +- `detect_debug_events` + +建议字段: + +- `id` +- `source_region` +- `node_code` +- `service` +- `event_type` +- `level` +- `message` +- `payload_json` +- `created_at` + +这样后面海外 Codex 只要直接查海外库,就能复原国内问题,不需要再人工复制日志。 + +--- + +## 8. 建议下一步实施顺序 + +### 方案 A:先收尾展示,再补日志回调 + +1. 修海外 cluster/readiness 的 worker 显示 +2. 清理 `mainland-control-imported` +3. 确认结果回流展示 +4. 再做日志回调 + +### 方案 B:先补日志回调,再继续深度调试 + +1. 海外新增 `debug-ingest` +2. 国内 controller / worker / sync-agent 回调关键事件 +3. 海外 Codex 直接观察结构化事件 +4. 再做 cluster/readiness 视图收尾 + +如果后续还要继续高频调试,我更推荐 **方案 B**。 + +--- + +## 9. 当前阶段结论 + +截至本交接文档生成时,可以确认: + +- 国内外任务下发链路已打通 +- 国内本地任务落库已打通 +- 国内 worker 已实际执行检测 +- 国内运行时已回传海外 +- 海外已能看到新的大陆 controller 节点 + +当前剩余问题已经从“主流程不通”降级为: + +- 海外 cluster/readiness 视图收尾 +- 结果回流展示补强 +- 调试手段升级为结构化日志回调 + +主流程已经不再是 blocker。 diff --git a/domain-api/app/services/cluster_runtime_service.py b/domain-api/app/services/cluster_runtime_service.py index 1feb409..0917722 100644 --- a/domain-api/app/services/cluster_runtime_service.py +++ b/domain-api/app/services/cluster_runtime_service.py @@ -161,6 +161,35 @@ def register_node_heartbeat( conn.commit() +def cleanup_imported_runtime_nodes(*, region: str, role: str, keep_node_code: str) -> None: + normalized_region = str(region or "").strip() or "unknown" + normalized_role = str(role or "").strip() or "unknown" + preserved_node_code = str(keep_node_code or "").strip() + if not preserved_node_code: + return + with get_db() as conn: + with conn.cursor() as cur: + cur.execute( + """ + DELETE FROM detect_worker_nodes + WHERE region = %s + AND role = %s + AND node_code <> %s + AND ( + node_code = %s + OR (metadata_json->>'service') = 'runtime-ingest' + ) + """, + ( + normalized_region, + normalized_role, + preserved_node_code, + f"{normalized_region}-{normalized_role}-imported", + ), + ) + conn.commit() + + def register_local_control_heartbeat() -> None: register_node_heartbeat( node_code=settings.node_code, diff --git a/domain-api/app/services/sync_push_service.py b/domain-api/app/services/sync_push_service.py index 002258f..5807643 100644 --- a/domain-api/app/services/sync_push_service.py +++ b/domain-api/app/services/sync_push_service.py @@ -10,7 +10,7 @@ from uuid import uuid4 from app.core.config import settings from app.core.db import get_db -from app.services.cluster_runtime_service import register_node_heartbeat +from app.services.cluster_runtime_service import cleanup_imported_runtime_nodes, register_node_heartbeat from app.services.sync_record_service import _decode_json, _normalize_region @@ -91,6 +91,43 @@ def _refresh_remote_runtime_node(*, source_region: str, projection: dict, receiv hostname_override=hostname, ip_override=ip, ) + cleanup_imported_runtime_nodes(region=region, role=role, keep_node_code=node_code) + + active_job = projection.get("active_job") or {} + for node_stat in list(active_job.get("node_stats") or []): + worker_node_code = str(node_stat.get("node_code") or "").strip() + if not worker_node_code or worker_node_code == "unassigned": + continue + items_running = int(node_stat.get("items_running", 0) or 0) + items_claimed = int(node_stat.get("items_claimed", 0) or 0) + items_total = int(node_stat.get("items_total", 0) or 0) + worker_status = "busy" if (items_running > 0 or items_claimed > 0) else "online" + worker_load = max(items_running, items_claimed, 0) + worker_metadata = { + "service": "runtime-ingest", + "projection_source_region": source_region, + "worker_mode": projection.get("worker_mode", ""), + "phase_label": projection.get("phase_label", ""), + "phase_detail": projection.get("phase_detail", ""), + "proxy_runtime_label": projection.get("proxy_runtime_label", ""), + "proxy_runtime_reason": projection.get("proxy_runtime_reason", ""), + "updated_at": _format_time(received_at or datetime.now()), + "job_items_total": items_total, + "job_items_running": items_running, + "job_items_claimed": items_claimed, + "derived_from": node_code, + } + register_node_heartbeat( + node_code=worker_node_code, + region=region, + role="worker", + status=worker_status, + current_load=worker_load, + metadata=worker_metadata, + hostname_override=hostname, + ip_override=ip, + ) + cleanup_imported_runtime_nodes(region=region, role="worker", keep_node_code=worker_node_code) def _load_latest_projection(sync_type: str) -> dict | None: diff --git a/domain-api/app/services/sync_record_service.py b/domain-api/app/services/sync_record_service.py index 4399006..8d5217a 100644 --- a/domain-api/app/services/sync_record_service.py +++ b/domain-api/app/services/sync_record_service.py @@ -435,6 +435,7 @@ def append_runtime_projection_if_changed( "items_pending": active_job.get("items_pending", 0), "items_running": active_job.get("items_running", 0), "items_failed": active_job.get("items_failed", 0), + "node_stats": list(active_job.get("node_stats") or []), }, "cluster_summary": { "nodes_total": int(cluster.get("nodes_total", 0) or 0),