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

Всем привет! В предыдущих статьях (часть 1 и часть 2) я описывал мой опыт в части "наколенной" разработки системы алертинга и проверки состояния для сервиса, работающего на удаленном сервере, коммуникации с которым происходят через телеграм бота. Такой способ коммуникации удобен, потому что телефон с телегой всегда под рукой, а ноутбук иногда даже доставать лень, когда все можно быстро проверить в телеге.

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

После пары дней использования сервиса, описанного в предыдущих статьях, я ощутил необходимость также получать графики сразу в телеграме, а не идти за ноутбуком, открывать csv файлы в jupyter notobook и т.д. И подумал, почему бы не сделать.

У меня были сомнения насчет того, что графики будут не очень хорошо выглядеть, если создавать их не глядя, автоматически. Например, надписи могли бы наезжать на линии или оси. Но в итоге при использовании библиотеки seaborn вместо привычного matplotlib в целом без каких-то особых настроек получается добиться вполне себе приличных графиков.

Начнем как обычно создания виртуального окружения:

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 seaborn # необходима для построения графиков

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

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

Итак, код скрипта для сбора данных data_collection.py:

import datetime 
import csv

# импортируем необходимые функции
from functions import get_last_block_once, check_service

# путь к файлу для логгирования
LOG_FILE = '../logs.csv'
# Адрес ноды, состояние которой я отслеживаю (публичная нода в данном случае)
OBJECT_OF_CHECKING = 'https://polygon-mainnet.chainstacklabs.com'

# функция сохранения одного измерения в csv файл
def save_log(log_data):
    with open(LOG_FILE, mode='a', newline='') as log_file:
        log_writer = csv.writer(log_file)
        log_writer.writerow(log_data)


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

    # текущие дата-время
    timestamp_string = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    # формируем строку данных и сохраняем в файл
    log_data = [timestamp_string, max_val, max_support, med_val, med_support, last_block]
    save_log(log_data)

Данный скрипт запускается через cron c регулярностью, например, в 5 минут. Чтобы открыть настройки cron'a:

crontab -e 

В файле настроек сохраняем строку для запуска скрипта в окружении в созданном окружении up_env, а ошибки отдельно логгируем в другой файл:

* * * * *  cd ~; source up_env/bin/activate; cd /path/to/script; python data_collection.py >> ~/collect.log 2>&1

Как указано в файле data_collection.py данные будут сохраняться в файле ../logs.csv. Их то мы и будем использовать для построения графиков. Теперь перейдем к скрипту с описанием работы телеграм-бота. Импортируем необходимые зависимости и задаем константы:

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

# для считывания данных и построения графиков
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

import datetime
import io

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

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

def send_pics(update, context, interval):

    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

    # чтение данных из файла (в файле отсутствует header)
    df = pd.read_csv(LOG_FILE, header=None, names=[
        'timestamp_string', 'max_val', 'max_support', 'med_val', 'med_support', 'block_number'
    ])
    # преобразование строки в тип даты/времени
    df['timestamp'] = pd.to_datetime(df['timestamp_string'])

    # определение момента, начиная с которого данные будут отражены на графике
    now = datetime.datetime.now()
    # доступны несоклько интервалов - неделя, день, час
    if interval == 'week':
        one_x_ago = now - datetime.timedelta(weeks=1)
    elif interval == 'day':  # day
        one_x_ago = now - datetime.timedelta(days=1)
    else:
        one_x_ago = now - datetime.timedelta(hours=1)
    # фильтрация датафрейма
    df = df[df['timestamp'] >= one_x_ago]

    # формирование колонок с лагами для исследуемой ноды и для ноды с наибольшим значением блока
    cols_to_show = ['node_lag', 'best_node_lag']
    df['node_lag'] = df['block_number'] - df['med_val']
    df['best_node_lag'] = df['max_val'] - df['med_val']

    # создание графика с seaborn
    plt.figure()
    sns.set(rc={'figure.figsize': (11, 4)})  # set figure size
    sns.lineplot(x='timestamp', y='value', hue='variable', data=df[['timestamp']+cols_to_show].melt('timestamp', var_name='variable', value_name='value'))
    # добавление "коридора" для удобства сравнения с порогом
    plt.axhline(y=THRESHOLD, color='black', linestyle='--')
    plt.axhline(y=-THRESHOLD, color='black', linestyle='--')

    # Сохранение картинки в буффер
    buf = io.BytesIO()
    plt.savefig(buf, format='png')
    buf.seek(0)

    # Отправка картинки пользователю
    context.bot.send_photo(chat_id=user.id, photo=buf)

    # "Закрытие" объекта графика
    plt.close()

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

• час

• день

• неделя

Их выбор осуществляется через соотвествующие команды бота /hour, /day, /week, которые в коде выглядят следуюшим образом:

# хэндлер команды вызова графика за последний час
def hour(update, context):
    send_pics(update, context, 'hour')

# хэндлер команды вызова графика за последний день
def day(update, context):
    send_pics(update, context, 'day')

# хэндлер команды вызова графика за последнюю неделю
def week(update, context):
    send_pics(update, context, 'week')

Остается только инициализировать бота и "привязать" команды к их хэндлерам:

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

# создание экземпляра бота
bot = telegram.Bot(token=token)
updater = Updater(token=token, use_context=True)
dispatcher = updater.dispatcher

# подключение хэндлеров обработки команд к командам бота
dispatcher.add_handler(CommandHandler("hour", hour, filters=Filters.chat_type.private))
dispatcher.add_handler(CommandHandler("day", day, filters=Filters.chat_type.private))
dispatcher.add_handler(CommandHandler("week", week, filters=Filters.chat_type.private))

# запуск бота
updater.start_polling()

Помимо этого для удобства можно добавить реальные кнопки для команд через BotFather. В итоге это все добро выглядит следующим образом:

На этом все, если остались вопросы, буду рад ответить.

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