#!/usr/bin/env python3 """ Mithal.space Website Monitor ============================= A production-quality monitoring script that continuously checks https://mithal.space for uptime, latency, DNS, SSL, and search performance. Usage: python monitor.py python monitor.py --interval 30 python monitor.py --interval 120 --target https://mithal.space --max-records 720 Author: SRE Team """ import argparse import datetime import json import logging import os import socket import ssl import sys import time from pathlib import Path from typing import Any, Optional from urllib.parse import urlparse import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # --------------------------------------------------------------------------- # Configuration Defaults # --------------------------------------------------------------------------- DEFAULT_TARGET_URL: str = "https://mithal.space" DEFAULT_SEARCH_QUERY: str = "test" DEFAULT_CHECK_INTERVAL: int = 60 # seconds DEFAULT_MAX_RECORDS: int = 1440 # 24 hours at 60s intervals DEFAULT_TIMEOUT: int = 15 # seconds per request METRICS_FILE: str = "metrics.json" LOG_FILE: str = "monitor.log" # --------------------------------------------------------------------------- # Logging Setup # --------------------------------------------------------------------------- def setup_logging(log_file: str = LOG_FILE, level: int = logging.INFO) -> logging.Logger: """Configure and return the application logger.""" logger = logging.getLogger("mithal_monitor") logger.setLevel(level) # Console handler console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(level) console_fmt = logging.Formatter( "[%(asctime)s] %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) console_handler.setFormatter(console_fmt) logger.addHandler(console_handler) # File handler file_handler = logging.FileHandler(log_file, encoding="utf-8") file_handler.setLevel(level) file_fmt = logging.Formatter( "[%(asctime)s] %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) file_handler.setFormatter(file_fmt) logger.addHandler(file_handler) return logger logger = setup_logging() # --------------------------------------------------------------------------- # HTTP Session Factory # --------------------------------------------------------------------------- def create_session() -> requests.Session: """Create a requests session with retry and timeout defaults.""" session = requests.Session() retries = Retry( total=2, backoff_factor=1, status_forcelist=[502, 503, 504], ) adapter = HTTPAdapter(max_retries=retries) session.mount("https://", adapter) session.mount("http://", adapter) return session # --------------------------------------------------------------------------- # Metric Collection Functions # --------------------------------------------------------------------------- def check_dns(hostname: str) -> float: """ Measure DNS lookup time in milliseconds. Returns the time taken to resolve the hostname. Raises SocketError on failure. """ start = time.monotonic() socket.getaddrinfo(hostname, None) elapsed_ms = (time.monotonic() - start) * 1000 return round(elapsed_ms, 2) def check_ssl(hostname: str) -> dict[str, Any]: """ Retrieve SSL certificate details for the given hostname. Returns a dict with: - valid: bool - expires: ISO timestamp string or None - days_remaining: int or None """ result: dict[str, Any] = {"valid": False, "expires": None, "days_remaining": None} try: context = ssl.create_default_context() with socket.create_connection((hostname, 443), timeout=DEFAULT_TIMEOUT) as sock: with context.wrap_socket(sock, server_hostname=hostname) as ssock: cert = ssock.getpeercert() if not cert: return result not_after = cert.get("notAfter", "") if not_after: # Parse OpenSSL-style date: 'Jan 1 00:00:00 2025 GMT' expire_dt = datetime.datetime.strptime( not_after, "%b %d %H:%M:%S %Y %Z" ) now = datetime.datetime.utcnow() days_left = (expire_dt - now).days result["valid"] = days_left > 0 result["expires"] = expire_dt.isoformat() + "Z" result["days_remaining"] = days_left except (ssl.SSLError, socket.error, OSError) as exc: logger.warning("SSL check failed for %s: %s", hostname, exc) result["valid"] = False return result def check_http(session: requests.Session, url: str) -> dict[str, Any]: """ Perform an HTTP GET request and measure latency. Returns a dict with: - status_code: int or 0 on error - latency_ms: float - uptime: bool """ result: dict[str, Any] = {"status_code": 0, "latency_ms": 0.0, "uptime": False} try: start = time.monotonic() resp = session.get(url, timeout=DEFAULT_TIMEOUT, allow_redirects=True) elapsed_ms = (time.monotonic() - start) * 1000 result["status_code"] = resp.status_code result["latency_ms"] = round(elapsed_ms, 2) result["uptime"] = 200 <= resp.status_code < 400 except requests.exceptions.Timeout: logger.warning("HTTP request timed out for %s", url) except requests.exceptions.ConnectionError as exc: logger.warning("Connection error for %s: %s", url, exc) except requests.exceptions.RequestException as exc: logger.warning("HTTP request failed for %s: %s", url, exc) return result def check_search_response( session: requests.Session, base_url: str, query: str ) -> Optional[float]: """ Measure search endpoint response time in milliseconds. mithal.space exposes a public search endpoint at /search?q=. We send a real search request and measure the response time. Returns: Response time in milliseconds, or None on failure. """ search_url = f"{base_url}/search?q={query}" try: start = time.monotonic() resp = session.get(search_url, timeout=DEFAULT_TIMEOUT, allow_redirects=True) elapsed_ms = (time.monotonic() - start) * 1000 if resp.status_code == 200: return round(elapsed_ms, 2) logger.warning("Search endpoint returned status %d", resp.status_code) return None except requests.exceptions.RequestException as exc: logger.warning("Search request failed: %s", exc) return None # --------------------------------------------------------------------------- # Metrics Persistence # --------------------------------------------------------------------------- def load_metrics(metrics_path: str) -> list[dict[str, Any]]: """Load existing metrics from the JSON file.""" if not os.path.exists(metrics_path): return [] try: with open(metrics_path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): return data return [] except (json.JSONDecodeError, IOError) as exc: logger.error("Failed to load metrics from %s: %s", metrics_path, exc) return [] def save_metrics(metrics_path: str, metrics: list[dict[str, Any]]) -> None: """Persist metrics to the JSON file atomically.""" tmp_path = metrics_path + ".tmp" try: with open(tmp_path, "w", encoding="utf-8") as f: json.dump(metrics, f, indent=2, ensure_ascii=False) os.replace(tmp_path, metrics_path) except IOError as exc: logger.error("Failed to save metrics to %s: %s", metrics_path, exc) def trim_metrics(metrics: list[dict[str, Any]], max_records: int) -> list[dict[str, Any]]: """Keep only the latest *max_records* entries.""" if len(metrics) > max_records: return metrics[-max_records:] return metrics # --------------------------------------------------------------------------- # Single Check # --------------------------------------------------------------------------- def perform_check( session: requests.Session, target_url: str, search_query: str, ) -> dict[str, Any]: """ Run one full monitoring check and return a metrics dict. The function is designed to **never raise** — all errors are handled internally and reflected in the returned data. """ parsed = urlparse(target_url) hostname = parsed.hostname or "mithal.space" base_url = f"{parsed.scheme}://{parsed.netloc}" timestamp = datetime.datetime.utcnow().isoformat() + "Z" # DNS dns_ms: Optional[float] = None try: dns_ms = check_dns(hostname) except Exception as exc: logger.error("DNS lookup failed: %s", exc) # SSL ssl_info = check_ssl(hostname) # HTTP http_info = check_http(session, target_url) # Search search_ms = check_search_response(session, base_url, search_query) record: dict[str, Any] = { "timestamp": timestamp, "status_code": http_info["status_code"], "uptime": http_info["uptime"], "latency_ms": http_info["latency_ms"], "dns_ms": dns_ms, "ssl": ssl_info, "search_ms": search_ms, } return record # --------------------------------------------------------------------------- # CLI Argument Parsing # --------------------------------------------------------------------------- def parse_args() -> argparse.Namespace: """Parse command-line arguments.""" parser = argparse.ArgumentParser( description="Monitor https://mithal.space uptime, latency, DNS, SSL, and search.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=( "Examples:\n" " python monitor.py\n" " python monitor.py --interval 30\n" " python monitor.py --interval 120 --max-records 720\n" ), ) parser.add_argument( "--interval", type=int, default=DEFAULT_CHECK_INTERVAL, help=f"Check interval in seconds (default: {DEFAULT_CHECK_INTERVAL})", ) parser.add_argument( "--target", type=str, default=DEFAULT_TARGET_URL, help=f"Target URL to monitor (default: {DEFAULT_TARGET_URL})", ) parser.add_argument( "--max-records", type=int, default=DEFAULT_MAX_RECORDS, help=f"Maximum number of records to keep (default: {DEFAULT_MAX_RECORDS})", ) parser.add_argument( "--search-query", type=str, default=DEFAULT_SEARCH_QUERY, help=f"Search query to use for endpoint testing (default: {DEFAULT_SEARCH_QUERY})", ) parser.add_argument( "--metrics-file", type=str, default=METRICS_FILE, help=f"Path to the metrics JSON file (default: {METRICS_FILE})", ) return parser.parse_args() # --------------------------------------------------------------------------- # Main Loop # --------------------------------------------------------------------------- def main() -> None: """Entry point: run the monitoring loop.""" args = parse_args() logger.info("=" * 60) logger.info("Mithal.space Monitor started") logger.info("Target URL : %s", args.target) logger.info("Search query : %s", args.search_query) logger.info("Check interval : %d seconds", args.interval) logger.info("Max records : %d", args.max_records) logger.info("Metrics file : %s", args.metrics_file) logger.info("Log file : %s", LOG_FILE) logger.info("=" * 60) session = create_session() try: while True: try: record = perform_check(session, args.target, args.search_query) # Load, append, trim, save metrics = load_metrics(args.metrics_file) metrics.append(record) metrics = trim_metrics(metrics, args.max_records) save_metrics(args.metrics_file, metrics) # Summary log status_emoji = "✓" if record["uptime"] else "✗" logger.info( "%s status=%d latency=%.0fms dns=%.0fms ssl_days=%s search=%s", status_emoji, record["status_code"], record["latency_ms"], record["dns_ms"] or 0, record["ssl"]["days_remaining"] if record["ssl"] else "N/A", f"{record['search_ms']:.0f}ms" if record["search_ms"] else "N/A", ) except Exception as exc: logger.exception("Unexpected error during check: %s", exc) time.sleep(args.interval) except KeyboardInterrupt: logger.info("Monitor stopped by user (Ctrl+C)") finally: session.close() logger.info("Session closed. Goodbye.") if __name__ == "__main__": main()