Uptime Monitor

In einem Hotel, einem Restaurant oder einem kleinen Büro fällt eine ausgefallene Kamera oder ein WLAN-Access-Point meist erst Tage später auf: wenn jemand eine Aufnahme braucht, die es nicht gibt, oder wenn sich Gäste über das WLAN beschweren.

Große Monitoring-Systeme wie Zabbix oder PRTG sind zu schwer für einen Ort ohne IT-Abteilung. Ich installiere beruflich Kameras, Rekorder und MikroTik-Netzwerke, deshalb wollte ich einen kleinen Prozess und eine YAML-Datei.

• Die Geräte stehen in devices.yaml: Name, IP-Adresse, Ping an oder aus und die zu prüfenden TCP-Ports (554 für eine IP-Kamera, 9100 für einen Drucker).
• Alle 60 Sekunden werden alle Geräte mit asyncio parallel geprüft: ein Ping plus eine TCP-Verbindung zu jedem Port.
• Ein Gerät gilt erst nach zwei fehlgeschlagenen Runden in Folge als DOWN und erst nach zwei erfolgreichen wieder als UP. So flutet weder ein verlorener Ping noch ein flackernder Access Point den Chat.
• Telegram-Alarme sagen, was ausgefallen ist („ping: no reply; tcp/554: timed out"), und bei der Wiederherstellung, wie lange der Ausfall dauerte.
• Eine Statusseite (FastAPI + HTMX, alle 10 Sekunden aktualisiert) zeigt die Verfügbarkeit über 24 Stunden und 7 Tage, Stundenbalken und die letzten Ausfälle.
• Gerätezustände werden gespeichert, damit ein Neustart denselben Ausfall nie zweimal meldet.

• Python 3.12
• asyncio
• FastAPI
• HTMX
• Jinja2
• aiogram 3
• SQLite
• pytest

Ping nutzt den Systembefehl ping statt Raw-Sockets, für die Administratorrechte nötig wären. Geräteadressen werden geprüft, bevor sie in die Kommandozeile gelangen, damit eine Adresse, die mit „-" beginnt, nicht als Ping-Option gelesen wird. Verfügbarkeit ist die Zahl erfolgreicher Prüfungen geteilt durch alle Prüfungen im Zeitraum — eine SQL-Abfrage.

Die UP-/DOWN-Entscheidung ist eine reine Funktion. Der Ausfall beginnt bei der ersten fehlgeschlagenen Prüfung und endet bei der ersten erfolgreichen Prüfung der Erholung, nicht in dem Moment, in dem die Alarme verschickt wurden:

def apply(state, *, device, ok, at, rules, problems=()):
    if ok:
        first_success = state.first_success_at or at
        state = replace(state, failures=0, first_failure_at=None,
                        successes=state.successes + 1, first_success_at=first_success)
        if state.status is Status.UNKNOWN:
            return replace(state, status=Status.UP), None
        if state.status is Status.DOWN and state.successes >= rules.recover_after:
            event = Event(EventKind.UP, device, first_success, state.down_since)
            return replace(state, status=Status.UP, down_since=None), event
        return state, None

    first_failure = state.first_failure_at or at
    state = replace(state, successes=0, first_success_at=None,
                    failures=state.failures + 1, first_failure_at=first_failure)
    if state.status is not Status.DOWN and state.failures >= rules.fail_after:
        event = Event(EventKind.DOWN, device, at, first_failure, problems)
        return replace(state, status=Status.DOWN, down_since=first_failure), event
    return state, None

Auf einem nicht-englischen Windows wurde die Ping-Latenz nie gemessen

_LATENCY = re.compile(r"[=<]\s*(\d+(?:[.,]\d+)?)\s*(?:ms|мс)", re.IGNORECASE)

match = _LATENCY.search(output) if ok else None

_LATENCY_LATIN = re.compile(r"time[=<]\s*(\d+(?:[.,]\d+)?)\s*ms", re.IGNORECASE)
_LATENCY_BEFORE_TTL = re.compile(r"[=<]\s*(\d+(?:[.,]\d+)?)\S*\s+TTL=")

match = _LATENCY_LATIN.search(output) or _LATENCY_BEFORE_TTL.search(output)

• 52 automatische Tests, darunter echte Ping-Ausgaben von russischem Windows, englischem Windows und Linux.
• In einem echten Netzwerk getestet: Heimrouter, Cloudflare-DNS, ein geschlossener Port und ein unbekannter Hostname, jeweils mit dem richtigen Ergebnis und Grund.
• Der Demo-Modus simuliert acht Geräte und erzeugt eine Woche Historie in etwa einer Sekunde.

• Quellcode auf GitHub↗
• Live-Demo→