test: add go systemd production smoke

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:]))