Merge pull request #440 from yungwine/stats

add new node metric stats

Author: igroman787

modules/alert_bot.py

@@ -116,6 +116,12 @@ def init_alerts():
             "Found proposals with hashes `{hashes}` that have significant amount of votes, but current validator didn't vote for them. Please check @tonstatus for more details.",
             VALIDATION_PERIOD
         ),
+        "initial_sync_completed": Alert(
+            "info",
+            "Initial sync has been completed (info alert with no sound)",
+            "Node initial sync has been completed",
+            0
+        )
     }
 
 
@@ -134,6 +140,7 @@ def __init__(self, ton, local, *args, **kwargs):
         self.token = None
         self.chat_id = None
         self.last_db_check = 0
+        self.initial_sync = None
 
     def send_message(self, text: str, silent: bool = False, disable_web_page_preview: bool = False):
         if self.token is None:
@@ -207,6 +214,7 @@ def init(self):
         self.wallet = self.ton.GetValidatorWallet().addrB64
         self.ip = self.ton.get_node_ip()
         self.set_global_vars()
+        self.initial_sync = self.ton.in_initial_sync()
         init_alerts()
         self.inited = True
 
@@ -334,12 +342,12 @@ def check_efficiency(self):
 
     def check_validator_working(self):
         validator_status = self.ton.GetValidatorStatus()
-        if not validator_status.is_working:
+        if not self.initial_sync and not validator_status.is_working:
             self.send_alert("service_down")
 
     def check_sync(self):
         validator_status = self.ton.GetValidatorStatus()
-        if validator_status.is_working and validator_status.out_of_sync >= 20:
+        if not self.initial_sync and validator_status.is_working and validator_status.out_of_sync >= 20:
             self.send_alert("out_of_sync", sync=validator_status.out_of_sync)
 
     def check_zero_blocks_created(self):
@@ -435,6 +443,13 @@ def check_voting(self):
         if need_to_vote:
             self.send_alert("voting", hashes=' '.join(need_to_vote))
 
+    def check_initial_sync(self):
+        if not self.initial_sync:
+            return
+        if not self.ton.in_initial_sync():
+            self.initial_sync = False
+            self.send_alert("initial_sync_completed")
+
     def check_status(self):
         if not self.ton.using_alert_bot():
             return
@@ -453,6 +468,7 @@ def check_status(self):
         self.local.try_function(self.check_stake_sent)
         self.local.try_function(self.check_stake_returned)
         self.local.try_function(self.check_voting)
+        self.local.try_function(self.check_initial_sync)
 
     def add_console_commands(self, console):
         console.AddItem("enable_alert", self.enable_alert, self.local.translate("enable_alert_cmd"))

modules/prometheus.py

@@ -26,6 +26,18 @@ def to_format(self, value):
     'stake': Metric('validator_stake', 'Validator stake', 'gauge'),
     'celldb_gc_block': Metric('validator_celldb_gc_block', 'Celldb GC block latency', 'gauge'),
     'celldb_gc_state': Metric('validator_celldb_gc_state', 'Celldb GC queue size', 'gauge'),
+    'collated_master_ok': Metric('validator_blocks_collated_master_ok', 'Number of masterchain blocks successfully collated', 'gauge'),
+    'collated_master_err': Metric('validator_blocks_collated_master_err', 'Number of masterchain blocks failed to collate', 'gauge'),
+    'collated_shard_ok': Metric('validator_blocks_collated_shard_ok', 'Number of shardchain blocks successfully collated', 'gauge'),
+    'collated_shard_err': Metric('validator_blocks_collated_shard_err', 'Number of shardchain blocks failed to collate', 'gauge'),
+    'validated_master_ok': Metric('validator_blocks_validated_master_ok', 'Number of masterchain blocks successfully validated', 'gauge'),
+    'validated_master_err': Metric('validator_blocks_validated_master_err', 'Number of masterchain blocks failed to validate', 'gauge'),
+    'validated_shard_ok': Metric('validator_blocks_validated_shard_ok', 'Number of shardchain blocks successfully validated', 'gauge'),
+    'validated_shard_err': Metric('validator_blocks_validated_shard_err', 'Number of shardchain blocks failed to validate', 'gauge'),
+    'validator_groups_master': Metric('validator_active_groups_master', 'Number of masterchain validation groups validator participates in', 'gauge'),
+    'validator_groups_shard': Metric('validator_active_groups_shard', 'Number of shardchain validation groups validator participates in', 'gauge'),
+    'ls_queries_ok': Metric('validator_ls_queries_ok', 'Number of Liteserver successful queries', 'gauge'),
+    'ls_queries_err': Metric('validator_ls_queries_err', 'Number of Liteserver failed queries', 'gauge'),
 }
 
 
