怎样实现流量突增预警脚本

本文目录导读：

1. 基于Nginx日志的实时监控方案
2. 基于系统监控的轻量级方案
3. 基于API调用的监控方案
4. 配置文件示例 (config.yaml)
5. 部署建议
6. 优化建议

我来帮你设计一个流量突增预警脚本,包含多种实现方案：

基于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

优化建议

1. 数据处理优化：

   • 使用Redis存储实时数据
   • 采用时间序列数据库（如InfluxDB）
   • 使用异步IO处理大量日志

2. 算法优化：

   • 使用指数移动平均计算基线
   • 考虑周期性规律（日/周/月）
   • 异常检测算法（如3-sigma）

3. 性能优化：

   • 使用多线程处理不同来源数据
   • 批量处理日志条目
   • 缓存计算结果

4. 可观测性：

   • 集成Prometheus指标导出
   • 添加Grafana仪表板
   • 记录详细的告警日志

这个脚本可以根据你的具体需求进行调整,选择适合的监控方式和告警渠道。