140 lines
4.1 KiB
Python
Executable File
140 lines
4.1 KiB
Python
Executable File
#!/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:]))
|