Codex Release Bot
225 lines
7.8 KiB
Python
225 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Batch SSH diagnostics for Nexus Agent on sub-servers (run from central / dev with .env).
|
|
|
|
Collects per server:
|
|
- Central: Redis heartbeat, MySQL last_heartbeat, agent_api_key_set
|
|
- Remote (SSH): systemctl, redacted config.json, recent 401/heartbeat errors, /health curl
|
|
|
|
Usage:
|
|
python scripts/agent_remote_diagnose.py --id 95
|
|
python scripts/agent_remote_diagnose.py --name 颜又文
|
|
python scripts/agent_remote_diagnose.py --offline-only --limit 20
|
|
python scripts/agent_remote_diagnose.py --ids 95,12,33
|
|
|
|
Requires .env with DATABASE_URL + REDIS_URL (or NEXUS_*). SSH creds from servers table.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from server.application.agent_diagnose_service import ( # noqa: E402
|
|
REDIS_KEY_PREFIX,
|
|
central_status,
|
|
diagnose_server_record,
|
|
has_agent_monitoring,
|
|
)
|
|
|
|
|
|
def _load_dotenv() -> None:
|
|
env_path = ROOT / ".env"
|
|
if not env_path.is_file():
|
|
return
|
|
for line in env_path.read_text(encoding="utf-8", errors="replace").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
key, value = key.strip(), value.strip().strip("\"'")
|
|
if key.startswith("NEXUS_") and key[6:] not in os.environ:
|
|
os.environ[key[6:]] = value
|
|
elif key and key not in os.environ:
|
|
os.environ[key] = value
|
|
|
|
|
|
async def _fetch_servers(
|
|
session: Any,
|
|
*,
|
|
server_id: int | None,
|
|
server_ids: list[int] | None,
|
|
name_like: str | None,
|
|
offline_only: bool,
|
|
limit: int,
|
|
) -> list[Any]:
|
|
from sqlalchemy import or_, select
|
|
|
|
from server.domain.models import Server
|
|
from server.infrastructure.redis.client import get_redis
|
|
|
|
q = select(Server).order_by(Server.id)
|
|
if server_id is not None:
|
|
q = q.where(Server.id == server_id)
|
|
elif server_ids:
|
|
q = q.where(Server.id.in_(server_ids))
|
|
if name_like:
|
|
q = q.where(or_(Server.name.like(f"%{name_like}%"), Server.domain.like(f"%{name_like}%")))
|
|
|
|
result = await session.execute(q)
|
|
servers = list(result.scalars().all())
|
|
|
|
if offline_only:
|
|
redis = get_redis()
|
|
filtered = []
|
|
for s in servers:
|
|
if not has_agent_monitoring(s):
|
|
continue
|
|
hb = await redis.hgetall(f"{REDIS_KEY_PREFIX}{s.id}")
|
|
if central_status(True, hb or None) == "offline":
|
|
filtered.append(s)
|
|
servers = filtered
|
|
|
|
if limit > 0:
|
|
servers = servers[:limit]
|
|
return servers
|
|
|
|
|
|
async def run_diagnose(args: argparse.Namespace) -> list[dict[str, Any]]:
|
|
from server.config import settings
|
|
from server.infrastructure.database.session import AsyncSessionLocal, init_db
|
|
from server.infrastructure.redis.client import close_redis, get_redis, init_redis
|
|
|
|
await init_db()
|
|
await init_redis(settings.REDIS_URL)
|
|
redis = get_redis()
|
|
|
|
ids: list[int] | None = None
|
|
if args.ids:
|
|
ids = [int(x.strip()) for x in args.ids.split(",") if x.strip()]
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
servers = await _fetch_servers(
|
|
session,
|
|
server_id=args.id,
|
|
server_ids=ids,
|
|
name_like=args.name,
|
|
offline_only=args.offline_only,
|
|
limit=args.limit,
|
|
)
|
|
|
|
sem = asyncio.Semaphore(max(1, args.concurrency))
|
|
|
|
async def one(server: Any) -> dict[str, Any]:
|
|
async with sem:
|
|
hb = await redis.hgetall(f"{REDIS_KEY_PREFIX}{server.id}")
|
|
row = await diagnose_server_record(
|
|
server,
|
|
hb or None,
|
|
ssh=not args.central_only,
|
|
timeout=args.timeout,
|
|
)
|
|
return {
|
|
"id": row["server_id"],
|
|
"name": row["server_name"],
|
|
"domain": row.get("server_domain", ""),
|
|
"central_status": row["central_status"],
|
|
"has_agent": row["has_agent"],
|
|
"agent_version_db": row["agent_version_db"],
|
|
"last_heartbeat_db": row["last_heartbeat_db"],
|
|
"last_heartbeat_redis": row["last_heartbeat_redis"],
|
|
"remote": row.get("remote"),
|
|
}
|
|
|
|
try:
|
|
return list(await asyncio.gather(*(one(s) for s in servers)))
|
|
finally:
|
|
await close_redis()
|
|
|
|
|
|
def _format_report(rows: list[dict[str, Any]], *, verbose: bool) -> str:
|
|
lines: list[str] = []
|
|
ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
|
|
lines.append(f"# Nexus Agent remote diagnose @ {ts}")
|
|
lines.append(f"servers: {len(rows)}")
|
|
lines.append("")
|
|
|
|
for row in rows:
|
|
lines.append(f"## [{row['id']}] {row['name']} ({row['domain']})")
|
|
lines.append(
|
|
f"- central: **{row['central_status']}** | "
|
|
f"redis_hb={row['last_heartbeat_redis'] or '—'} | "
|
|
f"mysql_hb={row['last_heartbeat_db'] or '—'} | "
|
|
f"agent_db={row['agent_version_db'] or '—'}"
|
|
)
|
|
remote = row.get("remote")
|
|
if remote:
|
|
if remote.get("error"):
|
|
lines.append(f"- SSH: **FAIL** — {remote['error']}")
|
|
elif not remote.get("ssh_ok"):
|
|
lines.append(
|
|
f"- SSH: exit {remote.get('exit_code')} stderr={remote.get('stderr', '')[:200]}"
|
|
)
|
|
if verbose or remote.get("stdout"):
|
|
body = remote.get("log_errors") or remote.get("systemctl_detail") or "(empty)"
|
|
lines.append("```")
|
|
lines.append(str(body)[:4000])
|
|
lines.append("```")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(description="SSH batch diagnostics for Nexus Agent on sub-servers")
|
|
p.add_argument("--id", type=int, help="Single server id")
|
|
p.add_argument("--ids", help="Comma-separated server ids")
|
|
p.add_argument("--name", help="Filter name/domain substring (SQL LIKE)")
|
|
p.add_argument("--offline-only", action="store_true", help="Only servers with Agent but no Redis heartbeat")
|
|
p.add_argument("--limit", type=int, default=50, help="Max servers (default 50)")
|
|
p.add_argument("--concurrency", type=int, default=5, help="Parallel SSH probes (default 5)")
|
|
p.add_argument("--timeout", type=int, default=45, help="SSH command timeout seconds")
|
|
p.add_argument("--central-only", action="store_true", help="Skip SSH; only Redis/MySQL snapshot")
|
|
p.add_argument("-o", "--output", help="Write report markdown to file")
|
|
p.add_argument("-q", "--quiet", action="store_true", help="Only print summary lines")
|
|
return p
|
|
|
|
|
|
def main() -> int:
|
|
_load_dotenv()
|
|
args = build_parser().parse_args()
|
|
if not any([args.id, args.ids, args.name, args.offline_only]):
|
|
print("Specify --id, --ids, --name, or --offline-only", file=sys.stderr)
|
|
return 2
|
|
|
|
rows = asyncio.run(run_diagnose(args))
|
|
if not rows:
|
|
print("No matching servers.")
|
|
return 0
|
|
|
|
report = _format_report(rows, verbose=not args.quiet)
|
|
if args.output:
|
|
Path(args.output).write_text(report, encoding="utf-8")
|
|
print(f"Wrote {args.output} ({len(rows)} servers)")
|
|
else:
|
|
print(report)
|
|
|
|
offline = sum(1 for r in rows if r["central_status"] == "offline")
|
|
ssh_fail = sum(
|
|
1 for r in rows if r.get("remote") and (r["remote"].get("error") or not r["remote"].get("ssh_ok"))
|
|
)
|
|
print(f"Summary: total={len(rows)} central_offline={offline} ssh_fail={ssh_fail}", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|