KeyValueFix

README.md

@@ -1,5 +1,62 @@
 # Thornode Monitor
 
+The Monitor dashboard is used to collate information on active, standby and old nodes on the rune network it collates information
+from multiple sources and displays them on a single dashboard. This repo consists of both the front end (js react app) 
+and the backend module (a python api server).
+
+1) Front end:
+  - This is all the code in this repository, minus the lambda functions folder and python backend folder
+  - Built using react
+  - It pulls data from the backend api
+
+2) Back end:
+  - Several parts to the backend
+    1) mySQL database to store all the data
+    2) python app responsible to collect data and provide an api access point to the front end
+
+
+  - The database is fairly simple and consists of 2 tables with the following structure
+
+
+    - thornode_monitor_global - holds network information
+      - primary_key - int
+      - maxHeight - int (current blockheight used to pull other data)
+      - retiring - boolean (indicates if a vault is retiring)
+      - coingecko - string (json string containing price/trading info pulled from coingecko api)
+      - lastChurn - int (block of last churn)
+      - secondsPerBlock - float
+      - churnInterval - int
+      - BadValidatorRedline - int
+
+
+    - thornode_monitor - holds information on each node in the network
+      - node_address - string (primary key)
+      - active_block_height - int (block at which the node became active)
+      - bond_provider - string (json string containing information on bonds and providers)
+      - bond -int (amount bonded)
+      - current_award - int (amount awarded this epoch)
+      - slash_points - int (slash points this epoch)
+      - forced_to_leave - boolean
+      - requested_to_leave - boolean
+      - jail - string (json string containing information on any past jailing)
+      - observe_chains - string (json string containing block heights for downstream chains)
+      - preflight_status - string
+      - status - string
+      - status_since - int
+      - version - string
+      - ip_address - string
+      - location - string
+      - isp - string
+      - rpc - string (json string of response from health rpc calls)
+      - bifrost - string (bifrost p2pid)
+
+
+  - The python app runs off a main loop which pulls data every 20 seconds and updates the DB, it listens to api calls to
+  /api/grabData and returns the data in json.  
+
+
+# Old deployment via AWS lambda (no longer maintained in this fork)
+
 There are several moving parts to this dashboard
 
 1) Front end:

backend python/backend.py

@@ -0,0 +1,55 @@
+import time
+
+from common import grabQuery
+from flask import Flask
+from flask_cors import CORS, cross_origin
+from thormonitor_collect_data import gradDataAndSaveToDB
+from thormonitor_update_ips import updateIPs
+from thornode_collect_data_global import collectDataGlobal
+from thormonitor_collect_data_rpc_bifrost import biFrostGrabDataAndSaveToDB
+
+from threading import Thread
+
+app = Flask(__name__)
+cors = CORS(app)
+app.config['CORS_HEADERS'] = 'Content-Type'
+
+
+def flaskThread():
+    app.run(host='0.0.0.0', port=6000)
+
+
+@app.route('/thor/api/grabData', methods=['GET'])
+@cross_origin()
+def grabData():
+    """
+    grabData is used to output the DB in json format, fires on api accesses
+
+    return: json containing the current data from thornode_monitor and thornode_monitor_global tables
+    """
+    currentDBData = (grabQuery('SELECT * FROM noderunner.thornode_monitor'))
+    globalData = (grabQuery('SELECT * FROM noderunner.thornode_monitor_global'))
+
+    return {'data': currentDBData, 'globalData': globalData[0]}
+
+
+def main():
+    """
+    main contains the main loop which simply spins every minuite and update the various DBs
+    """
+    worker = Thread(target=flaskThread)
+    worker.start()
+
+    while (1):
+        try:
+            gradDataAndSaveToDB()
+            updateIPs()
+            collectDataGlobal()
+            biFrostGrabDataAndSaveToDB()
+        except Exception as e:
+            print(e)
+        time.sleep(60)
+
+
+if __name__ == "__main__":
+    main()

backend python/common.py

