Как сделать Telegram-бота для проверки аптайма своего сервиса на Python (ч.2 алертинг)

Всем привет, в предыдущей статье я начал описывать мой опыт разработки Python телеграм-бота для проверки работоспособности и мониторинга моего сервиса, расположенного на удаленом сервере. Если в двух словах, то когда делаешь какой-нибудь пет-проект, да и в некоторых рабочих задачах, бывает хочется иметь всю текущую информацию по состоянию системы под рукой (особенно мне нравится возможность видеть все и управлять через телеграм бота), при этом не хочется тратить на разработку много времени.

В предыдущей части мы рассматривали способ получить мгновенные метрики по запросу. В этой части мы будем делать простой алертинг, то есть получение сообщения в боте, когда в системе начались сбои. А в следующей третьей части разберем еще случай сбора аналитики и получения онлайн графиков.

Как и в прошлой части пример будет основан на реальной задаче, но я буду помечать те места в коде, которые вы можете поменять на свою логику, чтобы основную часть примера можно было переиспользовать.

В данном случае мне необходимо получить алерт в виде сообщения в телеграм боте в том случае, если интересующая меня нода потеряла связь с сетью (или по какой-то другой причине), но ее последний засинхронизированный блок отстает от последнего блока сети.

Как и в прошлый раз сначала необходимо настроить виртуальное окружение:

cd ~
virtualenv -p python3.8 up_env  # создаем окружение
source ~/up_env/bin/activate  # активируем окружение

и устанавливить необходимые зависимости:

pip install python-telegram-bot
pip install "python-telegram-bot[job-queue]" --pre
pip install --upgrade python-telegram-bot==13.6.0  # код написан во времена до версии 20, поэтому здесь версия указывается явно
pip install numpy # нужна для функции получения медианного значения
pip install web3 # нужна для запросов к нодам (замените на то, что необходимо вам)

Файл с функциями functions.py не претерпевает в данном случае изменений и остается таким же, как в предыдущей части:

# импортируем необходимые библиотеки
import numpy as np
from web3 import Web3
import multiprocessing

# Вспомогательная функция, которая проверяет отдельно одну ноду
def get_last_block_once(rpc):
    try:
        w3 = Web3(Web3.HTTPProvider(rpc))
        block_number = w3.eth.block_number
        if isinstance(block_number, int):
            return block_number
        else:
            return None
    except Exception as e:
        print(f'{rpc} - {repr(e)}')
        return None

# Основная функция проверки состояния сервиса, которая будет вызываться 
# из основного потока бота
def check_service():
    # заранее подготовленный список референсных нод
    # для любой сети его можно найти на сайте https://chainlist.org/
    list_of_public_nodes = [
        'https://polygon.llamarpc.com',
        'https://polygon.rpc.blxrbdn.com',
        'https://polygon.blockpi.network/v1/rpc/public',
        'https://polygon-mainnet.public.blastapi.io',
        'https://rpc-mainnet.matic.quiknode.pro',
        'https://polygon-bor.publicnode.com',
        'https://poly-rpc.gateway.pokt.network',
        'https://rpc.ankr.com/polygon',
        'https://polygon-rpc.com'
    ]
    
    # параллельная обработка запросов ко всем нодам
    with multiprocessing.Pool(processes=len(list_of_public_nodes)) as pool:
        results = pool.map(get_last_block_once, list_of_public_nodes)
        last_blocks = [b for b in results if b is not None and isinstance(b, int)]
        
    # определени максимального и мединного значения текущего блока
    med_val = int(np.median(last_blocks))
    max_val = int(np.max(last_blocks))

    # определение количества нод с максимальным и медианным значением
    med_support = np.sum([1 for x in last_blocks if x == med_val])
    max_support = np.sum([1 for x in last_blocks if x == max_val])

    return max_val, max_support, med_val, med_support

Теперь рассмотрим основной файл бота alert_bot.py. Так как в разных задачах может потребоваться только алертинг или только запрос мгновенных значений, то я не собираю бота, который может все и сразу, а наоборот разделяю этот функционал на разные маленькие примеры. В данном случае код основного файла бота будет включать в себя только алертинг, но вы можете объединить все, что необходимо в один бот.

Итак, импортируем библиотеки и функции из файла выше и задаем необходимые константы:

import telegram
from telegram.ext import Updater
from functions import get_last_block_once, check_service 

# Адрес ноды, состояние которой я отслеживаю (тоже публичная нода в данном случае)
OBJECT_OF_CHECKING = 'https://polygon-mainnet.chainstacklabs.com'
# Порог для подсвечивания критического отставания
THRESHOLD = 5
# ID вашего аккаунта в телеграм. Проще всего узнать через бота @chatIDrobot
USER_ID = 123456789 

Далее описываем функцию, которая будет вызвана регулярно по таймеру:

def check_for_alert(context):

    # Вызов основной функции проверки состояния сети
    max_val, max_support, med_val, med_support = check_service()
    # Вызов функции проверки состояния проверяемой ноды
    last_block = get_last_block_once(OBJECT_OF_CHECKING)

    # Формирование сообщения для отправки в телеграм
    message = ""
    # Информация о состоянии нод во внешней сети (медиана, максимум и количество нод)
    message += f"Public median block number {med_val} (on {med_support}) RPCsn"
    message += f"Public maximum block number +{max_val - med_val} (on {max_support}) PRCsn"

    # данная переменная будет хранить решение, отправлять ли алерт
    # в случае, если нода отстает или не ответила
    to_send = False

    # проверка состояния
    if last_block is not None:
        out_text = str(last_block - med_val) if last_block - med_val < 0 else '+' + str(last_block - med_val)
        # Сравнение с порогом
        if abs(last_block - med_val) > THRESHOLD:
            to_send = True
            message += f"The node block number shift ⚠️<b>{out_text}</b>⚠️"
        else:
            message += f"The node block number shift {out_text}"
    else: # Обработка исключения, если нода не ответила
        to_send = True
        message += f"The node has ⚠️<b>not responded</b>⚠️"

    # срабатываение алерта и отправка сообщения пользователю
    if to_send:
        context.bot.send_message(chat_id=USER_ID, text=message, parse_mode="HTML")

Далее остается только дописать часть, в которой инициализируется бот и к нему подключается регулярная "джоба", проверяющая состояние ноды:

# Токен вашего телеграм бота, полученный через BotFather
token = "xxx"

# создание экземпляра бота
bot = telegram.Bot(token=token)
updater = Updater(token=token, use_context=True)
dispatcher = updater.dispatcher
job_queue = updater.job_queue
# Здесь переменной interval (в секундах) задается периодичность 
# запуска проверки состояния - в данном случае каждые 10 минут
job_queue.run_repeating(check_for_alert, interval=10.0 * 60.0, first=0.0)
# запуск бота
updater.start_polling()

Далее код можно запустить на любом VPS сервере через:

source ~/up_env/bin/activate
python uptime_bot.py

Предварительно настроив systemd unit-файл.

В итоге работа бота будет выглядеть следующим образом - если алерт сработал, я получаю сообщение с информацией о проблеме:

В следующей статье я опишу, как можно реализовать оставшуюся задачу:

• По запросу получить графики, как все происходило в течение последних X часов. Она будет состоять из двух частей: скрипта для логгирования запускаемого по cron'у и бота, собирающего графики из логов по запросу пользователя

Исходный код проекта доступен в репозитории на GitHub. Присоединяйтесь, делайте форк и предлагайте свои улучшения!