#!/usr/bin/env python3 """ Monitoring Collector for mithal.space Collects metrics every 60 seconds: Latency, Uptime, SSL, DNS, Search Response. Stores results in data/mithal-metrics.json. """ import urllib.request import urllib.error import ssl import socket import json import time import datetime import os import sys # ───────────────────────────────────────────── # Configuration # ───────────────────────────────────────────── TARGET_URL = "https://mithal.space" SEARCH_URL = "https://mithal.space" # Search/query endpoint (adjust if different) CHECK_INTERVAL = 60 # seconds (1 minute) DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") DATA_FILE = os.path.join(DATA_DIR, "mithal-metrics.json") MAX_RECORDS = 1440 # Keep 24 hours of data (1 per minute) TIMEOUT = 10 # seconds # ANSI colors GREEN = "\033[92m" RED = "\033[91m" YELLOW = "\033[93m" CYAN = "\033[96m" BOLD = "\033[1m" RESET = "\033[0m" def ensure_data_dir(): """Create data directory if it doesn't exist.""" os.makedirs(DATA_DIR, exist_ok=True) if not os.path.exists(DATA_FILE): with open(DATA_FILE, "w", encoding="utf-8") as f: json.dump({"records": []}, f) def load_data(): """Load existing metrics data.""" try: with open(DATA_FILE, "r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, FileNotFoundError): return {"records": []} def save_data(data): """Save metrics data, keeping only the last MAX_RECORDS entries.""" data["records"] = data["records"][-MAX_RECORDS:] with open(DATA_FILE, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) # ───────────────────────────────────────────── # Metric Collectors # ───────────────────────────────────────────── def check_latency(url): """Measure HTTP response time (latency) in milliseconds.""" try: start = time.time() req = urllib.request.Request(url, method="GET") req.add_header("User-Agent", "GhaymaMonitor/1.0") response = urllib.request.urlopen(req, timeout=TIMEOUT) latency_ms = round((time.time() - start) * 1000) status_code = response.status response.close() return { "latency_ms": latency_ms, "status_code": status_code, "is_up": 200 <= status_code < 400, "error": None } except urllib.error.HTTPError as e: latency_ms = round((time.time() - start) * 1000) return { "latency_ms": latency_ms, "status_code": e.code, "is_up": False, "error": f"HTTP {e.code}: {e.reason}" } except Exception as e: return { "latency_ms": None, "status_code": None, "is_up": False, "error": str(e) } def check_ssl(hostname): """Check SSL certificate status and expiration date.""" try: context = ssl.create_default_context() with socket.create_connection((hostname, 443), timeout=TIMEOUT) as sock: with context.wrap_socket(sock, server_hostname=hostname) as ssock: cert = ssock.getpeercert() # Parse expiration date expire_str = cert.get("notAfter", "") expire_date = datetime.datetime.strptime( expire_str, "%b %d %H:%M:%S %Y %Z" ) now = datetime.datetime.utcnow() days_remaining = (expire_date - now).days # Get issuer issuer_parts = dict(x[0] for x in cert.get("issuer", [])) issuer = issuer_parts.get("organizationName", "Unknown") return { "valid": True, "issuer": issuer, "expires": expire_date.strftime("%Y-%m-%d %H:%M:%S UTC"), "days_remaining": days_remaining, "error": None } except Exception as e: return { "valid": False, "issuer": None, "expires": None, "days_remaining": None, "error": str(e) } def check_dns(hostname): """Measure DNS resolution time in milliseconds.""" try: start = time.time() ip_address = socket.gethostbyname(hostname) dns_time_ms = round((time.time() - start) * 1000) return { "dns_time_ms": dns_time_ms, "resolved_ip": ip_address, "error": None } except Exception as e: return { "dns_time_ms": None, "resolved_ip": None, "error": str(e) } def check_search_response(url): """Send a search/query request and measure response time.""" try: search_url = url.rstrip("/") + "/" start = time.time() req = urllib.request.Request(search_url, method="GET") req.add_header("User-Agent", "GhaymaMonitor/1.0") response = urllib.request.urlopen(req, timeout=TIMEOUT) search_time_ms = round((time.time() - start) * 1000) content_length = len(response.read()) response.close() return { "search_time_ms": search_time_ms, "content_length": content_length, "status_code": response.status, "error": None } except Exception as e: return { "search_time_ms": None, "content_length": None, "status_code": None, "error": str(e) } # ───────────────────────────────────────────── # Main Collector # ───────────────────────────────────────────── def collect_metrics(): """Run all metric checks and return a combined record.""" timestamp = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") hostname = TARGET_URL.replace("https://", "").replace("http://", "").split("/")[0] print(f"\n{CYAN}[{timestamp}] Collecting metrics for {TARGET_URL}...{RESET}") # 1. DNS Resolution dns_result = check_dns(hostname) status = f"{GREEN}✅ {dns_result['dns_time_ms']}ms{RESET}" if not dns_result["error"] else f"{RED}❌ {dns_result['error']}{RESET}" print(f" DNS Resolution: {status}") # 2. HTTP Latency & Uptime latency_result = check_latency(TARGET_URL) if latency_result["is_up"]: status = f"{GREEN}✅ {latency_result['status_code']} in {latency_result['latency_ms']}ms{RESET}" else: status = f"{RED}❌ {latency_result.get('error', 'Down')}{RESET}" print(f" HTTP Latency: {status}") # 3. SSL Certificate ssl_result = check_ssl(hostname) if ssl_result["valid"]: days = ssl_result["days_remaining"] color = GREEN if days > 30 else (YELLOW if days > 7 else RED) status = f"{color}✅ Valid, {days} days remaining{RESET}" else: status = f"{RED}❌ {ssl_result.get('error', 'Invalid')}{RESET}" print(f" SSL Status: {status}") # 4. Search Response search_result = check_search_response(SEARCH_URL) if search_result["search_time_ms"]: status = f"{GREEN}✅ {search_result['search_time_ms']}ms ({search_result['content_length']} bytes){RESET}" else: status = f"{RED}❌ {search_result.get('error', 'Failed')}{RESET}" print(f" Search Response: {status}") # Combine into a single record record = { "timestamp": timestamp, "latency": latency_result, "uptime": { "is_up": latency_result["is_up"], "status_code": latency_result["status_code"] }, "ssl": ssl_result, "dns": dns_result, "search": search_result } return record def main(): ensure_data_dir() print(f"{BOLD}{CYAN}") print("=" * 58) print(" mithal.space — Monitoring Collector") print(f" Target: {TARGET_URL}") print(f" Interval: {CHECK_INTERVAL}s (every minute)") print(f" Storage: {DATA_FILE}") print("=" * 58) print(f"{RESET}") try: while True: record = collect_metrics() # Load, append, save data = load_data() data["records"].append(record) save_data(data) up_count = sum(1 for r in data["records"] if r["uptime"]["is_up"]) total = len(data["records"]) uptime_pct = round((up_count / total) * 100, 2) if total > 0 else 0 print(f" {BOLD}Uptime (session): {uptime_pct}% ({up_count}/{total} checks){RESET}") print(f" {CYAN}Next check in {CHECK_INTERVAL}s...{RESET}") time.sleep(CHECK_INTERVAL) except KeyboardInterrupt: print(f"\n{YELLOW}Collector stopped.{RESET}") sys.exit(0) if __name__ == "__main__": main()