怎样实现告警级别划分脚本

wen 实用脚本 2026-06-27 231

本文目录导读：

怎样实现告警级别划分脚本

基础版 - 基于阈值的告警级别
中级版 - 多维度告警级别
高级版 - 可配置规则引擎
简易Shell脚本版本

我来介绍几种实现告警级别划分脚本的方法,从简单到复杂：

基础版 - 基于阈值的告警级别

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
基础告警级别划分脚本
"""
import json
from datetime import datetime
class AlertLevel:
    """告警级别枚举"""
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"
class AlertManager:
    def __init__(self):
        self.levels = {
            AlertLevel.INFO: 1,
            AlertLevel.WARNING: 2,
            AlertLevel.ERROR: 3,
            AlertLevel.CRITICAL: 4
        }
    def determine_level_basic(self, metric_value, thresholds):
        """
        基于阈值的简单告警级别划分
        :param metric_value: 指标值
        :param thresholds: 阈值配置 {level: value}
        """
        for level in [AlertLevel.CRITICAL, AlertLevel.ERROR, AlertLevel.WARNING]:
            if metric_value >= thresholds.get(level, float('inf')):
                return level
        return AlertLevel.INFO
    def create_alert(self, metric_name, value, level, message=""):
        """创建告警对象"""
        alert = {
            "id": f"alert_{datetime.now().timestamp()}",
            "metric": metric_name,
            "value": value,
            "level": level,
            "level_num": self.levels[level],
            "message": message or f"{metric_name} 当前值: {value}, 级别: {level}",
            "timestamp": datetime.now().isoformat(),
            "status": "active"
        }
        return alert
# 使用示例
def threshold_example():
    alert_manager = AlertManager()
    # CPU使用率阈值配置
    cpu_thresholds = {
        AlertLevel.WARNING: 70,
        AlertLevel.ERROR: 85,
        AlertLevel.CRITICAL: 95
    }
    # 模拟监控数据
    cpu_usage = 88
    level = alert_manager.determine_level_basic(cpu_usage, cpu_thresholds)
    alert = alert_manager.create_alert("cpu_usage", cpu_usage, level)
    print(f"CPU使用率: {cpu_usage}%")
    print(f"告警级别: {level}")
    print(f"告警信息: {json.dumps(alert, indent=2)}")
if __name__ == "__main__":
    threshold_example()

中级版 - 多维度告警级别

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
多维度告警级别划分脚本
"""
import json
import hashlib
from datetime import datetime
from collections import defaultdict
class AdvancedAlertManager:
    def __init__(self):
        self.alert_history = defaultdict(list)
        self.rules = []
    def add_rule(self, rule):
        """添加告警规则"""
        self.rules.append(rule)
    def evaluate_rules(self, metric_data):
        """评估所有规则"""
        alerts = []
        for rule in self.rules:
            if rule['type'] == 'threshold':
                alert = self._evaluate_threshold_rule(metric_data, rule)
            elif rule['type'] == 'rate':
                alert = self._evaluate_rate_rule(metric_data, rule)
            elif rule['type'] == 'pattern':
                alert = self._evaluate_pattern_rule(metric_data, rule)
            else:
                continue
            if alert:
                alerts.append(alert)
        return self._aggregate_alerts(alerts)
    def _evaluate_threshold_rule(self, data, rule):
        """阈值规则评估"""
        value = data.get(rule['metric'], 0)
        for level, threshold in rule['thresholds'].items():
            if value >= threshold:
                return self._create_alert(
                    rule['metric'], 
                    value, 
                    level,
                    rule.get('message', f"超过{level}阈值")
                )
        return None
    def _evaluate_rate_rule(self, data, rule):
        """变化率规则评估"""
        metric = rule['metric']
        current_value = data.get(metric, 0)
        # 获取历史数据
        history = self.alert_history[metric][-10:] if self.alert_history[metric] else [current_value]
        if len(history) > 1:
            rate_of_change = (current_value - history[-1]) / history[-1] * 100
            for level, threshold in rule['thresholds'].items():
                if abs(rate_of_change) >= threshold:
                    return self._create_alert(
                        f"{metric}_rate",
                        rate_of_change,
                        level,
                        f"变化率: {rate_of_change:.2f}%"
                    )
        # 更新历史数据
        self.alert_history[metric].append(current_value)
        return None
    def _evaluate_pattern_rule(self, data, rule):
        """模式识别规则评估"""
        metric = rule['metric']
        current_value = data.get(metric, 0)
        # 获取近期数据点
        history = self.alert_history[metric][-rule.get('window', 5):]
        history.append(current_value)
        if len(history) >= rule.get('min_points', 3):
            # 检查连续上升趋势
            if rule['pattern'] == 'increasing':
                if all(history[i] < history[i+1] for i in range(len(history)-1)):
                    return self._create_alert(
                        f"{metric}_pattern",
                        current_value,
                        rule.get('level', 'warning'),
                        f"持续上升趋势 (len(history)}个数据点)"
                    )
            # 检查峰值
            elif rule['pattern'] == 'spike':
                avg = sum(history[:-1]) / (len(history) - 1)
                if current_value > avg * rule.get('multiplier', 3):
                    return self._create_alert(
                        f"{metric}_spike",
                        current_value,
                        rule.get('level', 'error'),
                        f"检测到异常峰值 (平均值: {avg:.2f})"
                    )
        self.alert_history[metric] = history
        return None
    def _create_alert(self, metric, value, level, message):
        """创建告警"""
        timestamp = datetime.now()
        alert_id = hashlib.md5(f"{metric}{timestamp.timestamp()}".encode()).hexdigest()[:8]
        return {
            "id": f"alert_{alert_id}",
            "metric": metric,
            "value": value,
            "level": level,
            "level_num": {"info": 1, "warning": 2, "error": 3, "critical": 4}.get(level, 0),
            "message": message,
            "timestamp": timestamp.isoformat(),
            "status": "active"
        }
    def _aggregate_alerts(self, alerts):
        """聚合告警"""
        if not alerts:
            return []
        # 按严重级别排序
        alerts.sort(key=lambda x: x['level_num'], reverse=True)
        # 去重（相同指标保留最高级别）
        deduplicated = {}
        for alert in alerts:
            key = alert['metric']
            if key not in deduplicated or alert['level_num'] > deduplicated[key]['level_num']:
                deduplicated[key] = alert
        return list(deduplicated.values())
