Ghaymah SRE exam submission - Mohamed Adel

q5-mithal-monitor/monitor.py

@@ -0,0 +1,172 @@
+"""
+monitor.py — Monitoring script for mithal.space (Q5)
+
+Collects every 60 seconds:
+  - Latency: HTTP request response time
+  - Uptime: is the site reachable? (status code)
+  - SSL: certificate validity + days until expiry
+  - DNS: DNS resolution time
+  - Search Response: time to send a query and get a response
+
+Stores results in both CSV and JSON so dashboard.html can read either.
+
+Usage:
+    python3 monitor.py --url https://mithal.space --interval 60
+    python3 monitor.py --once          # single run, useful for cron/testing
+"""
+
+import argparse
+import csv
+import json
+import socket
+import ssl
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import urlparse
+
+import urllib.request
+import urllib.error
+
+CSV_FILE = Path(__file__).parent / "metrics.csv"
+JSON_FILE = Path(__file__).parent / "metrics.json"
+CSV_FIELDS = [
+    "timestamp", "up", "status_code", "latency_ms",
+    "dns_ms", "ssl_valid", "ssl_days_remaining",
+    "search_ok", "search_response_ms", "error",
+]
+
+
+def check_dns(hostname: str):
+    """Returns DNS resolution time in ms, or None on failure."""
+    start = time.perf_counter()
+    try:
+        socket.gethostbyname(hostname)
+        return round((time.perf_counter() - start) * 1000, 2)
+    except socket.gaierror:
+        return None
+
+
+def check_ssl(hostname: str, port: int = 443):
+    """Returns (is_valid, days_remaining) for the site's SSL certificate."""
+    try:
+        ctx = ssl.create_default_context()
+        with socket.create_connection((hostname, port), timeout=5) as sock:
+            with ctx.wrap_socket(sock, server_hostname=hostname) as ssock:
+                cert = ssock.getpeercert()
+        expiry = datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z")
+        expiry = expiry.replace(tzinfo=timezone.utc)
+        days_remaining = (expiry - datetime.now(timezone.utc)).days
+        return True, days_remaining
+    except Exception:
+        return False, None
+
+
+def check_http(url: str, timeout: int = 10):
+    """Returns (up, status_code, latency_ms, error)."""
+    start = time.perf_counter()
+    try:
+        req = urllib.request.Request(url, headers={"User-Agent": "Ghaymah-SRE-Monitor/1.0"})
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            latency_ms = round((time.perf_counter() - start) * 1000, 2)
+            return True, resp.status, latency_ms, None
+    except urllib.error.HTTPError as e:
+        latency_ms = round((time.perf_counter() - start) * 1000, 2)
+        # site responded, just with an error status (e.g. 404/500)
+        return e.code < 500, e.code, latency_ms, str(e)
+    except Exception as e:
+        latency_ms = round((time.perf_counter() - start) * 1000, 2)
+        return False, None, latency_ms, str(e)
+
+
+def check_search(base_url: str, query: str = "test", timeout: int = 10):
+    """
+    Simulates sending a search query to the site and measuring response time.
+    Adjust `search_path` below to match mithal.space's actual search endpoint
+    once known; falls back to hitting the homepage with a query string.
+    """
+    search_path = "/search"
+    url = f"{base_url.rstrip('/')}{search_path}?q={query}"
+    start = time.perf_counter()
+    try:
+        req = urllib.request.Request(url, headers={"User-Agent": "Ghaymah-SRE-Monitor/1.0"})
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+            return resp.status < 400, elapsed_ms
+    except Exception:
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+        return False, elapsed_ms
+
+
+def run_check(url: str):
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+
+    dns_ms = check_dns(hostname)
+    up, status_code, latency_ms, error = check_http(url)
+    ssl_valid, ssl_days = check_ssl(hostname)
+    search_ok, search_ms = check_search(url)
+
+    return {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "up": up,
+        "status_code": status_code,
+        "latency_ms": latency_ms,
+        "dns_ms": dns_ms,
+        "ssl_valid": ssl_valid,
+        "ssl_days_remaining": ssl_days,
+        "search_ok": search_ok,
+        "search_response_ms": search_ms,
+        "error": error,
+    }
+
+
+def append_csv(row: dict):
+    is_new = not CSV_FILE.exists()
+    with open(CSV_FILE, "a", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
+        if is_new:
+            writer.writeheader()
+        writer.writerow(row)
+
+
+def append_json(row: dict, keep_last: int = 1440):
+    """Keeps a rolling window of the last N checks (1440 = 24h at 1/min)."""
+    data = []
+    if JSON_FILE.exists():
+        try:
+            data = json.loads(JSON_FILE.read_text())
+        except json.JSONDecodeError:
+            data = []
+    data.append(row)
+    data = data[-keep_last:]
+    JSON_FILE.write_text(json.dumps(data, indent=2))
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Monitor mithal.space")
+    parser.add_argument("--url", default="https://mithal.space")
+    parser.add_argument("--interval", type=int, default=60, help="seconds between checks")
+    parser.add_argument("--once", action="store_true", help="run a single check and exit")
+    args = parser.parse_args()
+
+    while True:
+        result = run_check(args.url)
+        append_csv(result)
+        append_json(result)
+
+        status = "UP" if result["up"] else "DOWN"
+        print(
+            f"[{result['timestamp']}] {status} | "
+            f"latency={result['latency_ms']}ms | dns={result['dns_ms']}ms | "
+            f"ssl_days_left={result['ssl_days_remaining']} | "
+            f"search={result['search_response_ms']}ms"
+        )
+
+        if args.once:
+            break
+        time.sleep(args.interval)
+
+
+if __name__ == "__main__":
+    main()