276 أسطر
9.1 KiB
Python
276 أسطر
9.1 KiB
Python
#!/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()
|