迷失博客 - 分享技术

AWS Lightsail 提供了极具性价比的轻量云服务器套餐，并且每个套餐都自带了丰厚的月度网络传输配额（通常为 1TB ~ 5TB 出方向流量）。

然而，一旦实例遭受网络攻击、异常跑流量或业务突发导致流量超出月度配额，AWS 将按照每 GB 额外计费（通常为 $0.09 ~ $0.12 / GB），极易产生高昂的账单风险。此外，Lightsail 原生监控仅提供简单的邮件提醒，无法在流量超标时自动关机止损，也无法在每月 1 号重置配额时自动拉起实例。

本文将分享一套经过生产验证的 AWS Serverless 全自动监控方案：无需在每台 VPS 安装任何探针或 Agent，完全利用 AWS 官方的原生服务（EventBridge + Lambda + SSM Parameter Store），实现全区域 Lightsail 实例流量巡检、超标自动关机、多渠道 Webhook/企业微信推送以及每月 1 号自动重置开机。

架构设计与拓扑

整个系统采用 100% 无服务器（Serverless）架构，运行时仅按调用次数计费（每月在 AWS 免费额度内，完全 0 成本运行）：

+------------------------+
| Amazon EventBridge     | (建议每 5 分钟高频定时触发)
| (Cron / Rate Rule)     |
+-----------+------------+
            |
            v
+-----------+------------+       +----------------------------+
| AWS Lambda 函数        | ----> | AWS Lightsail API          |
| (Python 3.12 核心计算) |       | (获取配额/统计指标/启停机)   |
+-----------+------------+       +----------------------------+
            |
            +------------------> +----------------------------+
            |                    | AWS SSM Parameter Store    |
            |                    | (持久化已停机/告警状态记录) |
            |                    +----------------------------+
            |
            +------------------> +----------------------------+
                                 | Webhook 消息网关 / 企微/SNS |
                                 | (即时推送告警与汇总日报)   |
                                 +----------------------------+

核心设计亮点

零受控端侵入（Agentless）：完全基于 AWS IAM 鉴权通过云端 API 交互，不在目标 Lightsail 机器内部安装任何软件，机器重装、改系统亦不受任何影响；
当月新建机器配额智能折算：对于月中新建的机器，脚本会自动计算当月剩余有效天数比例并折算实际可用配额，彻底杜绝月中新建机器误触发超标；
出入双向流量精准聚合：自动按天聚合查询 NetworkIn 与 NetworkOut 数据点求和，精准比对当月总配额；
防重复打扰与幂等关机：关机后将实例写入 AWS SSM Parameter Store，对于已经处于停止状态的实例保持静默，不反复推送骚扰通知；
每月 1 号自动重置与唤醒（带去重防抖）：北京时间每月 1 号 00:00，自动清空历史告警记录；对于上月因超标被系统关机且当月新流量未超标的实例，自动执行开机恢复服务，并写入 SSM 日期防重标记，杜绝高频触发下的重复通知轰炸；
每月首尾日全量用量报表（带当日去重）：在月初与月末自动汇总账号下所有 Lightsail 实例的流量消耗详情并推送全局报表，每日仅推送一次；
WAF 与 Cloudflare 防火墙兼容：内置标准浏览器 User-Agent，彻底避免 Python 默认 urllib 请求头被 Cloudflare WAF / 边缘防火墙拦截（如 Error 1010 / 403 Forbidden）。

接口与环境变量参数规范

Lambda 函数所有配置均采用环境变量驱动，便于在 AWS 控制台灵活调整：

环境变量名	类型	默认值	说明
`AWS_REGION`	String	`ap-southeast-1`	Lightsail 实例所在区域（与 Lambda 区域保持一致）
`ALERT_THRESHOLD`	Float	`0.95`	流量超标预警阈值（例如 `0.96` 代表达到 96% 即触发关机）
`WECHAT_NOTIFY_URL`	String	(可选)	Webhook 消息通知地址（含 Token，留空则不发送通知仅执行关机）
`SNS_TOPIC_ARN`	String	(可选)	AWS SNS 主题 ARN（留空则不使用 SNS 推送）
`METRIC_PERIOD`	Integer	`86400`	CloudWatch 指标聚合周期（默认 86400 秒即按天聚合）

