diff --git a/q5-mithal-monitor/dashboard.html b/q5-mithal-monitor/dashboard.html new file mode 100644 index 0000000..aab70b0 --- /dev/null +++ b/q5-mithal-monitor/dashboard.html @@ -0,0 +1,437 @@ + + + + + + Mithal Monitor Dashboard + + + + +
+
+
+
Mithal.space monitoring
+

Live uptime and latency dashboard

+

This page reads metrics.json, summarizes the last 24 hours, and refreshes automatically every 60 seconds.

+
+
Waiting for data...
+
+ +
+
+
Uptime percentage (24h)
+
--
+
+
+
+
Latest HTTP latency
+
--
+
+
+
+
Latest DNS lookup time
+
--
+
+
+
+
SSL status
+
--
+
+
+
+
SSL expiration date
+
--
+
+
+
+
Latest search response time
+
--
+
+
+
+ +
+

Latency line chart

+
+ +
+
+ +
+

Latest 10 monitoring records

+
+ + + + + + + + + + + + + + +
TimestampStatusUptimeLatencyDNSSSLSSL ExpirySearch
+
+ +
+
+ + + + \ No newline at end of file diff --git a/q5-mithal-monitor/metrics.json b/q5-mithal-monitor/metrics.json new file mode 100644 index 0000000..043b964 --- /dev/null +++ b/q5-mithal-monitor/metrics.json @@ -0,0 +1,34 @@ +[ + { + "timestamp": "2026-07-29T02:43:55.590275+00:00", + "latency_ms": 737.47, + "status_code": 200, + "uptime": true, + "ssl_valid": null, + "ssl_valid_error": "SSL certificate does not expose an expiration date", + "ssl_expiry": null, + "ssl_expiry_error": "SSL certificate does not expose an expiration date", + "dns_lookup_ms": 48.09, + "search_response_ms": 769.37 + }, + { + "timestamp": "2026-07-29T02:44:09.531381+00:00", + "latency_ms": 705.17, + "status_code": 200, + "uptime": true, + "ssl_valid": true, + "ssl_expiry": "2026-09-15T13:10:47+00:00", + "dns_lookup_ms": 48.28, + "search_response_ms": 763.97 + }, + { + "timestamp": "2026-07-29T02:46:01.686895+00:00", + "latency_ms": 708.52, + "status_code": 200, + "uptime": true, + "ssl_valid": true, + "ssl_expiry": "2026-09-15T13:10:47+00:00", + "dns_lookup_ms": 47.78, + "search_response_ms": 750.01 + } +] diff --git a/q5-mithal-monitor/monitor.py b/q5-mithal-monitor/monitor.py new file mode 100644 index 0000000..23d5627 --- /dev/null +++ b/q5-mithal-monitor/monitor.py @@ -0,0 +1,300 @@ +#!/usr/bin/env python3 + +"""Collect availability metrics for https://mithal.space every minute.""" + +from __future__ import annotations + +import argparse +import json +import socket +import ssl +import sys +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from html.parser import HTMLParser +from pathlib import Path +from typing import Any +import tempfile +from urllib import parse, request + + +TARGET_URL = "https://mithal.space" +METRICS_FILE = Path(__file__).with_name("metrics.json") +USER_AGENT = "Mozilla/5.0 (compatible; MithalMonitor/1.0)" + + +@dataclass +class SearchConfig: + """Discovered search request details from the homepage.""" + + action_url: str + method: str + query_param: str + fixed_params: dict[str, str] + + +class SearchFormParser(HTMLParser): + """Extract candidate search forms from the homepage HTML.""" + + def __init__(self) -> None: + super().__init__() + self.forms: list[dict[str, Any]] = [] + self._current_form: dict[str, Any] | None = None + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attributes = {key.lower(): value for key, value in attrs} + + if tag.lower() == "form": + self._current_form = { + "action": attributes.get("action", ""), + "method": (attributes.get("method") or "GET").upper(), + "inputs": [], + } + return + + if self._current_form is not None and tag.lower() == "input": + self._current_form["inputs"].append( + { + "type": (attributes.get("type") or "text").lower(), + "name": attributes.get("name", ""), + "value": attributes.get("value", ""), + } + ) + + def handle_endtag(self, tag: str) -> None: + if tag.lower() == "form" and self._current_form is not None: + self.forms.append(self._current_form) + self._current_form = None + + +def now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def http_get(url: str, timeout: int = 20) -> tuple[float, int, bytes]: + """Perform a full HTTP GET and return latency, status code, and body.""" + + started_at = time.perf_counter() + req = request.Request(url, headers={"User-Agent": USER_AGENT}) + + with request.urlopen(req, timeout=timeout) as response: + body = response.read() + latency_ms = (time.perf_counter() - started_at) * 1000 + return latency_ms, int(response.status), body + + +def fetch_homepage_html(base_url: str) -> str: + req = request.Request(base_url, headers={"User-Agent": USER_AGENT}) + with request.urlopen(req, timeout=20) as response: + return response.read().decode("utf-8", errors="replace") + + +def discover_search_config(base_url: str) -> SearchConfig: + """Inspect the homepage and derive the real search request shape.""" + + homepage_html = fetch_homepage_html(base_url) + parser = SearchFormParser() + parser.feed(homepage_html) + + for form in parser.forms: + action = (form.get("action") or "").strip() + method = (form.get("method") or "GET").upper() + inputs = form.get("inputs") or [] + + query_input = None + fixed_params: dict[str, str] = {} + + for input_item in inputs: + input_name = (input_item.get("name") or "").strip() + input_type = (input_item.get("type") or "text").lower() + input_value = input_item.get("value") or "" + + if not input_name: + continue + + if input_type in {"text", "search", "url", "email", "tel", "hidden"} and query_input is None: + if input_type != "hidden": + query_input = input_name + + if input_type == "hidden" and input_value: + fixed_params[input_name] = input_value + + if query_input is None: + for input_item in inputs: + input_name = (input_item.get("name") or "").strip() + input_type = (input_item.get("type") or "text").lower() + if input_name and input_type != "hidden": + query_input = input_name + break + + if query_input and ("search" in action.lower() or any("search" in (item.get("name") or "").lower() for item in inputs)): + return SearchConfig( + action_url=parse.urljoin(base_url, action or "/search"), + method=method, + query_param=query_input, + fixed_params=fixed_params, + ) + + return SearchConfig( + action_url=parse.urljoin(base_url, "/search"), + method="GET", + query_param="q", + fixed_params={}, + ) + + +def measure_dns_lookup(hostname: str) -> float: + started_at = time.perf_counter() + socket.getaddrinfo(hostname, 443, type=socket.SOCK_STREAM) + return (time.perf_counter() - started_at) * 1000 + + +def measure_ssl_certificate(hostname: str, port: int = 443) -> tuple[bool, str]: + pem_certificate = ssl.get_server_certificate((hostname, port), timeout=20) + + with tempfile.NamedTemporaryFile("w", delete=True) as temp_file: + temp_file.write(pem_certificate) + temp_file.flush() + certificate = ssl._ssl._test_decode_cert(temp_file.name) # type: ignore[attr-defined] + + not_after = certificate.get("notAfter") + if not not_after: + raise ValueError("SSL certificate does not expose an expiration date") + + expiry_timestamp = ssl.cert_time_to_seconds(not_after) + expiry_dt = datetime.fromtimestamp(expiry_timestamp, tz=timezone.utc) + is_valid = datetime.now(timezone.utc) < expiry_dt + return is_valid, expiry_dt.isoformat() + + +def measure_search_response(base_url: str, query: str = "test") -> float: + search_config = discover_search_config(base_url) + params = dict(search_config.fixed_params) + params[search_config.query_param] = query + + if search_config.method != "GET": + raise ValueError(f"Unsupported search method: {search_config.method}") + + search_url = f"{search_config.action_url}?{parse.urlencode(params)}" + latency_ms, _, _ = http_get(search_url) + return latency_ms + + +def metric_record(name: str, value: Any, error_message: str | None = None) -> dict[str, Any]: + record = {name: value} + if error_message is not None: + record[f"{name}_error"] = error_message + return record + + +def collect_metrics(base_url: str = TARGET_URL) -> dict[str, Any]: + hostname = parse.urlparse(base_url).hostname or "mithal.space" + record: dict[str, Any] = {"timestamp": now_iso()} + + try: + latency_ms, status_code, _ = http_get(base_url) + record.update(metric_record("latency_ms", round(latency_ms, 2))) + record.update(metric_record("status_code", status_code)) + record["uptime"] = status_code == 200 + except Exception as exc: # noqa: BLE001 - keep the monitor resilient. + record.update(metric_record("latency_ms", None, str(exc))) + record.update(metric_record("status_code", None, str(exc))) + record["uptime"] = None + record["uptime_error"] = str(exc) + + try: + ssl_valid, ssl_expiry = measure_ssl_certificate(hostname) + record["ssl_valid"] = ssl_valid + record["ssl_expiry"] = ssl_expiry + except Exception as exc: # noqa: BLE001 - keep the monitor resilient. + record["ssl_valid"] = None + record["ssl_valid_error"] = str(exc) + record["ssl_expiry"] = None + record["ssl_expiry_error"] = str(exc) + + try: + dns_lookup_ms = measure_dns_lookup(hostname) + record["dns_lookup_ms"] = round(dns_lookup_ms, 2) + except Exception as exc: # noqa: BLE001 - keep the monitor resilient. + record["dns_lookup_ms"] = None + record["dns_lookup_ms_error"] = str(exc) + + try: + search_response_ms = measure_search_response(base_url) + record["search_response_ms"] = round(search_response_ms, 2) + except Exception as exc: # noqa: BLE001 - keep the monitor resilient. + record["search_response_ms"] = None + record["search_response_ms_error"] = str(exc) + + return record + + +def load_metrics(path: Path) -> list[dict[str, Any]]: + if not path.exists(): + return [] + + try: + with path.open("r", encoding="utf-8") as file_handle: + data = json.load(file_handle) + except Exception: + return [] + + if isinstance(data, list): + return [record for record in data if isinstance(record, dict)] + + return [] + + +def save_metrics(path: Path, records: list[dict[str, Any]]) -> None: + temp_path = path.with_suffix(".tmp") + with temp_path.open("w", encoding="utf-8") as file_handle: + json.dump(records, file_handle, ensure_ascii=False, indent=2) + file_handle.write("\n") + temp_path.replace(path) + + +def append_metric_record(record: dict[str, Any], path: Path = METRICS_FILE) -> None: + records = load_metrics(path) + records.append(record) + save_metrics(path, records) + + +def run_monitor(interval_seconds: int, once: bool, target_url: str, metrics_file: Path) -> None: + while True: + record = collect_metrics(target_url) + append_metric_record(record, metrics_file) + print(json.dumps(record, ensure_ascii=False)) + + if once: + return + + time.sleep(interval_seconds) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Collect monitoring metrics for mithal.space") + parser.add_argument("--interval", type=int, default=60, help="Seconds between metric collections") + parser.add_argument("--once", action="store_true", help="Collect one record and exit") + parser.add_argument("--target-url", default=TARGET_URL, help="Website to monitor") + parser.add_argument("--metrics-file", default=str(METRICS_FILE), help="Path to metrics.json") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + metrics_file = Path(args.metrics_file) + + try: + run_monitor(args.interval, args.once, args.target_url, metrics_file) + except KeyboardInterrupt: + return 0 + except Exception as exc: # noqa: BLE001 - the outer loop should never crash silently. + print(f"monitor failed: {exc}", file=sys.stderr) + return 1 + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file