154 أسطر
4.8 KiB
Python
154 أسطر
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
monitor.py — Monitoring script for mithal.space (Q5)
|
|
|
|
Collects every minute:
|
|
- Latency: time-to-first-byte for the homepage
|
|
- Uptime: HTTP status code check
|
|
- SSL: certificate validity + days remaining until expiry
|
|
- DNS: DNS resolution time
|
|
- Search response: time to get a response from a search query
|
|
|
|
Stores results as newline-delimited JSON (JSONL) for easy append + parsing,
|
|
and a rolling `latest.json` snapshot consumed by dashboard.html.
|
|
|
|
Usage:
|
|
python3 monitor.py # runs forever, checks every 60s
|
|
python3 monitor.py --once # single check, useful for cron/testing
|
|
"""
|
|
import argparse
|
|
import json
|
|
import socket
|
|
import ssl
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
|
|
TARGET_URL = "https://mithal.space"
|
|
SEARCH_URL = "https://mithal.space/search?q=غيمة"
|
|
CHECK_INTERVAL_SECONDS = 60
|
|
DATA_DIR = Path(__file__).parent
|
|
JSONL_LOG = DATA_DIR / "metrics.jsonl"
|
|
LATEST_SNAPSHOT = DATA_DIR / "latest.json"
|
|
MAX_HISTORY_IN_SNAPSHOT = 1440 # 24h at 1-minute resolution
|
|
|
|
|
|
def check_dns(hostname: str) -> float:
|
|
"""Returns DNS resolution time in ms."""
|
|
start = time.perf_counter()
|
|
socket.gethostbyname(hostname)
|
|
return round((time.perf_counter() - start) * 1000, 2)
|
|
|
|
|
|
def check_http(url: str, timeout: int = 10):
|
|
"""Returns (status_code, latency_ms) or (None, None) on failure."""
|
|
try:
|
|
start = time.perf_counter()
|
|
resp = requests.get(url, timeout=timeout, headers={"User-Agent": "mithal-monitor/1.0"})
|
|
latency_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return resp.status_code, latency_ms
|
|
except requests.RequestException as exc:
|
|
return None, None
|
|
|
|
|
|
def check_ssl(hostname: str, port: int = 443, timeout: int = 10):
|
|
"""Returns dict with ssl validity + days_remaining, or None on failure."""
|
|
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 = datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z").replace(tzinfo=timezone.utc)
|
|
days_remaining = (not_after - datetime.now(timezone.utc)).days
|
|
return {
|
|
"valid": True,
|
|
"expires_at": not_after.isoformat(),
|
|
"days_remaining": days_remaining,
|
|
}
|
|
except Exception as exc:
|
|
return {"valid": False, "error": str(exc)}
|
|
|
|
|
|
def check_search(url: str, timeout: int = 15):
|
|
"""Sends a search query and measures response time."""
|
|
status, latency_ms = check_http(url, timeout=timeout)
|
|
return {"status_code": status, "latency_ms": latency_ms, "ok": status == 200}
|
|
|
|
|
|
def run_check() -> dict:
|
|
hostname = urlparse(TARGET_URL).hostname
|
|
timestamp = datetime.now(timezone.utc).isoformat()
|
|
|
|
dns_ms = None
|
|
try:
|
|
dns_ms = check_dns(hostname)
|
|
except Exception:
|
|
pass
|
|
|
|
status_code, latency_ms = check_http(TARGET_URL)
|
|
ssl_info = check_ssl(hostname)
|
|
search_info = check_search(SEARCH_URL)
|
|
|
|
return {
|
|
"timestamp": timestamp,
|
|
"target": TARGET_URL,
|
|
"uptime": {
|
|
"status_code": status_code,
|
|
"up": status_code == 200,
|
|
},
|
|
"latency_ms": latency_ms,
|
|
"dns_ms": dns_ms,
|
|
"ssl": ssl_info,
|
|
"search": search_info,
|
|
}
|
|
|
|
|
|
def append_jsonl(record: dict):
|
|
with open(JSONL_LOG, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
|
|
|
|
def update_snapshot(record: dict):
|
|
history = []
|
|
if LATEST_SNAPSHOT.exists():
|
|
try:
|
|
history = json.loads(LATEST_SNAPSHOT.read_text(encoding="utf-8")).get("history", [])
|
|
except json.JSONDecodeError:
|
|
history = []
|
|
history.append(record)
|
|
history = history[-MAX_HISTORY_IN_SNAPSHOT:]
|
|
|
|
up_count = sum(1 for h in history if h["uptime"]["up"])
|
|
uptime_pct = round((up_count / len(history)) * 100, 2) if history else 0.0
|
|
|
|
snapshot = {
|
|
"last_check": record,
|
|
"uptime_percent_24h": uptime_pct,
|
|
"history": history,
|
|
}
|
|
LATEST_SNAPSHOT.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Monitor mithal.space")
|
|
parser.add_argument("--once", action="store_true", help="Run a single check and exit")
|
|
parser.add_argument("--interval", type=int, default=CHECK_INTERVAL_SECONDS)
|
|
args = parser.parse_args()
|
|
|
|
while True:
|
|
record = run_check()
|
|
append_jsonl(record)
|
|
update_snapshot(record)
|
|
print(json.dumps(record, ensure_ascii=False))
|
|
|
|
if args.once:
|
|
break
|
|
time.sleep(args.interval)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|