Ghaymah intern

q5-mithal-monitor/monitor.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""
+Monitoring script for mithal.space
+Collects every 60 seconds:
+  - Latency: HTTP response time for the homepage
+  - Uptime: whether the site responds with a healthy status code
+  - SSL: certificate validity and days remaining until expiry
+  - DNS: DNS resolution time
+  - Search response: time to get a response from the search endpoint
+
+Storage:
+  - dashboard/data/metrics_log.csv     -> full historical log (append-only)
+  - dashboard/data/metrics_history.json -> rolling window (last 24h) used by the dashboard
+
+Only standard library is used - no extra dependencies to install.
+"""
+
+import csv
+import json
+import socket
+import ssl
+import time
+import urllib.request
+import urllib.error
+from datetime import datetime, timezone
+from pathlib import Path
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+TARGET_HOST = "mithal.space"
+TARGET_URL = "https://mithal.space/"
+SEARCH_URL = "https://mithal.space/search?q=test"   # adjust if the real search path differs
+CHECK_INTERVAL_SECONDS = 60
+REQUEST_TIMEOUT_SECONDS = 10
+ROLLING_WINDOW_MINUTES = 24 * 60   # keep last 24 hours (1 check/minute)
+
+DATA_DIR = Path(__file__).parent / "dashboard" / "data"
+CSV_LOG_PATH = DATA_DIR / "metrics_log.csv"
+HISTORY_JSON_PATH = DATA_DIR / "metrics_history.json"
+
+CSV_FIELDS = [
+    "timestamp", "up", "status_code", "latency_ms",
+    "dns_ms", "ssl_valid", "ssl_days_remaining", "ssl_expiry",
+    "search_status", "search_latency_ms", "error",
+]
+
+
+def now_iso():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def check_dns(host):
+    """Measure DNS resolution time in milliseconds."""
+    start = time.perf_counter()
+    try:
+        socket.getaddrinfo(host, 443)
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+        return elapsed_ms, None
+    except socket.gaierror as exc:
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+        return elapsed_ms, str(exc)
+
+
+def check_ssl(host, port=443):
+    """Connect to the host and read certificate expiry info."""
+    context = ssl.create_default_context()
+    try:
+        with socket.create_connection((host, port), timeout=REQUEST_TIMEOUT_SECONDS) as sock:
+            with context.wrap_socket(sock, server_hostname=host) as ssock:
+                cert = ssock.getpeercert()
+        not_after = cert.get("notAfter")
+        expiry_dt = datetime.strptime(not_after, "%b %d %H:%M:%S %Y %Z").replace(tzinfo=timezone.utc)
+        days_remaining = (expiry_dt - datetime.now(timezone.utc)).days
+        return {
+            "ssl_valid": True,
+            "ssl_days_remaining": days_remaining,
+            "ssl_expiry": expiry_dt.isoformat(),
+        }, None
+    except Exception as exc:
+        return {
+            "ssl_valid": False,
+            "ssl_days_remaining": None,
+            "ssl_expiry": None,
+        }, str(exc)
+
+
+def check_http(url):
+    """Measure HTTP response time and status code."""
+    start = time.perf_counter()
+    try:
+        req = urllib.request.Request(url, headers={"User-Agent": "mithal-monitor/1.0"})
+        with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT_SECONDS) as resp:
+            elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+            return resp.getcode(), elapsed_ms, None
+    except urllib.error.HTTPError as exc:
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+        return exc.code, elapsed_ms, None
+    except Exception as exc:
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+        return None, elapsed_ms, str(exc)
+
+
+def run_check():
+    record = {"timestamp": now_iso(), "error": None}
+
+    # 1) DNS resolution time
+    dns_ms, dns_err = check_dns(TARGET_HOST)
+    record["dns_ms"] = dns_ms
+
+    # 2) SSL certificate status
+    ssl_info, ssl_err = check_ssl(TARGET_HOST)
+    record.update(ssl_info)
+
+    # 3) HTTP latency + uptime (status code)
+    status_code, latency_ms, http_err = check_http(TARGET_URL)
+    record["status_code"] = status_code
+    record["latency_ms"] = latency_ms
+    record["up"] = status_code is not None and 200 <= status_code < 400
+
+    # 4) Search response time
+    search_status, search_latency_ms, search_err = check_http(SEARCH_URL)
+    record["search_status"] = search_status
+    record["search_latency_ms"] = search_latency_ms
+
+    errors = [e for e in (dns_err, ssl_err, http_err, search_err) if e]
+    if errors:
+        record["error"] = " | ".join(errors)
+
+    return record
+
+
+def append_csv(record):
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    file_exists = CSV_LOG_PATH.exists()
+    with CSV_LOG_PATH.open("a", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
+        if not file_exists:
+            writer.writeheader()
+        writer.writerow(record)
+
+
+def update_history_json(record):
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    history = []
+    if HISTORY_JSON_PATH.exists():
+        try:
+            history = json.loads(HISTORY_JSON_PATH.read_text(encoding="utf-8"))
+        except json.JSONDecodeError:
+            history = []
+    history.append(record)
+    history = history[-ROLLING_WINDOW_MINUTES:]
+    HISTORY_JSON_PATH.write_text(json.dumps(history, indent=2, ensure_ascii=False), encoding="utf-8")
+
+
+def main():
+    print(f"[{now_iso()}] Starting monitor for {TARGET_URL} (every {CHECK_INTERVAL_SECONDS}s)")
+    while True:
+        record = run_check()
+        append_csv(record)
+        update_history_json(record)
+
+        status_label = "UP" if record["up"] else "DOWN"
+        print(
+            f"[{record['timestamp']}] {status_label} "
+            f"status={record['status_code']} latency={record['latency_ms']}ms "
+            f"dns={record['dns_ms']}ms ssl_days={record['ssl_days_remaining']} "
+            f"search={record['search_latency_ms']}ms"
+        )
+        if record["error"]:
+            print(f"  -> issues: {record['error']}")
+
+        time.sleep(CHECK_INTERVAL_SECONDS)
+
+
+if __name__ == "__main__":
+    main()