Add more multi-threading to uptime query. Add batch processing of UPDATE. Processing update query to remote database or local depending on user.

Author: freQniK

meile_node_uptime.py

@@ -11,8 +11,8 @@
 import concurrent.futures
 from dbutils.pooled_db import PooledDB
 
-VERSION = 20250227.1502
-WORKERS = 30
+VERSION = 20250304.1738
+WORKERS = 50 # edit. This should be plenty.
 APIURL = 'https://api.sentinel.mathnodes.com'
 
 class UpdateNodeUptime():
@@ -21,7 +21,17 @@ def connDB(self):
 
         self.db_pool = PooledDB(
             creator=pymysql, 
-            maxconnections=30, 
+            maxconnections=50, 
+            host=scrtsxx.RHOST, 
+            port=scrtsxx.PORT,
+            user=scrtsxx.USERNAME,
+            passwd=scrtsxx.PASSWORD,
+            database=scrtsxx.DB,      
+        )
+        
+        self.local_db_pool = PooledDB(
+            creator=pymysql, 
+            maxconnections=50, 
             host=scrtsxx.HOST, 
             port=scrtsxx.PORT,
             user=scrtsxx.USERNAME,
@@ -52,28 +62,20 @@ def get_node_uptime_table(self,db):
     def get_remote_url_of_node(self, db, NodeData):
         NodeRemoteURL = {'address' : [], 'url' : []}
 
-        # Retrieve nodes with empty remote_url from the table
-        #query = "SELECT * FROM node_uptime WHERE remote_url = '';"
-        c = db.cursor()
-        #c.execute(query)
-        #nodes_without_remote_url = c.fetchall()
-        #print(nodes_without_remote_url)
-        for n in NodeData:
+        def get_remote_url(n):
             address = n['node_address']
-            
-            # Check if the node already has a remote_url in the table
-            #if any(node['node_address'] == address for node in nodes_without_remote_url):
-            #    continue
-            
-            # Retrieve remote_url from the table for nodes that have it stored
+            #print(address)
             query = f"SELECT remote_url FROM node_uptime WHERE node_address = '{address}';"
+            connection = self.local_db_pool.connection()
+            #c = db.cursor()
+            c = connection.cursor()
             c.execute(query)
             result = c.fetchone()
-            #print(result['remote_url'])
-            if not result['remote_url']:
-                
+            #print(result)
+            connection.close()
+            if not result[0]:
                 endpoint = APIURL + '/sentinel/nodes/' + address
-                remote_url = result['remote_url']
+                remote_url = ''
                 print(f"Getting remote_url of: {address}")
                 sys.stdout.flush()
 
@@ -82,59 +84,21 @@ def get_remote_url_of_node(self, db, NodeData):
                     remote_url = r.json()['node']['remote_url']
                 except Exception as e:
                     print(str(e))
-                    continue
-                #print(f"{remote_url}")
+                    return n['node_address'], None
             else:
-                remote_url = result['remote_url']
-                
-            NodeRemoteURL['address'].append(n['node_address'])
-            NodeRemoteURL['url'].append(remote_url)
-        
-        #print(NodeRemoteURL)
-        
-        return NodeRemoteURL
-
-    '''
-    def check_uptime(self, NodeRemoteURLs):
-        k=0
+                remote_url = result[0]
+            
+            return n['node_address'], remote_url
 
-        NodeUptimeBoolean = {'address' : [], 'up' : []}
+        with concurrent.futures.ThreadPoolExecutor(max_workers=WORKERS) as executor:
+            futures = [executor.submit(get_remote_url, n) for n in NodeData]
+            results = [future.result() for future in futures]
 
+        for address, remote_url in results:
+            NodeRemoteURL['address'].append(address)
+            NodeRemoteURL['url'].append(remote_url)
 
-        for n in NodeRemoteURLs['address']:
-            url = NodeRemoteURLs['url'][k]
-            parsed_url = urlparse(url)
-            netloc = parsed_url.netloc
-            #host, port = urlparse(url).netloc.split(":")
-            #print(f"host: {host}, port: {port}")
-            #print("Checking if up: ", end='')
-            
-            if '[' in netloc and ']' in netloc:
-                host = netloc.split(']', 1)[0][1:]
-                
-                if ':' in netloc.split(']', 1)[1]:
-                    port = int(netloc.split(']', 1)[1].split(':', 1)[1])
-                else:
-                    port = None
-            else:
-                parts = netloc.split(':')
-                if len(parts) > 2:
-                    raise ValueError("Invalid URL format")
-                host = parts[0]
-                if len(parts) == 2:
-                    port = int(parts[1])
-                else:
-                    port = None
-            
-            
-            up = self.check_socket(host, int(port))
-            #print(up)
-            NodeUptimeBoolean['address'].append(n)
-            NodeUptimeBoolean['up'].append(up)
-            k += 1
-            
-        return NodeUptimeBoolean
-    '''
+        return NodeRemoteURL
 
     def check_uptime(self, NodeRemoteURLs):
         k = 0
@@ -162,10 +126,10 @@ def check_uptime_for_node(n, url):
 
             up = self.check_socket(host, int(port) if port else None)
             result = (n, up)
-            print(result)
+            #print(result)
             return result
 
-        with concurrent.futures.ThreadPoolExecutor(max_workers=30) as executor:
+        with concurrent.futures.ThreadPoolExecutor(max_workers=WORKERS) as executor:
             futures = []
 
             for n in NodeRemoteURLs['address']:
@@ -194,48 +158,43 @@ def check_socket(self, host, port):
                 return False
 
 
-    def update_node_uptime(self, node, NodeUptimeData, NodeRemoteURLs, NodeUptimeBoolean):
-        try: 
-            index  = NodeUptimeBoolean['address'].index(node['node_address'])
-            rindex = NodeRemoteURLs['address'].index(node['node_address'])
-            remote_url = NodeRemoteURLs['url'][rindex]    
-        except Exception as e:
-            print(node)
-            print(str(e))
-            return
-        
-        tries = node['tries'] + 1
-        if NodeUptimeBoolean['up'][index]:
-            success = node['success'] + 1                
-        else:
-            success = node['success']
-        success_rate = round(float(success/tries),3)
-        
-        query = 'UPDATE node_uptime SET remote_url = "%s", tries = %d, success = %d, success_rate = "%.3f" WHERE node_address = "%s";' % (remote_url,
-                                                                                                                                          tries,
-                                                                                                                                          success,
-                                                                                                                                          success_rate,
-                                                                                                                                          node['node_address'])
+    def UpdateNodeUptimeTable(self, db, NodeUptimeData, NodeRemoteURLs, NodeUptimeBoolean):
 
-        #print(query)
-        #c = db.cursor()
-        #c.execute(query)
-        #db.commit()
-        with self.lock:
-            connection = self.db_pool.connection()
-            cursor     = connection.cursor()
-            cursor.execute(query)
-            connection.commit()
-            connection.close()
-                
-    def UpdateNodeUptimeTable(self, NodeUptimeData, NodeRemoteURLs, NodeUptimeBoolean):
-        with concurrent.futures.ThreadPoolExecutor(max_workers=WORKERS) as executor:
-            futures = []
-            for node in NodeUptimeData:
-                futures.append(executor.submit(self.update_node_uptime, node, NodeUptimeData, NodeRemoteURLs, NodeUptimeBoolean))
-            for future in concurrent.futures.as_completed(futures):
-                future.result()    
+        query = 'UPDATE node_uptime SET remote_url = "%s", tries = %s, success = %s, success_rate = "%s" WHERE node_address = "%s";'
+        update_data = []
+        for node in NodeUptimeData:
+            try: 
+                index  = NodeUptimeBoolean['address'].index(node['node_address'])
+                rindex = NodeRemoteURLs['address'].index(node['node_address'])
+                remote_url = NodeRemoteURLs['url'][rindex]    
+            except Exception as e:
+                print(node)
+                print(str(e))
+                return
 
+            tries = node['tries'] + 1
+            if NodeUptimeBoolean['up'][index]:
+                success = node['success'] + 1                
+            else:
+                success = node['success']
+            success_rate = round(float(success/tries),3)
+            
+            update_data.append((remote_url,
+                               tries,
+                               success,
+                               success_rate,
+                               node['node_address']))
+
+            if len(update_data) == 200:
+                # Use if not running from a remote server and comment the line after the next one. 
+                #connection = self.local_db_pool.connection() 
+                connection = self.db_pool.connection()
+                cursor     = connection.cursor()
+                cursor.executemany(query, update_data)
+                connection.commit()
+                connection.close()
+                update_data.clear()
+                        
 if __name__ == "__main__":
     NodeUptimeClass = UpdateNodeUptime()
     db = NodeUptimeClass.connDB()
@@ -262,7 +221,7 @@ def UpdateNodeUptimeTable(self, NodeUptimeData, NodeRemoteURLs, NodeUptimeBoolea
     print("It took %ss to check if all nodes are connectable" % (time3))
 
     start = timer()
-    NodeUptimeClass.UpdateNodeUptimeTable(NodeUptimeData, NodeRemoteURLs, NodeUptimeBoolean)
+    NodeUptimeClass.UpdateNodeUptimeTable(db, NodeUptimeData, NodeRemoteURLs, NodeUptimeBoolean)
     end = timer()
 
     time4 = round((end - start),4)

scrtsxx.py

@@ -2,4 +2,5 @@
 USERNAME = "username"
 PASSWORD = "password"
 HOST     = "localhost"
+RHOST    = "foo.bar" # edit if using remote query
 PORT     = 3306