How to Create a Telegram Bot for Monitoring Your Service Uptime in Python (Part 3: Getting Graphs)

Hello everyone! In the previous articles (part 1 and part 2), I described my experience with the development of an alerting and state checking system for a service operating on a remote server, with communication occurring through a Telegram bot. This way of communication is convenient because the phone with Telegram is always at hand, and sometimes it's even too much effort to take out the laptop when you can quickly check everything in Telegram.

In this part, I will describe the process of regularly collecting data and forming graphs on the functioning of the service, which can also be obtained through the bot.

After a couple of days of using the service described in previous articles, I felt the need to also get graphs directly in Telegram, rather than going for the laptop, opening csv files in Jupyter notebook, etc. And I thought, why not do it?

I had doubts that the graphs would not look very good if created blindly, automatically. For instance, labels might overlap with lines or axes. But in the end, using the Seaborn library instead of the usual Matplotlib, without any special settings, it is possible to achieve quite decent graphs.

Let's start as usual with creating a virtual environment:

cd ~
virtualenv -p python3.8 up_env 
source ~/up_env/bin/activate

and installing the necessary dependencies:

pip install python-telegram-bot
pip install "python-telegram-bot[job-queue]" --pre
pip install --upgrade python-telegram-bot==13.6.0 
pip install numpy 
pip install seaborn

pip install web3 # add here anything you need

As in the previous article, the functions.py file does not undergo any changes in this case and remains the same:

import numpy as np
import multiprocessing

from web3 import Web3 #  add those libraries needed for your task

# Helper function that checks a single node
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


# Main function to check the status of the service that will be called
def check_service():
    # pre-prepared list of reference nodes
    # for any network, it can be found on the website 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'
    ]
    
    # parallel processing of requests to all nodes
    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)]

    # define the maximum and median value of the current block
    med_val = int(np.median(last_blocks))
    max_val = int(np.max(last_blocks))
    # determine the number of nodes with the maximum and median value
    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

A significant difference from the two previous scripts to launch the bot is that to create a graph and then send it through the bot, you first need to collect data, or rather, collect it at some regularity. Since, unlike alerting, here I want the data to be written regardless of the operability of the bot itself (for example, downtime during bot updates), the script for data collection will be put into a separate file and will be run through cron at some regularity.

So, the code for the data collection script data_collection.py:

import datetime 
import csv

# Import necessary functions
from functions import get_last_block_once, check_service

# File path for logging
LOG_FILE = '../logs.csv'
# Address of the node whose state I am monitoring (public node in this case)
OBJECT_OF_CHECKING = 'https://polygon-mainnet.chainstacklabs.com'

# Function to save one measurement to a CSV file
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__':
    # Call the main function for checking the network state
    max_val, max_support, med_val, med_support = check_service()
    # Call the function for checking the state of the node being monitored
    last_block = get_last_block_once(OBJECT_OF_CHECKING)

    # Current date and time
    timestamp_string = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    # Formulate data string and save to file
    log_data = [timestamp_string, max_val, max_support, med_val, med_support, last_block]
    save_log(log_data)

This script is run through cron with a regularity, for example, every 5 minutes. To open cron settings:

crontab -e 

In the settings file, we save the line to run the script in the created up_env environment, and separately log errors in another file:

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

As stated in the data_collection.py file, data will be saved in the ../logs.csv file. We will use them to build graphs. Now let's move on to the script describing the work of the Telegram bot. Import the necessary dependencies and set the constants:

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

# For reading data and building charts
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

import datetime
import io

# Here I can specify a limited circle of bot users by listing their usernames
ALLOWED_USERS = ['your_telegram_account', 'someone_else']
# The node address whose status I am tracking (also a public node in this case)
OBJECT_OF_CHECKING = 'https://polygon-mainnet.chainstacklabs.com'
# Threshold for highlighting critical lag
THRESHOLD = 5
# Data file
LOG_FILE = '../logs.csv'

We describe a function that collects a graph and sends it to the user:

def send_pics(update, context, interval):

    try:
        # Get the user
        user = update.effective_user

        # Filter out bots (just in case)
        if user.is_bot:
            return

        # Check if the user is in the allowed list 
        username = str(user.username)
        if username not in ALLOWED_USERS:
            return
    except Exception as e:
        print(f'{repr(e)}')
        return

    # Read data from a file (file does not have a header)
    df = pd.read_csv(LOG_FILE, header=None, names=[
        'timestamp_string', 'max_val', 'max_support', 'med_val', 'med_support', 'block_number'
    ])
    # Convert a string to a date/time type
    df['timestamp'] = pd.to_datetime(df['timestamp_string'])

    # Determine the starting point from which the data will be reflected on the graph
    now = datetime.datetime.now()
    # Multiple intervals are available - week, day, hour
    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)
    # Filter the dataframe
    df = df[df['timestamp'] >= one_x_ago]

    # Create columns with lags for the selected node and for the node with the highest block value
    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']

    # Create a graph with 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'))
    # Add a "corridor" for easier comparison with the threshold
    plt.axhline(y=THRESHOLD, color='black', linestyle='--')
    plt.axhline(y=-THRESHOLD, color='black', linestyle='--')

    # Save the image to a buffer
    buf = io.BytesIO()
    plt.savefig(buf, format='png')
    buf.seek(0)

    # Send the image to the user
    context.bot.send_photo(chat_id=user.id, photo=buf)

    # "Close" the graph object
    plt.close()

In this case, in the code, you can choose intervals for display:

• hour
• day
• week

Their selection is carried out through the corresponding bot commands /hour, /day, /week, which look like this in the code:

# Handler for the command to display the chart for the last hour
def hour(update, context):
    send_pics(update, context, 'hour')

# Handler for the command to display the chart for the last day
def day(update, context):
    send_pics(update, context, 'day')

# Handler for the command to display the chart for the last week
def week(update, context):
    send_pics(update, context, 'week')

All that remains is to initialize the bot and "bind" the commands to their handlers:

# Your Telegram bot token obtained through BotFather
token = "xxx"

# Create a bot instance
bot = telegram.Bot(token=token)
updater = Updater(token=token, use_context=True)
dispatcher = updater.dispatcher


# Connect command handlers to the bot's commands
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))

# Start the bot
updater.start_polling()

In addition to this, for convenience, you can add real buttons for commands via BotFather. Finally the result looks like this:

PS. I noticed that the labels under the time axis were overlapping, but this problem can be easily solved by adding a rotation of 15 degrees with the command "plt.xticks(rotation=15)", executed after plotting the graphs.

That's all, if there are any questions, I will be glad to answer them.

The source code of the project is available in the repository on GitHub. If this tutorial seemed useful to you, feel free to put a star on GitHub, I would appreciate it 🙂