knowledge-governance/scripts/verify_cloud_release_target.py

#!/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())