6863d9c0c5
Some checks failed
Architecture Lint / Linter Tests (pull_request) Successful in 9s
PR Checklist / pr-checklist (pull_request) Failing after 1m16s
Smoke Test / smoke (pull_request) Failing after 6s
Validate Config / YAML Lint (pull_request) Failing after 6s
Validate Config / JSON Validate (pull_request) Successful in 6s
Validate Config / Python Syntax & Import Check (pull_request) Failing after 7s
Validate Config / Shell Script Lint (pull_request) Successful in 14s
Validate Config / Cron Syntax Check (pull_request) Successful in 5s
Validate Config / Deploy Script Dry Run (pull_request) Successful in 5s
Validate Config / Playbook Schema Validation (pull_request) Successful in 8s
Architecture Lint / Lint Repository (pull_request) Failing after 8s
One-page terminal dashboard for the Timmy Foundation fleet:
- Gitea: open PRs, issues, recent merges per repo
- VPS health: SSH reachability, service status, disk usage for ezra/allegro/bezalel
- Cron jobs: schedule, state, last run status from cron/jobs.json
Usage: python3 scripts/fleet-dashboard.py
python3 scripts/fleet-dashboard.py --json
Uses existing gitea_client.py patterns for Gitea API access.
No external dependencies -- stdlib only.
391 lines
14 KiB
Python
Executable File
391 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
fleet-dashboard.py -- Timmy Foundation Fleet Status Dashboard.
|
|
|
|
One-page terminal dashboard showing:
|
|
1. Gitea: open PRs, open issues, recent merges
|
|
2. VPS health: SSH reachability, service status, disk usage
|
|
3. Cron jobs: scheduled jobs, last run status
|
|
|
|
Usage:
|
|
python3 scripts/fleet-dashboard.py
|
|
python3 scripts/fleet-dashboard.py --json # machine-readable output
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime, timezone, timedelta
|
|
from pathlib import Path
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
GITEA_BASE = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
|
|
GITEA_API = f"{GITEA_BASE}/api/v1"
|
|
GITEA_ORG = "Timmy_Foundation"
|
|
|
|
# Key repos to check for PRs/issues
|
|
REPOS = [
|
|
"timmy-config",
|
|
"the-nexus",
|
|
"hermes-agent",
|
|
"the-forge",
|
|
"timmy-sandbox",
|
|
]
|
|
|
|
# VPS fleet
|
|
VPS_HOSTS = {
|
|
"ezra": {
|
|
"ip": "143.198.27.163",
|
|
"ssh_user": "root",
|
|
"services": ["nginx", "gitea", "docker"],
|
|
},
|
|
"allegro": {
|
|
"ip": "167.99.126.228",
|
|
"ssh_user": "root",
|
|
"services": ["hermes-agent"],
|
|
},
|
|
"bezalel": {
|
|
"ip": "159.203.146.185",
|
|
"ssh_user": "root",
|
|
"services": ["hermes-agent", "evennia"],
|
|
},
|
|
}
|
|
|
|
CRON_JOBS_FILE = Path(__file__).parent.parent / "cron" / "jobs.json"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gitea helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _gitea_token() -> str:
|
|
for p in [
|
|
Path.home() / ".hermes" / "gitea_token",
|
|
Path.home() / ".hermes" / "gitea_token_vps",
|
|
Path.home() / ".config" / "gitea" / "token",
|
|
]:
|
|
if p.exists():
|
|
return p.read_text().strip()
|
|
return ""
|
|
|
|
|
|
def _gitea_get(path: str, params: dict | None = None) -> list | dict:
|
|
url = f"{GITEA_API}{path}"
|
|
if params:
|
|
qs = "&".join(f"{k}={v}" for k, v in params.items() if v is not None)
|
|
if qs:
|
|
url += f"?{qs}"
|
|
req = urllib.request.Request(url)
|
|
token = _gitea_token()
|
|
if token:
|
|
req.add_header("Authorization", f"token {token}")
|
|
req.add_header("Accept", "application/json")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
return json.loads(resp.read())
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
def check_gitea_health() -> dict:
|
|
"""Ping Gitea and collect PR/issue stats."""
|
|
result = {"reachable": False, "version": "", "repos": {}, "totals": {}}
|
|
|
|
# Ping
|
|
data = _gitea_get("/version")
|
|
if isinstance(data, dict) and "error" not in data:
|
|
result["reachable"] = True
|
|
result["version"] = data.get("version", "unknown")
|
|
elif isinstance(data, dict) and "error" in data:
|
|
return result
|
|
|
|
total_open_prs = 0
|
|
total_open_issues = 0
|
|
total_recent_merges = 0
|
|
cutoff = (datetime.now(timezone.utc) - timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
for repo in REPOS:
|
|
repo_path = f"/repos/{GITEA_ORG}/{repo}"
|
|
repo_info = {"prs": [], "issues": [], "recent_merges": 0}
|
|
|
|
# Open PRs
|
|
prs = _gitea_get(f"{repo_path}/pulls", {"state": "open", "limit": "10", "sort": "newest"})
|
|
if isinstance(prs, list):
|
|
for pr in prs:
|
|
repo_info["prs"].append({
|
|
"number": pr.get("number"),
|
|
"title": pr.get("title", "")[:60],
|
|
"user": pr.get("user", {}).get("login", "unknown"),
|
|
"created": pr.get("created_at", "")[:10],
|
|
})
|
|
total_open_prs += len(prs)
|
|
|
|
# Open issues (excluding PRs)
|
|
issues = _gitea_get(f"{repo_path}/issues", {
|
|
"state": "open", "type": "issues", "limit": "10", "sort": "newest"
|
|
})
|
|
if isinstance(issues, list):
|
|
for iss in issues:
|
|
repo_info["issues"].append({
|
|
"number": iss.get("number"),
|
|
"title": iss.get("title", "")[:60],
|
|
"user": iss.get("user", {}).get("login", "unknown"),
|
|
"created": iss.get("created_at", "")[:10],
|
|
})
|
|
total_open_issues += len(issues)
|
|
|
|
# Recent merges (closed PRs)
|
|
merged = _gitea_get(f"{repo_path}/pulls", {"state": "closed", "limit": "20", "sort": "newest"})
|
|
if isinstance(merged, list):
|
|
recent = [p for p in merged if p.get("merged") and p.get("closed_at", "") >= cutoff]
|
|
repo_info["recent_merges"] = len(recent)
|
|
total_recent_merges += len(recent)
|
|
|
|
result["repos"][repo] = repo_info
|
|
|
|
result["totals"] = {
|
|
"open_prs": total_open_prs,
|
|
"open_issues": total_open_issues,
|
|
"recent_merges_7d": total_recent_merges,
|
|
}
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VPS health helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def check_ssh(ip: str, timeout: int = 5) -> bool:
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(timeout)
|
|
result = sock.connect_ex((ip, 22))
|
|
sock.close()
|
|
return result == 0
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def check_service(ip: str, user: str, service: str) -> str:
|
|
"""Check if a systemd service is active on remote host."""
|
|
cmd = f"ssh -o StrictHostKeyChecking=no -o ConnectTimeout=8 {user}@{ip} 'systemctl is-active {service} 2>/dev/null || echo inactive'"
|
|
try:
|
|
proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=15)
|
|
return proc.stdout.strip() or "unknown"
|
|
except subprocess.TimeoutExpired:
|
|
return "timeout"
|
|
except Exception:
|
|
return "error"
|
|
|
|
|
|
def check_disk(ip: str, user: str) -> dict:
|
|
cmd = f"ssh -o StrictHostKeyChecking=no -o ConnectTimeout=8 {user}@{ip} 'df -h / | tail -1'"
|
|
try:
|
|
proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=15)
|
|
if proc.returncode == 0:
|
|
parts = proc.stdout.strip().split()
|
|
if len(parts) >= 5:
|
|
return {"total": parts[1], "used": parts[2], "avail": parts[3], "pct": parts[4]}
|
|
except Exception:
|
|
pass
|
|
return {"total": "?", "used": "?", "avail": "?", "pct": "?"}
|
|
|
|
|
|
def check_vps_health() -> dict:
|
|
result = {}
|
|
for name, cfg in VPS_HOSTS.items():
|
|
ip = cfg["ip"]
|
|
ssh_up = check_ssh(ip)
|
|
entry = {"ip": ip, "ssh": ssh_up, "services": {}, "disk": {}}
|
|
if ssh_up:
|
|
for svc in cfg.get("services", []):
|
|
entry["services"][svc] = check_service(ip, cfg["ssh_user"], svc)
|
|
entry["disk"] = check_disk(ip, cfg["ssh_user"])
|
|
result[name] = entry
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cron job status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def check_cron_jobs() -> list[dict]:
|
|
jobs = []
|
|
if not CRON_JOBS_FILE.exists():
|
|
return [{"name": "jobs.json", "status": "FILE NOT FOUND"}]
|
|
try:
|
|
data = json.loads(CRON_JOBS_FILE.read_text())
|
|
for job in data.get("jobs", []):
|
|
jobs.append({
|
|
"name": job.get("name", "unnamed"),
|
|
"schedule": job.get("schedule_display", job.get("schedule", {}).get("display", "?")),
|
|
"enabled": job.get("enabled", False),
|
|
"state": job.get("state", "unknown"),
|
|
"completed": job.get("repeat", {}).get("completed", 0),
|
|
"last_status": job.get("last_status") or "never run",
|
|
"last_error": job.get("last_error"),
|
|
})
|
|
except Exception as e:
|
|
jobs.append({"name": "jobs.json", "status": f"PARSE ERROR: {e}"})
|
|
return jobs
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Terminal rendering
|
|
# ---------------------------------------------------------------------------
|
|
|
|
BOLD = "\033[1m"
|
|
DIM = "\033[2m"
|
|
GREEN = "\033[32m"
|
|
RED = "\033[31m"
|
|
YELLOW = "\033[33m"
|
|
CYAN = "\033[36m"
|
|
RESET = "\033[0m"
|
|
|
|
|
|
def _ok(val: bool) -> str:
|
|
return f"{GREEN}UP{RESET}" if val else f"{RED}DOWN{RESET}"
|
|
|
|
|
|
def _svc_icon(status: str) -> str:
|
|
s = status.lower().strip()
|
|
if s in ("active", "running"):
|
|
return f"{GREEN}active{RESET}"
|
|
elif s in ("inactive", "dead", "failed"):
|
|
return f"{RED}{s}{RESET}"
|
|
elif s == "timeout":
|
|
return f"{YELLOW}timeout{RESET}"
|
|
else:
|
|
return f"{YELLOW}{s}{RESET}"
|
|
|
|
|
|
def render_dashboard(gitea: dict, vps: dict, cron: list[dict]) -> str:
|
|
lines = []
|
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
|
lines.append("")
|
|
lines.append(f"{BOLD}{'=' * 72}{RESET}")
|
|
lines.append(f"{BOLD} TIMMY FOUNDATION -- FLEET STATUS DASHBOARD{RESET}")
|
|
lines.append(f"{DIM} Generated: {now}{RESET}")
|
|
lines.append(f"{BOLD}{'=' * 72}{RESET}")
|
|
|
|
# ── Section 1: Gitea ──────────────────────────────────────────────────
|
|
lines.append("")
|
|
lines.append(f"{BOLD}{CYAN} [1] GITEA{RESET}")
|
|
lines.append(f" {'-' * 68}")
|
|
if gitea.get("reachable"):
|
|
lines.append(f" Status: {GREEN}REACHABLE{RESET} (version {gitea.get('version', '?')})")
|
|
t = gitea.get("totals", {})
|
|
lines.append(f" Totals: {t.get('open_prs', 0)} open PRs | {t.get('open_issues', 0)} open issues | {t.get('recent_merges_7d', 0)} merges (7d)")
|
|
lines.append("")
|
|
for repo_name, repo in gitea.get("repos", {}).items():
|
|
prs = repo.get("prs", [])
|
|
issues = repo.get("issues", [])
|
|
merges = repo.get("recent_merges", 0)
|
|
lines.append(f" {BOLD}{repo_name}{RESET} ({len(prs)} PRs, {len(issues)} issues, {merges} merges/7d)")
|
|
for pr in prs[:5]:
|
|
lines.append(f" PR #{pr['number']:>4} {pr['title'][:50]:<50} {DIM}{pr['user']}{RESET} {pr['created']}")
|
|
for iss in issues[:3]:
|
|
lines.append(f" IS #{iss['number']:>4} {iss['title'][:50]:<50} {DIM}{iss['user']}{RESET} {iss['created']}")
|
|
else:
|
|
lines.append(f" Status: {RED}UNREACHABLE{RESET}")
|
|
|
|
# ── Section 2: VPS Health ─────────────────────────────────────────────
|
|
lines.append("")
|
|
lines.append(f"{BOLD}{CYAN} [2] VPS HEALTH{RESET}")
|
|
lines.append(f" {'-' * 68}")
|
|
lines.append(f" {'Host':<12} {'IP':<18} {'SSH':<8} {'Disk':<12} {'Services'}")
|
|
lines.append(f" {'-' * 12} {'-' * 17} {'-' * 7} {'-' * 11} {'-' * 30}")
|
|
for name, info in vps.items():
|
|
ssh_str = _ok(info["ssh"])
|
|
disk = info.get("disk", {})
|
|
disk_str = disk.get("pct", "?")
|
|
if disk_str != "?":
|
|
pct_val = int(disk_str.rstrip("%"))
|
|
if pct_val >= 90:
|
|
disk_str = f"{RED}{disk_str}{RESET}"
|
|
elif pct_val >= 75:
|
|
disk_str = f"{YELLOW}{disk_str}{RESET}"
|
|
else:
|
|
disk_str = f"{GREEN}{disk_str}{RESET}"
|
|
svc_parts = []
|
|
for svc, status in info.get("services", {}).items():
|
|
svc_parts.append(f"{svc}:{_svc_icon(status)}")
|
|
svc_str = " ".join(svc_parts) if svc_parts else f"{DIM}n/a{RESET}"
|
|
lines.append(f" {name:<12} {info['ip']:<18} {ssh_str:<18} {disk_str:<22} {svc_str}")
|
|
|
|
# ── Section 3: Cron Jobs ──────────────────────────────────────────────
|
|
lines.append("")
|
|
lines.append(f"{BOLD}{CYAN} [3] CRON JOBS{RESET}")
|
|
lines.append(f" {'-' * 68}")
|
|
lines.append(f" {'Name':<28} {'Schedule':<16} {'State':<12} {'Last':<12} {'Runs'}")
|
|
lines.append(f" {'-' * 27} {'-' * 15} {'-' * 11} {'-' * 11} {'-' * 5}")
|
|
for job in cron:
|
|
name = job.get("name", "?")[:27]
|
|
sched = job.get("schedule", "?")[:15]
|
|
state = job.get("state", "?")
|
|
if state == "scheduled":
|
|
state_str = f"{GREEN}{state}{RESET}"
|
|
elif state == "paused":
|
|
state_str = f"{YELLOW}{state}{RESET}"
|
|
else:
|
|
state_str = state
|
|
last = job.get("last_status", "never")[:11]
|
|
if last == "ok":
|
|
last_str = f"{GREEN}{last}{RESET}"
|
|
elif last in ("error", "never run"):
|
|
last_str = f"{RED}{last}{RESET}"
|
|
else:
|
|
last_str = last
|
|
runs = job.get("completed", 0)
|
|
enabled = job.get("enabled", False)
|
|
marker = " " if enabled else f"{DIM}(disabled){RESET}"
|
|
lines.append(f" {name:<28} {sched:<16} {state_str:<22} {last_str:<22} {runs} {marker}")
|
|
|
|
# ── Footer ────────────────────────────────────────────────────────────
|
|
lines.append("")
|
|
lines.append(f"{BOLD}{'=' * 72}{RESET}")
|
|
lines.append(f"{DIM} python3 scripts/fleet-dashboard.py | timmy-config{RESET}")
|
|
lines.append(f"{BOLD}{'=' * 72}{RESET}")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
json_mode = "--json" in sys.argv
|
|
|
|
if not json_mode:
|
|
print(f"\n {DIM}Collecting fleet data...{RESET}\n", file=sys.stderr)
|
|
|
|
gitea = check_gitea_health()
|
|
vps = check_vps_health()
|
|
cron = check_cron_jobs()
|
|
|
|
if json_mode:
|
|
output = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"gitea": gitea,
|
|
"vps": vps,
|
|
"cron": cron,
|
|
}
|
|
print(json.dumps(output, indent=2))
|
|
else:
|
|
print(render_dashboard(gitea, vps, cron))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|