Initial commitHEADmaster

3 files changed, 300 insertions, 0 deletions

diff --git a/README.md b/README.md
new file mode 100644
index 0000000..90220e6
--- /dev/null
+++ b/README.md
@@ -0,0 +1,45 @@
+# nanostatus
+
+A minimal service status page generator. It runs shell commands to check if your services are up, records the results, and generates a static HTML page showing uptime over the last 30 days.
+
+## Quick start
+
+Install Python 3 if you don't have it:
+
+```sh
+pkg install python3
+```
+
+Clone the repo and copy the example config:
+
+```sh
+git clone https://github.com/yourname/nanostatus
+cd nanostatus
+cp example.ini status.ini
+```
+
+Edit `status.ini` to add your services. Each service needs a name, a shell command that returns 0 on success, and optionally a URL:
+
+```ini
+[service:mysite]
+url = https://example.com
+cmd = curl --fail https://example.com
+```
+
+Run it:
+
+```sh
+python3 nanostatus.py -c status.ini
+```
+
+This writes check results to `./` and generates `./html/status.html`. To run it on a schedule, add a cron job:
+
+```sh
+crontab -e
+```
+
+```
+*/5 * * * * cd /path/to/nanostatus && python3 nanostatus.py -c status.ini
+```
+
+Serve `html/status.html` with any web server (nginx, caddy, etc.).
diff --git a/example.ini b/example.ini
new file mode 100644
index 0000000..2b66c04
--- /dev/null
+++ b/example.ini
@@ -0,0 +1,9 @@
+[service:website]
+url = https://example.com
+cmd = curl --fail https://example.com
+on_fail = mail -s "website is down" you@example.com
+on_recover = mail -s "website is back up" you@example.com
+
+[service:api]
+url = https://api.example.com
+cmd = curl --fail https://api.example.com/health
diff --git a/nanostatus.py b/nanostatus.py
new file mode 100644
index 0000000..31c5147
--- /dev/null
+++ b/nanostatus.py
@@ -0,0 +1,246 @@
+#!/usr/bin/env python3
+
+import argparse
+import configparser
+import subprocess
+import sys
+
+from dataclasses import dataclass
+from datetime import datetime
+from enum import Enum, auto
+from pathlib import Path
+
+
+class State(Enum):
+    UP = 1
+    DOWN = 2
+    UNSTABLE = 3
+    UNKNOWN = auto()
+
+    @staticmethod
+    def from_rc(rc: int) -> "State":
+        if rc == 0:
+            return State.UP
+        else:
+            return State.DOWN
+
+    def to_colour(self) -> str:
+        return {
+            State.UP: "green",
+            State.DOWN: "red",
+            State.UNSTABLE: "orange",
+            State.UNKNOWN: "gray"
+        }[self]
+
+
+class StateChange(Enum):
+    NONE = 0
+    FAIL = 1
+    RECOVER = 2
+
+
+@dataclass
+class Service:
+    name: str
+    cmd: str
+    url: str
+    on_fail: str | None = None
+    on_recover: str | None = None
+
+
+def get_services_from_config(config_path: Path) -> list[Service]:
+    config = configparser.ConfigParser()
+    config.read(config_path)
+    services = []
+    for section in config.sections():
+        if not section.startswith("service:"):
+            continue
+        name = section.split("service:")[1]
+        cmd = config[section].get("cmd")
+        url = config[section].get("url") or "#"
+        on_fail = config[section].get("on_fail")
+        on_recover = config[section].get("on_recover")
+        if cmd is None:
+            raise ValueError(f"Section service:{name} missing 'cmd'.")
+        services.append(Service(name=name, url=url, cmd=cmd, on_fail=on_fail, on_recover=on_recover))
+    return services
+
+
+def do_check(service: Service) -> int:
+    print(f"Checking {service.name}...", file=sys.stderr, end="")
+    rc = subprocess.run(
+        service.cmd,
+        shell=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    ).returncode
+    print(f" {State.from_rc(rc).name}", file=sys.stderr)
+    return rc
+
+
+def run_on_fail(service: Service):
+    if service.on_fail:
+        subprocess.run(service.on_fail, shell=True)
+
+
+def run_on_recover(service: Service):
+    if service.on_recover:
+        subprocess.run(service.on_recover, shell=True)
+
+
+def write_status(rc: int, status_file: Path):
+    now = datetime.now().isoformat(timespec="seconds")
+    with open(status_file, "a") as f:
+        f.write(f"{now} {rc}\n")
+
+
+def check_for_state_change(status_file: Path) -> StateChange:
+    with open(status_file, "r") as f:
+        lines = f.readlines()[-2:]
+
+    try:
+        prev_rc = int(lines[0].split()[1])
+        cur_rc = int(lines[1].split()[1])
+    except IndexError:
+        return StateChange.NONE
+
+    if prev_rc == 0 and cur_rc != 0:
+        return StateChange.FAIL
+    elif prev_rc != 0 and cur_rc == 0:
+        return StateChange.RECOVER
+    else:
+        return StateChange.NONE
+
+
+def check_service(service: Service, workdir: Path):
+    status_dir = workdir / service.name
+    if not status_dir.exists():
+        status_dir.mkdir(parents=True)
+    today = datetime.today().strftime("%Y-%m-%d")
+    status_file = status_dir / today
+
+    rc = do_check(service)
+    write_status(rc, status_file)
+
+    state_change = check_for_state_change(status_file)
+    if state_change == StateChange.FAIL:
+        run_on_fail(service)
+    elif state_change == StateChange.RECOVER:
+        run_on_recover(service)
+
+
+def write_report_header(f):
+    f.write("""<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Service Status</title>
+    <style>
+        body { font-family: system-ui; max-width: 600px; margin: 2rem auto; padding: 0 1rem; }
+        h1 { margin-bottom: 1rem; }
+        table { border-collapse: collapse; width: 100%; }
+        th, td { padding: .5rem .75rem; border-bottom: 1px solid #ddd; }
+        th { text-align: left; background: #f6f6f6; position: sticky; top: 0; }
+        td:last-child { font-size: 0; white-space: nowrap; }
+        td:last-child span { display: inline-block; font-size: 1rem; width: 6px; height: 1.4em; margin-right: 2px; border-radius: 2px; line-height: 1; background-color: currentColor; }
+        .badge { display: inline-block; font-size: .75rem; font-weight: bold; padding: .15em .5em; border-radius: 999px; color: #fff; }
+        .badge-green { background: green; }
+        .badge-red { background: red; }
+        .badge-orange { background: orange; color: #000; }
+        .badge-gray { background: gray; }
+        footer { margin-top: 1rem; font-size: .8rem; color: #999; }
+    </style>
+</head>
+<body>
+    <h1>Service Status</h1>
+    <table>
+        <tr>
+            <th>Service</th>
+            <th>Status</th>
+            <th>Last 30 days</th>
+        </tr>
+""")
+
+
+def write_service_row(f, service: Service, days: list[tuple[str, State, State]]):
+    latest_state = days[-1][2]
+    badge = f'<span class="badge badge-{latest_state.to_colour()}">{latest_state.name}</span>'
+    f.write(f'        <tr>\n            <td><a href="{service.url}">{service.name}</a></td><td>{badge}</td>\n            <td>\n')
+    for day, overall_state, _ in days:
+        f.write(f'                <span title="{day}: {overall_state.name}" style="color: {overall_state.to_colour()};"></span>\n')
+    f.write("            </td>\n        </tr>\n")
+
+
+def write_report_footer(f):
+    now = datetime.now().strftime("%Y-%m-%d %H:%M")
+    f.write(f"""
+    </table>
+    <footer>Updated: {now}</footer>
+</body>
+</html>
+""")
+
+
+def generate_report(services: list[Service], workdir: Path, output: Path):
+    if not output.exists():
+        output.mkdir(parents=True)
+    report_file = output / "status.html"
+
+    with open(report_file, "w") as report_f:
+        write_report_header(report_f)
+
+        for service in services:
+            status_dir = workdir / service.name
+            # Last 30 days
+            status_files = sorted(status_dir.glob("*"))[-30:]
+            days: list[tuple[str, State, State]] = []  # (date, overall_state, latest_state)
+            for status_file in status_files:
+                with open(status_file, "r") as status_f:
+                    overall_state = State.UNKNOWN
+                    for line in status_f:
+                        when, raw_rc = line.strip().split()
+                        rc = int(raw_rc)
+                        latest_state = State.from_rc(rc)
+                        if overall_state == State.UNKNOWN:
+                            overall_state = latest_state
+                        else:
+                            if latest_state != overall_state:
+                                overall_state = State.UNSTABLE
+                    days.append((status_file.name, overall_state, latest_state))
+            write_service_row(report_f, service, days)
+
+        write_report_footer(report_f)
+
+
+def run(args: argparse.Namespace) -> int:
+    services = get_services_from_config(args.config)
+    for service in services:
+        check_service(service, args.workdir)
+    generate_report(services, args.workdir, args.output)
+
+    return 0
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Monitor the uptime of your services.")
+    parser.add_argument(
+        "-c", "--config", type=Path, required=True, help="Path to the configuration file."
+    )
+    parser.add_argument(
+            # TODO: Should probably be database
+        "-w", "--workdir", type=Path, default=Path.cwd(), help="Working directory."
+    )
+    parser.add_argument(
+        "-o", "--output", type=Path, default=Path.cwd() / "html", help="Output directory."
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    return run(args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())