feat: Ghayma Assessment - complete submission

q1-deploy-monitor/Health-check.py

@@ -0,0 +1,148 @@
+#!/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()