# 使用示例
def advanced_example():
    manager = AdvancedAlertManager()
    # 添加规则
    manager.add_rule({
        'type': 'threshold',
        'metric': 'cpu_usage',
        'thresholds': {
            'warning': 70,
            'error': 85,
            'critical': 95
        },
        'message': "CPU使用率过高"
    })
    manager.add_rule({
        'type': 'rate',
        'metric': 'memory_usage',
        'thresholds': {
            'warning': 20,
            'error': 50,
            'critical': 80
        }
    })
    manager.add_rule({
        'type': 'pattern',
        'metric': 'error_count',
        'pattern': 'increasing',
        'window': 5,
        'min_points': 3,
        'level': 'error'
    })
    # 模拟数据序列
    test_data = [
        {"cpu_usage": 75, "memory_usage": 60, "error_count": 5},
        {"cpu_usage": 82, "memory_usage": 65, "error_count": 8},
        {"cpu_usage": 88, "memory_usage": 70, "error_count": 12},
        {"cpu_usage": 92, "memory_usage": 75, "error_count": 15},
    ]
    for i, data in enumerate(test_data):
        print(f"\n--- 第 {i+1} 轮检测 ---")
        print(f"数据: {data}")
        alerts = manager.evaluate_rules(data)
        if alerts:
            print(f"产生的告警:")
            for alert in alerts:
                print(json.dumps(alert, indent=2))
        else:
            print("无告警")
if __name__ == "__main__":
    advanced_example()