@@ -0,0 +1,51 @@
+from dotenv import load_dotenv
+import mysql.connector
+import os
+
+HOST = os.getenv('host')
+USER = os.getenv('user')
+PASSWORD = os.getenv('password')
+DATABASE = os.getenv('db')
+
+load_dotenv()
+
+
+def getDB():
+    """
+    getDB is used to get a sql connector for the db defined in the env file
+
+    :return: db connection
+    """
+    dbConnection = mysql.connector.connect(host=HOST, user=USER, password=PASSWORD, database=DATABASE)
+    return dbConnection
+
+
+
+def commitQuery(query):
+    """
+    commitQuery is used to update items in the database
+
+    :param query: SQL query to be run against the DB
+    """
+    db = getDB()
+    cursor = db.cursor(prepared=True, dictionary=True)
+    cursor.execute(query)
+    db.commit()
+    cursor.close()
+    db.close()
+
+
+def grabQuery(query):
+    """
+    grabQuery is used to grab data from the DB
+
+    :param query: SQL query to be run against the DB
+    :return: returns the output of the query in dictionary format
+    """
+    db = getDB()
+    cursor = db.cursor(prepared=True, dictionary=True)
+    cursor.execute(query)
+    data = cursor.fetchall()
+    cursor.close()
+    db.close()
+    return data

backend python/thormonitor_collect_data.py

