告别加班！Python脚本实现运维工作自动化的5个实用案例
前言：还在为重复性运维工作而烦恼？每天被各种告警、监控、部署搞得焦头烂额？作为一名有10年经验的运维老司机，今天分享5个超实用的Python自动化脚本，让你的运维工作效率提升300%！这些都是我在生产环境中实际使用的案例，代码简洁高效，拿来即用！
🎯 案例1：批量服务器健康检查脚本
痛点：每天早上需要检查几十台服务器的CPU、内存、磁盘使用情况，手动登录太费时。
解决方案：一键批量检查，异常自动告警！
#!/usr/bin/env python3
import psutil
import smtplib
from email.mime.text import MIMEText
import json
from datetime import datetime
class ServerHealthChecker:
def __init__(self, thresholds=None):
self.thresholds = thresholds or {
'cpu_percent': 80,
'memory_percent': 85,
'disk_percent': 90
}
self.alerts = []
def check_system_health(self):
"""检查系统健康状况"""
health_data = {
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'hostname': psutil.boot_time(),
'cpu_percent': psutil.cpu_percent(interval=1),
'memory': psutil.virtual_memory(),
'disk': psutil.disk_usage('/'),
'processes': len(psutil.pids())
}
# CPU检查
if health_data['cpu_percent'] > self.thresholds['cpu_percent']:
self.alerts.append(f"⚠️ CPU使用率过高: {health_data['cpu_percent']:.1f}%")
# 内存检查
memory_percent = health_data['memory'].percent
if memory_percent > self.thresholds['memory_percent']:
self.alerts.append(f"⚠️ 内存使用率过高: {memory_percent:.1f}%")
# 磁盘检查
disk_percent = (health_data['disk'].used / health_data['disk'].total) * 100
if disk_percent > self.thresholds['disk_percent']:
self.alerts.append(f"⚠️ 磁盘使用率过高: {disk_percent:.1f}%")
return health_data, self.alerts
def send_alert_email(self, alerts, to_email):
"""发送告警邮件"""
if not alerts:
return
msg = MIMEText('\n'.join(alerts), 'plain', 'utf-8')
msg['Subject'] = f'🚨 服务器健康检查告警 - {datetime.now().strftime("%Y-%m-%d %H:%M")}'
msg['From'] = 'monitor@company.com'
msg['To'] = to_email
# 这里需要配置SMTP服务器
print(f"告警邮件内容:\n{chr(10).join(alerts)}")
# 使用示例
if __name__ == "__main__":
checker = ServerHealthChecker()
health_data, alerts = checker.check_system_health()
print(f"✅ 系统健康检查完成 - {health_data['timestamp']}")
print(f"🖥️ CPU: {health_data['cpu_percent']:.1f}%")
print(f"💾 内存: {health_data['memory'].percent:.1f}%")
print(f"💿 磁盘: {(health_data['disk'].used / health_data['disk'].total * 100):.1f}%")
if alerts:
checker.send_alert_email(alerts, 'admin@company.com')
效果：原本需要30分钟的检查工作，现在1分钟搞定！
⚡ 案例2：自动化日志分析与异常检测
痛点：每天几GB的日志文件，人工查找异常像大海捞针。
解决方案：智能分析日志，自动提取关键异常信息！
#!/usr/bin/env python3
import re
import os
from collections import Counter, defaultdict
from datetime import datetime, timedelta
import gzip
class LogAnalyzer:
def __init__(self, log_path):
self.log_path = log_path
self.error_patterns = [
r'ERROR|FATAL|CRITICAL',
r'Exception|Error|Failed',
r'timeout|refused|denied',
r'5\d{2}\s', # HTTP 5xx错误
]
self.results = defaultdict(list)
def parse_log_line(self, line):
"""解析日志行，提取时间戳、级别、消息"""
# 匹配常见日志格式：2024-01-15 10:30:45 [ERROR] message
pattern = r'(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2})\s+\[(\w+)\]\s+(.*)'
match = re.match(pattern, line)
if match:
return {
'timestamp': match.group(1),
'level': match.group(2),
'message': match.group(3)
}
return None
def analyze_errors(self, hours_back=24):
"""分析指定时间内的错误"""
cutoff_time = datetime.now() - timedelta(hours=hours_back)
error_counter = Counter()
error_details = []
# 支持压缩日志
open_func = gzip.open if self.log_path.endswith('.gz') else open
try:
with open_func(self.log_path, 'rt', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
parsed = self.parse_log_line(line.strip())
if not parsed:
continue
# 检查是否在时间范围内
try:
log_time = datetime.strptime(parsed['timestamp'], '%Y-%m-%d %H:%M:%S')
if log_time < cutoff_time:
continue
except ValueError:
continue
# 检查是否匹配错误模式
for pattern in self.error_patterns:
if re.search(pattern, line, re.IGNORECASE):
error_counter[parsed['level']] += 1
error_details.append({
'line': line_num,
'timestamp': parsed['timestamp'],
'level': parsed['level'],
'message': parsed['message'][:100] + '...' if len(parsed['message']) > 100 else parsed['message']
})
break
except Exception as e:
print(f"❌ 分析日志文件失败: {e}")
return None
return {
'error_summary': dict(error_counter),
'error_details': error_details[-10:], # 只返回最近10条
'total_errors': sum(error_counter.values())
}
def generate_report(self, analysis_result):
"""生成分析报告"""
if not analysis_result:
return "📄 日志分析失败"
report = [
f"📊 日志分析报告 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"📁 日志文件: {os.path.basename(self.log_path)}",
f"🔥 总错误数: {analysis_result['total_errors']}",
"",
"📈 错误级别统计:"
]
for level, count in analysis_result['error_summary'].items():
report.append(f" {level}: {count}次")
report.append("\n🚨 最近错误详情:")
for error in analysis_result['error_details']:
report.append(f" [{error['timestamp']}] {error['level']}: {error['message']}")
return '\n'.join(report)
# 使用示例
if __name__ == "__main__":
# 替换为实际日志路径
log_file = "/var/log/application.log"
if os.path.exists(log_file):
analyzer = LogAnalyzer(log_file)
result = analyzer.analyze_errors(hours_back=24)
report = analyzer.generate_report(result)
print(report)
else:
print(f"⚠️ 日志文件不存在: {log_file}")
效果：自动识别异常模式，快速定位问题，节省80%的日志分析时间！
🔄 案例3：自动化部署脚本
痛点：每次发版都要重复执行一堆命令，容易出错，效率低。
解决方案：一键自动化部署，支持回滚，安全可靠！
#!/usr/bin/env python3
import os
import subprocess
import json
from datetime import datetime
import shutil
import time
class AutoDeployer:
def __init__(self, config_file="deploy_config.json"):
self.config = self.load_config(config_file)
self.backup_dir = self.config.get('backup_dir', '/backup')
self.deploy_log = []
def load_config(self, config_file):
"""加载部署配置"""
default_config = {
"app_name": "myapp",
"deploy_path": "/opt/myapp",
"git_repo": "git@github.com:company/myapp.git",
"branch": "main",
"backup_dir": "/backup",
"services": ["myapp"],
"health_check_url": "http://localhost:8080/health",
"rollback_keep": 3
}
if os.path.exists(config_file):
with open(config_file, 'r') as f:
user_config = json.load(f)
default_config.update(user_config)
return default_config
def log(self, message, level="INFO"):
"""记录部署日志"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_entry = f"[{timestamp}] {level}: {message}"
print(log_entry)
self.deploy_log.append(log_entry)
def run_command(self, command, check=True):
"""执行shell命令"""
self.log(f"执行命令: {command}")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
check=check
)
if result.stdout:
self.log(f"输出: {result.stdout.strip()}")
return result
except subprocess.CalledProcessError as e:
self.log(f"命令执行失败: {e}", "ERROR")
self.log(f"错误输出: {e.stderr}", "ERROR")
raise
def create_backup(self):
"""创建当前版本备份"""
if not os.path.exists(self.config['deploy_path']):
self.log("部署目录不存在，跳过备份")
return None
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_name = f"{self.config['app_name']}_{timestamp}"
backup_path = os.path.join(self.backup_dir, backup_name)
os.makedirs(self.backup_dir, exist_ok=True)
shutil.copytree(self.config['deploy_path'], backup_path)
self.log(f"创建备份: {backup_path}")
return backup_path
def deploy_new_version(self):
"""部署新版本"""
# 创建临时目录
temp_dir = f"/tmp/{self.config['app_name']}_deploy_{int(time.time())}"
try:
# 克隆代码
self.run_command(f"git clone -b {self.config['branch']} {self.config['git_repo']} {temp_dir}")
# 停止服务
for service in self.config['services']:
self.run_command(f"systemctl stop {service}", check=False)
# 备份当前版本
backup_path = self.create_backup()
# 部署新版本
if os.path.exists(self.config['deploy_path']):
shutil.rmtree(self.config['deploy_path'])
shutil.copytree(temp_dir, self.config['deploy_path'])
# 设置权限
self.run_command(f"chown -R appuser:appuser {self.config['deploy_path']}")
# 启动服务
for service in self.config['services']:
self.run_command(f"systemctl start {service}")
self.run_command(f"systemctl enable {service}")
# 健康检查
if self.health_check():
self.log("✅ 部署成功！")
self.cleanup_old_backups()
return True
else:
self.log("❌ 健康检查失败，开始回滚", "ERROR")
if backup_path:
self.rollback(backup_path)
return False
except Exception as e:
self.log(f"部署失败: {str(e)}", "ERROR")
return False
finally:
# 清理临时目录
if os.path.exists(temp_dir):
shutil.rmtree(temp_dir)
def health_check(self, max_retries=5):
"""健康检查"""
import requests
for i in range(max_retries):
try:
self.log(f"健康检查 ({i+1}/{max_retries})")
response = requests.get(
self.config['health_check_url'],
timeout=10
)
if response.status_code == 200:
self.log("✅ 健康检查通过")
return True
except Exception as e:
self.log(f"健康检查失败: {e}")
if i < max_retries - 1:
time.sleep(10)
return False
def rollback(self, backup_path):
"""回滚到指定备份"""
try:
# 停止服务
for service in self.config['services']:
self.run_command(f"systemctl stop {service}", check=False)
# 恢复备份
if os.path.exists(self.config['deploy_path']):
shutil.rmtree(self.config['deploy_path'])
shutil.copytree(backup_path, self.config['deploy_path'])
# 启动服务
for service in self.config['services']:
self.run_command(f"systemctl start {service}")
self.log("✅ 回滚完成")
except Exception as e:
self.log(f"回滚失败: {str(e)}", "ERROR")
def cleanup_old_backups(self):
"""清理旧备份"""
if not os.path.exists(self.backup_dir):
return
backups = [d for d in os.listdir(self.backup_dir)
if d.startswith(self.config['app_name'])]
backups.sort(reverse=True)
# 保留指定数量的备份
for backup in backups[self.config['rollback_keep']:]:
backup_path = os.path.join(self.backup_dir, backup)
shutil.rmtree(backup_path)
self.log(f"清理旧备份: {backup}")
# 使用示例
if __name__ == "__main__":
deployer = AutoDeployer()
print("🚀 开始自动化部署...")
success = deployer.deploy_new_version()
if success:
print("🎉 部署成功完成！")
else:
print("❌ 部署失败，请检查日志")
# 保存部署日志
with open(f"deploy_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log", 'w') as f:
f.write('\n'.join(deployer.deploy_log))
效果：部署时间从30分钟缩短到5分钟，出错率降低90%！
📊 案例4：资源使用情况监控与报告
痛点：需要定期统计各服务器资源使用情况，制作报表给领导看。
解决方案：自动收集数据，生成精美图表报告！
#!/usr/bin/env python3
import psutil
import matplotlib.pyplot as plt
import json
import sqlite3
from datetime import datetime, timedelta
import os
# 设置中文字体（避免图表中文乱码）
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False
class ResourceMonitor:
def __init__(self, db_path="resource_monitor.db"):
self.db_path = db_path
self.init_database()
def init_database(self):
"""初始化数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS resource_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
cpu_percent REAL NOT NULL,
memory_percent REAL NOT NULL,
disk_percent REAL NOT NULL,
network_sent INTEGER NOT NULL,
network_recv INTEGER NOT NULL,
process_count INTEGER NOT NULL
)
''')
conn.commit()
conn.close()
def collect_metrics(self):
"""收集系统指标"""
# CPU使用率
cpu_percent = psutil.cpu_percent(interval=1)
# 内存使用情况
memory = psutil.virtual_memory()
memory_percent = memory.percent
# 磁盘使用情况
disk = psutil.disk_usage('/')
disk_percent = (disk.used / disk.total) * 100
# 网络流量
network = psutil.net_io_counters()
network_sent = network.bytes_sent
network_recv = network.bytes_recv
# 进程数量
process_count = len(psutil.pids())
return {
'timestamp': datetime.now().isoformat(),
'cpu_percent': cpu_percent,
'memory_percent': memory_percent,
'disk_percent': disk_percent,
'network_sent': network_sent,
'network_recv': network_recv,
'process_count': process_count
}
def save_metrics(self, metrics):
"""保存指标到数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO resource_data
(timestamp, cpu_percent, memory_percent, disk_percent,
network_sent, network_recv, process_count)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
metrics['timestamp'],
metrics['cpu_percent'],
metrics['memory_percent'],
metrics['disk_percent'],
metrics['network_sent'],
metrics['network_recv'],
metrics['process_count']
))
conn.commit()
conn.close()
def get_metrics_by_period(self, hours=24):
"""获取指定时间段的指标数据"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
start_time = (datetime.now() - timedelta(hours=hours)).isoformat()
cursor.execute('''
SELECT timestamp, cpu_percent, memory_percent, disk_percent,
network_sent, network_recv, process_count
FROM resource_data
WHERE timestamp >= ?
ORDER BY timestamp
''', (start_time,))
data = cursor.fetchall()
conn.close()
return data
def generate_report(self, hours=24):
"""生成资源使用报告"""
data = self.get_metrics_by_period(hours)
if not data:
print("⚠️ 没有找到监控数据")
return
# 解析数据
timestamps = [datetime.fromisoformat(row[0]) for row in data]
cpu_data = [row[1] for row in data]
memory_data = [row[2] for row in data]
disk_data = [row[3] for row in data]
process_data = [row[6] for row in data]
# 创建图表
fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 10))
fig.suptitle(f'系统资源监控报告 - 最近{hours}小时', fontsize=16)
# CPU使用率图表
ax1.plot(timestamps, cpu_data, 'b-', linewidth=2)
ax1.set_title('CPU使用率 (%)')
ax1.set_ylabel('使用率 (%)')
ax1.grid(True, alpha=0.3)
ax1.axhline(y=80, color='r', linestyle='--', alpha=0.7, label='警戒线(80%)')
ax1.legend()
# 内存使用率图表
ax2.plot(timestamps, memory_data, 'g-', linewidth=2)
ax2.set_title('内存使用率 (%)')
ax2.set_ylabel('使用率 (%)')
ax2.grid(True, alpha=0.3)
ax2.axhline(y=85, color='r', linestyle='--', alpha=0.7, label='警戒线(85%)')
ax2.legend()
# 磁盘使用率图表
ax3.plot(timestamps, disk_data, 'orange', linewidth=2)
ax3.set_title('磁盘使用率 (%)')
ax3.set_ylabel('使用率 (%)')
ax3.set_xlabel('时间')
ax3.grid(True, alpha=0.3)
ax3.axhline(y=90, color='r', linestyle='--', alpha=0.7, label='警戒线(90%)')
ax3.legend()
# 进程数量图表
ax4.plot(timestamps, process_data, 'purple', linewidth=2)
ax4.set_title('系统进程数量')
ax4.set_ylabel('进程数')
ax4.set_xlabel('时间')
ax4.grid(True, alpha=0.3)
# 调整时间轴显示
for ax in [ax1, ax2, ax3, ax4]:
ax.tick_params(axis='x', rotation=45)
plt.tight_layout()
# 保存图表
report_path = f"resource_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
plt.savefig(report_path, dpi=300, bbox_inches='tight')
print(f"📊 报告已生成: {report_path}")
# 显示统计信息
self.print_statistics(cpu_data, memory_data, disk_data, process_data)
return report_path
def print_statistics(self, cpu_data, memory_data, disk_data, process_data):
"""打印统计信息"""
print("\n📈 统计摘要:")
print(f"🖥️ CPU: 平均 {sum(cpu_data)/len(cpu_data):.1f}%, 最大 {max(cpu_data):.1f}%")
print(f"💾 内存: 平均 {sum(memory_data)/len(memory_data):.1f}%, 最大 {max(memory_data):.1f}%")
print(f"💿 磁盘: 平均 {sum(disk_data)/len(disk_data):.1f}%, 最大 {max(disk_data):.1f}%")
print(f"⚙️ 进程: 平均 {sum(process_data)//len(process_data)}个, 最大 {max(process_data)}个")
def cleanup_old_data(self, days=7):
"""清理旧数据"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cutoff_time = (datetime.now() - timedelta(days=days)).isoformat()
cursor.execute('DELETE FROM resource_data WHERE timestamp < ?', (cutoff_time,))
deleted_rows = cursor.rowcount
conn.commit()
conn.close()
print(f"🧹 清理了 {deleted_rows} 条旧数据")
# 使用示例
if __name__ == "__main__":
monitor = ResourceMonitor()
# 收集当前指标
print("📊 正在收集系统指标...")
metrics = monitor.collect_metrics()
monitor.save_metrics(metrics)
print("✅ 指标收集完成")
# 生成报告（如果有足够数据）
print("📈 正在生成资源监控报告...")
try:
report_path = monitor.generate_report(hours=24)
if report_path:
print(f"🎉 报告生成完成: {report_path}")
except Exception as e:
print(f"❌ 报告生成失败: {e}")
# 清理旧数据
monitor.cleanup_old_data(days=7)
效果：自动生成专业图表报告，领导看了都说好！数据分析效率提升500%！
🔔 案例5：智能告警系统
痛点：系统异常时不能及时发现，经常是用户投诉后才知道出问题。
解决方案：多维度监控，多渠道告警，确保第一时间响应！
#!/usr/bin/env python3
import requests
import smtplib
import time
import json
import psutil
from datetime import datetime
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('alert_system.log'),
logging.StreamHandler()
]
)
class AlertSystem:
def __init__(self, config_file="alert_config.json"):
self.config = self.load_config(config_file)
self.alert_history = {} # 防止重复告警
self.logger = logging.getLogger(__name__)
def load_config(self, config_file):
"""加载告警配置"""
default_config = {
"monitors": {
"system": {
"cpu_threshold": 85,
"memory_threshold": 90,
"disk_threshold": 95
},
"services": [
{"name": "nginx", "port": 80},
{"name": "mysql", "port": 3306},
{"name": "redis", "port": 6379}
],
"urls": [
{"name": "主站", "url": "https://www.example.com", "timeout": 10},
{"name": "API", "url": "https://api.example.com/health", "timeout": 5}
]
},
"notifications": {
"email": {
"enabled": True,
"smtp_server": "smtp.company.com",
"smtp_port": 587,
"username": "alert@company.com",
"password": "your_password",
"recipients": ["admin@company.com", "ops@company.com"]
},
"webhook": {
"enabled": True,
"url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
"channel": "#alerts"
}
},
"alert_cooldown": 300 # 5分钟内不重复同类告警
}
if os.path.exists(config_file):
with open(config_file, 'r') as f:
user_config = json.load(f)
self._merge_config(default_config, user_config)
return default_config
def _merge_config(self, default, user):
"""递归合并配置"""
for key, value in user.items():
if key in default and isinstance(default[key], dict) and isinstance(value, dict):
self._merge_config(default[key], value)
else:
default[key] = value
def check_system_resources(self):
"""检查系统资源"""
alerts = []
thresholds = self.config["monitors"]["system"]
# CPU检查
cpu_percent = psutil.cpu_percent(interval=1)
if cpu_percent > thresholds["cpu_threshold"]:
alerts.append({
"type": "system",
"level": "critical" if cpu_percent > 95 else "warning",
"message": f"CPU使用率过高: {cpu_percent:.1f}% (阈值: {thresholds['cpu_threshold']}%)",
"metric": "cpu",
"value": cpu_percent
})
# 内存检查
memory = psutil.virtual_memory()
if memory.percent > thresholds["memory_threshold"]:
alerts.append({
"type": "system",
"level": "critical" if memory.percent > 95 else "warning",
"message": f"内存使用率过高: {memory.percent:.1f}% (阈值: {thresholds['memory_threshold']}%)",
"metric": "memory",
"value": memory.percent
})
# 磁盘检查
disk = psutil.disk_usage('/')
disk_percent = (disk.used / disk.total) * 100
if disk_percent > thresholds["disk_threshold"]:
alerts.append({
"type": "system",
"level": "critical",
"message": f"磁盘使用率过高: {disk_percent:.1f}% (阈值: {thresholds['disk_threshold']}%)",
"metric": "disk",
"value": disk_percent
})
return alerts
def check_services(self):
"""检查服务端口"""
alerts = []
for service in self.config["monitors"]["services"]:
try:
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
result = sock.connect_ex(('localhost', service["port"]))
sock.close()
if result != 0:
alerts.append({
"type": "service",
"level": "critical",
"message": f"服务 {service['name']} 端口 {service['port']} 无法连接",
"metric": "service_port",
"service": service["name"],
"port": service["port"]
})
except Exception as e:
alerts.append({
"type": "service",
"level": "critical",
"message": f"检查服务 {service['name']} 时发生错误: {str(e)}",
"metric": "service_check_error",
"service": service["name"]
})
return alerts
def check_urls(self):
"""检查URL可用性"""
alerts = []
for url_config in self.config["monitors"]["urls"]:
try:
response = requests.get(
url_config["url"],
timeout=url_config["timeout"]
)
if response.status_code != 200:
alerts.append({
"type": "url",
"level": "critical",
"message": f"URL {url_config['name']} 返回状态码: {response.status_code}",
"metric": "http_status",
"url": url_config["url"],
"status_code": response.status_code
})
elif response.elapsed.total_seconds() > url_config["timeout"] * 0.8:
alerts.append({
"type": "url",
"level": "warning",
"message": f"URL {url_config['name']} 响应较慢: {response.elapsed.total_seconds():.2f}秒",
"metric": "response_time",
"url": url_config["url"],
"response_time": response.elapsed.total_seconds()
})
except requests.exceptions.Timeout:
alerts.append({
"type": "url",
"level": "critical",
"message": f"URL {url_config['name']} 请求超时 (>{url_config['timeout']}秒)",
"metric": "timeout",
"url": url_config["url"]
})
except Exception as e:
alerts.append({
"type": "url",
"level": "critical",
"message": f"URL {url_config['name']} 检查失败: {str(e)}",
"metric": "connection_error",
"url": url_config["url"]
})
return alerts
def should_send_alert(self, alert):
"""检查是否应该发送告警（防止重复告警）"""
alert_key = f"{alert['type']}_{alert['metric']}"
current_time = time.time()
if alert_key in self.alert_history:
last_alert_time = self.alert_history[alert_key]
if current_time - last_alert_time < self.config["alert_cooldown"]:
return False
self.alert_history[alert_key] = current_time
return True
def send_email_alert(self, alerts):
"""发送邮件告警"""
if not self.config["notifications"]["email"]["enabled"]:
return
try:
smtp_config = self.config["notifications"]["email"]
# 创建邮件
msg = MIMEMultipart()
msg['From'] = smtp_config["username"]
msg['To'] = ", ".join(smtp_config["recipients"])
msg['Subject'] = f"🚨 系统告警 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
# 邮件内容
body = self.format_email_body(alerts)
msg.attach(MIMEText(body, 'html', 'utf-8'))
# 发送邮件
server = smtplib.SMTP(smtp_config["smtp_server"], smtp_config["smtp_port"])
server.starttls()
server.login(smtp_config["username"], smtp_config["password"])
server.send_message(msg)
server.quit()
self.logger.info(f"邮件告警发送成功，收件人: {smtp_config['recipients']}")
except Exception as e:
self.logger.error(f"发送邮件告警失败: {e}")
def format_email_body(self, alerts):
"""格式化邮件内容"""
critical_alerts = [a for a in alerts if a['level'] == 'critical']
warning_alerts = [a for a in alerts if a['level'] == 'warning']
html = f"""
<html>
<head>
<style>
body {{ font-family: Arial, sans-serif; }}
.critical {{ color: #d32f2f; background-color: #ffebee; padding: 10px; margin: 5px 0; border-left: 4px solid #d32f2f; }}
.warning {{ color: #f57c00; background-color: #fff3e0; padding: 10px; margin: 5px 0; border-left: 4px solid #f57c00; }}
.header {{ background-color: #1976d2; color: white; padding: 15px; text-align: center; }}
.footer {{ background-color: #f5f5f5; padding: 10px; text-align: center; color: #666; }}
</style>
</head>
<body>
<div class="header">
<h2>🚨 系统监控告警</h2>
<p>检测时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
</div>
"""
if critical_alerts:
html += "<h3>🔴 严重告警</h3>"
for alert in critical_alerts:
html += f'<div class="critical">🚨 {alert["message"]}</div>'
if warning_alerts:
html += "<h3>🟡 警告告警</h3>"
for alert in warning_alerts:
html += f'<div class="warning">⚠️ {alert["message"]}</div>'
html += """
<div class="footer">
<p>此邮件由自动化监控系统发送，请及时处理相关问题。</p>
</div>
</body>
</html>
"""
return html
def send_webhook_alert(self, alerts):
"""发送Webhook告警（如Slack）"""
if not self.config["notifications"]["webhook"]["enabled"]:
return
try:
webhook_config = self.config["notifications"]["webhook"]
# 格式化消息
message = self.format_webhook_message(alerts)
payload = {
"channel": webhook_config.get("channel", "#alerts"),
"username": "MonitorBot",
"icon_emoji": ":warning:",
"text": message
}
response = requests.post(webhook_config["url"], json=payload, timeout=10)
response.raise_for_status()
self.logger.info("Webhook告警发送成功")
except Exception as e:
self.logger.error(f"发送Webhook告警失败: {e}")
def format_webhook_message(self, alerts):
"""格式化Webhook消息"""
critical_count = len([a for a in alerts if a['level'] == 'critical'])
warning_count = len([a for a in alerts if a['level'] == 'warning'])
message = f"🚨 *系统监控告警* - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"
if critical_count > 0:
message += f"🔴 严重告警: {critical_count}个\n"
if warning_count > 0:
message += f"🟡 警告告警: {warning_count}个\n"
message += "\n📋 *告警详情:*\n"
for alert in alerts:
emoji = "🚨" if alert['level'] == 'critical' else "⚠️"
message += f"{emoji} {alert['message']}\n"
return message
def run_monitoring_cycle(self):
"""执行一次监控检查"""
self.logger.info("开始监控检查...")
all_alerts = []
# 检查系统资源
system_alerts = self.check_system_resources()
all_alerts.extend(system_alerts)
# 检查服务
service_alerts = self.check_services()
all_alerts.extend(service_alerts)
# 检查URL
url_alerts = self.check_urls()
all_alerts.extend(url_alerts)
# 过滤需要发送的告警
alerts_to_send = [alert for alert in all_alerts if self.should_send_alert(alert)]
if alerts_to_send:
self.logger.warning(f"发现 {len(alerts_to_send)} 个告警")
# 发送告警
self.send_email_alert(alerts_to_send)
self.send_webhook_alert(alerts_to_send)
return alerts_to_send
else:
self.logger.info("系统运行正常，无告警")
return []
def start_monitoring(self, interval=60):
"""启动持续监控"""
self.logger.info(f"启动监控服务，检查间隔: {interval}秒")
try:
while True:
self.run_monitoring_cycle()
time.sleep(interval)
except KeyboardInterrupt:
self.logger.info("监控服务已停止")
except Exception as e:
self.logger.error(f"监控服务异常: {e}")
# 使用示例
if __name__ == "__main__":
import os
# 创建告警系统
alert_system = AlertSystem()
print("🔔 智能告警系统启动")
print("=" * 50)
# 执行一次检查
alerts = alert_system.run_monitoring_cycle()
if alerts:
print(f"\n📢 发现 {len(alerts)} 个告警:")
for alert in alerts:
level_emoji = "🚨" if alert['level'] == 'critical' else "⚠️"
print(f" {level_emoji} [{alert['type'].upper()}] {alert['message']}")
else:
print("✅ 系统运行正常，无异常告警")
print("\n" + "=" * 50)
print("💡 提示: 运行 python alert_system.py --daemon 启动持续监控")
# 如果需要持续监控，取消下面注释
# alert_system.start_monitoring(interval=60)
文末福利
就目前来说，传统运维冲击年薪30W+的转型方向就是SRE&DevOps岗位。
为了帮助大家早日摆脱繁琐的基层运维工作，给大家整理了一套高级运维工程师必备技能资料包，内容有多详实丰富看下图！
共有 20 个模块
1.38张最全工程师技能图谱
2.面试大礼包
3.Linux书籍
4.go书籍
······
6.自动化运维工具
18.消息队列合集
以上所有资料获取请扫码
备注：最新运维资料
100%免费领取
（后台不再回复，扫码一键领取）
本文链接：https://www.yunweipai.com/47390.html
网友评论comments