姚章浩
e6472b0ec6
为什么改：发布执行链是"与目标同机"假设写死的（本机 docker.sock、本机 drain.json、 本机 Redis/PG），把目标换成云测试不能只改展示 URL；直接搬整套 worker 又会抢走 本机封存/QA/周报队列并按本机配置写错目标。所以先把"谁消费哪段队列"变成显式角色， 再补齐跨主机搬运与目标侧门禁。 角色（LIBRARY_WORKER_ROLE=full/release-executor/capture）不只影响 handler 列表， 还决定是否拥有 DDL（release-executor 只读校验迁移版本，落后即拒绝启动）、 收拾哪些 library_jobs.kind、是否跑 release 接管扫描、是否起 scheduler/QA/周报/探测。 capture 保留本机原有全部业务；release-executor 只挂 sync/restore/recover/baseline。 快照跨主机：新增内容寻址的 bundle（版本清单先落、blob 按 sha256 命名、清单最后 Link 作提交点），pack/unpack 幂等且拒绝覆盖与同名不同内容；执行器使用前仍逐 blob 重算 sha256，因此传输不可变有两道证据。 目标侧门禁：目标知识库 graph_enabled=true 时拒绝发布（发布只写 PG/ES/OSS，不写 Neo4j；未知也拒绝），只读检测、不改目标配置、不做图谱迁移。 云端入口：新增 drain 感知的 ingress 闸门（建连超时与读超时分离，半关闭单向传播， SIGTERM 排空后写 drain.json 并 exit 0/2）、compose 叠加片段（app 不再发布端口、 frontend 与 adapter 的入口收敛到闸门、指向真实网络 weknora-test-v071-network）、 执行器启动脚本与治理 PG 反向隧道脚本（默认 unix socket，不占 TCP）。 验证：deploy/cloud-test/test_gate.py 13 项全 PASS（含 12 秒静默流、半关闭、2MiB 完整、 排空 exit 0/active_at_close=1、排空超时 exit 2、上游不可达不挂住）；go test 相关包全绿； scripts/ci_checks.py 16 项全部 success（EXIT=0，1241 用例通过/0 失败/191 已声明可选集成跳过）； 云测试只读预检 17 项 / 6 FAIL，FAIL 项即当前部署缺口（无闸门、app 仍发布端口、 adapter 与 frontend 仍直连 app、云端归档根未建），未部署、未改云端任何状态。
252 lines
13 KiB
Python
Executable File
252 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""云测试发布目标：只读预检。
|
||
|
||
只做 ssh 上的只读命令（docker inspect/ps/exec 的只读查询、curl GET、test、stat），
|
||
不启停容器、不写文件、不改配置、不连治理库。每行一个判据：PASS / FAIL / WARN / MANUAL。
|
||
|
||
用法：
|
||
python3 scripts/verify_cloud_release_target.py --host weknora-test
|
||
python3 scripts/verify_cloud_release_target.py --host weknora-test --snapshot <64位ID>
|
||
|
||
退出码：有 FAIL 即 1，否则 0（WARN/MANUAL 不影响退出码，但必须人读）。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import shlex
|
||
import subprocess
|
||
import sys
|
||
|
||
PROJECT_DEFAULT = "weknora-test-v071"
|
||
APP_DEFAULT = "app"
|
||
GATE_SERVICE_DEFAULT = "ingress"
|
||
DRAIN_DIR_DEFAULT = "/srv/kgtest-gate/state"
|
||
ARCHIVE_DEFAULT = "/var/lib/knowledge-governance/library"
|
||
GATE_URL_DEFAULT = "http://127.0.0.1:8080/health"
|
||
EXECUTOR_DEFAULT = "kgtest-release-executor"
|
||
RDS_HOST_DEFAULT = "pgm-wz9ohr9gk919pr1q.pg.rds.aliyuncs.com"
|
||
|
||
results = []
|
||
|
||
|
||
def record(level, name, detail):
|
||
results.append((level, name, detail))
|
||
print(f"{level:<6} {name}: {detail}", flush=True)
|
||
|
||
|
||
def ssh(host, command, timeout=25):
|
||
argv = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", host, command]
|
||
try:
|
||
completed = subprocess.run(argv, capture_output=True, text=True, timeout=timeout)
|
||
except subprocess.TimeoutExpired:
|
||
return 124, "", "ssh timeout"
|
||
return completed.returncode, completed.stdout.strip(), completed.stderr.strip()
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--host", default="weknora-test")
|
||
parser.add_argument("--project", default=PROJECT_DEFAULT)
|
||
parser.add_argument("--app-service", default=APP_DEFAULT)
|
||
parser.add_argument("--gate-service", default=GATE_SERVICE_DEFAULT)
|
||
parser.add_argument("--drain-dir", default=DRAIN_DIR_DEFAULT)
|
||
parser.add_argument("--archive", default=ARCHIVE_DEFAULT)
|
||
parser.add_argument("--gate-url", default=GATE_URL_DEFAULT)
|
||
parser.add_argument("--executor", default=EXECUTOR_DEFAULT)
|
||
parser.add_argument("--rds-host", default=RDS_HOST_DEFAULT)
|
||
parser.add_argument("--snapshot", default="")
|
||
args = parser.parse_args()
|
||
host = args.host
|
||
|
||
code, _, err = ssh(host, "true")
|
||
if code != 0:
|
||
record("FAIL", "ssh", f"{host} 不可达（{err or code}）")
|
||
return finish()
|
||
record("PASS", "ssh", f"{host} 可达")
|
||
|
||
# 1) 项目容器清单：每个服务都必须是写入者或"声明为非写入者"，否则窗口拒绝开。
|
||
code, out, err = ssh(
|
||
host,
|
||
"docker ps -a --filter label=com.docker.compose.project=%s "
|
||
"--format '{{.Label \"com.docker.compose.service\"}}'" % shlex.quote(args.project),
|
||
)
|
||
services = sorted({line.strip() for line in out.splitlines() if line.strip()})
|
||
if code != 0:
|
||
record("FAIL", "project-containers", f"读不到项目容器（{err or code}）")
|
||
services = []
|
||
else:
|
||
record("PASS" if services else "FAIL", "project-containers",
|
||
f"project={args.project} services={','.join(services) or '<empty>'}")
|
||
|
||
# 2) 写入者不得发布宿主机端口（否则存在绕开入口闸门的直连通道）。
|
||
containers = {service: f"{args.project}-{service}-1" for service in services}
|
||
app_container = containers.get(args.app_service, "")
|
||
if app_container:
|
||
code, out, err = ssh(host, "docker inspect %s --format '{{json .HostConfig.PortBindings}}'"
|
||
% shlex.quote(app_container))
|
||
if code != 0:
|
||
record("FAIL", "writer-no-publish", f"inspect {app_container} 失败（{err or code}）")
|
||
else:
|
||
bindings = json.loads(out) if out else None
|
||
if bindings in (None, {}, "null"):
|
||
record("PASS", "writer-no-publish", f"{app_container} 未发布宿主机端口")
|
||
else:
|
||
record("FAIL", "writer-no-publish",
|
||
f"{app_container} 仍发布端口 {out}：拓扑校验会拒绝开维护窗口")
|
||
else:
|
||
record("FAIL", "writer-no-publish", f"项目里找不到写入者服务 {args.app_service}")
|
||
|
||
# 3) 入口闸门存在、健康、发布端口与入口地址端口一致。
|
||
code, out, err = ssh(host, "docker inspect %s --format '{{json .State.Health.Status}} {{json .HostConfig.PortBindings}} {{json .State.Running}}'"
|
||
% shlex.quote(f"{args.project}-{args.gate_service}-1"))
|
||
if code != 0:
|
||
record("FAIL", "gate", f"找不到闸门容器 {args.project}-{args.gate_service}-1（{err or code}）")
|
||
else:
|
||
parts = out.split(" ", 2)
|
||
health = parts[0] if parts else ""
|
||
bindings = parts[1] if len(parts) > 1 else ""
|
||
running = parts[2] if len(parts) > 2 else ""
|
||
if running.strip() == "true" and health.strip('"') == "healthy" and ":8080->" in bindings:
|
||
record("PASS", "gate", f"healthy，发布端口 {bindings}")
|
||
else:
|
||
record("FAIL", "gate", f"running={running} health={health} bindings={bindings}")
|
||
|
||
# 4) 经闸门的健康检查：这是"入口在服务"的实际证据，而不是容器状态。
|
||
code, out, err = ssh(host, "curl -s -o /dev/null -w '%%{http_code}' -m 8 %s" % shlex.quote(args.gate_url))
|
||
if code == 0 and out.strip() == "200":
|
||
record("PASS", "gate-health", f"{args.gate_url} -> 200")
|
||
else:
|
||
record("FAIL", "gate-health", f"{args.gate_url} -> {out or err or code}")
|
||
|
||
# 5) 闸门写排空证据的目录必须存在且可写（协调器要在同一真实文件上删/读）。
|
||
code, out, err = ssh(host, "test -d %s && test -w %s && echo ok" % (shlex.quote(args.drain_dir), shlex.quote(args.drain_dir)))
|
||
if code == 0 and out.strip() == "ok":
|
||
record("PASS", "drain-dir", f"{args.drain_dir} 存在且可写")
|
||
else:
|
||
record("FAIL", "drain-dir", f"{args.drain_dir} 不存在或不可写（{err or code}）")
|
||
code, out, _ = ssh(host, "stat -c '%%y %%s' %s 2>/dev/null" % shlex.quote(args.drain_dir + "/drain.json"))
|
||
if code == 0 and out:
|
||
record("WARN", "drain-file", f"已存在 drain.json（{out}）：协调器开窗口前会先删除并验证不存在")
|
||
else:
|
||
record("PASS", "drain-file", "无残留 drain.json")
|
||
|
||
# 6) 队列：redis 在本项目网络内可达（只读 PING；有口令时 NOAUTH 也算可达）。
|
||
redis_container = containers.get("redis", "")
|
||
if redis_container:
|
||
code, out, err = ssh(host, "docker exec %s redis-cli ping 2>&1" % shlex.quote(redis_container))
|
||
text = (out or err).strip()
|
||
if "PONG" in text or "NOAUTH" in text:
|
||
record("PASS", "redis", f"{redis_container}: {text}")
|
||
else:
|
||
record("FAIL", "redis", f"{redis_container}: {text or code}")
|
||
else:
|
||
record("FAIL", "redis", "项目里没有 redis 服务")
|
||
|
||
# 7) 索引：ES 可达且目标索引存在（只读 HEAD/GET）。
|
||
es_container = containers.get("elasticsearch", "")
|
||
if es_container:
|
||
code, out, err = ssh(host, "docker exec %s curl -s -o /dev/null -w '%%{http_code}' -m 8 http://localhost:9200/_cluster/health"
|
||
% shlex.quote(es_container))
|
||
record("PASS" if out.strip() == "200" else "FAIL", "elasticsearch",
|
||
f"{es_container} cluster/health -> {out.strip() or err or code}")
|
||
else:
|
||
record("FAIL", "elasticsearch", "项目里没有 elasticsearch 服务")
|
||
|
||
# 8) RDS：宿主机到 5432 的 TCP 可达（不登录、不带凭据）。
|
||
code, out, err = ssh(host, "timeout 5 bash -c 'cat < /dev/null > /dev/tcp/%s/5432' && echo open"
|
||
% shlex.quote(args.rds_host))
|
||
record("PASS" if out.strip() == "open" else "FAIL", "rds-reachable",
|
||
f"{args.rds_host}:5432 {'可达' if out.strip() == 'open' else '不可达'}")
|
||
|
||
# 9) 归档根：执行器要在这里读快照、写预备份与写集合台账。
|
||
code, out, err = ssh(host, "test -d %s && stat -c '%%a' %s" % (shlex.quote(args.archive), shlex.quote(args.archive)))
|
||
if code == 0:
|
||
record("PASS" if out.strip() == "700" else "WARN", "archive-root",
|
||
f"{args.archive} 权限 {out.strip()}（要求 0700）")
|
||
else:
|
||
record("FAIL", "archive-root", f"{args.archive} 不存在")
|
||
|
||
# 10) 可选的快照包：给出 ID 时只核对版本清单是否已落位（不读 blob）。
|
||
if args.snapshot:
|
||
version = f"{args.archive}/versions/{args.snapshot}.json"
|
||
code, out, err = ssh(host, "test -f %s && sha256sum %s | cut -d' ' -f1"
|
||
% (shlex.quote(version), shlex.quote(version)))
|
||
if code == 0 and out.strip() == args.snapshot:
|
||
record("PASS", "snapshot", f"{args.snapshot} 已导入且清单摘要自证一致")
|
||
else:
|
||
record("FAIL", "snapshot", f"{args.snapshot} 未导入或清单摘要不符（{out.strip() or err or code}）")
|
||
|
||
# 11) 其它入口：adapter / DobotAI / frontend 都能到 app。只改 frontend 不够，
|
||
# "唯一入口"必须逐条核对：谁能绕过闸门直连 app，就等于闸门关不干净。
|
||
code, out, _ = ssh(host, "docker network ls --format '{{.Name}}'")
|
||
networks = set(out.split())
|
||
record("PASS" if "weknora-test-v071-network" in networks else "FAIL", "target-network",
|
||
f"weknora-test-v071-network {'存在' if 'weknora-test-v071-network' in networks else '不存在'}")
|
||
|
||
def inspect_env(container):
|
||
code, env_raw, _ = ssh(host, "docker inspect %s --format '{{json .Config.Env}}'" % shlex.quote(container))
|
||
if code != 0 or not env_raw:
|
||
return None
|
||
values = {}
|
||
for entry in json.loads(env_raw):
|
||
key, _, value = entry.partition("=")
|
||
values[key] = value
|
||
return values
|
||
|
||
code, out, _ = ssh(host, "docker ps --format '{{.Names}}'")
|
||
running = set(out.split())
|
||
adapter = next((name for name in running if name.startswith("knowledge-adapter-adapter")), "")
|
||
if adapter:
|
||
env = inspect_env(adapter) or {}
|
||
base = env.get("WEKNORA_BASE_URL", "")
|
||
ok = base.startswith("http://ingress:") or base.startswith("http://kgtest-ingress:")
|
||
record("PASS" if ok else "FAIL", "entry/adapter",
|
||
f"{adapter} WEKNORA_BASE_URL={base or '<unset>'}（要求指向 ingress，否则 adapter 绕过闸门直连 app）")
|
||
else:
|
||
record("WARN", "entry/adapter", "未找到 adapter 容器（若它就是知识问答入口，必须改成指向 ingress）")
|
||
|
||
dobotai = next((name for name in running if name.startswith("dobotai-dingtalk")), "")
|
||
if dobotai:
|
||
env = inspect_env(dobotai) or {}
|
||
base = env.get("WEKNORA_BASE_URL", "")
|
||
code, mode, _ = ssh(host, "docker inspect %s --format '{{.HostConfig.NetworkMode}}'" % shlex.quote(dobotai))
|
||
host_network = mode.strip() == "host"
|
||
ok = host_network and base in ("http://127.0.0.1:8080", "http://localhost:8080")
|
||
record("PASS" if ok else "FAIL", "entry/dobotai",
|
||
f"{dobotai} net={mode.strip()} WEKNORA_BASE_URL={base or '<unset>'}（host 网络 + 127.0.0.1:8080 才恰好走闸门）")
|
||
else:
|
||
record("WARN", "entry/dobotai", "未找到 dobotai-dingtalk 容器")
|
||
|
||
if "frontend" in containers:
|
||
env = inspect_env(containers["frontend"]) or {}
|
||
app_host = env.get("APP_HOST", "")
|
||
record("PASS" if app_host == "ingress" else "FAIL", "entry/frontend",
|
||
f"frontend APP_HOST={app_host or '<unset>'}（要求 ingress）")
|
||
|
||
# 11) 发布执行器：在不在跑、角色对不对（只读这一个变量，不打印 env）。
|
||
code, out, err = ssh(host, "docker ps --filter name=%s --format '{{.Status}}'" % shlex.quote(args.executor))
|
||
if code == 0 and out.strip():
|
||
record("PASS", "executor", f"{args.executor} {out.strip()}")
|
||
code, role, _ = ssh(host, "docker exec %s printenv LIBRARY_WORKER_ROLE" % shlex.quote(args.executor))
|
||
if code == 0 and role.strip() == "release-executor":
|
||
record("PASS", "executor-role", "LIBRARY_WORKER_ROLE=release-executor")
|
||
else:
|
||
record("FAIL", "executor-role", f"LIBRARY_WORKER_ROLE={role.strip() or '<unset>'}（要求 release-executor）")
|
||
else:
|
||
record("WARN", "executor", f"{args.executor} 未运行（父部署后才会出现）")
|
||
|
||
# 12) 图谱：目标 KB 若开了 graph_enabled，发布只写 PG/ES/OSS，图谱会静默变旧。
|
||
record("MANUAL", "target-graph",
|
||
"需人工确认目标 KB 的 indexing_strategy.graph_enabled=false（发布路径不写 Neo4j）")
|
||
return finish()
|
||
|
||
|
||
def finish():
|
||
failures = [row for row in results if row[0] == "FAIL"]
|
||
print()
|
||
print(f"summary: {len(results)} checks, {len(failures)} FAIL")
|
||
return 1 if failures else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |