laily
10d8800f07
Give ack, builder, and deployer an explicit init/check mode that reports missing project config instead of failing mid-work. Point builder at makefile.builder so its contract targets do not collide with an existing Makefile.
286 lines
9.4 KiB
Python
286 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Readiness checker for deployer project layout.
|
|
|
|
Usage:
|
|
python3 -I -S check.py [--project DIR]
|
|
|
|
Resolves the deploy root as DEPLOYER_ROOT, else <project>/.pouch/deployer
|
|
(or .skiff/deployer), else walking up from cwd. Does not SSH, rsync, or
|
|
start containers. Exit 0 = PASS (SKIP allowed), 1 = FAIL, 2 = usage.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
_SCRIPT_DIR = Path(__file__).resolve().parent
|
|
if str(_SCRIPT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(_SCRIPT_DIR))
|
|
|
|
import lib # noqa: E402
|
|
|
|
PASS = "PASS"
|
|
FAIL = "FAIL"
|
|
SKIP = "SKIP"
|
|
_PROJECT_LAYOUT_DIRS = (".pouch", ".skiff")
|
|
_REPO_OR_DIR = re.compile(r"^(repo|repo_dir)\s*:", re.MULTILINE)
|
|
|
|
|
|
class Report:
|
|
def __init__(self) -> None:
|
|
self.failures = 0
|
|
|
|
def add(self, status: str, number: int, title: str, detail: str) -> None:
|
|
print(f"[{status}] {number}. {title}")
|
|
for line in detail.splitlines():
|
|
print(f" {line}")
|
|
if status == FAIL:
|
|
self.failures += 1
|
|
|
|
|
|
def node_in_ssh_config(node: str, hosts: set[str]) -> bool:
|
|
if node in hosts:
|
|
return True
|
|
if "@" in node:
|
|
_, host = node.rsplit("@", 1)
|
|
return host in hosts
|
|
return False
|
|
|
|
|
|
def resolve_deploy_root(project: Path | None) -> tuple[Path | None, str]:
|
|
env = os.environ.get("DEPLOYER_ROOT", "").strip()
|
|
if env:
|
|
path = Path(env).expanduser().resolve()
|
|
return (path if path.is_dir() else None), "DEPLOYER_ROOT"
|
|
if project is not None:
|
|
root = project.resolve()
|
|
for dirname in _PROJECT_LAYOUT_DIRS:
|
|
candidate = root / dirname / "deployer"
|
|
if candidate.is_dir():
|
|
return candidate, "project"
|
|
return None, "project"
|
|
found = lib._find_project_root()
|
|
if found.name == "deployer" and found.parent.name in _PROJECT_LAYOUT_DIRS:
|
|
return found, "project"
|
|
if found == lib._SKILL_DIR.parent:
|
|
return None, "cwd"
|
|
return found, "cwd"
|
|
|
|
|
|
def compose_service_dirs(root: Path) -> list[Path]:
|
|
dirs: list[Path] = []
|
|
for compose in sorted(root.glob("**/compose.yaml")):
|
|
service_dir = compose.parent
|
|
if not lib.is_deployable_dir(service_dir, root):
|
|
continue
|
|
dirs.append(service_dir)
|
|
return dirs
|
|
|
|
|
|
def check_layout(report: Report, root: Path | None, source: str, project: Path | None) -> bool:
|
|
if root is None:
|
|
if source == "DEPLOYER_ROOT":
|
|
report.add(
|
|
FAIL,
|
|
1,
|
|
"部署根存在",
|
|
"DEPLOYER_ROOT is set but is not a directory",
|
|
)
|
|
else:
|
|
hint_root = project.resolve() if project is not None else Path.cwd()
|
|
report.add(
|
|
FAIL,
|
|
1,
|
|
"部署根存在",
|
|
"\n".join(
|
|
[
|
|
f"no .pouch/deployer under {hint_root}",
|
|
"Fix: run deployer 初始化 and create .pouch/deployer/",
|
|
" _config.yaml # node: <ssh-host-alias>",
|
|
" test/compose.yaml # default env for ACK",
|
|
]
|
|
),
|
|
)
|
|
return False
|
|
kind = "project layout" if lib.in_project_layout(root) else "standalone deploy root"
|
|
report.add(PASS, 1, "部署根存在", f"{root} ({kind}, via {source})")
|
|
return True
|
|
|
|
|
|
def check_toolchain(report: Report) -> None:
|
|
ssh_ok = shutil.which("ssh") is not None
|
|
rsync_ok = shutil.which("rsync") is not None
|
|
lines = [
|
|
f"ssh: {'found' if ssh_ok else 'MISSING (blocks compose deploy)'}",
|
|
f"rsync: {'found' if rsync_ok else 'MISSING (tar-over-SSH fallback)'}",
|
|
]
|
|
if not ssh_ok:
|
|
lines.append("install openssh-client")
|
|
report.add(FAIL, 2, "工具链", "\n".join(lines))
|
|
return
|
|
report.add(PASS if rsync_ok else SKIP, 2, "工具链", "\n".join(lines))
|
|
|
|
|
|
def check_argocd(report: Report, root: Path) -> bool:
|
|
path = root / "argocd.yaml"
|
|
if not path.is_file():
|
|
report.add(SKIP, 3, "Argo CD 指针", "no argocd.yaml")
|
|
return False
|
|
text = path.read_text(encoding="utf-8")
|
|
if _REPO_OR_DIR.search(text):
|
|
report.add(PASS, 3, "Argo CD 指针", "argocd.yaml has repo or repo_dir")
|
|
return True
|
|
report.add(
|
|
FAIL,
|
|
3,
|
|
"Argo CD 指针",
|
|
"argocd.yaml exists but has neither repo: nor repo_dir:\n"
|
|
"Fix: repo: git@host:org/infra-gitops.git",
|
|
)
|
|
return True
|
|
|
|
|
|
def check_services(report: Report, root: Path) -> None:
|
|
services = compose_service_dirs(root)
|
|
if not services:
|
|
report.add(
|
|
FAIL,
|
|
4,
|
|
"至少有一个 compose.yaml",
|
|
"no compose.yaml under the deploy root\n"
|
|
"Fix: add .pouch/deployer/<env>/compose.yaml (env usually test)",
|
|
)
|
|
report.add(SKIP, 5, "每个服务能解析 node", "(no compose.yaml)")
|
|
report.add(SKIP, 6, "node 出现在 SSH config", "(no compose.yaml)")
|
|
report.add(SKIP, 7, "list 可发现服务", "(no compose.yaml)")
|
|
return
|
|
|
|
rels = [str(path.relative_to(root)) for path in services]
|
|
report.add(PASS, 4, "至少有一个 compose.yaml", "\n".join(rels))
|
|
|
|
hosts = lib.ssh_config_hosts()
|
|
node_lines = []
|
|
ssh_lines = []
|
|
node_fail = False
|
|
ssh_fail = False
|
|
for path in services:
|
|
rel = str(path.relative_to(root))
|
|
info = lib.service_info(rel, strict=False)
|
|
if info is None:
|
|
node_lines.append(f"{rel}: MISSING node")
|
|
ssh_lines.append(f"{rel}: skipped (no node)")
|
|
node_fail = True
|
|
continue
|
|
node = str(info["node"])
|
|
node_lines.append(f"{rel}: node={node}")
|
|
if node_in_ssh_config(node, hosts):
|
|
ssh_lines.append(f"{rel}: {node} in ~/.ssh/config")
|
|
else:
|
|
ssh_lines.append(f"{rel}: {node} NOT in ~/.ssh/config")
|
|
ssh_fail = True
|
|
|
|
if node_fail:
|
|
node_lines.extend(
|
|
[
|
|
"",
|
|
"Fix: write node in _config.yaml (deploy root or env dir).",
|
|
"Example:",
|
|
" node: my-vps",
|
|
" base_path: /opt/app",
|
|
"Do not invent a hostname. It must be an SSH Host alias.",
|
|
]
|
|
)
|
|
report.add(FAIL, 5, "每个服务能解析 node", "\n".join(node_lines))
|
|
else:
|
|
report.add(PASS, 5, "每个服务能解析 node", "\n".join(node_lines))
|
|
|
|
if ssh_fail or node_fail:
|
|
if ssh_fail:
|
|
ssh_lines.extend(
|
|
[
|
|
"",
|
|
"Fix: add a Host entry to ~/.ssh/config for the node alias.",
|
|
"This check does not open an SSH connection.",
|
|
]
|
|
)
|
|
report.add(FAIL, 6, "node 出现在 SSH config", "\n".join(ssh_lines))
|
|
else:
|
|
report.add(PASS, 6, "node 出现在 SSH config", "\n".join(ssh_lines))
|
|
|
|
found = lib.discover_services()
|
|
if not found:
|
|
report.add(
|
|
FAIL,
|
|
7,
|
|
"list 可发现服务",
|
|
"compose.yaml exists but discover_services found none "
|
|
"(need resolvable node)",
|
|
)
|
|
return
|
|
names = [os.path.relpath(item["service_dir"], root) for item in found]
|
|
report.add(PASS, 7, "list 可发现服务", f"{len(found)} service(s): " + ", ".join(names))
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
|
|
)
|
|
parser.add_argument(
|
|
"--project",
|
|
type=Path,
|
|
default=None,
|
|
help="project root (looks for .pouch/deployer); ignored when DEPLOYER_ROOT is set",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
report = Report()
|
|
project = args.project.resolve() if args.project is not None else None
|
|
root, source = resolve_deploy_root(project)
|
|
layout_ok = check_layout(report, root, source, project)
|
|
check_toolchain(report)
|
|
if not layout_ok:
|
|
report.add(SKIP, 3, "Argo CD 指针", "(no deploy root)")
|
|
report.add(SKIP, 4, "至少有一个 compose.yaml", "(no deploy root)")
|
|
report.add(SKIP, 5, "每个服务能解析 node", "(no deploy root)")
|
|
report.add(SKIP, 6, "node 出现在 SSH config", "(no deploy root)")
|
|
report.add(SKIP, 7, "list 可发现服务", "(no deploy root)")
|
|
print()
|
|
print(f"RESULT: FAILED ({report.failures} check(s) failed)")
|
|
return 1
|
|
|
|
assert root is not None
|
|
lib.PROJECT_ROOT = root
|
|
lib._SSH_HOSTS = None
|
|
previous_cwd = Path.cwd()
|
|
try:
|
|
os.chdir(root)
|
|
has_argocd = check_argocd(report, root)
|
|
services = compose_service_dirs(root)
|
|
if services:
|
|
check_services(report, root)
|
|
elif has_argocd:
|
|
report.add(SKIP, 4, "至少有一个 compose.yaml", "Argo CD only; no compose env")
|
|
report.add(SKIP, 5, "每个服务能解析 node", "Argo CD only")
|
|
report.add(SKIP, 6, "node 出现在 SSH config", "Argo CD only")
|
|
report.add(SKIP, 7, "list 可发现服务", "Argo CD only")
|
|
else:
|
|
check_services(report, root)
|
|
finally:
|
|
os.chdir(previous_cwd)
|
|
|
|
print()
|
|
if report.failures:
|
|
print(f"RESULT: FAILED ({report.failures} check(s) failed)")
|
|
return 1
|
|
print("RESULT: PASSED")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|