高级版 - 可配置规则引擎

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
可配置告警规则引擎
"""
import yaml
import json
from typing import Dict, List, Any
from datetime import datetime, timedelta
from enum import Enum
class Severity(Enum):
    INFO = (1, "信息")
    WARNING = (2, "警告")
    ERROR = (3, "错误")
    CRITICAL = (4, "严重")
    def __init__(self, level, description):
        self.level = level
        self.description = description
class AlertRuleEngine:
    """告警规则引擎"""
    def __init__(self, config_path=None):
        self.rules = []
        self.actions = {}
        self.history = {}
        if config_path:
            self.load_config(config_path)
    def load_config(self, config_path):
        """从配置文件加载规则"""
        with open(config_path, 'r', encoding='utf-8') as f:
            if config_path.endswith('.yaml') or config_path.endswith('.yml'):
                config = yaml.safe_load(f)
            else:
                config = json.load(f)
        self.rules = config.get('rules', [])
        self.actions = config.get('actions', {})
    def process_metric(self, metric_data: Dict[str, Any]) -> List[Dict]:
        """处理单个指标数据"""
        alerts = []
        for rule in self.rules:
            # 检查条件
            if self._match_rule_condition(rule, metric_data):
                alert = self._create_alert(rule, metric_data)
                alerts.append(alert)
                # 执行告警动作
                self._execute_actions(rule.get('actions', []), alert)
        return alerts
    def _match_rule_condition(self, rule: Dict, data: Dict) -> bool:
        """匹配规则条件"""
        condition = rule.get('condition', {})
        metric = condition.get('metric')
        operator = condition.get('operator', '>=')
        threshold = condition.get('threshold')
        if not metric or metric not in data:
            return False
        value = data[metric]
        # 支持多种比较操作
        operators = {
            '>': lambda v, t: v > t,
            '>=': lambda v, t: v >= t,
            '<': lambda v, t: v < t,
            '<=': lambda v, t: v <= t,
            '==': lambda v, t: v == t,
            '!=': lambda v, t: v != t,
            'between': lambda v, t: t[0] <= v <= t[1],
        }
        if operator in operators:
            return operators[operator](value, threshold)
        return False
    def _create_alert(self, rule: Dict, data: Dict) -> Dict:
        """创建告警"""
        severity = Severity[rule.get('severity', 'WARNING').upper()]
        return {
            'id': f"alert_{hash(json.dumps(data, sort_keys=True))}",
            'rule': rule.get('name', 'unknown'),
            'severity': severity.name,
            'level': severity.level,
            'metric': rule.get('condition', {}).get('metric'),
            'value': data.get(rule.get('condition', {}).get('metric')),
            'message': rule.get('message', '').format(**data),
            'timestamp': datetime.now().isoformat(),
            'context': data
        }
    def _execute_actions(self, actions: List[str], alert: Dict):
        """执行告警动作"""
        for action_name in actions:
            if action_name in self.actions:
                action_config = self.actions[action_name]
                self._execute_action(action_config, alert)
    def _execute_action(self, action_config: Dict, alert: Dict):
        """执行单个动作"""
        action_type = action_config.get('type')
        if action_type == 'log':
            self._log_action(alert)
        elif action_type == 'email':
            self._email_action(action_config, alert)
        elif action_type == 'webhook':
            self._webhook_action(action_config, alert)
    def _log_action(self, alert: Dict):
        """记录日志"""
        print(f"[ALERT] {alert['severity']}: {alert['message']}")
    def _email_action(self, config: Dict, alert: Dict):
        """发送邮件"""
        # 实际实现需要SMTP配置
        print(f"发送邮件到 {config.get('recipients')}")
    def _webhook_action(self, config: Dict, alert: Dict):
        """发送Webhook"""
        import requests
        url = config.get('url')
        if url:
            try:
                requests.post(url, json=alert, timeout=5)
            except Exception as e:
                print(f"Webhook失败: {e}")
# 配置文件示例
RULES_CONFIG = """
rules:
  - name: "cpu_high"
    severity: "WARNING"
    condition:
      metric: "cpu_usage"
      operator: ">="
      threshold: 80
    message: "CPU使用率: {cpu_usage}%, 超过80%"
    actions: ["log", "email"]
  - name: "memory_critical"
    severity: "CRITICAL"
    condition:
      metric: "memory_usage"
      operator: ">="
      threshold: 95
    message: "内存使用率: {memory_usage}%, 超过95%"
    actions: ["log", "email", "webhook"]
  - name: "disk_warning"
    severity: "WARNING"
    condition:
      metric: "disk_usage"
      operator: "between"
      threshold: [80, 90]
    message: "磁盘使用率: {disk_usage}%, 处于警告范围"
    actions: ["log"]
