diff --git a/docs/operations/current-ecs-deployment.md b/docs/operations/current-ecs-deployment.md index 3b834984..54168d87 100644 --- a/docs/operations/current-ecs-deployment.md +++ b/docs/operations/current-ecs-deployment.md @@ -155,10 +155,13 @@ curl -sS 'http://115.29.187.205:20210/api/stats/daily-metrics?dateFrom=2026-07-0 部署后推荐直接运行 Go 原生生产 smoke： ```bash +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 ``` -该脚本通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询，三类协议的位置和里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。 +`go_systemd_prod_smoke.py` 通过 SSH 检查四个 Go systemd 服务是否 `active`，并确认 `808`、`32960` 由 `gateway` 监听、`20210` 由 `realtime-api` 监听，避免旧 Java/Docker 进程占用生产端口。运行环境需要已配置 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。 ## 部署命令 @@ -189,6 +192,8 @@ systemctl restart lingniu-go-realtime-api.service ## 运行检查 ```bash +python3 tools/go_systemd_prod_smoke.py --host 115.29.187.205 --user root + systemctl is-active lingniu-go-gateway lingniu-go-history-writer lingniu-go-stat-writer lingniu-go-realtime-api ss -lntp | egrep ':(808|32960|20210)\b' journalctl -u lingniu-go-gateway --since '5 minutes ago' --no-pager diff --git a/docs/operations/go-vehicle-gateway-verification.md b/docs/operations/go-vehicle-gateway-verification.md index b2f1b842..d4b94c94 100644 --- a/docs/operations/go-vehicle-gateway-verification.md +++ b/docs/operations/go-vehicle-gateway-verification.md @@ -18,32 +18,26 @@ go build ./cmd/stat-writer go build ./cmd/realtime-api ``` -## 镜像构建 +## 生产原生部署验证 ```bash -cd go/vehicle-gateway -docker build -t vehicle-gateway-go:local . +python3 tools/go_systemd_prod_smoke.py --host 115.29.187.205 --user root ``` -同一个镜像包含四个二进制： +脚本通过 SSH 检查： -- `/app/gateway` -- `/app/history-writer` -- `/app/stat-writer` -- `/app/realtime-api` +- `lingniu-go-gateway.service` +- `lingniu-go-history-writer.service` +- `lingniu-go-stat-writer.service` +- `lingniu-go-realtime-api.service` +- `808`、`32960` 由 `gateway` 监听 +- `20210` 由 `realtime-api` 监听 -## Portainer 部署 +## 生产环境变量 -Go 旁路 stack 文件： +当前 goal 的生产面使用 ECS 原生 systemd 部署，不再以 Docker/Portainer 作为生产运行方式。以下环境变量维护在 `/opt/lingniu-go-native/env/*.env`： ```bash -deploy/portainer/docker-compose-go.yml -``` - -必填变量： - -```bash -LINGNIU_GO_IMAGE_VERSION= KAFKA_BROKERS=172.17.111.56:9092 TDENGINE_DSN=root:@ws(172.17.111.57:6041)/lingniu_vehicle_ts MYSQL_DSN=lingniu_vehicle:@tcp(rm-bp179zbv481rnw3e2.mysql.rds.aliyuncs.com:3306)/lingniu_vehicle_data?parseTime=true&charset=utf8mb4,utf8&loc=Asia%2FShanghai @@ -68,8 +62,7 @@ YUTONG_MQTT_PASSWORD= ## ECS 进程检查 ```bash -docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}' \ - | egrep 'go-vehicle-gateway|go-history-writer|go-stat-writer|go-realtime-api|NAMES' +systemctl is-active lingniu-go-gateway lingniu-go-history-writer lingniu-go-stat-writer lingniu-go-realtime-api ss -lntp | egrep ':(808|32960|20210)\b' ``` diff --git a/tools/go_systemd_prod_smoke.py b/tools/go_systemd_prod_smoke.py new file mode 100755 index 00000000..e3145f55 --- /dev/null +++ b/tools/go_systemd_prod_smoke.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""Systemd and port smoke checks for the Go native production ECS.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from dataclasses import asdict, dataclass + + +DEFAULT_HOST = "115.29.187.205" +DEFAULT_USER = "root" +DEFAULT_SERVICES = [ + "lingniu-go-gateway.service", + "lingniu-go-history-writer.service", + "lingniu-go-stat-writer.service", + "lingniu-go-realtime-api.service", +] +DEFAULT_PORTS = { + 808: "gateway", + 32960: "gateway", + 20210: "realtime-api", +} + + +@dataclass(frozen=True) +class Check: + name: str + status: str + message: str + + +def parse_service_states(output: str) -> dict[str, str]: + states: dict[str, str] = {} + for line in output.splitlines(): + parts = line.split() + if len(parts) >= 2: + states[parts[0]] = parts[1] + return states + + +def service_checks(output: str, services: list[str]) -> list[Check]: + states = parse_service_states(output) + checks: list[Check] = [] + for service in services: + state = states.get(service, "missing") + status = "pass" if state == "active" else "fail" + checks.append(Check("service." + service, status, "state=" + state)) + return checks + + +def port_checks(output: str, ports: dict[int, str]) -> list[Check]: + lines = output.splitlines() + checks: list[Check] = [] + for port, process in sorted(ports.items()): + matching = [line.strip() for line in lines if has_port(line, port)] + if not matching: + checks.append(Check("port." + str(port), "fail", "not listening")) + continue + owned = [line for line in matching if process in line] + if owned: + checks.append(Check("port." + str(port), "pass", owned[0])) + else: + checks.append(Check("port." + str(port), "fail", matching[0])) + return checks + + +def has_port(line: str, port: int) -> bool: + needle = ":" + str(port) + return needle + " " in line or needle + "\t" in line + + +def ssh(host: str, user: str, command: str, timeout: float) -> str: + target = user + "@" + host if user else host + completed = subprocess.run( + ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", target, command], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=timeout, + ) + return completed.stdout + + +def remote_command(services: list[str]) -> str: + quoted = " ".join(services) + return ( + "for svc in " + quoted + "; do " + "printf '%s ' \"$svc\"; systemctl is-active \"$svc\" || true; " + "done; " + "printf '\\n--PORTS--\\n'; " + "ss -lntp" + ) + + +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 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) + status = "fail" if any(check.status == "fail" for check in checks) else "pass" + return status, checks + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--host", default=DEFAULT_HOST) + parser.add_argument("--user", default=DEFAULT_USER) + parser.add_argument("--timeout", type=float, default=20.0) + return parser.parse_args(argv) + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + try: + status, checks = run(args) + except Exception as exc: + status = "fail" + checks = [Check("systemd.ssh", "fail", str(exc))] + print(json.dumps({ + "status": status, + "host": args.host, + "checks": [asdict(check) for check in checks], + }, ensure_ascii=False, indent=2)) + return 0 if status == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tools/test_go_systemd_prod_smoke.py b/tools/test_go_systemd_prod_smoke.py new file mode 100644 index 00000000..09eef163 --- /dev/null +++ b/tools/test_go_systemd_prod_smoke.py @@ -0,0 +1,57 @@ +import unittest + +from tools import go_systemd_prod_smoke as smoke + + +class GoSystemdProdSmokeTest(unittest.TestCase): + def test_service_checks_require_all_go_units_active(self): + output = """ +lingniu-go-gateway.service active +lingniu-go-history-writer.service active +lingniu-go-stat-writer.service active +lingniu-go-realtime-api.service active +""" + + checks = smoke.service_checks(output, smoke.DEFAULT_SERVICES) + + self.assertTrue(all(check.status == "pass" for check in checks)) + + def test_service_checks_fail_when_unit_is_missing_or_inactive(self): + output = """ +lingniu-go-gateway.service active +lingniu-go-history-writer.service failed +""" + + checks = smoke.service_checks(output, smoke.DEFAULT_SERVICES) + by_name = {check.name: check for check in checks} + + self.assertEqual(by_name["service.lingniu-go-history-writer.service"].status, "fail") + self.assertEqual(by_name["service.lingniu-go-stat-writer.service"].status, "fail") + + def test_port_checks_require_expected_go_processes(self): + output = """ +LISTEN 0 4096 *:32960 *:* users:(("gateway",pid=10,fd=3)) +LISTEN 0 4096 *:808 *:* users:(("gateway",pid=10,fd=4)) +LISTEN 0 4096 *:20210 *:* users:(("realtime-api",pid=11,fd=3)) +""" + + checks = smoke.port_checks(output, smoke.DEFAULT_PORTS) + + self.assertTrue(all(check.status == "pass" for check in checks)) + + def test_port_checks_fail_when_java_owns_ingest_port(self): + output = """ +LISTEN 0 4096 *:808 *:* users:(("java",pid=20,fd=4)) +LISTEN 0 4096 *:32960 *:* users:(("gateway",pid=10,fd=3)) +LISTEN 0 4096 *:20210 *:* users:(("realtime-api",pid=11,fd=3)) +""" + + checks = smoke.port_checks(output, smoke.DEFAULT_PORTS) + by_name = {check.name: check for check in checks} + + self.assertEqual(by_name["port.808"].status, "fail") + self.assertIn("java", by_name["port.808"].message) + + +if __name__ == "__main__": + unittest.main()