@@ -44,7 +56,7 @@ def get_validator_status_metrics(self, result: list):
             result.append(METRICS['master_out_of_sync'].to_format(status.masterchain_out_of_sync))
         if status.shardchain_out_of_sync is not None:
             result.append(METRICS['shard_out_of_sync'].to_format(status.shardchain_out_of_sync))
-        if status.masterchain_out_of_ser is not None and status.stateserializermasterchainseqno != 0:
+        if status.stateserializerenabled and status.masterchain_out_of_ser is not None and status.stateserializermasterchainseqno != 0:
             result.append(METRICS['out_of_ser'].to_format(status.masterchain_out_of_ser))
         if status.masterchainblock is not None and status.gcmasterchainblock is not None:
             result.append(METRICS['celldb_gc_block'].to_format(status.masterchainblock - status.gcmasterchainblock))
@@ -53,6 +65,9 @@ def get_validator_status_metrics(self, result: list):
                 result.append(METRICS['celldb_gc_state'].to_format(status.gcmasterchainblock - status.last_deleted_mc_state))
             else:
                 result.append(METRICS['celldb_gc_state'].to_format(-1))
+        if status.validator_groups_master is not None:
+            result.append(METRICS['validator_groups_master'].to_format(status.validator_groups_master))
+            result.append(METRICS['validator_groups_shard'].to_format(status.validator_groups_shard))
         result.append(METRICS['vc_up'].to_format(int(is_working)))
 
     def get_validator_validation_metrics(self, result: list):
@@ -67,6 +82,25 @@ def get_validator_validation_metrics(self, result: list):
             if stake:
                 result.append(METRICS['stake'].to_format(round(stake, 2)))
 
+    def get_node_stats_metrics(self, result: list):
+        stats = self.ton.get_node_statistics()
+        if stats and 'ls_queries' in stats:
+            if stats['ls_queries']['time'] < 50:
+                self.local.add_log(f'Liteserver queries time is too low: {stats}')
+                return
+            result.append(METRICS['ls_queries_ok'].to_format(stats['ls_queries']['ok']))
+            result.append(METRICS['ls_queries_err'].to_format(stats['ls_queries']['error']))
+        if stats and 'collated' in stats:
+            result.append(METRICS['collated_master_ok'].to_format(stats['collated']['master']['ok']))
+            result.append(METRICS['collated_master_err'].to_format(stats['collated']['master']['error']))
+            result.append(METRICS['collated_shard_ok'].to_format(stats['collated']['shard']['ok']))
+            result.append(METRICS['collated_shard_err'].to_format(stats['collated']['shard']['error']))
+        if stats and 'validated' in stats:
+            result.append(METRICS['validated_master_ok'].to_format(stats['validated']['master']['ok']))
+            result.append(METRICS['validated_master_err'].to_format(stats['validated']['master']['error']))
+            result.append(METRICS['validated_shard_ok'].to_format(stats['validated']['shard']['ok']))
+            result.append(METRICS['validated_shard_err'].to_format(stats['validated']['shard']['error']))
+
     def push_metrics(self):
         if not self.ton.using_prometheus():
             return
@@ -77,6 +111,7 @@ def push_metrics(self):
         metrics = []
         self.local.try_function(self.get_validator_status_metrics, args=[metrics])
         self.local.try_function(self.get_validator_validation_metrics, args=[metrics])
+        self.local.try_function(self.get_node_stats_metrics, args=[metrics])
         requests.post(url, data='\n'.join(metrics).encode())
 
     def add_console_commands(self, console):

mytoncore/functions.py

@@ -286,6 +286,64 @@ def CalculateNetworkStatistics(zerodata, data):
 # end define
 
 
