Uptime Monitor

В отеле, ресторане или небольшом офисе камеру или точку Wi-Fi, которая перестала работать, обычно замечают через несколько дней: когда понадобилась запись, которой нет, или когда гости жалуются на Wi-Fi.

Большие системы мониторинга вроде Zabbix или PRTG слишком тяжелы для места без IT-отдела. Я профессионально устанавливаю камеры, регистраторы и сети MikroTik, поэтому хотел один небольшой процесс и один YAML-файл.

• Устройства перечислены в devices.yaml: имя, IP-адрес, ping вкл./выкл. и TCP-порты для проверки (554 для IP-камеры, 9100 для принтера).
• Каждые 60 секунд все устройства проверяются параллельно через asyncio: ping плюс TCP-подключение к каждому порту.
• Устройство считается упавшим только после двух неудачных раундов подряд, а поднявшимся — после двух удачных. Один потерянный ping или «мигающая» точка доступа не засыпают чат.
• Тревоги в Telegram говорят, что именно не работает («ping: no reply; tcp/554: timed out»), а при восстановлении — сколько длился простой.
• Страница статуса (FastAPI + HTMX, обновляется каждые 10 секунд) показывает аптайм за 24 часа и 7 дней, полосы по часам и последние аварии.
• Состояния устройств сохраняются, поэтому после перезапуска одна и та же авария не объявляется дважды.

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

Ping использует системную команду ping, а не «сырые» сокеты, которым нужны права администратора. Адреса устройств проверяются до того, как попадут в командную строку, поэтому адрес, который начинается с «-», не может стать опцией ping. Аптайм — это удачные проверки, делённые на все проверки за период: один SQL-запрос.

Решение «упал / поднялся» — чистая функция. Простой начинается с первой неудачной проверки и заканчивается первой удачной проверкой восстановления, а не в моменты отправки тревог:

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

На неанглийской Windows не измерялась задержка ping

_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 автоматических теста, включая настоящий вывод ping с русской Windows, английской Windows и Linux.
• Проверено на настоящей сети: домашний роутер, DNS Cloudflare, закрытый порт и неизвестное имя хоста — у каждого правильный результат и причина.
• Демо-режим моделирует восемь устройств и генерирует неделю истории примерно за секунду.

• Исходный код на GitHub↗
• Живое демо→