From b4ecef96fae14a92d12279927986e69c2e223180 Mon Sep 17 00:00:00 2001 From: yyamashita Date: Sat, 31 Oct 2026 13:20:27 +0900 Subject: Add status pages, ntfy alerts, and license-slot-checker repo - status/: lightweight status checker (status.yyamashita.com, /api/status) with ntfy DOWN/recovered alerts after consecutive failures - uptime-kuma/: Uptime Kuma at uptime.yyamashita.com - hetzner-admin: authenticated /status page (status-app results + docker ps -a) - ntfy/: self-hosted ntfy at ntfy.yyamashita.com (deny-all auth, install.sh creates admin user and status-app token) - Caddyfile, DNS, post-receive hook, tests/check.sh, docs updated - license-slot-checker: add to repos.txt, hook, and Claude sessions Co-Authored-By: Claude Opus 5.5 --- status/app.py | 198 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 status/app.py (limited to 'status/app.py') diff --git a/status/app.py b/status/app.py new file mode 100644 index 0000000..9cde2e9 --- /dev/null +++ b/status/app.py @@ -0,0 +1,198 @@ +"""軽量ステータスページ。targets.txt の各サービスを定期チェックし HTML / JSON で公開する。""" +import json +import os +import socket +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from collections import deque +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from html import escape +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +TARGETS_FILE = os.environ.get("TARGETS_FILE", "/app/targets.txt") +INTERVAL = int(os.environ.get("CHECK_INTERVAL", "60")) +TIMEOUT = float(os.environ.get("CHECK_TIMEOUT", "10")) +HISTORY = int(os.environ.get("HISTORY_SIZE", "90")) +PORT = int(os.environ.get("PORT", "8080")) +NTFY_URL = os.environ.get("NTFY_URL", "") # 例: http://ntfy/status（空なら通知しない） +NTFY_TOKEN = os.environ.get("NTFY_TOKEN", "") +FAIL_THRESHOLD = int(os.environ.get("FAIL_THRESHOLD", "2")) # 連続失敗この回数で DOWN 通知 +PUBLIC_URL = os.environ.get("PUBLIC_URL", "https://status.yyamashita.com") + +lock = threading.Lock() +state = {} # name -> {"url", "up", "detail", "ms", "checked_at", "history": deque[bool], "fails", "alerted"} + + +def load_targets(): + targets = [] + with open(TARGETS_FILE) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + name, url = line.split(None, 1) + targets.append((name, url.strip())) + return targets + + +def check(url): + start = time.monotonic() + parsed = urllib.parse.urlparse(url) + try: + if parsed.scheme == "tcp": + with socket.create_connection((parsed.hostname, parsed.port), timeout=TIMEOUT): + pass + up, detail = True, "connected" + else: + req = urllib.request.Request(url, headers={"User-Agent": "status-check/1.0"}) + try: + with urllib.request.urlopen(req, timeout=TIMEOUT) as r: + code = r.status + except urllib.error.HTTPError as e: + code = e.code + up, detail = code < 500, f"HTTP {code}" + except Exception as e: + up, detail = False, f"{type(e).__name__}: {e}" + return up, detail, int((time.monotonic() - start) * 1000) + + +def notify(title, message, down): + if not NTFY_URL: + return + headers = { + "Title": title, + "Tags": "rotating_light" if down else "white_check_mark", + "Priority": "high" if down else "default", + "Click": PUBLIC_URL, + } + if NTFY_TOKEN: + headers["Authorization"] = f"Bearer {NTFY_TOKEN}" + req = urllib.request.Request(NTFY_URL, data=message.encode(), headers=headers, method="POST") + try: + urllib.request.urlopen(req, timeout=TIMEOUT).close() + except Exception as e: + print(f"ntfy notify failed: {e}", flush=True) + + +def run_checks(): + targets = load_targets() + with ThreadPoolExecutor(max_workers=max(1, len(targets))) as pool: + results = list(pool.map(lambda t: check(t[1]), targets)) + now = datetime.now(timezone.utc).isoformat(timespec="seconds") + alerts = [] + with lock: + for name in set(state) - {n for n, _ in targets}: + del state[name] + for (name, url), (up, detail, ms) in zip(targets, results): + entry = state.setdefault( + name, {"history": deque(maxlen=HISTORY), "fails": 0, "alerted": False} + ) + entry.update(url=url, up=up, detail=detail, ms=ms, checked_at=now) + entry["history"].append(up) + entry["fails"] = 0 if up else entry["fails"] + 1 + if not up and entry["fails"] >= FAIL_THRESHOLD and not entry["alerted"]: + entry["alerted"] = True + alerts.append((f"{name} is DOWN", f"{url}\n{detail}", True)) + elif up and entry["alerted"]: + entry["alerted"] = False + alerts.append((f"{name} recovered", f"{url}\n{detail} ({ms} ms)", False)) + for alert in alerts: + notify(*alert) + + +def checker_loop(): + while True: + try: + run_checks() + except Exception as e: + print(f"check failed: {e}", flush=True) + time.sleep(INTERVAL) + + +def snapshot(): + with lock: + services = [ + { + "name": name, + "url": e["url"], + "up": e["up"], + "detail": e["detail"], + "ms": e["ms"], + "checked_at": e["checked_at"], + "uptime": round(100 * sum(e["history"]) / len(e["history"]), 1), + "history": list(e["history"]), + } + for name, e in state.items() + ] + return {"all_up": all(s["up"] for s in services), "services": services} + + +def render_html(data): + rows = [] + for s in data["services"]: + bars = "".join( + f'' for ok in s["history"] + ) + rows.append( + f'{escape(s["name"])}' + f'{escape(s["detail"])}{s["ms"]} ms{s["uptime"]}%' + f'{bars}' + ) + if not data["services"]: + banner = '' + elif data["all_up"]: + banner = '' + else: + banner = '' + checked = data["services"][0]["checked_at"] if data["services"] else "-" + return f""" + +Status - yyamashita.com + +

yyamashita.com status

+{banner} +{"".join(rows)}
+

Last checked: {checked} (every {INTERVAL}s) · JSON

+""" + + +class Handler(BaseHTTPRequestHandler): + def do_GET(self): + data = snapshot() + if self.path == "/api/status": + body = json.dumps(data).encode() + ctype = "application/json" + elif self.path == "/healthz": + body, ctype = b"ok", "text/plain" + elif self.path == "/": + body = render_html(data).encode() + ctype = "text/html; charset=utf-8" + else: + self.send_error(404) + return + self.send_response(200) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + +if __name__ == "__main__": + threading.Thread(target=checker_loop, daemon=True).start() + ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever() -- cgit v1.3