5 步极速搭建实战

第一步：创建 IAM 权限角色

进入 AWS IAM 控制台 -> Roles (角色) -> Create role；
实体类型选择 AWS service，用例选择 Lambda；
创建自定义权限策略（Policy），策略 JSON 如下（仅赋予最小必需权限）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "LightsailOperations",
      "Effect": "Allow",
      "Action": [
        "lightsail:GetInstances",
        "lightsail:GetInstance",
        "lightsail:GetInstanceMetricData",
        "lightsail:StartInstance",
        "lightsail:StopInstance"
      ],
      "Resource": "*"
    },
    {
      "Sid": "SSMParameterAccess",
      "Effect": "Allow",
      "Action": [
        "ssm:GetParameter",
        "ssm:PutParameter",
        "ssm:DeleteParameter"
      ],
      "Resource": "arn:aws:ssm:*:*:parameter/lightsail/*"
    },
    {
      "Sid": "SNSPublish",
      "Effect": "Allow",
      "Action": [
        "sns:Publish"
      ],
      "Resource": "*"
    },
    {
      "Sid": "CloudWatchLogs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    }
  ]
}

将该策略命名为 LightsailTrafficMonitorPolicy 并绑定到新建的 Role（例如 LightsailMonitorRole）。

第二步：初始化 SSM Parameter Store 参数

进入 AWS Systems Manager -> Parameter Store，在目标区域创建两个初始参数（其余去重参数会在首次运行时由代码自动创建与维护）：

参数一（存储超标关机实例列表）：
- Name: /lightsail/stopped-by-quota
- Tier: Standard
- Type: StringList（或 String）
- Value: []
参数二（存储当前告警中实例状态字典）：
- Name: /lightsail/alerted-instances
- Tier: Standard
- Type: String
- Value: {}
参数三（可选，用于月初/月末汇总推送防重复打标）：
- Name: /lightsail/summary-last-date
- Type: String
- Value: 2026-01-01
参数四（可选，用于每月 1 号开机恢复防重复打标）：
- Name: /lightsail/start-last-date
- Type: String
- Value: 2026-01-01

第三步：创建并配置 Lambda 函数

打开 AWS Lambda 控制台，点击 Create function；
选择 Author from scratch：
- Function name: lightsail-traffic-monitor
- Runtime: Python 3.12
- Architecture: arm64（或 x86_64）
- Execution role: 选择刚刚创建的 LightsailMonitorRole；
进入 Configuration -> General configuration：
- Timeout (超时时间) 设置为 1 分钟 (60 秒)；
- Memory (内存) 设置为 128 MB；
进入 Configuration -> Environment variables，添加环境变量：
- ALERT_THRESHOLD: 0.96
- WECHAT_NOTIFY_URL: https://sms.example.com/message?token=YOUR_TOKEN
- AWS_REGION: ap-southeast-1

第四步：部署 Python 完整代码

在 Lambda 代码编辑器中，将 lambda_function.py 的内容替换为以下完整源码并点击 Deploy：

import json
import boto3
import calendar
import os
import time
from datetime import datetime, date, time as dt_time, timedelta
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError

# ------------------------------ 全局常量定义（纯环境变量驱动，无硬编码）------------------------------
# AWS Lightsail 流量监控与自动关机
AWS_REGION = os.getenv('AWS_REGION', os.getenv('AWS_DEFAULT_REGION', 'ap-southeast-1'))

# 环境变量读取配置
ALERT_THRESHOLD = float(os.getenv('ALERT_THRESHOLD', '0.95'))  # 超标阈值（默认 95%）
SNS_TOPIC_ARN = os.getenv('SNS_TOPIC_ARN', '')                # 可选 SNS 通知 ARN
METRIC_PERIOD = int(os.getenv('METRIC_PERIOD', '86400'))        # 指标聚合周期（秒，默认 1 天）
WECHAT_NOTIFY_URL = os.getenv('WECHAT_NOTIFY_URL', '')          # 消息通知 Webhook URL

