105 أسطر
3.7 KiB
Python
105 أسطر
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Lightweight uptime/health monitor.
|
|
|
|
Polls the target app's /health endpoint every CHECK_INTERVAL seconds,
|
|
logs status + response time, and writes the latest result to a JSON
|
|
file (status.json) that can be consumed by other tools or dashboards.
|
|
|
|
Usage:
|
|
python health-check.py --url https://your-app.ghaymah.systems
|
|
python health-check.py --url https://your-app.ghaymah.systems --interval 30
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime, timezone
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
handlers=[
|
|
logging.FileHandler("monitor.log"),
|
|
logging.StreamHandler(),
|
|
],
|
|
)
|
|
log = logging.getLogger("monitor")
|
|
|
|
STATUS_FILE = "status.json"
|
|
FAILURE_THRESHOLD = 3 # consecutive failures before raising an "ALERT"
|
|
|
|
|
|
def check_health(url: str, timeout: float = 5.0):
|
|
start = time.perf_counter()
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as resp:
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
body = json.loads(resp.read().decode())
|
|
return {
|
|
"ok": resp.status == 200,
|
|
"http_status": resp.status,
|
|
"response_time_ms": elapsed_ms,
|
|
"body": body,
|
|
}
|
|
except urllib.error.HTTPError as e:
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return {"ok": False, "http_status": e.code, "response_time_ms": elapsed_ms, "error": str(e)}
|
|
except Exception as e: # DNS errors, timeouts, connection refused, etc.
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return {"ok": False, "http_status": None, "response_time_ms": elapsed_ms, "error": str(e)}
|
|
|
|
|
|
def write_status(result: dict, consecutive_failures: int):
|
|
payload = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"status": "healthy" if result["ok"] else "unhealthy",
|
|
"http_status": result.get("http_status"),
|
|
"response_time_ms": result.get("response_time_ms"),
|
|
"consecutive_failures": consecutive_failures,
|
|
}
|
|
with open(STATUS_FILE, "w") as f:
|
|
json.dump(payload, f, indent=2)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Poll /health endpoint on an interval.")
|
|
parser.add_argument("--url", required=True, help="Base URL of the app, e.g. https://myapp.ghaymah.systems")
|
|
parser.add_argument("--interval", type=int, default=30, help="Seconds between checks (default: 30)")
|
|
args = parser.parse_args()
|
|
|
|
health_url = args.url.rstrip("/") + "/health"
|
|
log.info("Starting monitor for %s every %ss", health_url, args.interval)
|
|
|
|
consecutive_failures = 0
|
|
|
|
while True:
|
|
result = check_health(health_url)
|
|
write_status(result, consecutive_failures)
|
|
|
|
if result["ok"]:
|
|
if consecutive_failures > 0:
|
|
log.info("Service RECOVERED after %d failed check(s).", consecutive_failures)
|
|
consecutive_failures = 0
|
|
log.info("OK status=%s response_time=%sms", result["http_status"], result["response_time_ms"])
|
|
else:
|
|
consecutive_failures += 1
|
|
log.warning(
|
|
"FAIL attempt=%d status=%s error=%s",
|
|
consecutive_failures, result.get("http_status"), result.get("error"),
|
|
)
|
|
if consecutive_failures >= FAILURE_THRESHOLD:
|
|
log.error("ALERT: %s has failed %d consecutive health checks!", args.url, consecutive_failures)
|
|
|
|
time.sleep(args.interval)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
log.info("Monitor stopped by user.")
|