ghaymah-exam-mohamed-adel-sre/q5-mithal-monitor/monitor.py

"""
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()