logo NodeSeekbeta
search for post search for people use google search

阿里云ECS流量超阈值后使用tg通知

zombie
zombie楼主
in 技术
#0

1. 方案特点与架构说明

在 512MB 甚至更小内存的 VPS 上，部署常规的监控告警组件（如 Prometheus Alertmanager、Zabbix、Netdata 云告警等）往往需要常驻数十甚至数百兆内存。

本告警系统基于 vnStat 2.x 的 JSON 接口 + 纯 Python 3 原生标准库 构建：

  • 零外部依赖：不依赖 requests 等第三方库，直接使用标准库 urllib.request。
  • 常驻内存 0 MB：基于 Crontab 定时调度，执行耗时通常低于 0.2 秒，执行完毕立即释放系统资源。
  • 三维精细化阈值监控：
    1. 月度配额告警：防止 VPS 流量包超额扣取高额带宽费用。
    2. 单日异常突增告警：及时发现非预期的大额数据传输。
    3. 5分钟瞬时洪峰告警：识别短时间内的恶意刷量、CC 攻击或高并发下载。
  • 内置防轰炸锁（State Deduplication）：本地持久化记录告警状态，同一周期的超额事件仅通知一次，避免邮箱或聊天机器人被消息刷屏。

2. 推送渠道参数获取

本系统支持多渠道推送，选择你最常用的一种即可。

渠道 A：Telegram Bot（最推荐，无国内域名白名单限制）

1. 创建 Bot 获取 TG_BOT_TOKEN
  1. 打开 Telegram，搜索官方账号 @BotFather（注意认证蓝标）。
  2. 点击 Start 或发送 /start。
  3. 输入 /newbot，按提示设置：
    • Bot 昵称（如：My Traffic Alert）
    • Bot 用户名（必须以 bot 结尾，如：myserver_traffic_bot）
  4. 创建成功后，获取格式如下的 Token：
    1234567890:ABCdefGhIJklmnOpQRstUVwxyz_1234567
2. 获取接收通知的 TG_CHAT_ID
  1. 关键前提：在 Telegram 中找到你刚创建的 Bot，主动给它发送一条消息或点击 /start（未建立会话时 Bot 无法向你发送消息）。
  2. 搜索机器人 @userinfobot 并点击 Start。
  3. 它会返回你的个人数字 ID（Id: 123456789），这串纯数字即为 TG_CHAT_ID。

渠道 B：国内主流办公软件 Webhook（飞书 / 钉钉 / 企业微信）

  • 飞书：群设置 -> 添加机器人 -> 自定义机器人 -> 复制 Webhook 地址。
  • 钉钉：群设置 -> 智能群助手 -> 添加机器人 -> 自定义 -> 安全设置选择“自定义关键词”（填入关键词如 流量）或加签 -> 复制 Webhook 地址。
  • 企业微信：群设置 -> 添加群机器人 -> 复制 Webhook 地址。

3. 告警核心脚本部署

3.1 创建脚本文件

在服务器上创建并编辑 /usr/local/bin/vnstat_alert.py：

cat << 'EOF' > /usr/local/bin/vnstat_alert.py
#!/usr/bin/env python3
import json
import os
import sys
import urllib.request
import urllib.parse
from datetime import datetime

# ==================== 1. 阈值配置 (单位: GB / MB) ====================
MONTH_LIMIT_GB = 800.0    # 当月累计流量超标阈值 (例如套餐 1000G，设为 800G)
TODAY_LIMIT_GB = 30.0     # 单日流量突增阈值 (今日已用总流量超限则触发)
FIVE_MIN_LIMIT_MB = 200.0 # 5分钟瞬时流量阈值 (5分钟区间超限则触发)

# ==================== 2. 通知推送渠道配置 (填写需要的项即可) ====================
TG_BOT_TOKEN = ""     # 例: "1234567890:ABCdefGhIJklmnOpQRstUVwxyz_1234567"
TG_CHAT_ID = ""       # 例: "987654321"

FEISHU_WEBHOOK = ""   # 飞书机器人 Webhook URL
DINGTALK_WEBHOOK = "" # 钉钉机器人 Webhook URL
WECHAT_WEBHOOK = ""   # 企业微信机器人 Webhook URL

