120 أسطر
4.0 KiB
Python
120 أسطر
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple uptime/health monitor.
|
|
Checks the deployed app's /health endpoint every 30 seconds,
|
|
logs the result, and writes a rolling JSON history file that
|
|
the dashboard (or anything else) can read.
|
|
|
|
Usage:
|
|
python monitor.py --url https://your-app.ghaymah.systems
|
|
python monitor.py --url http://localhost:3000 --interval 30
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
LOG_FILE = Path("monitor.log")
|
|
HISTORY_FILE = Path("monitor_history.json")
|
|
MAX_HISTORY = 200 # keep the last N checks
|
|
|
|
|
|
def now_iso():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def load_history():
|
|
if HISTORY_FILE.exists():
|
|
try:
|
|
return json.loads(HISTORY_FILE.read_text())
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return []
|
|
|
|
|
|
def save_history(history):
|
|
HISTORY_FILE.write_text(json.dumps(history[-MAX_HISTORY:], indent=2))
|
|
|
|
|
|
def log_line(message):
|
|
line = f"[{now_iso()}] {message}"
|
|
print(line)
|
|
with LOG_FILE.open("a", encoding="utf-8") as f:
|
|
f.write(line + "\n")
|
|
|
|
|
|
def check_health(base_url, timeout, health_path="/health"):
|
|
url = base_url.rstrip("/") + health_path
|
|
start = time.monotonic()
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "ghaymah-monitor/1.0"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
elapsed_ms = round((time.monotonic() - start) * 1000, 2)
|
|
status_code = resp.getcode()
|
|
raw = resp.read().decode("utf-8", errors="ignore")
|
|
try:
|
|
body = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
# Not a JSON API (e.g. static site returning plain "ok") -
|
|
# any 200 response is considered healthy.
|
|
body = {"raw": raw.strip()[:200]}
|
|
healthy = status_code == 200
|
|
return {
|
|
"timestamp": now_iso(),
|
|
"healthy": healthy,
|
|
"status_code": status_code,
|
|
"response_time_ms": elapsed_ms,
|
|
"detail": body,
|
|
}
|
|
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, Exception) as exc:
|
|
elapsed_ms = round((time.monotonic() - start) * 1000, 2)
|
|
return {
|
|
"timestamp": now_iso(),
|
|
"healthy": False,
|
|
"status_code": None,
|
|
"response_time_ms": elapsed_ms,
|
|
"error": str(exc),
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Health monitor for the deployed API")
|
|
parser.add_argument("--url", required=True, help="Base URL of the deployed app, e.g. https://myapp.ghaymah.systems")
|
|
parser.add_argument("--health-path", default="/health", help="Path to check (default: /health)")
|
|
parser.add_argument("--interval", type=int, default=30, help="Seconds between checks (default: 30)")
|
|
parser.add_argument("--timeout", type=int, default=5, help="Request timeout in seconds (default: 5)")
|
|
parser.add_argument("--once", action="store_true", help="Run a single check and exit (useful for testing)")
|
|
args = parser.parse_args()
|
|
|
|
history = load_history()
|
|
consecutive_failures = 0
|
|
|
|
log_line(f"Starting monitor for {args.url} (interval={args.interval}s)")
|
|
|
|
while True:
|
|
result = check_health(args.url, args.timeout, args.health_path)
|
|
history.append(result)
|
|
save_history(history)
|
|
|
|
if result["healthy"]:
|
|
consecutive_failures = 0
|
|
log_line(f"OK status={result['status_code']} response_time={result['response_time_ms']}ms")
|
|
else:
|
|
consecutive_failures += 1
|
|
reason = result.get("error", f"status_code={result.get('status_code')}")
|
|
log_line(f"FAIL consecutive_failures={consecutive_failures} reason={reason}")
|
|
if consecutive_failures == 3:
|
|
log_line("ALERT: app has failed 3 consecutive health checks")
|
|
|
|
if args.once:
|
|
break
|
|
time.sleep(args.interval)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|