Python脚本如何监控分布式节点同步状态

wen python案例 2026-06-25 218

本文目录导读：

基础监控框架
针对特定系统的监控
告警通知集成
完整使用示例
注意事项

我来为您介绍几种监控分布式节点同步状态的Python方法：

基础监控框架

import time
import json
import requests
from datetime import datetime
from typing import Dict, List
import logging
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class SyncMonitor:
    def __init__(self, nodes: List[str], threshold_seconds: int = 300):
        """
        初始化监控器
        :param nodes: 节点列表 [('node1', 'http://ip:port'), ...]
        :param threshold_seconds: 同步延迟阈值（秒）
        """
        self.nodes = nodes
        self.threshold = threshold_seconds
        self.node_status = {}
    def check_node_health(self, node_name: str, node_url: str) -> Dict:
        """检查单个节点状态"""
        try:
            # 示例：检查节点API
            response = requests.get(
                f"{node_url}/health",
                timeout=5
            )
            if response.status_code == 200:
                data = response.json()
                return {
                    'status': 'healthy',
                    'last_block': data.get('last_block', 0),
                    'timestamp': data.get('timestamp', 0),
                    'peers': data.get('peers', 0)
                }
            else:
                return {'status': 'unhealthy', 'error': f'HTTP {response.status_code}'}
        except requests.exceptions.RequestException as e:
            return {'status': 'unreachable', 'error': str(e)}
    def check_sync_status(self) -> Dict:
        """检查所有节点的同步状态"""
        results = {}
        for node_name, node_url in self.nodes:
            status = self.check_node_health(node_name, node_url)
            results[node_name] = status
            self.node_status[node_name] = status
        return results
    def detect_sync_lag(self) -> List[Dict]:
        """检测同步延迟"""
        alerts = []
        # 找出最新的区块高度
        max_block = 0
        for name, status in self.node_status.items():
            if status['status'] == 'healthy':
                if status['last_block'] > max_block:
                    max_block = status['last_block']
        # 检查每个节点的延迟
        for name, status in self.node_status.items():
            if status['status'] == 'healthy':
                lag = max_block - status['last_block']
                if lag > self.threshold:
                    alerts.append({
                        'node': name,
                        'type': 'sync_lag',
                        'current_block': status['last_block'],
                        'latest_block': max_block,
                        'lag': lag
                    })
        return alerts
    def monitor_loop(self, interval: int = 60):
        """持续监控循环"""
        logger.info(f"开始监控 {len(self.nodes)} 个节点")
        while True:
            try:
                # 检查同步状态
                results = self.check_sync_status()
                # 检测延迟
                alerts = self.detect_sync_lag()
                # 输出状态
                timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                logger.info(f"\n=== 节点状态 {timestamp} ===")
                for node, status in results.items():
                    if status['status'] == 'healthy':
                        logger.info(f"{node}: 正常 (区块: {status['last_block']})")
                    else:
                        logger.warning(f"{node}: 异常 - {status.get('error', '未知')}")
                # 输出告警
                if alerts:
                    logger.warning(f"\n发现 {len(alerts)} 个同步延迟节点:")
                    for alert in alerts:
                        logger.warning(
                            f"{alert['node']}: 延迟 {alert['lag']} 个区块"
                        )
                # 等待下一次检查
                time.sleep(interval)
            except KeyboardInterrupt:
                logger.info("监控停止")
                break
            except Exception as e:
                logger.error(f"监控错误: {e}")
                time.sleep(interval)

针对特定系统的监控

1 区块链节点监控

class BlockchainSyncMonitor:
    """针对区块链节点的同步监控"""
    @staticmethod
    def check_ethereum_sync(node_url: str) -> Dict:
        """检查以太坊节点同步状态"""
        data = {
            'jsonrpc': '2.0',
            'method': 'eth_syncing',
            'params': [],
            'id': 1
        }
        try:
            response = requests.post(
                node_url,
                json=data,
                timeout=10
            )
            result = response.json().get('result', {})
            if result is False:
                # 已同步完成
                return {'status': 'synced', 'syncing': False}
            else:
                # 正在同步
                current_block = int(result.get('currentBlock', '0x0'), 16)
                highest_block = int(result.get('highestBlock', '0x0'), 16)
                return {
                    'status': 'syncing',
                    'current_block': current_block,
                    'highest_block': highest_block,
                    'progress': f"{current_block/highest_block*100:.1f}%" if highest_block > 0 else "0%"
                }
        except Exception as e:
            return {'status': 'error', 'error': str(e)}