# ==================== 3. 告警频率与状态控制 ====================
STATE_FILE = '/tmp/vnstat_alert_state.json'
VNSTAT_JSON = '/var/www/vnstat/api/vnstat.json'

def send_notification(title, message):
    full_msg = f"【服务器流量告警】{title}\n时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{message}"
    print(f"Triggering alert:\n{full_msg}")
    
    # 1. Telegram
    if TG_BOT_TOKEN and TG_CHAT_ID:
        try:
            url = f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage"
            payload = json.dumps({"chat_id": TG_CHAT_ID, "text": full_msg}).encode('utf-8')
            req = urllib.request.Request(url, data=payload, headers={'Content-Type': 'application/json'})
            urllib.request.urlopen(req, timeout=10)
            print("Telegram 发送成功")
        except Exception as e:
            print(f"Telegram 推送失败: {e}")

    # 2. 飞书
    if FEISHU_WEBHOOK:
        try:
            payload = json.dumps({"msg_type": "text", "content": {"text": full_msg}}).encode('utf-8')
            req = urllib.request.Request(FEISHU_WEBHOOK, data=payload, headers={'Content-Type': 'application/json'})
            urllib.request.urlopen(req, timeout=10)
            print("飞书发送成功")
        except Exception as e:
            print(f"飞书推送失败: {e}")

    # 3. 钉钉
    if DINGTALK_WEBHOOK:
        try:
            payload = json.dumps({"msgtype": "text", "text": {"content": full_msg}}).encode('utf-8')
            req = urllib.request.Request(DINGTALK_WEBHOOK, data=payload, headers={'Content-Type': 'application/json'})
            urllib.request.urlopen(req, timeout=10)
            print("钉钉发送成功")
        except Exception as e:
            print(f"钉钉推送失败: {e}")

    # 4. 企业微信
    if WECHAT_WEBHOOK:
        try:
            payload = json.dumps({"msgtype": "text", "text": {"content": full_msg}}).encode('utf-8')
            req = urllib.request.Request(WECHAT_WEBHOOK, data=payload, headers={'Content-Type': 'application/json'})
            urllib.request.urlopen(req, timeout=10)
            print("企业微信发送成功")
        except Exception as e:
            print(f"企业微信推送失败: {e}")

def load_state():
    if os.path.exists(STATE_FILE):
        try:
            with open(STATE_FILE, 'r') as f:
                return json.load(f)
        except Exception:
            return {}
    return {}

def save_state(state):
    with open(STATE_FILE, 'w') as f:
        json.dump(state, f)

def check_traffic():
    if not os.path.exists(VNSTAT_JSON):
        print(f"错误: 找不到文件 {VNSTAT_JSON}")
        return
    
    with open(VNSTAT_JSON, 'r') as f:
        data = json.load(f)
    
    interfaces = data.get('interfaces', [])
    if not interfaces:
        print("未在 vnstat.json 中找到有效 interfaces 数据")
        return

    iface = interfaces[0]
    traffic = iface.get('traffic', {})
    state = load_state()
    today_str = datetime.now().strftime('%Y-%m-%d')
    month_str = datetime.now().strftime('%Y-%m')

    # A. 检查当月流量
    months = traffic.get('month', [])
    if months:
        curr_m = months[-1]
        m_total_gb = (curr_m['rx'] + curr_m['tx']) / (1024 ** 3)
        if m_total_gb >= MONTH_LIMIT_GB:
            alert_key = f"month_{month_str}"
            if not state.get(alert_key):
                send_notification("月度流量超额告警", f"本月({month_str})累计已消耗流量: {m_total_gb:.2f} GB (设定的阈值: {MONTH_LIMIT_GB} GB)")
                state[alert_key] = True

    # B. 检查今日流量
    days = traffic.get('day', [])
    if days:
        curr_d = days[-1]
        d_total_gb = (curr_d['rx'] + curr_d['tx']) / (1024 ** 3)
        if d_total_gb >= TODAY_LIMIT_GB:
            alert_key = f"day_{today_str}"
            if not state.get(alert_key):
                send_notification("单日流量突增告警", f"今日({today_str})累计已消耗流量: {d_total_gb:.2f} GB (设定的阈值: {TODAY_LIMIT_GB} GB)")
                state[alert_key] = True

    # C. 检查 5 分钟级别突发流量
    fivemins = traffic.get('fiveminute', [])
    if fivemins:
        latest_fm = fivemins[-1]
        fm_total_mb = (latest_fm['rx'] + latest_fm['tx']) / (1024 ** 2)
        fm_time_key = f"{latest_fm['date']['year']}-{latest_fm['date']['month']}-{latest_fm['date']['day']} {latest_fm['time']['hour']}:{latest_fm['time']['minute']}"
        if fm_total_mb >= FIVE_MIN_LIMIT_MB:
            alert_key = f"5m_{fm_time_key}"
            if not state.get(alert_key):
                send_notification("突发网络洪峰告警 (5分钟)", f"区间 [{fm_time_key}] 内消耗流量: {fm_total_mb:.2f} MB (阈值: {FIVE_MIN_LIMIT_MB} MB)")
                state[alert_key] = True

    save_state(state)
    print("流量检查完毕，当前指标未触发告警。")

