Comment créer un bot Telegram pour surveiller la disponibilité de votre service en Python (Partie 1 : Instant Metrics)

Bonjour à tous! Depuis plusieurs années maintenant, j'écris pour moi-même divers robots de télégramme "assistants" en Python qui gèrent diverses petites tâches de routine pour moi - m'informer de quelque chose, vérifier la disponibilité du service, transférer du contenu intéressant à partir de canaux de télégramme et de chats, etc.

C'est pratique car le téléphone est toujours à portée de main, et pouvoir réparer quelque chose sur le serveur sans même ouvrir mon portable me procure un plaisir particulier.

En général, j'ai accumulé de nombreux modèles de petits projets différents que je souhaite partager avec les lecteurs de HackerNoon.

Je dirai tout de suite que les exemples peuvent être de niche en termes d'application "telle quelle", mais je marquerai les endroits où, en changeant quelques lignes de code par les vôtres, vous pourrez réutiliser la plupart des développements pour vos projets.

J'ai terminé ce projet spécifique il y a quelques jours, et il m'a déjà apporté beaucoup d'avantages. Je travaille chez un fournisseur d'infrastructure Web3 chainstack.com, traitant d'un service d'indexation des données des contrats intelligents sur les blockchains EVM.

Et la qualité du service en cours de développement dépend essentiellement du "bon" fonctionnement des nœuds à partir desquels le service récupère les données en ligne.

J'ai passé de nombreuses heures à essayer d'utiliser des outils prêts à l'emploi que notre division d'infrastructure utilise, tels que Grafana, BetterUptime et d'autres, mais comme je m'intéresse peu aux composants internes du système, l'objectif principal pour moi étant les métriques à l'entrée et la sortie, j'ai décidé d'écrire mon propre bot, qui ferait ce qui suit :

• À ma demande, il irait au service, vérifierait les mesures et m'enverrait un bref rapport sur la situation actuelle.

• À mon autre demande, il m'enverrait des graphiques de ce qui s'est passé au cours des X dernières heures.

• En cas de situation particulière, il m'enverrait une notification indiquant que quelque chose se passe à ce moment-là.

Dans cet article, je vais me concentrer sur la première partie, c'est-à-dire recevoir des métriques sur demande.

Nous aurons besoin d'un nouvel environnement virtuel pour travailler.

 cd ~ virtualenv -p python3.8 up_env # crete a virtualenv source ~/up_env/bin/activate # activate the virtualenl

Installez les dépendances :

 pip install python-telegram-bot pip install "python-telegram-bot[job-queue]" --pre pip install --upgrade python-telegram-bot==13.6.0 # the code was written before version 20, so here the version is explicitly specified pip install numpy # needed for the median value function pip install web3 # needed for requests to nodes (replace with what you need)

Fichier avec les fonctions functions.py (vous pouvez l'implémenter avec des classes, mais comme l'exemple est court, je n'avais pas prévu de le diviser en modules, mais une bibliothèque multi-thread nécessite que les fonctions soient déplacées dans un fichier séparé). Dépendances d'importation :

 import numpy as np import multiprocessing from web3 import Web3 # add those libraries needed for your task

Décrire une fonction pour vérifier l'état. Dans mon cas, cela impliquait de parcourir des nœuds publics présélectionnés, de récupérer leur dernier bloc, de prendre la valeur médiane pour filtrer tout écart, puis de vérifier notre propre nœud par rapport à cette médiane.

Fonction de vérification de l'état du service (vous pouvez la remplacer par la vôtre) :

 # 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

Le prochain fichier important du bot est uptime_bot.py . Nous importons des bibliothèques et des fonctions à partir du fichier ci-dessus et définissons les constantes nécessaires :

 import telegram from telegram.ext import Updater, CommandHandler, Filters from functions import get_last_block_once, check_service # Here one can to set a limited circle of bot users, # listing the usernames of the users ALLOWED_USERS = ['your_telegram_account', 'someone_else'] # The address of the node that I am monitoring (also a public node in this case) OBJECT_OF_CHECKING = 'https://polygon-mainnet.chainstacklabs.com' # Threshold for highlighting critical lag THRESHOLD = 5

Ensuite, décrivons une fonction qui sera appelée lorsque la commande sera émise depuis l'interface utilisateur du bot.

 def start(update, context): """Send a message when the command /start is issued.""" try: # Get the user user = update.effective_user # Filter out bots if user.is_bot: return # Check if the user is allowed username = str(user.username) if username not in ALLOWED_USERS: return except Exception as e: print(f'{repr(e)}') return # Call the main function to check the network status max_val, max_support, med_val, med_support = check_service() # Call the function to check the status of the specified node last_block = get_last_block_once(OBJECT_OF_CHECKING) # Create the message to send to Telegram message = "" # Information about the state of the nodes in the public network (median, maximum, and number of nodes) 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" # Compare with the threshold 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: # Exception processing if a node has not responded message += f"The node has ⚠️<b>not responded</b>⚠️" # Send the message to the user context.bot.send_message(chat_id=user.id, text=message, parse_mode="HTML")

Maintenant, il ne reste plus qu'à ajouter la partie où le bot est initialisé et la fonction de gestionnaire est connectée :

 token = "xxx" # Bot token obtained from 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()

Enfin, vous pouvez exécuter le code sur un serveur VPS bon marché en utilisant :

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

Après avoir configuré le fichier d'unité systemd.

En conséquence, le travail du bot ressemblera à ceci.

1. Si tout va bien :

   
   

2. Et si le décalage devient trop important, alors comme suit :

   
   

Dans les articles suivants, je décrirai comment implémenter les deux tâches restantes :

1. Récupérez à la demande des graphiques montrant les événements survenus au cours des X dernières heures.

   
   

2. Recevez une alerte indiquant que quelque chose se passe actuellement et nécessite une action.

Le code source du projet est disponible dans le dépôt GitHub. Si vous avez trouvé ce tutoriel utile, n'hésitez pas à lui donner une étoile sur GitHub, je l'apprécierais🙂