# SSM 参数名称（持久化存储被停 / 已告警实例 / 汇总及开机防重标记）
SSM_PARAM_STOPPED = os.getenv('SSM_PARAM_STOPPED', '/lightsail/stopped-by-quota')
SSM_PARAM_ALERTED = os.getenv('SSM_PARAM_ALERTED', '/lightsail/alerted-instances')
SSM_PARAM_SUMMARY_DATE = os.getenv('SSM_PARAM_SUMMARY_DATE', '/lightsail/summary-last-date')
SSM_PARAM_START_DATE = os.getenv('SSM_PARAM_START_DATE', '/lightsail/start-last-date')

# 单位转换常量
GB_TO_BYTE = 10**9
MB_TO_BYTE = 1024 * 1024

# 实例状态常量
INSTANCE_STATE_STOPPED = 'stopped'
INSTANCE_STATE_RUNNING = 'running'
INSTANCE_STATE_STARTING = 'starting'
INSTANCE_STATE_STOPPING = 'stopping'

# 告警状态常量
ALERT_STATUS_TRIGGERED = 'triggered'
NOTIFY_RETRY_TIMES = 2

# ------------------------------ 全局客户端复用 ------------------------------
lightsail_client = boto3.client('lightsail', region_name=AWS_REGION)
ssm_client = boto3.client('ssm', region_name=AWS_REGION)

# ------------------------------ 通用辅助函数 ------------------------------
def json_serialize(obj):
    """通用 JSON 序列化，处理 boto3 响应对象"""
    return json.dumps(obj, default=str)

def update_ssm_instance_list(instance_name, operation='add', ssm_param=SSM_PARAM_STOPPED):
    """通用 SSM 列表更新函数，支持并发重试与状态清空"""
    try:
        # 每月 1 号清空整个 SSM 参数
        if instance_name == '' and operation == 'remove':
            ssm_client.put_parameter(
                Name=ssm_param,
                Value='{}' if ssm_param == SSM_PARAM_ALERTED else '[]',
                Type='StringList' if ssm_param == SSM_PARAM_STOPPED else 'String',
                Overwrite=True
            )
            print(f"SSM 参数 {ssm_param} 已全部清空（配额重置）")
            return True
        
        for _ in range(3):
            try:
                response = ssm_client.get_parameter(Name=ssm_param, WithDecryption=False)
                ssm_data = json.loads(response['Parameter']['Value'])
            except ssm_client.exceptions.ParameterNotFound:
                ssm_data = {} if ssm_param == SSM_PARAM_ALERTED else []
            except json.JSONDecodeError:
                ssm_data = {} if ssm_param == SSM_PARAM_ALERTED else []

            origin_data = ssm_data.copy()
            if ssm_param == SSM_PARAM_STOPPED:
                if operation == 'add' and instance_name not in ssm_data:
                    ssm_data.append(instance_name)
                    print(f"实例 {instance_name} 已添加到 SSM 被停列表")
                elif operation == 'remove' and instance_name in ssm_data:
                    ssm_data.remove(instance_name)
                    print(f"实例 {instance_name} 已从 SSM 被停列表移除")
                else:
                    return True
            elif ssm_param == SSM_PARAM_ALERTED:
                if operation == 'add':
                    ssm_data[instance_name] = ALERT_STATUS_TRIGGERED
                    print(f"实例 {instance_name} 已标记为 SSM 已告警")
                elif operation == 'remove' and instance_name in ssm_data:
                    del ssm_data[instance_name]
                    print(f"实例 {instance_name} 已从 SSM 告警列表移除")
                else:
                    return True

            if origin_data == ssm_data:
                return True
            
            ssm_client.put_parameter(
                Name=ssm_param,
                Value=json_serialize(ssm_data),
                Type='StringList' if ssm_param == SSM_PARAM_STOPPED else 'String',
                Overwrite=True
            )
            return True
        return False
    except Exception as e:
        print(f"更新 SSM 列表失败：{str(e)}")
        return False

