文件
hyb-oa/scripts/check-public-app-health.py
T

651 行
22 KiB
Python
可执行文件

#!/usr/bin/env python3
"""Probe the public OA readiness endpoint and drive a debounced 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 time
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
MAX_COUNTER = 1_000_000
MAX_TIMESTAMP = (1 << 53) - 1
STATE_SCHEMA_VERSION = 2
RECOVERY_THRESHOLD = 2
RECOVERY_WATCH_SECONDS = 60 * 60
FLAPPING_STABLE_SECONDS = 30 * 60
HEALTHY_STATUS = "PASS_APP_HEALTH_READY"
FAILURE_STATUS = "FAIL_APP_HEALTH"
FLAPPING_STATUS = "FAIL_APP_HEALTH_FLAPPING"
FAILURE_CATEGORIES = {
"dns_resolution_failed",
"tcp_connection_failed",
"tls_validation_failed",
"request_timeout",
"http_unavailable",
"response_contract_invalid",
"response_too_large",
}
LEGACY_FAILURE_CATEGORIES = FAILURE_CATEGORIES | {"dns_or_connect_failed"}
PHASES = {
"healthy",
"failure_pending",
"alerted",
"recovery_pending",
"recovery_watch",
"flapping",
}
STATE_KEYS = {
"schemaVersion",
"phase",
"activeCategory",
"lastRecoveredCategory",
"consecutiveFailures",
"consecutiveSuccesses",
"recoveredAt",
"healthySince",
"flapTransitions",
}
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 new_probe_state() -> dict[str, object]:
return {
"schemaVersion": STATE_SCHEMA_VERSION,
"phase": "healthy",
"activeCategory": None,
"lastRecoveredCategory": None,
"consecutiveFailures": 0,
"consecutiveSuccesses": 0,
"recoveredAt": None,
"healthySince": None,
"flapTransitions": 0,
}
def is_bounded_int(value: object, minimum: int = 0, maximum: int = MAX_COUNTER) -> bool:
return isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum
def is_category(value: object, allow_none: bool = True) -> bool:
if value is None:
return allow_none
return isinstance(value, str) and value in LEGACY_FAILURE_CATEGORIES and bool(SAFE_TOKEN.fullmatch(value))
def is_timestamp(value: object, allow_none: bool = True) -> bool:
if value is None:
return allow_none
return is_bounded_int(value, maximum=MAX_TIMESTAMP)
def validate_v2_state(payload: dict[str, object]) -> bool:
if set(payload) != STATE_KEYS or payload.get("schemaVersion") != STATE_SCHEMA_VERSION:
return False
phase = payload.get("phase")
active = payload.get("activeCategory")
recovered = payload.get("lastRecoveredCategory")
failures = payload.get("consecutiveFailures")
successes = payload.get("consecutiveSuccesses")
recovered_at = payload.get("recoveredAt")
healthy_since = payload.get("healthySince")
transitions = payload.get("flapTransitions")
if (
phase not in PHASES
or not is_category(active)
or not is_category(recovered)
or not is_bounded_int(failures)
or not is_bounded_int(successes)
or not is_timestamp(recovered_at)
or not is_timestamp(healthy_since)
or not is_bounded_int(transitions)
):
return False
if phase == "healthy":
return (
active is None
and recovered is None
and failures == 0
and successes == 0
and recovered_at is None
and healthy_since is None
and transitions == 0
)
if phase == "failure_pending":
return (
active is not None
and failures >= 1
and successes == 0
and healthy_since is None
and transitions == 0
and ((recovered is None and recovered_at is None) or (recovered is not None and recovered_at is not None))
)
if phase == "alerted":
return (
active is not None
and failures >= 1
and successes == 0
and recovered is None
and recovered_at is None
and healthy_since is None
and transitions == 0
)
if phase == "recovery_pending":
return (
active is not None
and failures == 0
and successes >= 1
and recovered is None
and recovered_at is None
and healthy_since is None
and transitions == 0
)
if phase == "recovery_watch":
return (
active is None
and recovered is not None
and failures == 0
and successes == 0
and recovered_at is not None
and healthy_since is None
and transitions == 0
)
if phase != "flapping" or active is None or active != recovered or recovered_at is None or transitions < 1:
return False
if healthy_since is None:
return failures >= 1 and successes == 0
return failures == 0 and successes >= 1
def migrate_v1_state(payload: dict[str, object]) -> dict[str, object] | None:
if set(payload) != {"schemaVersion", "consecutiveFailures", "lastCategory"}:
return None
count = payload.get("consecutiveFailures")
category = payload.get("lastCategory")
if payload.get("schemaVersion") != 1 or not is_bounded_int(count) or not is_category(category):
return None
state = new_probe_state()
if count == 0:
return state if category is None else None
if category is None:
return None
state.update(
{
"phase": "failure_pending",
"activeCategory": category,
"consecutiveFailures": count,
}
)
return state
def load_probe_state(path: Path) -> dict[str, object]:
if not path.exists():
return new_probe_state()
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")
if payload.get("schemaVersion") == STATE_SCHEMA_VERSION:
if not validate_v2_state(payload):
fail("local_state_invalid")
return payload
migrated = migrate_v1_state(payload)
if migrated is None:
fail("local_state_invalid")
return migrated
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 capped(value: int, maximum: int = MAX_COUNTER) -> int:
return min(value, maximum)
def elapsed(now: int, started_at: object) -> int:
if not isinstance(started_at, int) or isinstance(started_at, bool):
return 0
return max(0, now - started_at)
def transition_probe_state(
state: dict[str, object],
category: str | None,
now: int,
failure_threshold: int,
) -> tuple[dict[str, object], tuple[str, str] | None]:
"""Return the next redacted state and an optional notification candidate."""
if not validate_v2_state(state) or not is_timestamp(now, allow_none=False):
raise ValueError("invalid transition input")
if category is not None and category not in FAILURE_CATEGORIES:
raise ValueError("invalid transition category")
phase = str(state["phase"])
if category is None:
if phase == "healthy":
return state.copy(), None
if phase == "failure_pending":
if state["recoveredAt"] is not None:
next_state = state.copy()
next_state.update(
{
"phase": "recovery_watch",
"activeCategory": None,
"consecutiveFailures": 0,
}
)
return next_state, None
return new_probe_state(), None
if phase == "alerted":
next_state = state.copy()
next_state.update(
{
"phase": "recovery_pending",
"consecutiveFailures": 0,
"consecutiveSuccesses": 1,
}
)
return next_state, None
if phase == "recovery_pending":
successes = capped(int(state["consecutiveSuccesses"]) + 1)
if successes < RECOVERY_THRESHOLD:
next_state = state.copy()
next_state["consecutiveSuccesses"] = successes
return next_state, None
recovered_category = str(state["activeCategory"])
next_state = new_probe_state()
next_state.update(
{
"phase": "recovery_watch",
"lastRecoveredCategory": recovered_category,
"recoveredAt": now,
}
)
return next_state, (HEALTHY_STATUS, "public_ready_healthy")
if phase == "recovery_watch":
if elapsed(now, state["recoveredAt"]) >= RECOVERY_WATCH_SECONDS:
return new_probe_state(), None
return state.copy(), (HEALTHY_STATUS, "public_ready_healthy")
if phase == "flapping":
next_state = state.copy()
if state["healthySince"] is None:
next_state["healthySince"] = now
next_state["consecutiveFailures"] = 0
next_state["consecutiveSuccesses"] = 1
next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
else:
next_state["consecutiveSuccesses"] = capped(int(state["consecutiveSuccesses"]) + 1)
if elapsed(now, next_state["healthySince"]) >= FLAPPING_STABLE_SECONDS:
return new_probe_state(), None
return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))
if phase == "flapping":
next_state = state.copy()
if state["healthySince"] is not None:
next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
next_state.update(
{
"consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
"consecutiveSuccesses": 0,
"healthySince": None,
}
)
return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))
if phase == "alerted":
next_state = state.copy()
next_state.update(
{
"activeCategory": category,
"consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
"consecutiveSuccesses": 0,
}
)
return next_state, (FAILURE_STATUS, category)
if phase == "recovery_pending":
next_state = new_probe_state()
next_state.update(
{
"phase": "alerted",
"activeCategory": category,
"consecutiveFailures": 1,
}
)
return next_state, (FAILURE_STATUS, category)
if phase == "recovery_watch":
within_watch = elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
next_state = new_probe_state()
next_state.update(
{
"phase": "failure_pending",
"activeCategory": category,
"consecutiveFailures": 1,
"lastRecoveredCategory": state["lastRecoveredCategory"] if within_watch else None,
"recoveredAt": state["recoveredAt"] if within_watch else None,
}
)
return next_state, None
if phase == "healthy":
next_state = new_probe_state()
next_state.update(
{
"phase": "failure_pending",
"activeCategory": category,
"consecutiveFailures": 1,
}
)
return next_state, None
failures = capped(int(state["consecutiveFailures"]) + 1)
if failures < failure_threshold:
next_state = state.copy()
next_state.update({"activeCategory": category, "consecutiveFailures": failures})
return next_state, None
same_recent_failure = (
state["lastRecoveredCategory"] == category
and state["recoveredAt"] is not None
and elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
)
if same_recent_failure:
next_state = state.copy()
next_state.update(
{
"phase": "flapping",
"activeCategory": category,
"consecutiveFailures": failures,
"consecutiveSuccesses": 0,
"healthySince": None,
"flapTransitions": 1,
}
)
return next_state, (FLAPPING_STATUS, category)
next_state = new_probe_state()
next_state.update(
{
"phase": "alerted",
"activeCategory": category,
"consecutiveFailures": failures,
}
)
return next_state, (FAILURE_STATUS, category)
def classify_connection_error(reason: object) -> str:
if isinstance(reason, (ssl.SSLError, ssl.CertificateError)):
return "tls_validation_failed"
if isinstance(reason, (socket.timeout, TimeoutError)):
return "request_timeout"
if isinstance(reason, socket.gaierror):
return "dns_resolution_failed"
return "tcp_connection_failed"
def classify_url_error(error: urllib.error.URLError) -> str:
return classify_connection_error(error.reason)
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 socket.gaierror:
raise ProbeFailure("dns_resolution_failed") from None
except (http.client.HTTPException, ConnectionError, OSError):
raise ProbeFailure("tcp_connection_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 in {FAILURE_STATUS, FLAPPING_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 persist_transition(
state_file: Path,
summary_file: Path,
alert_state_file: Path,
state: dict[str, object],
candidate: tuple[str, str] | None,
alert_helper: Path,
webhook_file: Path,
) -> None:
try:
atomic_write(state_file, state)
if candidate is not None:
status, category = candidate
write_summary(summary_file, status, category)
except OSError:
fail("local_state_invalid")
if candidate is not None:
evaluate_alert(alert_helper, webhook_file, summary_file, alert_state_file)
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)
now = int(time.time())
try:
probe(ready_url, args.timeout_seconds, args.ca_file)
except ProbeFailure as error:
try:
next_state, candidate = transition_probe_state(state, error.category, now, args.failure_threshold)
except ValueError:
fail("local_state_invalid")
persist_transition(
probe_state_file,
summary_file,
alert_state_file,
next_state,
candidate,
args.alert_helper,
args.webhook_file,
)
alert_result = "evaluated" if candidate is not None else "pending"
print(
f"[app-health][ERROR] health_check_failed category={error.category} "
f"consecutive={next_state['consecutiveFailures']} alert={alert_result}",
file=sys.stderr,
)
raise SystemExit(1)
try:
next_state, candidate = transition_probe_state(state, None, now, args.failure_threshold)
except ValueError:
fail("local_state_invalid")
persist_transition(
probe_state_file,
summary_file,
alert_state_file,
next_state,
candidate,
args.alert_helper,
args.webhook_file,
)
print(f"[app-health][PASS] public readiness is healthy phase={next_state['phase']}")
if __name__ == "__main__":
main()