ghaymah-exam-marwan-abdelmoneim-SRE/q5-mithal-monitor/monitor.py

#!/usr/bin/env python3
"""
Ghaymah Cloud SRE Engine — Q5 Telemetry Collector for mithal.space
Candidate: Marwan Abdelmoneim (marwanabdelmoneim / marwantamermo@gmail.com)

Measures:
1. HTTP Latency (ms) & Status Code
2. Uptime Percentage
3. SSL Certificate Expiry (Days Remaining) & Issuer
4. DNS Resolution Lookup Time (ms)
5. Search Endpoint Query Latency & Status
Saves output to metrics.json & metrics.csv.
"""

import time
import datetime
import json
import csv
import os
import sys
import socket
import ssl
import urllib.request
import urllib.parse
import urllib.error

# Force UTF-8 output encoding for Windows compatibility
if sys.stdout.encoding and sys.stdout.encoding.lower() != 'utf-8':
    try:
        sys.stdout.reconfigure(encoding='utf-8')
    except Exception:
        pass

# Configuration
TARGET_DOMAIN = "mithal.space"
TARGET_URL = f"https://{TARGET_DOMAIN}"
SEARCH_URL = f"https://{TARGET_DOMAIN}/search?q=test"
METRICS_JSON_FILE = os.path.join(os.path.dirname(__file__), "metrics.json")
METRICS_CSV_FILE = os.path.join(os.path.dirname(__file__), "metrics.csv")
MAX_HISTORY = 1440  # 24 Hours of 1-minute data points


def measure_dns(domain: str) -> float:
    """Measures DNS resolution duration in ms."""
    start = time.time()
    try:
        socket.gethostbyname(domain)
        return round((time.time() - start) * 1000, 2)
    except Exception:
        return -1.0


def check_ssl(domain: str) -> dict:
    """Inspects SSL certificate validity and days remaining."""
    context = ssl.create_default_context()
    try:
        with socket.create_connection((domain, 443), timeout=5) as sock:
            with context.wrap_socket(sock, server_hostname=domain) as ssock:
                cert = ssock.getpeercert()
                not_after_str = cert.get('notAfter')
                not_after = datetime.datetime.strptime(not_after_str, '%b %d %H:%M:%S %Y %Z').replace(tzinfo=datetime.timezone.utc)
                now = datetime.datetime.now(datetime.timezone.utc)
                days_left = (not_after - now).days
                issuer = dict(x[0] for x in cert.get('issuer', [])).get('organizationName', 'Let\'s Encrypt')
                return {"valid": True, "days_remaining": days_left, "issuer": issuer, "error": None}
    except Exception as e:
        return {"valid": False, "days_remaining": 0, "issuer": "N/A", "error": str(e)}


def measure_http(url: str) -> dict:
    """Measures HTTP GET request round-trip latency and status code."""
    start = time.time()
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "Ghaymah-SRE-Telemetry/1.0"})
        with urllib.request.urlopen(req, timeout=8) as response:
            latency = (time.time() - start) * 1000
            return {"status_code": response.status, "latency_ms": round(latency, 2), "success": True, "error": None}
    except urllib.error.HTTPError as e:
        latency = (time.time() - start) * 1000
        return {"status_code": e.code, "latency_ms": round(latency, 2), "success": (e.code == 200), "error": f"HTTP {e.code}"}
    except Exception as e:
        latency = (time.time() - start) * 1000
        return {"status_code": 0, "latency_ms": round(latency, 2), "success": False, "error": str(e)}


def collect_metrics() -> dict:
    """Runs full telemetry check against mithal.space."""
    timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
    dns_time = measure_dns(TARGET_DOMAIN)
    ssl_info = check_ssl(TARGET_DOMAIN)
    main_site = measure_http(TARGET_URL)
    search_site = measure_http(SEARCH_URL)

    return {
        "timestamp": timestamp,
        "domain": TARGET_DOMAIN,
        "uptime": 100 if main_site["success"] else 0,
        "status_code": main_site["status_code"],
        "http_latency_ms": main_site["latency_ms"],
        "dns_lookup_ms": dns_time,
        "search_latency_ms": search_site["latency_ms"],
        "search_status_code": search_site["status_code"],
        "ssl": ssl_info,
        "error": main_site["error"] or search_site["error"] or ssl_info["error"]
    }


def save_metrics(record: dict):
    """Saves telemetry to JSON and CSV data stores."""
    # JSON Update
    history = []
    if os.path.exists(METRICS_JSON_FILE):
        try:
            with open(METRICS_JSON_FILE, "r", encoding="utf-8") as f: history = json.load(f)
        except Exception: history = []
    history.append(record)
    history = history[-MAX_HISTORY:]
    with open(METRICS_JSON_FILE, "w", encoding="utf-8") as f:
        json.dump(history, f, indent=2)

    # CSV Update
    file_exists = os.path.exists(METRICS_CSV_FILE)
    fieldnames = ["timestamp", "domain", "uptime", "status_code", "http_latency_ms", "dns_lookup_ms", "search_latency_ms", "ssl_valid", "ssl_days_remaining", "error"]
    with open(METRICS_CSV_FILE, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if not file_exists: writer.writeheader()
        writer.writerow({
            "timestamp": record["timestamp"], "domain": record["domain"], "uptime": record["uptime"],
            "status_code": record["status_code"], "http_latency_ms": record["http_latency_ms"],
            "dns_lookup_ms": record["dns_lookup_ms"], "search_latency_ms": record["search_latency_ms"],
            "ssl_valid": record["ssl"]["valid"], "ssl_days_remaining": record["ssl"]["days_remaining"],
            "error": record["error"] or ""
        })


if __name__ == "__main__":
    print(f"📡 Ghaymah Telemetry Collection for: {TARGET_DOMAIN}")
    rec = collect_metrics()
    save_metrics(rec)
    print("✅ Check Completed Successfully!")
    print(json.dumps(rec, indent=2))