def get_general_instance_usage(instance_name, start_time, end_time, data_types=None):
    """通用流量查询函数，支持单/多指标查询"""
    if data_types is None:
        data_types = ['NetworkIn', 'NetworkOut']
    start_time_str = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
    end_time_str = end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
    total_usage = 0.0

    for data_type in data_types:
        try:
            response = lightsail_client.get_instance_metric_data(
                instanceName=instance_name,
                metricName=data_type,
                period=METRIC_PERIOD,
                unit='Bytes',
                statistics=['Sum'],
                startTime=start_time_str,
                endTime=end_time_str
            )
            metric_data = response.get('metricData', [])
            if not metric_data:
                continue
            metric_sum = sum(dp.get('sum', 0.0) for dp in metric_data)
            total_usage += metric_sum
        except Exception as e:
            print(f"获取实例 {instance_name} {data_type} 指标失败：{str(e)}")
    
    print(f"实例 {instance_name} 流量使用量：{total_usage:.2f} Byte（{total_usage/GB_TO_BYTE:.2f} GB）")
    return total_usage

def get_alerted_instances():
    """获取已告警实例列表，用于判断是否重复通知"""
    try:
        response = ssm_client.get_parameter(Name=SSM_PARAM_ALERTED, WithDecryption=False)
        alerted_instances = json.loads(response['Parameter']['Value'])
        return alerted_instances if isinstance(alerted_instances, dict) else {}
    except (ssm_client.exceptions.ParameterNotFound, json.JSONDecodeError):
        return {}

def retry_notify(func):
    """通知重试装饰器，网络抖动时自动重试 2 次"""
    def wrapper(title, message):
        for retry in range(NOTIFY_RETRY_TIMES + 1):
            try:
                result = func(title, message)
                if result:
                    return True
                time.sleep(1)
            except Exception as e:
                print(f"通知重试第 {retry+1} 次异常：{str(e)}")
                time.sleep(1)
        return False
    return wrapper

# AWS Lightsail 流量监控与自动关机
def stop_instance(instance_name):
    """停止实例，带幂等性校验"""
    instance_state = get_instance_state(instance_name)
    if instance_state not in [INSTANCE_STATE_RUNNING, INSTANCE_STATE_STARTING]:
        print(f"实例 {instance_name} 当前状态 {instance_state}，无需停止")
        return True
    try:
        response = lightsail_client.stop_instance(instanceName=instance_name, force=True)
        print(f"实例 {instance_name} 停止成功：{json_serialize(response)}")
        update_ssm_instance_list(instance_name, operation='add', ssm_param=SSM_PARAM_STOPPED)
        return True
    except Exception as e:
        print(f"实例 {instance_name} 停止失败：{str(e)}")
        return False

def start_instance(instance_name):
    """开启实例，带幂等性校验"""
    instance_state = get_instance_state(instance_name)
    if instance_state != INSTANCE_STATE_STOPPED:
        print(f"实例 {instance_name} 当前状态 {instance_state}，无需开启")
        return True
    try:
        response = lightsail_client.start_instance(instanceName=instance_name)
        print(f"实例 {instance_name} 开启成功：{json_serialize(response)}")
        return True
    except Exception as e:
        print(f"实例 {instance_name} 开启失败：{str(e)}")
        return False

def list_instances():
    """获取所有 Lightsail 实例名称"""
    instances_list = []
    try:
        paginator = lightsail_client.get_paginator('get_instances')
        for page in paginator.paginate(PaginationConfig={'MaxItems': 50}):
            instances_list.extend([instance.get('name') for instance in page.get('instances', []) if instance.get('name')])
        return instances_list
    except Exception as e:
        print(f"获取实例列表失败：{str(e)}")
        return []

def get_instance_state(instance_name):
    """获取实例当前状态"""
    try:
        response = lightsail_client.get_instance(instanceName=instance_name)
        return response.get('instance', {}).get('state', {}).get('name', 'unknown').lower()
    except Exception as e:
        print(f"获取实例 {instance_name} 状态失败：{str(e)}")
        return 'unknown'

