From 6495b95c8d2a13b99010810998f58d0ac1565e51 Mon Sep 17 00:00:00 2001 From: vvzvlad Date: Mon, 31 Mar 2025 01:12:21 +0300 Subject: [PATCH] Add state management --- checker.py | 157 ++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 124 insertions(+), 33 deletions(-) diff --git a/checker.py b/checker.py index 6580f7a..f046fc4 100644 --- a/checker.py +++ b/checker.py @@ -41,6 +41,8 @@ import colorama import logging import socket +STATE_FILE_PATH = '/root/node/checker_state.json' + def self_update(logger): logger.info("Checking for updates..") script_path = os.path.abspath(__file__) @@ -130,50 +132,139 @@ class GRIST: return record.Value raise ValueError(f"Setting {key} not found") +def read_state(logger): + """Reads the last known state (status and sync count) from the state file.""" + default_state = {"last_status": "Idle", "sync_count": 0} + if not os.path.exists(STATE_FILE_PATH): + logger.info(f"State file {STATE_FILE_PATH} not found, starting with default state.") + return default_state + try: + with open(STATE_FILE_PATH, 'r', encoding='utf-8') as f: + state = json.load(f) + # Validate state structure + if "last_status" not in state or "sync_count" not in state: + logger.warning(f"Invalid state file format in {STATE_FILE_PATH}, using default state.") + return default_state + # Ensure sync_count is an integer + state["sync_count"] = int(state["sync_count"]) + return state + except (json.JSONDecodeError, ValueError, IOError) as e: + logger.error(f"Error reading or parsing state file {STATE_FILE_PATH}: {e}. Using default state.") + return default_state + +def write_state(state, logger): + """Writes the current state to the state file.""" + try: + with open(STATE_FILE_PATH, 'w', encoding='utf-8') as f: + json.dump(state, f, indent=4) + except IOError as e: + logger.error(f"Error writing state file {STATE_FILE_PATH}: {e}") + def clean_ansi(text): ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') return ansi_escape.sub('', text) def format_number(number_str): - number = int(number_str) - if number >= 1000: - return f"{number//1000}k" - return str(number) + try: + number = int(number_str) + if number >= 1000: + return f"{number//1000}k" + return str(number) + except (ValueError, TypeError): + return "NaN" # Or some other indicator of invalid input def check_logs(logger): + state = read_state(logger) + original_sync_count = state["sync_count"] + previous_status = state["last_status"] + try: logs = subprocess.run(['docker', 'logs', '--since', '10m', 'infernet-node'], capture_output=True, text=True, check=True) log_content = clean_ansi(logs.stdout) - - last_subscription_id = None - head_sub_id = None - - for line in log_content.splitlines(): - if "Ignored subscription creation" in line and "id=" in line: - id_match = re.search(r'id=(\d+)', line) - if id_match: - last_subscription_id = id_match.group(1) - - if "head sub id is:" in line: - id_match = re.search(r'head sub id is:\s*(\d+)', line) - if id_match: - head_sub_id = id_match.group(1) - - if last_subscription_id: - logger.info(f"Subscription: {last_subscription_id}") - return {"status": f"Sync: {format_number(last_subscription_id)}"} - - if head_sub_id: - logger.info(f"Head sub id: {head_sub_id}") - return {"status": f"OK: {head_sub_id}"} - - - logger.info("Not found subscription") - return {"status": "Idle"} - - except subprocess.CalledProcessError as e: - raise RuntimeError(f"Error running docker logs: {e}") + last_checking_info = None + last_ignored_id = None + last_head_sub_id = None + + # Regex patterns + checking_pattern = re.compile(r'Checking subscriptions.*last_sub_id=(\d+).*head_sub_id=(\d+).*num_subs_to_sync=(\d+)') + ignored_pattern = re.compile(r'Ignored subscription creation.*id=(\d+)') + head_sub_pattern = re.compile(r'head sub id is:\s*(\d+)') + + # Use deque to efficiently get the last few relevant lines if needed, + # but processing all lines and keeping the last match is simpler here. + for line in log_content.splitlines(): + match = checking_pattern.search(line) + if match: + last_checking_info = { + "last_sub_id": match.group(1), + "head_sub_id": match.group(2), + "num_subs_to_sync": int(match.group(3)) + } + continue # Prioritize checking_info + + match = ignored_pattern.search(line) + if match: + last_ignored_id = match.group(1) + continue + + match = head_sub_pattern.search(line) + if match: + last_head_sub_id = match.group(1) + # No continue here, allows checking_info from same timeframe to override + + current_status_type = "Idle" + status_message = "Idle" + + if last_checking_info: + formatted_id = format_number(last_checking_info["last_sub_id"]) + if last_checking_info["num_subs_to_sync"] > 0: + current_status_type = "Sync" + status_message = f"Sync: {formatted_id} ({state['sync_count']})" + logger.info(f"Node is syncing. Last sub ID: {last_checking_info['last_sub_id']}, Num subs to sync: {last_checking_info['num_subs_to_sync']}") + else: + current_status_type = "OK" + # Increment count only on transition from Sync to OK based on the *last* status type + if previous_status == "Sync": + state["sync_count"] += 1 + logger.info(f"Sync completed. Sync count incremented to {state['sync_count']}.") + status_message = f"OK: {formatted_id} ({state['sync_count']})" + logger.info(f"Node is OK. Last sub ID: {last_checking_info['last_sub_id']}") + + elif last_ignored_id: + # Fallback to "Ignored" logs if "Checking" is missing + formatted_id = format_number(last_ignored_id) + current_status_type = "Sync" # Assume sync if we only see ignored creations recently + status_message = f"Sync: {formatted_id} ({state['sync_count']})" + logger.info(f"Node possibly syncing (based on ignored logs). Last ignored ID: {last_ignored_id}") + + elif last_head_sub_id: + # Fallback to "head sub id" if others are missing + formatted_id = format_number(last_head_sub_id) + current_status_type = "OK" # Assume OK if this is the latest relevant info + status_message = f"OK: {formatted_id} ({state['sync_count']})" + logger.info(f"Node status based on head sub id. Head sub ID: {last_head_sub_id}") + + else: + logger.info("No relevant subscription log entries found in the last 10 minutes. Status: Idle.") + status_message = "Idle" + current_status_type = "Idle" + + # Update state and write back if changed + state["last_status"] = current_status_type + if state["sync_count"] != original_sync_count or state["last_status"] != previous_status: + write_state(state, logger) + + return {"status": status_message} + + except subprocess.CalledProcessError as e: + logger.error(f"Error running docker logs command: {e.stderr or e.stdout or e}") + # Don't update state file on command error, keep previous state + return {"status": f"Error: Docker logs failed ({e.returncode})"} + except Exception as e: + logger.error(f"Unexpected error processing logs: {e}", exc_info=True) + # Don't update state file on unexpected error + return {"status": f"Error: Log processing failed"} if __name__ == "__main__": colorama.init(autoreset=True)