+def save_node_statistics(local, ton):
+    status = ton.GetValidatorStatus(no_cache=True)
+    if status.unixtime is None:
+        return
+    data = {'timestamp': status.unixtime}
+
+    def get_ok_error(value: str):
+        ok, error = value.split()
+        return int(ok.split(':')[1]), int(error.split(':')[1])
+
+    if 'total.collated_blocks.master' in status:
+        master_ok, master_error = get_ok_error(status['total.collated_blocks.master'])
+        shard_ok, shard_error = get_ok_error(status['total.collated_blocks.shard'])
+        data['collated_blocks'] = {
+            'master': {'ok': master_ok, 'error': master_error},
+            'shard': {'ok': shard_ok, 'error': shard_error},
+        }
+    if 'total.validated_blocks.master' in status:
+        master_ok, master_error = get_ok_error(status['total.validated_blocks.master'])
+        shard_ok, shard_error = get_ok_error(status['total.validated_blocks.shard'])
+        data['validated_blocks'] = {
+            'master': {'ok': master_ok, 'error': master_error},
+            'shard': {'ok': shard_ok, 'error': shard_error},
+        }
+    if 'total.ext_msg_check' in status:
+        ok, error = get_ok_error(status['total.ext_msg_check'])
+        data['ext_msg_check'] = {'ok': ok, 'error': error}
+    if 'total.ls_queries_ok' in status and 'total.ls_queries_error' in status:
+        data['ls_queries'] = {}
+        for k in status['total.ls_queries_ok'].split():
+            if k.startswith('TOTAL'):
+                data['ls_queries']['ok'] = int(k.split(':')[1])
+        for k in status['total.ls_queries_error'].split():
+            if k.startswith('TOTAL'):
+                data['ls_queries']['error'] = int(k.split(':')[1])
+    statistics = local.db.get("statistics", dict())
+
+    if time.time() - int(status.start_time) <= 60:  # was node restart <60 sec ago, resetting node statistics
+        statistics['node'] = []
+
+    # statistics['node'] = [stats_from_election_id, stats_from_prev_min, stats_now]
+
+    election_id = ton.GetConfig34()['startWorkTime']
+    if 'node' not in statistics or len(statistics['node']) == 0:
+        statistics['node'] = [None, data]
+    elif len(statistics['node']) < 3:
+        statistics['node'].append(data)
+    if len(statistics['node']) == 3:
+        if statistics['node'][0] is None:
+            if 0 < data['timestamp'] - election_id < 90:
+                statistics['node'][0] = data
+        elif statistics['node'][0]['timestamp'] < election_id:
+            statistics['node'][0] = data
+        statistics['node'] = statistics.get('node', []) + [data]
+        statistics['node'].pop(1)
+    local.db["statistics"] = statistics
+
+
 def ReadTransData(local, scanner):
     transData = local.buffer.transData
     SetToTimeData(transData, scanner.transNum)
@@ -544,6 +602,17 @@ def ScanLiteServers(local, ton):
 # end define
 
 
+def check_initial_sync(local, ton):
+    if not ton.in_initial_sync():
+        return
+    validator_status = ton.GetValidatorStatus()
+    if validator_status.initial_sync:
+        return
+    if validator_status.out_of_sync < 20:
+        ton.set_initial_sync_off()
+        return
+
+
 def General(local):
     local.add_log("start General function", "debug")
     ton = MyTonCore(local)
@@ -570,6 +639,8 @@ def General(local):
 
     local.start_cycle(ScanLiteServers, sec=60, args=(local, ton,))
 
+    local.start_cycle(save_node_statistics, sec=60, args=(local, ton, ))
+
     from modules.custom_overlays import CustomOverlayModule
     local.start_cycle(CustomOverlayModule(ton, local).custom_overlays, sec=60, args=())
 
@@ -579,6 +650,9 @@ def General(local):
     from modules.prometheus import PrometheusModule
     local.start_cycle(PrometheusModule(ton, local).push_metrics, sec=30, args=())
 
+    if ton.in_initial_sync():
+        local.start_cycle(check_initial_sync, sec=120, args=(local, ton))
+
     thr_sleep()
 # end define
 

mytoncore/mytoncore.py

@@ -776,16 +776,24 @@ def GetShardsNumber(self, block=None):
 		return shardsNum
 	#end define
 
