Last active
May 23, 2020 21:32
-
-
Save novy4/4d2be87886b073d96f4b533a49e0ef45 to your computer and use it in GitHub Desktop.
Celo missed minet block alert
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python3 | |
### Monitor a celo validator and send telegram notification if validator fails to mine a block in a timely fashion. | |
# | |
# Suggest you run this from screen so that it will keep an eye | |
# on your validator even when your terminal is detached. | |
# | |
### | |
import sys | |
import json | |
import requests | |
import os | |
from time import sleep | |
from datetime import datetime | |
from datetime import timedelta | |
CHECK_INTERVAL = 300 # how often to check for validator failures | |
MINING_THRESHOLD = 10 # minutes we must mine a block else trigger an alert | |
base_url = "https://explorer.celo.org/api?module=account&action=getminedblocks&page=1&offset=1&address=" | |
address = "YOUR_VALIDATOR_SIGNER_ADDRESS" | |
def get_stake_data(): | |
try: | |
response = requests.get(base_url + address) | |
stake_data = response.json() | |
return(stake_data) | |
except: | |
e = sys.exc_info()[0] | |
write_to_page( "<p>Error: %s</p>" % e ) | |
def telegram_bot_sendtext(bot_message): | |
bot_token = '' | |
bot_chatID = '' | |
send_text = 'https://api.telegram.org/bot' + bot_token + '/sendMessage?chat_id=' + bot_chatID + '&parse_mode=Markdown&text=' + bot_message | |
response = requests.get(send_text) | |
return response.json() | |
#main | |
while True: | |
stake_data = get_stake_data() | |
print(stake_data); | |
print("Last block mined at: %s " % stake_data['result'][0]['timeStamp']); | |
# had to strip the 'Z' off the timestamp we get from Celo's API. <shrug> | |
lastBlockTime = datetime.strptime((stake_data['result'][0]['timeStamp'])[:-1], '%Y-%m-%d %H:%M:%S.%f') | |
print(lastBlockTime) | |
timeNow=datetime.utcnow(); | |
timeDiff=(timeNow - lastBlockTime) | |
print("Time now: %s" % timeNow); | |
print("Time since last block mined: %s" % timeDiff) | |
threshold = timedelta(minutes = MINING_THRESHOLD); | |
if timeDiff > threshold: | |
print("No block mined in %s minutes" % threshold) | |
message_content = "Celo monitor found no blocks mined for past %s minutes. Last block mined at %s GMT" % (threshold, lastBlockTime) | |
print("message_content = %s" % message_content) | |
telegram_bot_sendtext(message_content) | |
else: | |
print("Found a block mined within the last %s minutes" % threshold) | |
print("Sleeping for %i seconds" % CHECK_INTERVAL); | |
sleep(CHECK_INTERVAL) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment