Add state management
This commit is contained in:
vvzvlad
157
checker.py
157
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)
|
||||
|
||||
Reference in New Issue
Block a user