Как сделать 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}) RPCs\n"     message += f"Public maximum block number +{max_val - med_val} (on {max_support}) PRCs\n"      # данная переменная будет хранить решение, отправлять ли алерт     # в случае, если нода отстает или не ответила     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. Если этот туториал показался вам полезным, не стесняйтесь поставить звезду на гитхабе, мне будет приятно?