本文目录导读：
我来为您介绍几种监控分布式节点同步状态的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()
注意事项
- 错误处理: 对网络超时、节点宕机等情况做好处理
- 性能优化: 使用异步IO处理大量节点
- 数据持久化: 将监控数据存入数据库
- 告警分级: 根据延迟程度分级告警
- 资源限制: 控制请求频率，避免影响节点
这个框架可以根据您的具体分布式系统类型（区块链、数据库、微服务等）进行调整和扩展。
