From 601150b36418f0da3cc9d226a7dfe71bd7267962 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 01:57:53 +0800 Subject: [PATCH] test: verify gateway spool is drained --- docs/operations/current-ecs-deployment.md | 3 +- .../go-vehicle-gateway-verification.md | 3 +- tools/go_systemd_prod_smoke.py | 53 +++++++++++++++---- tools/test_go_systemd_prod_smoke.py | 12 +++++ 4 files changed, 60 insertions(+), 11 deletions(-) diff --git a/docs/operations/current-ecs-deployment.md b/docs/operations/current-ecs-deployment.md index c800e7ae..da4b724b 100644 --- a/docs/operations/current-ecs-deployment.md +++ b/docs/operations/current-ecs-deployment.md @@ -167,7 +167,7 @@ python3 tools/go_systemd_prod_smoke.py --host 115.29.187.205 --user root python3 tools/go_native_prod_smoke.py --date 2026-07-02 --timeout 8 ``` -`go_systemd_prod_smoke.py` 通过 SSH 检查四个 Go systemd 服务是否 `active` 且 `enabled`，并确认 `808`、`32960` 由 `gateway` 监听、`20210` 由 `realtime-api` 监听，避免旧 Java/Docker 进程占用生产端口。运行环境需要已配置 SSH 免密或已建立可用的 SSH 认证方式。 +`go_systemd_prod_smoke.py` 通过 SSH 检查四个 Go systemd 服务是否 `active` 且 `enabled`，并确认 `808`、`32960` 由 `gateway` 监听、`20210` 由 `realtime-api` 监听，同时要求 `/opt/lingniu-go-native/spool/gateway` 没有待回放文件，避免旧 Java/Docker 进程占用生产端口或 Kafka spool 静默堆积。运行环境需要已配置 SSH 免密或已建立可用的 SSH 认证方式。 `go_native_prod_smoke.py` 通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询，三类协议的位置和里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。 @@ -215,6 +215,7 @@ journalctl -u lingniu-go-realtime-api --since '5 minutes ago' --no-pager - 四个 Go systemd 服务均为 `active` 且 `enabled`。 - `gateway` 正在监听 `32960` 和 `808`。 - `realtime-api` 正在监听 `20210`。 +- gateway Kafka spool 当前无待回放文件。 - GB32960、JT808、YUTONG_MQTT 的东八区 RAW 查询均可命中生产数据。 - GB32960、JT808、YUTONG_MQTT 的最新 RAW 样本均在 15 分钟内。 - GB32960、JT808、YUTONG_MQTT 的 `vehicle_locations`、`vehicle_mileage_points` 查询均可命中生产数据。 diff --git a/docs/operations/go-vehicle-gateway-verification.md b/docs/operations/go-vehicle-gateway-verification.md index f2da60b7..7212f5f8 100644 --- a/docs/operations/go-vehicle-gateway-verification.md +++ b/docs/operations/go-vehicle-gateway-verification.md @@ -26,7 +26,7 @@ python3 tools/go_prod_acceptance.py --date 2026-07-02 该命令会聚合执行： -- `go_systemd_prod_smoke.py`：应用 ECS systemd 服务 active/enabled 和端口归属 +- `go_systemd_prod_smoke.py`：应用 ECS systemd 服务 active/enabled、端口归属和 gateway spool - `go_kafka_prod_smoke.py`：Kafka topic 和消费组 lag - `go_native_prod_smoke.py`：HTTP API、TDengine/MySQL/Redis 数据链路 @@ -45,6 +45,7 @@ python3 tools/go_systemd_prod_smoke.py --host 115.29.187.205 --user root - 上述服务均为 `active` 且 `enabled` - `808`、`32960` 由 `gateway` 监听 - `20210` 由 `realtime-api` 监听 +- `/opt/lingniu-go-native/spool/gateway` 无待回放文件 ## 生产环境变量 diff --git a/tools/go_systemd_prod_smoke.py b/tools/go_systemd_prod_smoke.py index c693a3d4..8f88fa20 100755 --- a/tools/go_systemd_prod_smoke.py +++ b/tools/go_systemd_prod_smoke.py @@ -23,6 +23,7 @@ DEFAULT_PORTS = { 32960: "gateway", 20210: "realtime-api", } +DEFAULT_SPOOL_DIR = "/opt/lingniu-go-native/spool/gateway" @dataclass(frozen=True) @@ -83,6 +84,24 @@ def has_port(line: str, port: int) -> bool: return needle + " " in line or needle + "\t" in line +def spool_check(output: str) -> Check: + values = parse_key_values(output) + files = int(values.get("files", "-1")) + recent = int(values.get("recent", "-1")) + status = "pass" if files == 0 and recent == 0 else "fail" + return Check("spool.gateway", status, output.strip() or "missing") + + +def parse_key_values(output: str) -> dict[str, str]: + values: dict[str, str] = {} + for part in output.split(): + if "=" not in part: + continue + key, value = part.split("=", 1) + values[key] = value + return values + + def ssh(host: str, user: str, command: str, timeout: float) -> str: target = user + "@" + host if user else host completed = subprocess.run( @@ -105,22 +124,38 @@ def remote_command(services: list[str]) -> str: "systemctl is-enabled \"$svc\" 2>/dev/null || true; " "done; " "printf '\\n--PORTS--\\n'; " - "ss -lntp" + "ss -lntp; " + "printf '\\n--SPOOL--\\n'; " + "spool='" + DEFAULT_SPOOL_DIR + "'; " + "if test -d \"$spool\"; then " + "files=$(find \"$spool\" -type f | wc -l); " + "bytes=$(du -sb \"$spool\" 2>/dev/null | awk '{print $1}'); " + "recent=$(find \"$spool\" -type f -mmin -5 | wc -l); " + "printf 'files=%s bytes=%s recent=%s\\n' \"$files\" \"${bytes:-0}\" \"$recent\"; " + "else printf 'files=-1 bytes=0 recent=-1 missing=%s\\n' \"$spool\"; fi" ) -def split_remote_output(output: str) -> tuple[str, str]: - marker = "\n--PORTS--\n" - if marker not in output: - return output, "" - before, after = output.split(marker, 1) - return before, after +def split_remote_output(output: str) -> tuple[str, str, str]: + ports_marker = "\n--PORTS--\n" + spool_marker = "\n--SPOOL--\n" + if ports_marker not in output: + return output, "", "" + service_output, rest = output.split(ports_marker, 1) + if spool_marker not in rest: + return service_output, rest, "" + port_output, spool_output = rest.split(spool_marker, 1) + return service_output, port_output, spool_output def run(args: argparse.Namespace) -> tuple[str, list[Check]]: output = ssh(args.host, args.user, remote_command(DEFAULT_SERVICES), args.timeout) - service_output, port_output = split_remote_output(output) - checks = service_checks(service_output, DEFAULT_SERVICES) + port_checks(port_output, DEFAULT_PORTS) + service_output, port_output, spool_output = split_remote_output(output) + checks = ( + service_checks(service_output, DEFAULT_SERVICES) + + port_checks(port_output, DEFAULT_PORTS) + + [spool_check(spool_output)] + ) status = "fail" if any(check.status == "fail" for check in checks) else "pass" return status, checks diff --git a/tools/test_go_systemd_prod_smoke.py b/tools/test_go_systemd_prod_smoke.py index 7d7f5b5e..535b3bfc 100644 --- a/tools/test_go_systemd_prod_smoke.py +++ b/tools/test_go_systemd_prod_smoke.py @@ -66,6 +66,18 @@ LISTEN 0 4096 *:20210 *:* users:(("realtime-api",pid=11,fd=3)) self.assertEqual(by_name["port.808"].status, "fail") self.assertIn("java", by_name["port.808"].message) + def test_spool_check_passes_when_no_pending_files(self): + check = smoke.spool_check("files=0 bytes=208896 recent=0") + + self.assertEqual(check.status, "pass") + self.assertIn("files=0", check.message) + + def test_spool_check_fails_when_pending_or_recent_files_exist(self): + for output in ["files=2 bytes=300 recent=0", "files=0 bytes=208896 recent=1"]: + check = smoke.spool_check(output) + + self.assertEqual(check.status, "fail") + if __name__ == "__main__": unittest.main()