-	def GetValidatorStatus(self):
+	def parse_stats_from_vc(self, output: str, result: dict):
+		for line in output.split('\n'):
+			if len(line.split('\t\t\t')) == 2:
+				name, value = line.split('\t\t\t')  # https://github.com/ton-blockchain/ton/blob/master/validator-engine-console/validator-engine-console-query.cpp#L648
+				if name not in result:
+					result[name] = value
+
+	def GetValidatorStatus(self, no_cache=False):
 		# Get buffer
 		bname = "validator_status"
 		buff = self.GetFunctionBuffer(bname)
-		if buff:
+		if buff and not no_cache:
 			return buff
 		#end if
 
 		self.local.add_log("start GetValidatorStatus function", "debug")
 		status = Dict()
+		result = None
 		try:
 			# Parse
 			status.is_working = True
@@ -809,10 +817,19 @@ def GetValidatorStatus(self):
 			status.out_of_sync = status.masterchain_out_of_sync if status.masterchain_out_of_sync > status.shardchain_out_of_sync else status.shardchain_out_of_sync
 			status.out_of_ser = status.masterchain_out_of_ser
 			status.last_deleted_mc_state = int(parse(result, "last_deleted_mc_state", '\n'))
+			status.stateserializerenabled = parse(result, "stateserializerenabled", '\n') == "true"
+			self.local.try_function(self.parse_stats_from_vc, args=[result, status])
+			if 'active_validator_groups' in status:
+				groups = status.active_validator_groups.split()  # master:1 shard:2
+				status.validator_groups_master = int(groups[0].split(':')[1])
+				status.validator_groups_shard = int(groups[1].split(':')[1])
 		except Exception as ex:
 			self.local.add_log(f"GetValidatorStatus warning: {ex}", "warning")
 			status.is_working = False
+			if result is not None:
+				self.local.try_function(self.parse_stats_from_vc, args=[result, status])
 		#end try
+		status.initial_sync = status.get("process.initial_sync")
 
 		# old vars
 		status.outOfSync = status.out_of_sync
@@ -3037,6 +3054,48 @@ def GetStatistics(self, name, statistics=None):
 		return data
 	#end define
 
+	def get_node_statistics(self):
+		"""
+		:return: stats for collated/validated blocks since round beggining and stats for ls queries for the last minute
+		"""
+		stats = self.local.db.get('statistics', {}).get('node')
+		result = {}
+		if stats is not None and len(stats) == 3 and stats[0] is not None:
+			for k in ['master', 'shard']:
+				result = {
+					'collated': {
+						'ok': 0,
+						'error': 0,
+					},
+					'validated': {
+						'ok': 0,
+						'error': 0,
+					}
+				}
+				collated_ok = stats[2]['collated_blocks'][k]['ok'] - stats[0]['collated_blocks'][k]['ok']
+				collated_error = stats[2]['collated_blocks'][k]['error'] - stats[0]['collated_blocks'][k]['error']
+				validated_ok = stats[2]['validated_blocks'][k]['ok'] - stats[0]['validated_blocks'][k]['ok']
+				validated_error = stats[2]['validated_blocks'][k]['error'] - stats[0]['validated_blocks'][k]['error']
+				result['collated'][k] = {
+					'ok': collated_ok,
+					'error': collated_error,
+				}
+				result['validated'][k] = {
+					'ok': validated_ok,
+					'error': validated_error,
+				}
+				result['collated']['ok'] += collated_ok
+				result['collated']['error'] += collated_error
+				result['validated']['ok'] += validated_ok
+				result['validated']['error'] += validated_error
+		if stats is not None and len(stats) >= 2 and stats[0] is not None:
+			result['ls_queries'] = {
+				'ok': stats[-1]['ls_queries']['ok'] - stats[-2]['ls_queries']['ok'],
+				'error': stats[-1]['ls_queries']['error'] - stats[-2]['ls_queries']['error'],
+				'time': stats[-1].get('timestamp', 0) - stats[-2].get('timestamp', 0),
+			}
+		return result
+
 	def GetSettings(self, name):
 		# self.local.load_db()
 		result = self.local.db.get(name)
@@ -3140,6 +3199,13 @@ def using_alert_bot(self):
 	def using_prometheus(self):
 		return self.get_mode_value('prometheus')
 
+	def in_initial_sync(self):
+		return self.local.db.get('initialSync', False)
+
+	def set_initial_sync_off(self):
+		self.local.db.pop('initialSync', None)
+		self.local.save()
+
 	def Tlb2Json(self, text):
 		# Заменить скобки
 		start = 0