Всем привет! Я уже несколько лет пишу для себя разных телеграм-ботов "помощников" на python, которые делают за меня ту или иную маленькую рутинную задачу - уведомляют меня о чем-то, проверяют аптайм сервисов, пересылают интересное из телеграм-каналов и чатов и тд. Это удобно, потому что телефон всегда под рукой, а возможность что-то поправить на сервере даже не открывая ноутбук, мне доставляет особое удовольствие.

В общем, у меня накопилось много разных небольших шаблонов проектов, которыми я хочу поделиться с хабровчанами. Сразу скажу, что примеры могут быть узкими с точки зрения применения "как есть", но я буду помечать те места, где, меняя несколько строк кода на свои, вы сможете переиспользовать основную часть наработок для своих проектов.

Конкретно этот проект я сделал несколько дней назад и он уже принес мне много пользы. Я работаю в Web3 инфраструктурном провайдере и занимаюсь сервисом для индексирования данных смарт-контрактов на EVM-блокчейнах. И качество разрабатываемого сервиса критически зависит от того, насколько "хорошо" работают ноды, с которых сервис в онлайне забирает данные.

Я провел много часов в попытках использования готовых инструментов, которыми пользуется наше инфраструктурное подразделение, типа grafana, betteruptime и прочие, но так как для меня мало интересного во внутренностях системы, при этом много основной интерес для меня представляют именно метрики на входе и на выходе, то я решил написать свой велосипед бота, который бы делал следующее:

• По моему запросу ходил на сервис, проверял метрики и присылал мне короткий отчет, о текущем положении дел.

• По моему другому запросу присылал мне графики, как все происходило в течение последних X часов

• При возникновении особой ситуации присылал мне уведомление, что что-то в данный момент происходит

В данной статье я сосредоточусь на первой части, то есть получении метрик по запросу.

Для работы нам понадобится новое виртуальное окружение

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
import multiprocessing

from web3 import Web3 # добавьте те библиотеки, которые необходимы для вашей задачи

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

Функция провеки состояния сервиса (можете заменить ее на свою):

# Вспомогательная функция, которая проверяет отдельно одну ноду
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

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

import telegram
from telegram.ext import Updater, CommandHandler, Filters

from functions import get_last_block_once, check_service

# Здесь я могу задать ограниченный круг пользователей бота, 
# перечислив username пользователей
ALLOWED_USERS = ['your_telegram_account', 'someone_else']
# Адрес ноды, состояние которой я отслеживаю (тоже публичная нода в данном случае)
OBJECT_OF_CHECKING = 'https://polygon-mainnet.chainstacklabs.com'
# Порог для подсвечивания критического отставания
THRESHOLD = 5

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

def start(update, context):
    """Send a message when the command /start is issued."""
  
    try:
        # Получить user'а
        user = update.effective_user

        # Отфильтровать ботов (а вдруг)
        if user.is_bot:
            return

        # Проверить, есть ли пользователь в списке разрешенных 
        username = str(user.username)
        if username not in ALLOWED_USERS:
            return
    except Exception as e:
        print(f'{repr(e)}')
        return

    # Вызов основной функции проверки состояния сети
    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"

    # Сравнение с порогом
    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:
            message += f"The node block number shift ⚠️<b>{out_text}</b>⚠️"
        else:
            message += f"The node block number shift {out_text}"
    else: # Обработка исключения, если нода не ответила
        message += f"The node has ⚠️<b>not responded</b>⚠️"

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

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

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

# set up the bot
bot = telegram.Bot(token=token)
updater = Updater(token=token, use_context=True)
dispatcher = updater.dispatcher

# bind the handler function
dispatcher.add_handler(CommandHandler("start", start, filters=Filters.chat_type.private))

# run the bot
updater.start_polling()

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

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

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

В итоге работа бота будет выглядеть следующим образом:

Если все в порядке:

А если отставание становится слишком большим, то следующим образом:

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

• По запросу получить графики, как все происходило в течение последних X часов.

• Получить алерт, что что-то в данный момент происходит и требуются действия.

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