本文目录导读：
我来帮你设计一个流量突增预警脚本,包含多种实现方案：
基于Nginx日志的实时监控方案
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
流量突增预警脚本 - 基于Nginx日志
"""
import re
import time
import logging
import smtplib
from collections import deque, Counter
from datetime import datetime, timedelta
from email.mime.text import MIMEText
import requests
import json
class TrafficMonitor:
def __init__(self, config):
self.config = config
self.log_path = config.get('log_path', '/var/log/nginx/access.log')
self.window_size = config.get('window_size', 60) # 窗口大小（秒）
self.baseline_ratio = config.get('baseline_ratio', 3) # 触发阈值倍数
self.check_interval = config.get('check_interval', 10) # 检查间隔（秒）
# 滑动窗口存储时间戳
self.timestamps = deque()
self.ip_counter = Counter()
# 告警状态
self.last_alert_time = 0
self.alert_cooldown = config.get('alert_cooldown', 300) # 告警冷却时间
# 日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('traffic_monitor.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def tail_log(self):
"""实时读取日志文件"""
with open(self.log_path, 'r') as f:
# 移动到文件末尾
f.seek(0, 2)
while True:
line = f.readline()
if line:
yield line
else:
time.sleep(0.1)
def parse_log_line(self, line):
"""解析日志行，提取时间戳和IP"""
# 适配常见Nginx日志格式
patterns = [
r'(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2})', # Nginx默认格式
r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})' # JSON格式
]
for pattern in patterns:
match = re.search(pattern, line)
if match:
try:
if '/' in match.group(1):
dt = datetime.strptime(match.group(1), '%d/%b/%Y:%H:%M:%S')
else:
dt = datetime.strptime(match.group(1), '%Y-%m-%d %H:%M:%S')
# 提取IP
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', line)
ip = ip_match.group(1) if ip_match else 'unknown'
return dt.timestamp(), ip
except:
continue
return None, None
def calculate_baseline(self):
"""计算基线流量（过去N分钟的平均值）"""
# 从数据库中或文件中读取历史数据
try:
# 示例：从Redis或文件读取过去一小时的统计数据
baseline = 100 # 单位：请求/分钟
return baseline
except:
return 100 # 默认值
def check_traffic_spike(self):
"""检测流量异常突增"""
current_time = time.time()
# 清理过期数据
while self.timestamps and current_time - self.timestamps[0] > self.window_size:
old_ts = self.timestamps.popleft()
# 清理对应IP的计数（简化处理）
# 获取当前窗口内的请求数
current_window_count = len(self.timestamps)
current_qps = current_window_count / self.window_size
# 获取基线
baseline_qps = self.calculate_baseline() / 60 # 转换为QPS
# 计算突增比例
if baseline_qps > 0:
increase_ratio = current_qps / baseline_qps
else:
increase_ratio = float('inf')
# 判断是否触发告警
if (increase_ratio >= self.baseline_ratio and
current_time - self.last_alert_time > self.alert_cooldown):
self.send_alert(current_qps, baseline_qps, increase_ratio)
self.last_alert_time = current_time
self.logger.warning(f"流量突增告警！倍数: {increase_ratio:.2f}x")
return current_qps, baseline_qps, increase_ratio
def process_logs(self):
"""主处理循环"""
self.logger.info("开始监控流量...")
last_check_time = 0
for line in self.tail_log():
timestamp, ip = self.parse_log_line(line)
if timestamp:
self.timestamps.append(timestamp)
# 定期检查流量
current_time = time.time()
if current_time - last_check_time >= self.check_interval:
current_qps, baseline_qps, ratio = self.check_traffic_spike()
# 输出状态信息
self.logger.info(
f"当前QPS: {current_qps:.2f}, "
f"基线QPS: {baseline_qps:.2f}, "
f"比率: {ratio:.2f}"
)
last_check_time = current_time
def send_alert(self, current_qps, baseline_qps, ratio):
"""发送告警通知"""
subject = f"[告警] 流量突增 - {ratio:.2f}倍"
body = f"""
流量突增告警！
当前QPS: {current_qps:.2f}
基线QPS: {baseline_qps:.2f}
异常比率: {ratio:.2f}x
触发阈值: {self.baseline_ratio}x
时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
# 发送邮件
if self.config.get('email_enabled'):
self._send_email(subject, body)
# 发送Slack通知
if self.config.get('slack_webhook'):
self._send_slack(body)
# 发送HTTP回调
self._send_webhook(body)
def _send_email(self, subject, body):
"""发送邮件告警"""
try:
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = self.config['email_from']
msg['To'] = self.config['email_to']
with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
server.starttls()
server.login(self.config['smtp_user'], self.config['smtp_password'])
server.send_message(msg)
except Exception as e:
self.logger.error(f"邮件发送失败: {e}")
def _send_slack(self, message):
"""发送Slack通知"""
try:
data = {'text': message}
requests.post(
self.config['slack_webhook'],
data=json.dumps(data),
headers={'Content-Type': 'application/json'}
)
except Exception as e:
self.logger.error(f"Slack发送失败: {e}")
def _send_webhook(self, data):
"""发送HTTP回调"""
if self.config.get('webhook_url'):
try:
requests.post(
self.config['webhook_url'],
json={'message': data, 'timestamp': datetime.now().isoformat()}
)
except Exception as e:
self.logger.error(f"Webhook发送失败: {e}")
# 配置文件示例
config = {
'log_path': '/var/log/nginx/access.log',
'window_size': 60, # 检查窗口（秒）
'baseline_ratio': 3, # 触发倍数
'check_interval': 10, # 检查间隔
'alert_cooldown': 300, # 告警冷却时间（秒）
# 告警方式
'email_enabled': False,
'email_from': 'monitor@example.com',
'email_to': 'admin@example.com',
'smtp_server': 'smtp.example.com',
'smtp_port': 587,
'smtp_user': 'monitor@example.com',
'smtp_password': 'your_password',
'slack_webhook': 'https://hooks.slack.com/services/xxx',
'webhook_url': 'http://your-webhook/api/alert'
}
if __name__ == "__main__":
monitor = TrafficMonitor(config)
monitor.process_logs()
基于系统监控的轻量级方案
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
流量监控脚本 - 基于系统网络接口
"""
import subprocess
import time
import json
from datetime import datetime
class NetworkMonitor:
def __init__(self, interface='eth0', threshold_mbps=100):
self.interface = interface
self.threshold_mbps = threshold_mbps # 触发阈值(Mbps)
def get_network_stats(self):
"""获取网络接口统计"""
with open(f'/sys/class/net/{self.interface}/statistics/rx_bytes', 'r') as f:
rx_bytes = int(f.read().strip())
with open(f'/sys/class/net/{self.interface}/statistics/tx_bytes', 'r') as f:
tx_bytes = int(f.read().strip())
return rx_bytes, tx_bytes
def calculate_bandwidth(self, interval=1):
"""计算带宽使用率"""
rx1, tx1 = self.get_network_stats()
time.sleep(interval)
rx2, tx2 = self.get_network_stats()
rx_speed = (rx2 - rx1) * 8 / interval / 1000000 # Mbps
tx_speed = (tx2 - tx1) * 8 / interval / 1000000 # Mbps
return rx_speed, tx_speed
def monitor(self):
"""监控循环"""
print(f"开始监控 {self.interface} 接口流量...")
print(f"触发阈值: {self.threshold_mbps} Mbps")
while True:
try:
rx_speed, tx_speed = self.calculate_bandwidth()
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
status = {
'time': timestamp,
'rx_mbps': round(rx_speed, 2),
'tx_mbps': round(tx_speed, 2),
'total_mbps': round(rx_speed + tx_speed, 2)
}
# 检查是否超过阈值
if rx_speed > self.threshold_mbps or tx_speed > self.threshold_mbps:
print(f"[告警] {json.dumps(status)} - 流量超过阈值!")
self.trigger_alert(status)
else:
print(f"[正常] {json.dumps(status)}")
except KeyboardInterrupt:
print("\n监控停止")
break
except Exception as e:
print(f"错误: {e}")
time.sleep(5)
def trigger_alert(self, data):
"""触发告警动作"""
# 可以添加钉钉/企业微信通知
# 记录日志
with open('traffic_alerts.log', 'a') as f:
f.write(json.dumps(data) + '\n')
# 使用示例
if __name__ == "__main__":
monitor = NetworkMonitor(
interface='eth0', # 根据实际情况修改
threshold_mbps=100 # 100Mbps阈值
)
monitor.monitor()
基于API调用的监控方案
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
流量突增监控 - 基于API每秒调用次数
"""
import time
import threading
from collections import defaultdict
from datetime import datetime
class APIMonitor:
def __init__(self, threshold_per_second=100, alert_cooldown=60):
self.threshold = threshold_per_second
self.alert_cooldown = alert_cooldown
self.request_counts = defaultdict(int)
self.last_alert_time = 0
self.lock = threading.Lock()
def record_request(self, endpoint='/api/data'):
"""记录一次API请求"""
current_second = int(time.time())
with self.lock:
self.request_counts[current_second] += 1
def check_traffic(self):
"""定期检查流量"""
while True:
time.sleep(1)
with self.lock:
current_second = int(time.time())
current_count = self.request_counts.get(current_second, 0)
# 清理旧数据（保留60秒）
old_seconds = [s for s in self.request_counts
if s < current_second - 60]
for s in old_seconds:
del self.request_counts[s]
# 检查当前QPS
if current_count > self.threshold:
now = time.time()
if now - self.last_alert_time > self.alert_cooldown:
self.trigger_alert(current_count, self.threshold)
self.last_alert_time = now
def trigger_alert(self, current, threshold):
"""触发告警"""
message = f"[告警] API请求突增! 当前: {current}/s, 阈值: {threshold}/s"
print(f"{datetime.now()}: {message}")
# 这里可以添加实际的通知逻辑
# send_email(message)
# send_slack(message)
def start_monitoring(self):
"""启动监控"""
monitor_thread = threading.Thread(target=self.check_traffic, daemon=True)
monitor_thread.start()
print(f"API监控已启动，阈值: {self.threshold} QPS")
# 使用示例
if __name__ == "__main__":
monitor = APIMonitor(threshold_per_second=100)
monitor.start_monitoring()
# 模拟API请求
try:
while True:
monitor.record_request('/api/endpoint')
time.sleep(0.01) # 模拟请求间隔
except KeyboardInterrupt:
print("监控停止")
配置文件示例 (config.yaml)
# 流量监控配置文件
traffic_monitor:
# 数据源配置
source: nginx_log # nginx_log | network_interface | api
# Nginx日志配置
log_path: /var/log/nginx/access.log
log_format: combined # combined | json | custom
# 网络接口配置
interface: eth0
threshold_mbps: 100
# API监控配置
api_endpoints:
- /api/v1/users
- /api/v1/orders
# 通用配置
check_interval: 10 # 检查间隔（秒）
window_size: 60 # 滑动窗口（秒）
baseline_ratio: 3 # 异常触发倍数
alert_cooldown: 300 # 告警冷却（秒）
# 基线配置
baseline:
method: historical # historical | static | dynamic
historical_window: 3600 # 历史窗口（秒）
static_value: 100 # 静态基线值（请求/分钟）
# 告警通知
notifications:
email:
enabled: false
smtp_server: smtp.gmail.com
smtp_port: 587
from: monitor@example.com
to: admin@example.com
username: your_email
password: your_password
slack:
enabled: true
webhook_url: https://hooks.slack.com/services/xxx
webhook:
enabled: false
url: http://your-system/api/alert
部署建议
作为systemd服务运行
# /etc/systemd/system/traffic-monitor.service [Unit] Description=Traffic Monitor Service After=network.target [Service] Type=simple User=root ExecStart=/usr/bin/python3 /opt/traffic-monitor/monitor.py Restart=always RestartSec=10 [Install] WantedBy=multi-user.target
启动服务
sudo systemctl enable traffic-monitor sudo systemctl start traffic-monitor sudo systemctl status traffic-monitor
优化建议
-
数据处理优化：
- 使用Redis存储实时数据
- 采用时间序列数据库（如InfluxDB）
- 使用异步IO处理大量日志
-
算法优化：
- 使用指数移动平均计算基线
- 考虑周期性规律（日/周/月）
- 异常检测算法（如3-sigma）
-
性能优化：
- 使用多线程处理不同来源数据
- 批量处理日志条目
- 缓存计算结果
-
可观测性：
- 集成Prometheus指标导出
- 添加Grafana仪表板
- 记录详细的告警日志
这个脚本可以根据你的具体需求进行调整,选择适合的监控方式和告警渠道。
