405 أسطر
13 KiB
Python
405 أسطر
13 KiB
Python
#!/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=<query>.
|
|
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()
|