本文目录导读：
我来介绍几种实现告警级别划分脚本的方法,从简单到复杂：
基础版 - 基于阈值的告警级别
#!/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
这些脚本提供了不同复杂度的告警级别划分方案：
- 基础版: 适合简单阈值监控
- 中级版: 支持多维度分析和模式识别
- 高级版: 可配置规则引擎，支持YAML/JSON配置
- Shell版本: 适合Linux系统快速部署
选择哪种方案取决于你的具体需求：
- 简单监控：使用基础版或Shell版本
- 复杂场景：使用中级或高级版本
- 生产环境：推荐高级版规则引擎