def get_month_dto_quota(instance_name):
    """计算实例当月网络配额"""
    try:
        response = lightsail_client.get_instance(instanceName=instance_name)
        instance = response.get('instance', {})
        base_quota_gb = instance.get('networking', {}).get('monthlyTransfer', {}).get('gbPerMonthAllocated', 0.0)
        
        if base_quota_gb <= 0:
            return 0.0
        
        beijing_today = get_beijing_current_date()
        instance_created_datetime = instance.get('createdAt')
        if instance_created_datetime:
            created_bj = instance_created_datetime + timedelta(hours=8)
            if not (created_bj.year == beijing_today.year and created_bj.month == beijing_today.month):
                return base_quota_gb
        else:
            return base_quota_gb
        
        first_day_zero = get_current_month_first_day_zero_time()
        last_day_last = get_current_month_last_day_last_time()
        month_ts = last_day_last.timestamp() - first_day_zero.timestamp()
        instance_valide_ts = last_day_last.timestamp() - instance_created_datetime.timestamp()
        
        if month_ts <= 0 or instance_valide_ts <= 0:
            return base_quota_gb
        
        return min((instance_valide_ts / month_ts) * base_quota_gb, base_quota_gb)
    except Exception as e:
        print(f"计算实例 {instance_name} 配额失败：{str(e)}")
        return 0.0

def get_instance_current_month_usage(instance_name):
    """获取实例当月（北京时间当月 1 号 0 点至当前 UTC）累计流量使用量"""
    start_time = get_current_month_first_day_zero_time()
    end_time = datetime.utcnow()
    if end_time < start_time:
        end_time = start_time
    return get_general_instance_usage(instance_name, start_time, end_time)

# ------------------------------ 通知推送函数 ------------------------------
def push_notification(msg):
    """SNS 告警通知"""
    if not SNS_TOPIC_ARN:
        return False
    try:
        sns_client = boto3.client('sns', region_name=AWS_REGION)
        sns_client.publish(TopicArn=SNS_TOPIC_ARN, Message=msg, Subject='Lightsail 流量配额告警')
        return True
    except Exception as e:
        print(f"SNS 通知发送失败：{str(e)}")
        return False

@retry_notify
def send_wechat_notification(title, message):
    """消息推送通知（带重试，URL 从环境变量读取，带 User-Agent 防拦截）"""
    if not WECHAT_NOTIFY_URL:
        print("未配置 WECHAT_NOTIFY_URL 环境变量，跳过消息通知")
        return False
    try:
        payload_bytes = json.dumps({
            "title": title,
            "content": message,
            "message": message
        }, ensure_ascii=False).encode('utf-8')
        request = Request(
            url=WECHAT_NOTIFY_URL,
            data=payload_bytes,
            headers={
                "Content-Type": "application/json",
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
            },
            method='POST'
        )
        with urlopen(request, timeout=10) as response:
            res_text = response.read().decode('utf-8')
            print(f"消息通知发送成功，响应：{res_text}")
        return True
    except (HTTPError, URLError) as e:
        print(f"消息通知发送失败：{str(e)}")
        return False
    except Exception as e:
        print(f"消息通知发送失败（未知错误）：{str(e)}")
        return False

# ------------------------------ 时间工具函数（北京时间与 UTC 精确对齐）------------------------------
def get_beijing_now():
    return datetime.utcnow() + timedelta(hours=8)

def get_beijing_current_date():
    return get_beijing_now().date()

def get_current_month_first_day_zero_time():
    """获取北京时间当月 1 号 00:00:00 对应的 UTC datetime（供 CloudWatch/Lightsail metric 查询）"""
    beijing_today = get_beijing_current_date()
    beijing_first_day = datetime(beijing_today.year, beijing_today.month, 1, 0, 0, 0)
    return beijing_first_day - timedelta(hours=8)

def get_current_month_last_day_last_time():
    """获取北京时间当月最后一天 23:59:59 对应的 UTC datetime"""
    beijing_today = get_beijing_current_date()
    last_day_num = calendar.monthrange(beijing_today.year, beijing_today.month)[1]
    beijing_last_day = datetime(beijing_today.year, beijing_today.month, last_day_num, 23, 59, 59)
    return beijing_last_day - timedelta(hours=8)

def is_beijing_month_first_day():
    return get_beijing_current_date().day == 1

def is_beijing_month_first_or_last_day():
    beijing_today = get_beijing_current_date()
    last_day_of_month = calendar.monthrange(beijing_today.year, beijing_today.month)[1]
    return beijing_today.day in (1, last_day_of_month)

