本文目录导读：
针对Python脚本告警同步任务性能异常,我推荐以下几种实现方案：
基础指标采集方案
import time
import threading
from datetime import datetime, timedelta
import smtplib
import json
from email.mime.text import MIMEText
from typing import Dict, List
class SyncTaskMonitor:
def __init__(self, config: dict):
self.config = config
self.metrics = {
'processing_time': [], # 处理时间记录
'batch_size': [], # 批处理大小
'error_count': 0, # 错误计数
'success_count': 0, # 成功计数
'last_execution_time': None,
'throughput': [] # 吞吐量记录
}
self.thresholds = config.get('thresholds', {
'max_processing_time': 30, # 最大处理时间(秒)
'min_batch_size': 100, # 最小批处理大小
'max_error_rate': 0.05, # 最大错误率
'throughput_window': 300 # 吞吐量统计窗口(5分钟)
})
def record_execution(self, execution_time: float, batch_size: int, errors: int):
"""记录执行指标"""
self.metrics['processing_time'].append({
'time': datetime.now(),
'value': execution_time
})
self.metrics['batch_size'].append({
'time': datetime.now(),
'value': batch_size
})
self.metrics['error_count'] += errors
self.metrics['success_count'] += batch_size
self.metrics['last_execution_time'] = datetime.now()
# 计算吞吐量
throughput = batch_size / execution_time if execution_time > 0 else 0
self.metrics['throughput'].append({
'time': datetime.now(),
'value': throughput
})
# 清理旧数据
self._clean_old_metrics()
def _clean_old_metrics(self, retention_minutes: int = 60):
"""清理超过保留时间的指标数据"""
cutoff_time = datetime.now() - timedelta(minutes=retention_minutes)
for key in self.metrics:
if isinstance(self.metrics[key], list):
self.metrics[key] = [
item for item in self.metrics[key]
if item['time'] > cutoff_time
]
异常检测与告警逻辑
class PerformanceAlert:
def __init__(self, notifiers: list):
self.notifiers = notifiers # 告警通知器列表
def check_performance_anomalies(self, monitor: 'SyncTaskMonitor') -> list:
"""检查性能异常"""
alerts = []
# 1. 检测处理时间异常(滑动窗口)
recent_times = [p['value'] for p in monitor.metrics['processing_time'][-10:]]
if recent_times:
avg_time = sum(recent_times) / len(recent_times)
if avg_time > monitor.thresholds['max_processing_time']:
alerts.append({
'type': 'processing_time_high',
'severity': 'warning',
'message': f'平均处理时间 {avg_time:.2f}s 超过阈值 {monitor.thresholds["max_processing_time"]}s',
'timestamp': datetime.now()
})
# 2. 检测批处理大小异常
recent_batches = [b['value'] for b in monitor.metrics['batch_size'][-5:]]
if recent_batches:
avg_batch = sum(recent_batches) / len(recent_batches)
if avg_batch < monitor.thresholds['min_batch_size']:
alerts.append({
'type': 'batch_size_low',
'severity': 'warning',
'message': f'平均批处理大小 {avg_batch:.0f} 低于阈值 {monitor.thresholds["min_batch_size"]}',
'timestamp': datetime.now()
})
# 3. 检测错误率异常
total_ops = monitor.metrics['success_count'] + monitor.metrics['error_count']
if total_ops > 0:
error_rate = monitor.metrics['error_count'] / total_ops
if error_rate > monitor.thresholds['max_error_rate']:
alerts.append({
'type': 'error_rate_high',
'severity': 'critical',
'message': f'错误率 {error_rate:.2%} 超过阈值 {monitor.thresholds["max_error_rate"]:.0%}',
'timestamp': datetime.now()
})
# 4. 检测吞吐量下降
throughput_window = monitor.thresholds['throughput_window']
recent_throughput = [
t['value'] for t in monitor.metrics['throughput']
if t['time'] > datetime.now() - timedelta(seconds=throughput_window)
]
if len(recent_throughput) >= 2:
current_avg = sum(recent_throughput[-3:]) / min(len(recent_throughput[-3:]), 3)
historical_avg = sum(recent_throughput[:-3]) / max(len(recent_throughput[:-3]), 1)
if historical_avg > 0 and current_avg / historical_avg < 0.7:
alerts.append({
'type': 'throughput_decrease',
'severity': 'warning',
'message': f'吞吐量下降30%，当前: {current_avg:.2f} 条/秒, 历史: {historical_avg:.2f} 条/秒',
'timestamp': datetime.now()
})
return alerts
def send_alerts(self, alerts: list):
"""发送告警通知"""
for alert in alerts:
for notifier in self.notifiers:
try:
notifier.send(alert)
except Exception as e:
print(f"发送告警失败: {e}")
告警通知器实现
import requests
import logging
from abc import ABC, abstractmethod
class BaseNotifier(ABC):
@abstractmethod
def send(self, alert: dict):
pass
class EmailNotifier(BaseNotifier):
def __init__(self, smtp_config: dict):
self.smtp_config = smtp_config
def send(self, alert: dict):
msg = MIMEText(
f"告警类型: {alert['type']}\n"
f"告警级别: {alert['severity']}\n"
f"告警时间: {alert['timestamp']}\n"
f"告警内容: {alert['message']}"
)
msg['Subject'] = f"[告警] 同步任务性能异常 - {alert['severity']}"
msg['From'] = self.smtp_config['from']
msg['To'] = ','.join(self.smtp_config['to'])
with smtplib.SMTP(self.smtp_config['host'], self.smtp_config['port']) as server:
server.starttls()
server.login(self.smtp_config['user'], self.smtp_config['password'])
server.send_message(msg)
class WebhookNotifier(BaseNotifier):
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send(self, alert: dict):
payload = {
'msgtype': 'text',
'text': {
'content': f"【同步任务告警】\n级别: {alert['severity']}\n内容: {alert['message']}"
}
}
requests.post(self.webhook_url, json=payload)
class LogNotifier(BaseNotifier):
def __init__(self, logger: logging.Logger):
self.logger = logger
def send(self, alert: dict):
log_method = getattr(self.logger, alert['severity'].lower(), self.logger.warning)
log_method(f"{alert['type']}: {alert['message']}")
集成使用示例
import logging
def sync_task_worker(monitor, alert_manager):
"""同步任务工作线程"""
while True:
try:
start_time = time.time()
# 执行同步任务
batch_size, errors = execute_sync_batch()
execution_time = time.time() - start_time
monitor.record_execution(execution_time, batch_size, errors)
# 检查性能异常
alerts = alert_manager.check_performance_anomalies(monitor)
if alerts:
alert_manager.send_alerts(alerts)
except Exception as e:
logging.error(f"同步任务执行异常: {e}")
# 发送任务异常告警
alert_manager.send_alerts([{
'type': 'task_exception',
'severity': 'critical',
'message': f'同步任务执行异常: {str(e)}',
'timestamp': datetime.now()
}])
def execute_sync_batch():
"""执行同步批次(示例)"""
# 实际业务逻辑
import random
time.sleep(random.uniform(1, 5))
batch_size = random.randint(50, 200)
errors = int(batch_size * random.uniform(0, 0.1))
return batch_size, errors
if __name__ == "__main__":
# 配置
config = {
'thresholds': {
'max_processing_time': 30,
'min_batch_size': 100,
'max_error_rate': 0.05,
'throughput_window': 300
}
}
# 初始化
monitor = SyncTaskMonitor(config)
logger = logging.getLogger('sync_task')
# 创建告警管理器(支持多个通知方式)
alert_manager = PerformanceAlert([
LogNotifier(logger),
EmailNotifier({
'host': 'smtp.example.com',
'port': 587,
'user': 'user@example.com',
'password': 'password',
'from': 'monitor@example.com',
'to': ['admin@example.com']
}),
WebhookNotifier('https://hooks.example.com/webhook')
])
# 启动监控线程
monitor_thread = threading.Thread(
target=sync_task_worker,
args=(monitor, alert_manager),
daemon=True
)
monitor_thread.start()
# 保持主线程运行
monitor_thread.join()
高级告警策略
class AdvancedAlertStrategy:
"""高级告警策略"""
def __init__(self):
self.alert_history = []
self.cooldown_period = 300 # 5分钟冷却期
self.escalation_threshold = 3 # 连续告警升级阈值
def should_alert(self, alert: dict) -> bool:
"""判断是否需要发送告警(防重复告警)"""
current_time = datetime.now()
# 检查冷却期
recent_alerts = [a for a in self.alert_history
if a['timestamp'] > current_time - timedelta(seconds=self.cooldown_period)
and a['type'] == alert['type']]
if recent_alerts:
# 检查是否有相同类型的告警在冷却期内
return False
# 记录告警
self.alert_history.append(alert)
# 检查是否需要升级(连续相同告警)
similar_alerts = [a for a in self.alert_history
if a['type'] == alert['type']
and a['timestamp'] > current_time - timedelta(hours=1)]
if len(similar_alerts) >= self.escalation_threshold:
alert['severity'] = self._escalate_severity(alert['severity'])
return True
def _escalate_severity(self, severity: str) -> str:
"""升级告警级别"""
escalation_map = {
'info': 'warning',
'warning': 'error',
'error': 'critical'
}
return escalation_map.get(severity, 'critical')
部署建议
- 数据存储: 使用时序数据库(如InfluxDB)存储指标数据
- 可视化: 集成Grafana展示性能趋势
- 告警聚合: 使用AlertManager处理告警风暴
- 分布式监控: 对大规模部署使用Prometheus采集指标
这个方案可以根据实际需求调整阈值和告警策略,并支持多种通知方式。
