Ghaymah-Exam-GamalMohamed-SRE/Q5/monitor.py

#!/usr/bin/env python3
"""
monitor.py — Lightweight uptime / latency / SSL monitor.

Standard-library only (urllib, ssl, socket, json, time). Runs an infinite
loop, checks a target URL (and optionally a search endpoint) every
CHECK_INTERVAL seconds, and persists a rolling RETENTION_HOURS window of
results to a JSON file that the static dashboard reads.

Configuration is via environment variables so the same image can monitor
any site without a rebuild:

    TARGET_URL        Full URL to monitor            (default: https://mithal.space)
    SEARCH_PATH        Path appended to origin for a  (default: /search?q=test)
                        secondary "search" check. Set to "" to disable.
    CHECK_INTERVAL     Seconds between checks         (default: 60)
    RETENTION_HOURS    How much history to keep       (default: 24)
    REQUEST_TIMEOUT    Per-request timeout, seconds   (default: 10)
    DATA_FILE          Where to write the JSON log    (default: /app/web/data/metrics.json)
"""

import json
import os
import socket
import ssl
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from urllib.parse import urlparse

# --------------------------------------------------------------------------
# Configuration
# --------------------------------------------------------------------------

TARGET_URL = os.environ.get("TARGET_URL", "https://mithal.space")
SEARCH_PATH = os.environ.get("SEARCH_PATH", "/search?q=test")
CHECK_INTERVAL = int(os.environ.get("CHECK_INTERVAL", "60"))
RETENTION_HOURS = float(os.environ.get("RETENTION_HOURS", "24"))
REQUEST_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", "10"))
DATA_FILE = os.environ.get("DATA_FILE", "/app/web/data/metrics.json")
USER_AGENT = "uptime-monitor/1.0 (+standard-library)"

_parsed = urlparse(TARGET_URL)
HOSTNAME = _parsed.hostname
PORT = _parsed.port or (443 if _parsed.scheme == "https" else 80)
SEARCH_URL = f"{_parsed.scheme}://{_parsed.netloc}{SEARCH_PATH}" if SEARCH_PATH else None


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def measure_dns(hostname: str):
    """Return DNS resolution time in milliseconds, or None on failure."""
    start = time.perf_counter()
    try:
        socket.getaddrinfo(hostname, None)
    except socket.gaierror as exc:
        return None, str(exc)
    elapsed_ms = (time.perf_counter() - start) * 1000
    return round(elapsed_ms, 2), None


def timed_get(url: str, timeout: float):
    """
    Perform an HTTP GET and return (status_code, latency_ms, error_str).
    latency_ms measures time-to-first-byte-of-full-response (connect + TLS
    + request + response), matching what a real visitor experiences.
    """
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    start = time.perf_counter()
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            resp.read(1)  # confirm the body actually starts streaming
            status = resp.status
        latency_ms = (time.perf_counter() - start) * 1000
        return status, round(latency_ms, 2), None
    except urllib.error.HTTPError as exc:
        # Still a "successful" connection from a monitoring standpoint —
        # the server responded, just with an error status.
        latency_ms = (time.perf_counter() - start) * 1000
        return exc.code, round(latency_ms, 2), None
    except (urllib.error.URLError, socket.timeout, TimeoutError) as exc:
        latency_ms = (time.perf_counter() - start) * 1000
        return None, round(latency_ms, 2), str(getattr(exc, "reason", exc))


def measure_ssl_expiry(hostname: str, port: int, timeout: float):
    """Return (days_remaining, error_str) for the TLS certificate."""
    if not hostname:
        return None, "no hostname"
    try:
        ctx = ssl.create_default_context()
        with socket.create_connection((hostname, port), timeout=timeout) as sock:
            with ctx.wrap_socket(sock, server_hostname=hostname) as ssock:
                cert = ssock.getpeercert()
        not_after = cert.get("notAfter")
        if not not_after:
            return None, "no notAfter field in certificate"
        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)).total_seconds() / 86400
        return round(days_remaining, 1), None
    except Exception as exc:  # noqa: BLE001 — monitoring must never crash the loop
        return None, str(exc)


def run_check() -> dict:
    dns_ms, dns_err = measure_dns(HOSTNAME)
    status, latency_ms, http_err = timed_get(TARGET_URL, REQUEST_TIMEOUT)

    search_status, search_latency_ms, search_err = None, None, None
    if SEARCH_URL:
        search_status, search_latency_ms, search_err = timed_get(SEARCH_URL, REQUEST_TIMEOUT)

    ssl_days, ssl_err = (None, None)
    if _parsed.scheme == "https":
        ssl_days, ssl_err = measure_ssl_expiry(HOSTNAME, PORT, REQUEST_TIMEOUT)

    success = status is not None and 200 <= status < 400

    return {
        "timestamp": now_iso(),
        "target": TARGET_URL,
        "http_status": status,
        "success": success,
        "latency_ms": latency_ms,
        "dns_ms": dns_ms,
        "search_status": search_status,
        "search_latency_ms": search_latency_ms,
        "ssl_days_remaining": ssl_days,
        "error": http_err or dns_err or ssl_err or search_err,
    }


def load_history(path: str) -> list:
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
            return data.get("checks", []) if isinstance(data, dict) else data
    except (FileNotFoundError, json.JSONDecodeError):
        return []


def prune(history: list, retention_hours: float) -> list:
    cutoff = time.time() - retention_hours * 3600
    pruned = []
    for entry in history:
        try:
            ts = datetime.fromisoformat(entry["timestamp"]).timestamp()
        except (KeyError, ValueError):
            continue
        if ts >= cutoff:
            pruned.append(entry)
    return pruned


def save_history(path: str, history: list) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    payload = {
        "target": TARGET_URL,
        "search_url": SEARCH_URL,
        "updated_at": now_iso(),
        "check_interval_seconds": CHECK_INTERVAL,
        "retention_hours": RETENTION_HOURS,
        "checks": history,
    }
    tmp_path = f"{path}.tmp"
    with open(tmp_path, "w", encoding="utf-8") as f:
        json.dump(payload, f, indent=2)
    os.replace(tmp_path, path)  # atomic write so the dashboard never reads a half-written file


def main() -> None:
    print(f"[monitor] target={TARGET_URL} interval={CHECK_INTERVAL}s "
          f"retention={RETENTION_HOURS}h data_file={DATA_FILE}", flush=True)

    history = load_history(DATA_FILE)

    while True:
        cycle_start = time.time()
        result = run_check()
        history.append(result)
        history = prune(history, RETENTION_HOURS)
        save_history(DATA_FILE, history)

        status_str = result["http_status"] if result["http_status"] is not None else "ERR"
        print(
            f"[monitor] {result['timestamp']} status={status_str} "
            f"latency={result['latency_ms']}ms dns={result['dns_ms']}ms "
            f"ssl_days={result['ssl_days_remaining']} "
            f"search_latency={result['search_latency_ms']}ms "
            f"{'error=' + result['error'] if result['error'] else 'ok'}",
            flush=True,
        )

        elapsed = time.time() - cycle_start
        time.sleep(max(0, CHECK_INTERVAL - elapsed))


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(0)