def has_summary_sent_today():
    """检查今天是否已推送过汇总通知（防 5 分钟高频重复推送）"""
    try:
        response = ssm_client.get_parameter(Name=SSM_PARAM_SUMMARY_DATE, WithDecryption=False)
        last_date_str = response['Parameter']['Value'].strip()
        today_str = get_beijing_current_date().strftime('%Y-%m-%d')
        if last_date_str == today_str:
            print(f"今日汇总已推送过（{today_str}），跳过")
            return True
        return False
    except ssm_client.exceptions.ParameterNotFound:
        return False
    except Exception as e:
        print(f"检查汇总推送标记失败：{str(e)}")
        return False

def mark_summary_sent_today():
    """标记今天已推送汇总通知"""
    try:
        today_str = get_beijing_current_date().strftime('%Y-%m-%d')
        ssm_client.put_parameter(
            Name=SSM_PARAM_SUMMARY_DATE,
            Value=today_str,
            Type='String',
            Overwrite=True
        )
        print(f"已标记今日汇总推送完成：{today_str}")
        return True
    except Exception as e:
        print(f"标记汇总推送失败：{str(e)}")
        return False

def has_month_first_started_today():
    """检查今天（每月 1 号）是否已执行过开机流程（防 5 分钟高频重复触发）"""
    try:
        response = ssm_client.get_parameter(Name=SSM_PARAM_START_DATE, WithDecryption=False)
        last_date_str = response['Parameter']['Value'].strip()
        today_str = get_beijing_current_date().strftime('%Y-%m-%d')
        if last_date_str == today_str:
            print(f"今日 1 号开机流程已执行过（{today_str}），跳过")
            return True
        return False
    except ssm_client.exceptions.ParameterNotFound:
        return False
    except Exception as e:
        print(f"检查开机执行标记失败：{str(e)}")
        return False

def mark_month_first_started_today():
    """标记今日 1 号开机流程已完成"""
    try:
        today_str = get_beijing_current_date().strftime('%Y-%m-%d')
        ssm_client.put_parameter(
            Name=SSM_PARAM_START_DATE,
            Value=today_str,
            Type='String',
            Overwrite=True
        )
        print(f"已标记今日 1 号开机完成：{today_str}")
        return True
    except Exception as e:
        print(f"标记开机完成失败：{str(e)}")
        return False

def get_all_instances_summary(instance_names):
    """生成所有实例流量使用汇总"""
    summary = [
        "=== Lightsail 实例当月网络流量汇总 ===",
        f"统计日期：{get_beijing_current_date().strftime('%Y-%m-%d')}（北京时间）",
        ""
    ]
    for instance_name in instance_names:
        quota_gb = get_month_dto_quota(instance_name)
        quota_byte = quota_gb * GB_TO_BYTE
        if quota_byte <= 0:
            summary.append(f"实例名称：{instance_name} - 无有效配额，跳过统计\n")
            continue
        total_usage_byte = get_instance_current_month_usage(instance_name)
        usage_percent = (total_usage_byte / quota_byte) * 100 if quota_byte > 0 else 0.0
        instance_info = (
            f"实例名称：{instance_name}\n"
            f"  当月用量：{total_usage_byte/GB_TO_BYTE:.2f} GB\n"
            f"  配额总量：{quota_gb:.2f} GB\n"
            f"  当前占比：{usage_percent:.2f} %\n"
            f"  告警阈值：{ALERT_THRESHOLD * 100:.1f} %\n"
        )
        summary.append(instance_info)
    return "\n".join(summary)