actions:
  email:
    type: "email"
    recipients: ["admin@example.com"]
  webhook:
    type: "webhook"
    url: "http://alert-manager/webhook"
"""
def config_file_example():
    """配置文件示例用法"""
    import tempfile
    import os
    # 创建临时配置文件
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
        f.write(RULES_CONFIG)
        config_path = f.name
    try:
        # 使用配置初始化引擎
        engine = AlertRuleEngine(config_path)
        # 模拟监控数据
        test_metrics = [
            {"cpu_usage": 85, "memory_usage": 70, "disk_usage": 75},
            {"cpu_usage": 92, "memory_usage": 96, "disk_usage": 82},
        ]
        for metrics in test_metrics:
            print(f"\n处理指标: {metrics}")
            alerts = engine.process_metric(metrics)
            for alert in alerts:
                print(f"告警: {json.dumps(alert, indent=2)}")
    finally:
        os.unlink(config_path)
if __name__ == "__main__":
    config_file_example()

简易Shell脚本版本

#!/bin/bash
# alert_level.sh - 简易告警级别脚本
# 颜色定义
RED='\033[0;31m'
YELLOW='\033[1;33m'
GREEN='\033[0;32m'
NC='\033[0m' # No Color
# 告警级别函数
check_cpu() {
    local cpu_usage=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)
    if (( $(echo "$cpu_usage > 90" | bc -l) )); then
        echo -e "${RED}CRITICAL: CPU使用率 ${cpu_usage}%${NC}"
    elif (( $(echo "$cpu_usage > 70" | bc -l) )); then
        echo -e "${YELLOW}WARNING: CPU使用率 ${cpu_usage}%${NC}"
    else
        echo -e "${GREEN}OK: CPU使用率 ${cpu_usage}%${NC}"
    fi
}
check_memory() {
    local mem_total=$(free -m | awk '/Mem:/ {print $2}')
    local mem_used=$(free -m | awk '/Mem:/ {print $3}')
    local mem_usage=$(echo "scale=2; $mem_used * 100 / $mem_total" | bc)
    if (( $(echo "$mem_usage > 95" | bc -l) )); then
        echo -e "${RED}CRITICAL: 内存使用率 ${mem_usage}%${NC}"
    elif (( $(echo "$mem_usage > 80" | bc -l) )); then
        echo -e "${YELLOW}WARNING: 内存使用率 ${mem_usage}%${NC}"
    else
        echo -e "${GREEN}OK: 内存使用率 ${mem_usage}%${NC}"
    fi
}
check_disk() {
    local disk_usage=$(df -h / | awk 'NR==2 {print $5}' | cut -d'%' -f1)
    if (( disk_usage > 95 )); then
        echo -e "${RED}CRITICAL: 磁盘使用率 ${disk_usage}%${NC}"
    elif (( disk_usage > 80 )); then
        echo -e "${YELLOW}WARNING: 磁盘使用率 ${disk_usage}%${NC}"
    else
        echo -e "${GREEN}OK: 磁盘使用率 ${disk_usage}%${NC}"
    fi
}
# 主函数
main() {
    echo "=== 系统资源监控 ==="
    echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
    echo "-------------------"
    check_cpu
    check_memory
    check_disk
    echo "-------------------"
}
main

这些脚本提供了不同复杂度的告警级别划分方案：