2 数据库同步监控

import pymongo
import mysql.connector
class DatabaseSyncMonitor:
    """数据库同步监控"""
    def __init__(self, primary_config: Dict, replica_configs: List[Dict]):
        self.primary = primary_config
        self.replicas = replica_configs
    def check_mysql_replication(self, replica_config: Dict) -> Dict:
        """检查MySQL主从同步状态"""
        try:
            conn = mysql.connector.connect(**replica_config)
            cursor = conn.cursor()
            # 检查从库状态
            cursor.execute("SHOW SLAVE STATUS")
            slave_status = cursor.fetchone()
            if slave_status:
                return {
                    'status': 'running' if slave_status[10] == 'Yes' else 'stopped',
                    'seconds_behind': slave_status[32],  # Seconds_Behind_Master
                    'last_error': slave_status[31]
                }
            else:
                return {'status': 'not_replicated'}
        except Exception as e:
            return {'status': 'error', 'error': str(e)}
    def check_mongodb_replication(self, replica_config: Dict) -> Dict:
        """检查MongoDB复制集状态"""
        try:
            client = pymongo.MongoClient(**replica_config)
            status = client.admin.command('replSetGetStatus')
            members = status.get('members', [])
            results = []
            for member in members:
                results.append({
                    'name': member.get('name'),
                    'state': member.get('stateStr'),
                    'optime_date': member.get('optimeDate'),
                    'lag': member.get('optimeDate', 0) - member.get('lastHeartbeat', 0) if member.get('optimeDate') else 0
                })
            return {'status': 'connected', 'members': results}
        except Exception as e:
            return {'status': 'error', 'error': str(e)}

告警通知集成

import smtplib
from email.mime.text import MIMEText
class AlertNotifier:
    """告警通知类"""
    def __init__(self, config: Dict):
        self.config = config
    def send_email_alert(self, subject: str, message: str):
        """发送邮件告警"""
        msg = MIMEText(message)
        msg['Subject'] = f"[告警] {subject}"
        msg['From'] = self.config['email_from']
        msg['To'] = self.config['email_to']
        with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
            server.login(self.config['smtp_user'], self.config['smtp_password'])
            server.send_message(msg)
    def send_webhook_alert(self, webhook_url: str, message: Dict):
        """发送Webhook告警"""
        try:
            requests.post(
                webhook_url,
                json=message,
                headers={'Content-Type': 'application/json'}
            )
        except Exception as e:
            logger.error(f"发送Webhook告警失败: {e}")

完整使用示例

def main():
    """主函数示例"""
    # 配置节点
    nodes = [
        ('node1', 'http://192.168.1.10:8545'),
        ('node2', 'http://192.168.1.11:8545'),
        ('node3', 'http://192.168.1.12:8545')
    ]
    # 创建监控器
    monitor = SyncMonitor(
        nodes=nodes,
        threshold_seconds=300
    )
    # 创建告警器
    notifier = AlertNotifier({
        'email_from': 'monitor@example.com',
        'email_to': 'admin@example.com',
        'smtp_server': 'smtp.example.com',
        'smtp_port': 587,
        'smtp_user': 'monitor@example.com',
        'smtp_password': 'your_password'
    })
    # 增强的监控循环
    while True:
        try:
            # 获取状态
            status = monitor.check_sync_status()
            # 检查延迟
            alerts = monitor.detect_sync_lag()
            # 如果有严重告警，发送通知
            if alerts:
                for alert in alerts:
                    if alert['lag'] > 1000:  # 严重延迟
                        notifier.send_email_alert(
                            subject=f"节点 {alert['node']} 严重同步延迟",
                            message=f"当前延迟: {alert['lag']} 个区块"
                        )
            # 写入状态到文件或数据库
            with open('sync_status.json', 'w') as f:
                json.dump({
                    'timestamp': datetime.now().isoformat(),
                    'status': status,
                    'alerts': alerts
                }, f, indent=2)
            time.sleep(60)
        except Exception as e:
            logger.error(f"监控循环错误: {e}")
            time.sleep(60)
if __name__ == "__main__":
    main()