if __name__ == '__main__':
    check_traffic()
EOF

chmod +x /usr/local/bin/vnstat_alert.py

3.2 权限与执行验证

赋予执行权限后进行初次执行：

/usr/local/bin/vnstat_alert.py

正常情况下会输出：

流量检查完毕，当前指标未触发告警。

4. 自动化调度配置（Crontab）

告警检查必须紧跟在 vnstat 的 JSON 数据更新之后。建议设置为 每 2 分钟检查一次：

执行：

crontab -e

将任务定义配置为链式触发：

# 每 2 分钟先导出最新流量数据，然后立即执行告警核对
*/2 * * * * /usr/bin/vnstat --json > /var/www/vnstat/api/vnstat.json 2>/dev/null && /usr/local/bin/vnstat_alert.py >/dev/null 2>&1

5. 告警全链路联调与维护

5.1 模拟告警测试（端到端验证）

要验证网络连通性以及通知是否能送达客户端，无需等待真实流量跑满：

  1. 编辑 /usr/local/bin/vnstat_alert.py，将单日阈值调至极小值：
    TODAY_LIMIT_GB = 0.0001
  2. 手动运行脚本触发报警：
    /usr/local/bin/vnstat_alert.py
  3. 检查手机 Telegram / 办公工具是否成功收到消息。
  4. 清理测试状态：
    测试成功后，恢复 TODAY_LIMIT_GB 的真实值，并清理临时去重缓存文件：
    rm -f /tmp/vnstat_alert_state.json

5.2 状态持久化机制解析

告警状态保存在 /tmp/vnstat_alert_state.json 中，结构如下：

{
  "day_2026-09-28": true,
  "month_2026-09": true,
  "5m_2026-9-28 19:45": true
}
  • 按天重置：次日日期变更（如变为 2026-09-29），旧键不再匹配，自动开启新一天的监控。
  • 按月重置：次月月份变更（如变为 2026-10），自动进入下一周期。
  • 紧急重置：如果想让已经报警过的周期在调整阈值后重新激活告警，只需手动执行 rm -f /tmp/vnstat_alert_state.json。
  • nyrvs
    nyrvs
    edited 2h 5min ago
    #1

    好活，支持一下，🍗奉上

  • roy
    roy
    #2

    零依赖好评👍

  • 3vpcowner
    3vpcowner
    #3

    支持一下。要是能做到不直接部署到阿里云的面板上就好了

  • snnae
    snnae
    #4

    阿里的CDT的流量是几号重置的？是固定1号吗

你好啊，陌生人!

我的朋友，看起来你是新来的，如果想参与到讨论中，点击下面的按钮！
登录 注册

快捷功能区

所有版块

📈用户数目📈

目前论坛共有72088位seeker

🎉欢迎新用户🎉

suzen
suzen
larita
larita
y5866y
y5866y
owoshuangshi
owoshuangshi
soulgoodman
soulgoodman
WaterApple
WaterApple
qingby
qingby
Hruodland
Hruodland