#!/usr/bin/env python3 """ UnderPass Intelligence - Production Monitor v2 Chain Integrity + TPS + Fork Detection + Alert Dedup + WAL SQLite """ import time import json import sqlite3 import threading import logging import requests from collections import deque, defaultdict from typing import Dict from enum import Enum logging.basicConfig(level=logging.INFO) logger = logging.getLogger("monitor") # ============================= # Enums # ============================= class AlertLevel(str, Enum): INFO = "info" WARNING = "warning" ERROR = "error" CRITICAL = "critical" # ============================= # Metrics Engine # ============================= class MetricsEngine: ALERT_COOLDOWN = 60 # seconds per message def __init__(self, db_path="monitor.db"): self.db = sqlite3.connect(db_path, check_same_thread=False) self.db.execute("PRAGMA journal_mode=WAL;") self.lock = threading.RLock() self._init_db() self.last_alert_time = {} def _init_db(self): cur = self.db.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS metrics ( name TEXT, value REAL, timestamp INTEGER ) """) cur.execute(""" CREATE TABLE IF NOT EXISTS alerts ( level TEXT, message TEXT, details TEXT, timestamp INTEGER ) """) self.db.commit() def record_metric(self, name: str, value: float): with self.lock: self.db.execute( "INSERT INTO metrics VALUES (?,?,?)", (name, value, int(time.time())) ) def add_alert(self, level: AlertLevel, message: str, details=None): now = time.time() # Dedup cooldown last = self.last_alert_time.get(message) if last and now - last < self.ALERT_COOLDOWN: return self.last_alert_time[message] = now with self.lock: self.db.execute( "INSERT INTO alerts VALUES (?,?,?,?)", (level.value, message, json.dumps(details or {}), int(now)) ) self.db.commit() getattr(logger, level.value)(f"[{level.value}] {message} {details}") # ============================= # Chain Monitor # ============================= class ChainMonitor: def __init__(self, api_url: str, metrics: MetricsEngine): self.api = api_url self.metrics = metrics self.running = False self.thread = None self.last_block = None self.last_hash_by_height = {} self.tx_window = deque(maxlen=200) def start(self): self.running = True self.thread = threading.Thread(target=self._loop, daemon=True) self.thread.start() def stop(self): self.running = False if self.thread: self.thread.join(timeout=5) def _loop(self): while self.running: try: self._check_chain() time.sleep(2) except Exception as e: logger.error(f"Monitor error: {e}") time.sleep(5) def _check_chain(self): r = requests.get(f"{self.api}/api/v1/blocks/latest", timeout=5) if r.status_code != 200: self.metrics.add_alert(AlertLevel.ERROR, "API unreachable") return data = r.json() if not data.get("success"): return block = data["data"] height = block["index"] block_hash = block["hash"] block_ts = block.get("timestamp", int(time.time())) tx_count = block.get("tx_count", 0) # Fork detection if height in self.last_hash_by_height: if self.last_hash_by_height[height] != block_hash: self.metrics.add_alert( AlertLevel.CRITICAL, "FORK DETECTED", {"height": height} ) self.last_hash_by_height[height] = block_hash # Block interval calculation using timestamps if self.last_block: prev_ts = self.last_block["timestamp"] interval = block_ts - prev_ts self.metrics.record_metric("block_interval", interval) if interval > 10: self.metrics.add_alert( AlertLevel.WARNING, "Slow block interval", {"interval": interval} ) # TPS calculation based on time span now = time.time() self.tx_window.append((now, tx_count)) self._calculate_tps() self.metrics.record_metric("block_height", height) self.last_block = block def _calculate_tps(self): if len(self.tx_window) < 2: return oldest_time = self.tx_window[0][0] newest_time = self.tx_window[-1][0] span = newest_time - oldest_time if span <= 0: return total_txs = sum(x[1] for x in self.tx_window) tps = total_txs / span self.metrics.record_metric("tps", tps) if tps < 0.5 and span > 30: self.metrics.add_alert( AlertLevel.WARNING, "Low TPS detected", {"tps": tps} ) # ============================= # Bridge Invariant Monitor # ============================= class BridgeMonitor: def __init__(self, bridge_api: str, metrics: MetricsEngine): self.api = bridge_api self.metrics = metrics self.running = False self.thread = None def start(self): self.running = True self.thread = threading.Thread(target=self._loop, daemon=True) self.thread.start() def stop(self): self.running = False def _loop(self): while self.running: try: self._check_bridge() time.sleep(10) except Exception as e: logger.error(f"Bridge monitor error: {e}") time.sleep(10) def _check_bridge(self): try: r = requests.get(f"{self.api}/state", timeout=5) if r.status_code != 200: return state = r.json() if not state.get("invariant_ok", True): self.metrics.add_alert( AlertLevel.CRITICAL, "Bridge invariant violated", state ) except Exception: pass # ============================= # Main Service # ============================= class MonitorService: def __init__(self, chain_api, bridge_api=None): self.metrics = MetricsEngine() self.chain_monitor = ChainMonitor(chain_api, self.metrics) self.bridge_monitor = None if bridge_api: self.bridge_monitor = BridgeMonitor(bridge_api, self.metrics) def start(self): logger.info("Monitor starting...") self.chain_monitor.start() if self.bridge_monitor: self.bridge_monitor.start() def stop(self): self.chain_monitor.stop() if self.bridge_monitor: self.bridge_monitor.stop() # ============================= # Entry # ============================= if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--chain-api", default="http://localhost:8080") parser.add_argument("--bridge-api", default=None) args = parser.parse_args() service = MonitorService(args.chain_api, args.bridge_api) try: service.start() while True: time.sleep(1) except KeyboardInterrupt: service.stop()