# ------------------------------ 每月 1 号自动开机逻辑 ------------------------------
def start_stopped_instances_on_month_first_day():
    """每月 1 号自动开启被限额停止且当月流量未超标的实例"""
    if not is_beijing_month_first_day():
        return
    
    if has_month_first_started_today():
        return

    print("=== 开始执行每月 1 号实例开启逻辑 ===")
    
    try:
        response = ssm_client.get_parameter(Name=SSM_PARAM_STOPPED, WithDecryption=False)
        stopped_instances = json.loads(response['Parameter']['Value'])
    except (ssm_client.exceptions.ParameterNotFound, json.JSONDecodeError):
        stopped_instances = []
    
    # 清空告警记录，允许新月份重新告警
    update_ssm_instance_list('', operation='remove', ssm_param=SSM_PARAM_ALERTED)
    
    if not stopped_instances:
        print("SSM 中无被限额停止的实例，无需开启")
        mark_month_first_started_today()
        return
    
    start_results = []
    for instance_name in list(stopped_instances):
        instance_state = get_instance_state(instance_name)
        if instance_state != INSTANCE_STATE_STOPPED:
            start_results.append(f"实例 {instance_name}：当前状态 {instance_state}，跳过")
            update_ssm_instance_list(instance_name, operation='remove', ssm_param=SSM_PARAM_STOPPED)
            continue
        
        total_usage = get_instance_current_month_usage(instance_name)
        # 新月份当月用量只要低于 100MB（或正常未超标），执行开启
        if total_usage > 100 * MB_TO_BYTE:
            start_results.append(f"实例 {instance_name}：当月用量 {total_usage/MB_TO_BYTE:.2f}MB > 100MB，跳过开启")
            continue
        
        start_success = start_instance(instance_name)
        if start_success:
            update_ssm_instance_list(instance_name, operation='remove', ssm_param=SSM_PARAM_STOPPED)
            start_results.append(f"实例 {instance_name}：开启成功")
        else:
            start_results.append(f"实例 {instance_name}：开启失败")
    
    summary_msg = "\n".join([
        "=== 每月 1 号 Lightsail 实例开启执行结果 ===",
        f"执行时间：{get_beijing_current_date().strftime('%Y-%m-%d')}（北京时间）",
        f"待处理实例数：{len(stopped_instances)}",
        "执行结果：",
    ] + start_results)
    send_wechat_notification("每月 1 号 Lightsail 实例开启结果通知", summary_msg)
    mark_month_first_started_today()

# ------------------------------ Lambda 入口主函数 ------------------------------
def lambda_handler(event, context):
    try:
        # 1. 每月 1 号开机与配额重置检查
        start_stopped_instances_on_month_first_day()
        
        # 2. 获取所有实例
        instance_names = list_instances()
        if not instance_names:
            return {'statusCode': 200, 'body': json_serialize('No Lightsail instances found!')}
        
        # 3. 获取已告警实例列表
        alerted_instances = get_alerted_instances()

        # 4. 逐个检查实例流量
        for instance_name in instance_names:
            quota_gb = get_month_dto_quota(instance_name)
            quota_byte = quota_gb * GB_TO_BYTE
            if quota_byte <= 0:
                continue
            
            total_usage_byte = get_instance_current_month_usage(instance_name)
            usage_percent = (total_usage_byte / quota_byte) * 100 if quota_byte > 0 else 0.0
            is_over_quota = total_usage_byte > (quota_byte * ALERT_THRESHOLD)
            is_alerted = instance_name in alerted_instances
            current_state = get_instance_state(instance_name)
            is_running = current_state in [INSTANCE_STATE_RUNNING, INSTANCE_STATE_STARTING]

            print(f"实例: {instance_name}, 超标: {is_over_quota}, 已标记告警: {is_alerted}, 运行中: {is_running}")

            # 超标停机与通知逻辑
            if is_over_quota:
                if is_running:
                    stop_result = stop_instance(instance_name)
                    action_type = "首次超标自动停机" if not is_alerted else "超标保护（检测到运行中，再次停机）"
                    alert_msg = (
                        f"实例名称：{instance_name}\n"
                        f"触发类型：{action_type}\n"
                        f"总使用量：{total_usage_byte/GB_TO_BYTE:.2f} GB\n"
                        f"配额总量：{quota_gb:.2f} GB\n"
                        f"使用占比：{usage_percent:.2f} %\n"
                        f"告警阈值：{ALERT_THRESHOLD * 100:.1f} %\n"
                        f"处理结果：{'实例已成功停止' if stop_result else '实例停止失败'}"
                    )
                    push_notification(alert_msg)
                    send_wechat_notification(f"Lightsail 流量超标停机通知 [{instance_name}]", alert_msg)
                    
                    if not is_alerted:
                        update_ssm_instance_list(instance_name, operation='add', ssm_param=SSM_PARAM_ALERTED)
                else:
                    print(f"实例 {instance_name} 已超标且处于停止状态，静默跳过")
            else:
                if is_alerted:
                    print(f"实例 {instance_name} 流量在阈值内，恢复正常状态")
                    update_ssm_instance_list(instance_name, operation='remove', ssm_param=SSM_PARAM_ALERTED)

        # 5. 每月首尾日汇总通知（带当日去重标记，防止 EventBridge 每 5 分钟重复推送）
        if is_beijing_month_first_or_last_day():
            if not has_summary_sent_today():
                send_wechat_notification(
                    "Lightsail 实例当月网络使用汇总",
                    get_all_instances_summary(instance_names)
                )
                mark_summary_sent_today()
            else:
                print("今日汇总通知已推送，跳过重复推送")
        
        return {
            'statusCode': 200,
            'body': json_serialize('Lightsail quota check and auto operation completed successfully!')
        }
    except Exception as e:
        error_msg = f"Lambda 执行异常：{str(e)}"
        print(error_msg)
        send_wechat_notification("Lightsail 流量监控异常告警", error_msg)
        return {
            'statusCode': 500,
            'body': json_serialize(error_msg)
        }

