149 أسطر
4.2 KiB
Python
149 أسطر
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Monitoring Script for Ghayma REST API
|
|
Scans the application health endpoint every 30 seconds and logs results.
|
|
"""
|
|
|
|
import urllib.request
|
|
import urllib.error
|
|
import json
|
|
import time
|
|
import datetime
|
|
import os
|
|
import sys
|
|
|
|
# Configuration
|
|
APP_URL = os.environ.get("APP_URL", "http://localhost:3000")
|
|
HEALTH_ENDPOINT = f"{APP_URL}/health"
|
|
CHECK_INTERVAL = 30 # seconds
|
|
LOG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "monitor.log")
|
|
TIMEOUT = 5 # seconds
|
|
|
|
# ANSI colors for terminal output
|
|
GREEN = "\033[92m"
|
|
RED = "\033[91m"
|
|
YELLOW = "\033[93m"
|
|
CYAN = "\033[96m"
|
|
RESET = "\033[0m"
|
|
BOLD = "\033[1m"
|
|
|
|
|
|
def log(message, level="INFO"):
|
|
"""Log a message to both console and log file."""
|
|
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
log_entry = f"[{timestamp}] [{level}] {message}"
|
|
|
|
# Color-coded console output
|
|
if level == "OK":
|
|
print(f"{GREEN}{log_entry}{RESET}")
|
|
elif level == "ERROR":
|
|
print(f"{RED}{BOLD}{log_entry}{RESET}")
|
|
elif level == "WARN":
|
|
print(f"{YELLOW}{log_entry}{RESET}")
|
|
else:
|
|
print(f"{CYAN}{log_entry}{RESET}")
|
|
|
|
# Append to log file
|
|
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
|
f.write(log_entry + "\n")
|
|
|
|
|
|
def check_health():
|
|
"""Send a GET request to the health endpoint and return the result."""
|
|
try:
|
|
req = urllib.request.Request(HEALTH_ENDPOINT, method="GET")
|
|
start_time = time.time()
|
|
response = urllib.request.urlopen(req, timeout=TIMEOUT)
|
|
response_time_ms = round((time.time() - start_time) * 1000)
|
|
status_code = response.status
|
|
body = json.loads(response.read().decode("utf-8"))
|
|
|
|
return {
|
|
"healthy": status_code == 200 and body.get("status") == "UP",
|
|
"status_code": status_code,
|
|
"response_time_ms": response_time_ms,
|
|
"uptime": body.get("uptime"),
|
|
"error": None,
|
|
}
|
|
|
|
except urllib.error.HTTPError as e:
|
|
return {
|
|
"healthy": False,
|
|
"status_code": e.code,
|
|
"response_time_ms": None,
|
|
"uptime": None,
|
|
"error": f"HTTP {e.code}: {e.reason}",
|
|
}
|
|
|
|
except urllib.error.URLError as e:
|
|
return {
|
|
"healthy": False,
|
|
"status_code": None,
|
|
"response_time_ms": None,
|
|
"uptime": None,
|
|
"error": f"Connection failed: {e.reason}",
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"healthy": False,
|
|
"status_code": None,
|
|
"response_time_ms": None,
|
|
"uptime": None,
|
|
"error": str(e),
|
|
}
|
|
|
|
|
|
def main():
|
|
consecutive_failures = 0
|
|
|
|
print(f"{BOLD}{CYAN}")
|
|
print("=" * 55)
|
|
print(" Ghayma REST API - Health Monitor")
|
|
print(f" Target: {HEALTH_ENDPOINT}")
|
|
print(f" Interval: {CHECK_INTERVAL}s")
|
|
print(f" Log file: {LOG_FILE}")
|
|
print("=" * 55)
|
|
print(f"{RESET}")
|
|
|
|
log(f"Monitor started | Target: {HEALTH_ENDPOINT} | Interval: {CHECK_INTERVAL}s")
|
|
|
|
try:
|
|
while True:
|
|
result = check_health()
|
|
|
|
if result["healthy"]:
|
|
consecutive_failures = 0
|
|
uptime_str = f"{result['uptime']:.1f}s" if result["uptime"] else "N/A"
|
|
log(
|
|
f"HEALTHY | Status: {result['status_code']} | "
|
|
f"Response: {result['response_time_ms']}ms | "
|
|
f"Uptime: {uptime_str}",
|
|
level="OK",
|
|
)
|
|
else:
|
|
consecutive_failures += 1
|
|
log(
|
|
f"UNHEALTHY | Error: {result['error']} | "
|
|
f"Consecutive failures: {consecutive_failures}",
|
|
level="ERROR",
|
|
)
|
|
|
|
if consecutive_failures >= 3:
|
|
log(
|
|
f"ALERT: {consecutive_failures} consecutive failures! "
|
|
f"Application may be DOWN.",
|
|
level="WARN",
|
|
)
|
|
|
|
time.sleep(CHECK_INTERVAL)
|
|
|
|
except KeyboardInterrupt:
|
|
log("Monitor stopped by user.", level="INFO")
|
|
print(f"\n{YELLOW}Monitor stopped.{RESET}")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|