@@ -0,0 +1,148 @@
+import time
+
+import requests
+import json
+import random
+from common import commitQuery, grabQuery
+
+
+def grabLatestBlockHeight(nodes):
+    """
+    grabLatestBlockHeight looks at 3 random nodes in the active pool and returns the max height from those 3 nodes
+
+    :param nodes: all thor nodes currently on the network pulled from ninerelms api
+
+    :return: the latest block height
+    """
+    activeNodes = [x for x in nodes if "Active" == x['status']]
+
+    status_code = 0
+    while status_code != 200:
+        randomOffsets = [random.randint(0, len(activeNodes) - 1), random.randint(0, len(activeNodes) - 1),
+                         random.randint(0, len(activeNodes) - 1)]
+
+        status = []
+        for i in range(0, len(randomOffsets)):
+            try:
+                state = requests.get('http://' + activeNodes[randomOffsets[i]]['ip_address'] + ":27147/status?",
+                                     timeout=5)
+                if state.status_code == 200:
+                    status.append(json.loads(state.text))
+            except Exception as e:
+                print("timed out")
+
+        #check if we have any blocks, if yes break from loop if not try another 3 random nodes
+        if len(status) != 0:
+            status_code = 200
+
+    blocks = [x['result']['sync_info']['latest_block_height'] for x in status]
+
+    return max(blocks)
+
+
+def splitNodes(nodes):
+    """
+    splitNodes compares the list of nodes currently in the DB and what is returned by ninerelms API
+
+    :param nodes: all thor nodes currently on the network pulled from ninerelms api
+
+    :return dataForExistingNodes: list of nodes already in our DB
+    :return dataForNewNodes: list of nodes not already in our DB
+    """
+    currentDBData = (grabQuery('SELECT * FROM noderunner.thornode_monitor'))
+    fullAddrList = [x['node_address'] for x in nodes]
+    currentAddrList = [x['node_address'] for x in currentDBData]
+    newList = list(set(fullAddrList).symmetric_difference(set(currentAddrList)))
+
+    dataForExistingNodes = [x for x in nodes if x['node_address'] in currentAddrList]
+    dataForNewNodes = [x for x in nodes if x['node_address'] in newList]
+
+    return dataForExistingNodes, dataForNewNodes
+
+
+def gradDataAndSaveToDB():
+    """
+    gradDataAndSaveToDB used to update thornode_monitor_global database
+    """
+    response_API = requests.get('https://thornode.ninerealms.com/thorchain/nodes')
+    data = json.loads(response_API.text)
+    # sanitise data remove any empty elements
+    nodes = [x for x in data if '' != x['node_address']]
+
+    maxHeight = grabLatestBlockHeight(nodes)
+
+    query = 'UPDATE noderunner.thornode_monitor_global SET maxHeight = {field} WHERE primary_key = 1;'.format(
+        field=maxHeight)
+    commitQuery(query)
+
+    # check if we are in a churn
+
+    dataForExistingNodes, dataForNewNodes = splitNodes(nodes)
+    for node in dataForExistingNodes:
+        query = "UPDATE noderunner.thornode_monitor SET " \
+                "active_block_height = '{active_block_height}'," \
+                "bond_providers = '{bond_providers}'," \
+                "bond = '{bond}'," \
+                "current_award = '{current_award}'," \
+                "slash_points = '{slash_points}'," \
+                "forced_to_leave = '{forced_to_leave}'," \
+                "requested_to_leave = '{requested_to_leave}'," \
+                "jail = '{jail}' ," \
+                "bond_address = '{bond_address}'," \
+                "observe_chains = '{observe_chains}'," \
+                "preflight_status = '{preflight_status}'," \
+                "status = '{status}'," \
+                "status_since = '{status_since}'," \
+                "version = '{version}' WHERE (node_address = '{node_address}');".format(
+            active_block_height=node['active_block_height'],
+            bond_providers=json.dumps(node['bond_providers']),
+            bond=int(node['total_bond']),
+            current_award=int(node['current_award']),
+            slash_points=node['slash_points'],
+            forced_to_leave=int(node['forced_to_leave']),
+            requested_to_leave=int(node['requested_to_leave']),
+            jail=json.dumps(node['jail']), bond_address='',
+            observe_chains=json.dumps(node['observe_chains']),
+            preflight_status=json.dumps(node['preflight_status']),
+            status=node['status'],
+            status_since=node['status_since'], version=node['version'], node_address=node['node_address'])
+        commitQuery(query)
+
+    # Loop over new nodes and grab IP addr
+    for node in dataForNewNodes:
+        if node['ip_address'] != "":
+            response_code = 0
+            while response_code != 200:
+                response = requests.get("http://ip-api.com/json/" + node['ip_address'])
+                response_code = response.status_code
+                if response_code == 429:
+                    print("rate limited wait 60seconds")
+                    time.sleep(60)
+                elif response_code == 200:
+                    ip_data = json.loads(response.text)
+                    node['ip_data'] = ip_data
+
+        else:
+            node['ip_data'] = {}
+            node['ip_data']['city'] = ""
+            node['ip_data']['isp'] = ""
+
+        query = "INSERT INTO noderunner.thornode_monitor (node_address, ip_address, location, isp, " \
+                "active_block_height, bond_providers, bond, current_award, slash_points,forced_to_leave, " \
+                "requested_to_leave, jail, bond_address, observe_chains, preflight_status, status, " \
+                "status_since, version) VALUES ('{node_address}', '{ip_address}','{city}','{isp}'," \
+                "'{active_block_height}','{bond_providers}','{bond}','{current_award}','{slash_points}'," \
+                "'{forced_to_leave}','{requested_to_leave}', '{jail}','{bond_address}'," \
+                "'{observe_chains}','{preflight_status}','{status}','{status_since}'," \
+                "'{version}')".format(node_address=node['node_address'], ip_address=node['ip_address'],
+                                      city=node['ip_data']['city'], isp=node['ip_data']['isp'],
+                                      active_block_height=node['active_block_height'],
+                                      bond_providers=json.dumps(node['bond_providers']), bond=int(node['total_bond']),
+                                      current_award=int(node['current_award']), slash_points=node['slash_points'],
+                                      forced_to_leave=int(node['forced_to_leave']),
+                                      requested_to_leave=int(node['requested_to_leave']),
+                                      jail=json.dumps(node['jail']), bond_address='',
+                                      observe_chains=json.dumps(node['observe_chains']),
+                                      preflight_status=json.dumps(node['preflight_status']), status=node['status'],
+                                      status_since=node['status_since'], version=node['version'])
+        commitQuery(query)