第五步：设置 EventBridge 定时触发器（建议 5 分钟高频巡检）

进入 Lambda 函数的 Add trigger 页面；
触发器源选择 EventBridge (CloudWatch Events)；
选择 Create a new rule：
- Rule name: Every5MinLightsailCheck
- Rule type: Schedule expression
- Schedule expression: rate(5 minutes)（或者 cron(*/5 * * * ? *)，每 5 分钟执行一次高频巡检）；
点击 Add 保存触发器。

💡 为什么推荐 5 分钟高频巡检？

防止突发流量跑超：遭遇异常大流量或恶意刷量时，若 1 小时检测一次可能已跑超数百 GB 造成巨额账单，5 分钟巡检可在超标瞬间极速关机止损；

完全 0 费用运行：全月每 5 分钟执行一次仅约 8,640 次调用，而 AWS Lambda 永久免费额度提供每月 1,000,000 次调用与 320 万秒计算时间，完全处于免费层，零额外成本。

运维避坑与经验总结

UTC 与北京时间跨月时区陷阱： AWS CloudWatch / Lightsail Metric 接口以 UTC 时间作为标准入参。若以北京时间（UTC+8）作为自然月界定，月初凌晨（00:00~08:00）时 UTC 实际仍处于上月月末。代码中必须将北京时间当月 1 号 00:00:00 精确换算为 UTC 时间（即上月最后一日 16:00:00 UTC），方可正确聚合当月流量，避免误读上月整月流量导致月初判定超标。
5 分钟高频触发下的幂等去重保护：在 5 分钟高频巡检场景下，月初 1 号开机任务与月末/月初全网流量用量大屏通知必须引入 SSM 日期去重标记（如 /lightsail/start-last-date 与 /lightsail/summary-last-date）。执行完成后打上当日日期标记，后续循环巡检自动静默跳过，彻底杜绝微信或 Webhook 被高频轰炸。
User-Agent 拦截问题：如果通知 Webhook 部署在 Cloudflare 之后，Python 原生 urllib 默认的 Python-urllib/3.x 请求头会触发 Cloudflare 浏览器完整性检查（Error 1010: 403 Forbidden）。因此在 Request Headers 中必须显式注入标准浏览器 User-Agent。
状态防死锁与清理：如果测试期间曾手动在 SSM Parameter Store 中写入过测试数据，可以通过将 /lightsail/alerted-instances 重置为 {}，以及将 /lightsail/stopped-by-quota 重置为 [] 即可一键恢复全新检测状态。
跨区域监控注意： Lightsail 的网络配额是按地域（Region）独立的。如果有多个区域的 Lightsail 实例，只需在对应区域分别部署该 Lambda 函数，或在代码中循环遍历区域列表即可。

AWS Lightsail 流量监控与自动关机