实现应用健康监控（07-27-app-health-monitoring）

scripts/check-public-app-health.py

@@ -0,0 +1,283 @@
+#!/usr/bin/env python3
+"""Probe the public OA readiness endpoint and drive a deduplicated alert."""
+
+from __future__ import annotations
+
+import argparse
+import http.client
+import json
+import os
+import re
+import socket
+import ssl
+import stat
+import subprocess
+import sys
+import tempfile
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+
+SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
+SAFE_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+MAX_RESPONSE_BYTES = 64 * 1024
+HEALTHY_STATUS = "PASS_APP_HEALTH_READY"
+FAILURE_STATUS = "FAIL_APP_HEALTH"
+
+
+class ProbeFailure(Exception):
+    def __init__(self, category: str):
+        super().__init__(category)
+        self.category = category
+
+
+def fail(category: str) -> "NoReturn":
+    print(f"[app-health][ERROR] {category}", file=sys.stderr)
+    raise SystemExit(1)
+
+
+def validate_regular_file(path: Path, label: str, private: bool = False) -> None:
+    try:
+        info = path.lstat()
+    except FileNotFoundError:
+        fail(f"{label}_missing")
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        fail(f"{label}_invalid")
+    if info.st_uid != os.geteuid():
+        fail(f"{label}_owner_invalid")
+    if private and stat.S_IMODE(info.st_mode) != 0o600:
+        fail(f"{label}_mode_invalid")
+    if not private and stat.S_IMODE(info.st_mode) & 0o022:
+        fail(f"{label}_mode_invalid")
+
+
+def prepare_state_dir(path: Path) -> None:
+    try:
+        path.mkdir(parents=True, exist_ok=True, mode=0o700)
+        info = path.lstat()
+    except OSError:
+        fail("local_state_invalid")
+    if (
+        stat.S_ISLNK(info.st_mode)
+        or not stat.S_ISDIR(info.st_mode)
+        or info.st_uid != os.geteuid()
+        or stat.S_IMODE(info.st_mode) != 0o700
+    ):
+        fail("local_state_invalid")
+
+
+def validate_ready_url(value: str) -> str:
+    try:
+        parsed = urllib.parse.urlsplit(value)
+        port = parsed.port
+    except ValueError:
+        fail("ready_url_invalid")
+    if (
+        parsed.scheme != "https"
+        or not parsed.hostname
+        or not SAFE_HOST.fullmatch(parsed.hostname)
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.path != "/ready"
+        or parsed.query
+        or parsed.fragment
+        or (port is not None and not 1 <= port <= 65535)
+    ):
+        fail("ready_url_invalid")
+    return value
+
+
+def load_probe_state(path: Path) -> dict[str, object]:
+    if not path.exists():
+        return {"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None}
+    validate_regular_file(path, "local_state", private=True)
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        fail("local_state_invalid")
+    if not isinstance(payload, dict):
+        fail("local_state_invalid")
+    count = payload.get("consecutiveFailures")
+    category = payload.get("lastCategory")
+    if (
+        payload.get("schemaVersion") != 1
+        or not isinstance(count, int)
+        or isinstance(count, bool)
+        or count < 0
+        or (category is not None and (not isinstance(category, str) or not SAFE_TOKEN.fullmatch(category)))
+    ):
+        fail("local_state_invalid")
+    return payload
+
+
+def atomic_write(path: Path, payload: dict[str, object]) -> None:
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        os.fchmod(fd, 0o600)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(payload, handle, ensure_ascii=True, indent=2)
+            handle.write("\n")
+        os.replace(temporary, path)
+    except BaseException:
+        try:
+            os.unlink(temporary)
+        except FileNotFoundError:
+            pass
+        raise
+
+
+def classify_url_error(error: urllib.error.URLError) -> str:
+    reason = error.reason
+    if isinstance(reason, (ssl.SSLError, ssl.CertificateError)):
+        return "tls_validation_failed"
+    if isinstance(reason, (socket.timeout, TimeoutError)):
+        return "request_timeout"
+    return "dns_or_connect_failed"
+
+
+def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
+    if ca_file is not None:
+        validate_regular_file(ca_file, "ca_file")
+    try:
+        context = ssl.create_default_context(cafile=str(ca_file) if ca_file else None)
+        request = urllib.request.Request(
+            ready_url,
+            headers={"Accept": "application/json", "User-Agent": "hyb-oa-app-health/1"},
+            method="GET",
+        )
+        with urllib.request.urlopen(request, timeout=timeout_seconds, context=context) as response:
+            if not 200 <= response.status < 300:
+                raise ProbeFailure("http_unavailable")
+            content_length = response.headers.get("Content-Length")
+            if content_length:
+                try:
+                    if int(content_length) > MAX_RESPONSE_BYTES:
+                        raise ProbeFailure("response_too_large")
+                except ValueError:
+                    raise ProbeFailure("response_contract_invalid") from None
+            body = response.read(MAX_RESPONSE_BYTES + 1)
+    except urllib.error.HTTPError:
+        raise ProbeFailure("http_unavailable") from None
+    except urllib.error.URLError as error:
+        raise ProbeFailure(classify_url_error(error)) from None
+    except (socket.timeout, TimeoutError):
+        raise ProbeFailure("request_timeout") from None
+    except (ssl.SSLError, ssl.CertificateError):
+        raise ProbeFailure("tls_validation_failed") from None
+    except (http.client.HTTPException, ConnectionError, OSError):
+        raise ProbeFailure("dns_or_connect_failed") from None
+    if len(body) > MAX_RESPONSE_BYTES:
+        raise ProbeFailure("response_too_large")
+    try:
+        payload = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise ProbeFailure("response_contract_invalid") from None
+    if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict) or payload["data"].get("ok") is not True:
+        raise ProbeFailure("response_contract_invalid")
+
+
+def write_summary(path: Path, status: str, category: str) -> None:
+    finding = status == FAILURE_STATUS
+    payload = {
+        "issueId": "APP_HEALTH",
+        "status": status,
+        "conclusion": status,
+        "findingCount": 1 if finding else 0,
+        "findings": (
+            [{"source": "public_ready", "status": status, "severity": "P0", "detail": category}]
+            if finding
+            else []
+        ),
+        "privacy": "status/count/category only; no credentials, addresses, paths, or payloads",
+    }
+    atomic_write(path, payload)
+
+
+def evaluate_alert(alert_helper: Path, webhook_file: Path, summary_file: Path, alert_state_file: Path) -> None:
+    validate_regular_file(alert_helper, "alert_helper")
+    try:
+        result = subprocess.run(
+            [
+                sys.executable,
+                str(alert_helper),
+                "--webhook-file",
+                str(webhook_file),
+                "--summary-file",
+                str(summary_file),
+                "--state-file",
+                str(alert_state_file),
+            ],
+            check=False,
+        )
+    except OSError:
+        fail("alert_evaluation_failed")
+    if result.returncode != 0:
+        fail("alert_evaluation_failed")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--ready-url", required=True)
+    parser.add_argument("--state-dir", required=True, type=Path)
+    parser.add_argument("--webhook-file", required=True, type=Path)
+    parser.add_argument("--alert-helper", required=True, type=Path)
+    parser.add_argument("--failure-threshold", type=int, default=2)
+    parser.add_argument("--timeout-seconds", type=float, default=10.0)
+    parser.add_argument("--ca-file", type=Path)
+    args = parser.parse_args()
+
+    ready_url = validate_ready_url(args.ready_url)
+    if not 1 <= args.failure_threshold <= 10:
+        fail("failure_threshold_invalid")
+    if not 0.1 <= args.timeout_seconds <= 30:
+        fail("timeout_invalid")
+    prepare_state_dir(args.state_dir)
+    probe_state_file = args.state_dir / "app-health-probe-state.json"
+    summary_file = args.state_dir / "last-app-health.json"
+    alert_state_file = args.state_dir / "app-health-alert-state.json"
+    state = load_probe_state(probe_state_file)
+
+    try:
+        probe(ready_url, args.timeout_seconds, args.ca_file)
+    except ProbeFailure as error:
+        failures = min(int(state["consecutiveFailures"]) + 1, args.failure_threshold)
+        try:
+            atomic_write(
+                probe_state_file,
+                {"schemaVersion": 1, "consecutiveFailures": failures, "lastCategory": error.category},
+            )
+        except OSError:
+            fail("local_state_invalid")
+        if failures >= args.failure_threshold:
+            try:
+                write_summary(summary_file, FAILURE_STATUS, error.category)
+            except OSError:
+                fail("local_state_invalid")
+            evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
+            print(
+                f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=evaluated",
+                file=sys.stderr,
+            )
+        else:
+            print(
+                f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=pending",
+                file=sys.stderr,
+            )
+        raise SystemExit(1)
+
+    try:
+        atomic_write(
+            probe_state_file,
+            {"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None},
+        )
+        write_summary(summary_file, HEALTHY_STATUS, "public_ready_healthy")
+    except OSError:
+        fail("local_state_invalid")
+    evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
+    print("[app-health][PASS] public readiness is healthy")
+
+
+if __name__ == "__main__":
+    main()