Python脚本如何告警同步任务性能异常

wen python案例 2026-06-25 243

本文目录导读：

基础指标采集方案
异常检测与告警逻辑
告警通知器实现
集成使用示例
高级告警策略
部署建议

针对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')