AWS Lightsail 全球多区域共享流量监控与自动停机部署手册

[aitobox]

我用 AWS 的 lightsail ，每个 VPS 的流量是叠加的；但是 AWS 本身没有提供一种功能：即流量快要用完了，自动停机；我找了一下网上的脚本，基本上都是基于单台 VPS 的监控，于是自己动手实现一个监控及报警方案；以下内容是边操作边用 AI 记录，最后 AI 整理完成的文档；我 Review 了一遍，是无误的：

本文档提供了一套基于 AWS Lambda + Amazon SNS + Amazon EventBridge 的自动化流量监控方案。

核心特性

1. 全球区域自适应：内置全球 15 个 Lightsail 可用区并发扫描，后续跨区迁移、新增或删除 VPS 均无需修改任何代码。
2. 区域流量池共享计算：严格对齐 AWS 计费规则，自动累加同区域内所有 VPS 的流量配额与实际使用量（入站 + 出站）。
3. 首月配额按天折算：自动识别当月新建的实例，按当月剩余时间比例精准折算可用免费配额。
4. 防天价账单保护：当某区域总使用量超过该区域总配额的 95% 时，自动强制关停该区域所有实例，并通过 SNS 发送汇总邮件告警。
5. lambda 在一定额度内是免费的；这个小监控脚本消耗非常少，是 100%免费额度内的

第一步：配置 SNS 邮件告警通道

1. 登录 AWS 控制台，进入 Simple Notification Service (SNS)（建议选择 us-east-1 区域）。

2. 创建主题 (Topic)：

   • 左侧菜单点击 主题 (Topics) -> 创建主题 (Create topic)。

   • 类型：选择 标准 (Standard)。

   • 名称：输入 lightsailmonitor ，滚动到底部点击 创建主题。

   • 记录 ARN：复制页面中生成的主题 ARN （格式为：arn:aws:sns:us-east-1:你的账号 ID:lightsailmonitor ）。注意：不要复制后面带冒号和随机字符串的订阅 ARN 。

3. 创建邮件订阅 (Subscription)：

   • 在该主题页面下方，点击 创建订阅 (Create subscription)。

   • 协议 (Protocol)：选择 Email 。

   • 终端节点 (Endpoint)：输入接收告警的电子邮箱地址，点击 创建订阅。

   • 确认订阅：前往你的邮箱，打开来自 AWS Notifications 的邮件，点击正文中的 Confirm subscription 链接完成绑定。

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

1. 进入 AWS Lambda 控制台（建议与 SNS 同在 us-east-1 区域），点击 创建函数 (Create function)。
2. 基础设置：

   • 选择 从头开始创作 (Author from scratch)。

   • 函数名称：lightsailmonitor

   • 运行时 (Runtime)：选择最新版本的 Python （如 Python 3.12 / 3.13 / 3.14 ）。

   • 点击右下角 创建函数。

3. 调整超时时间与内存（必做）：

   • 由于需要并发扫描全球 15 个区域，耗时通常在 5~9 秒左右（超过默认的 3 秒限制）。

   • 在函数页面点击 配置 (Configuration) -> 常规配置 (General configuration) -> 编辑 (E dit)。

   • 内存 (Memory)：建议设置为 256 MB 或 512 MB 。

   • 超时 (Timeout)：修改为 0 分 30 秒，点击 保存。

第三步：配置 IAM 执行角色权限

Lambda 需要跨全球区域读取、关停 Lightsail 实例以及向 SNS 发送通知的权限。

1. 在 Lambda 函数的 配置 (Configuration) -> 权限 (Permissions) 页面，点击“执行角色 (Execution role)”下方的角色名称链接，进入 IAM 控制台。
2. 点击 添加权限 (Add permissions) -> 创建内联策略 (Create inline policy)。
3. 点击右上角切换到 JSON 视图，清空原有内容并粘贴以下最小权限策略（支持全区域操作）：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "LightsailGlobalMonitorAndStop",
            "Effect": "Allow",
            "Action": [
                "lightsail:GetInstances",
                "lightsail:GetInstance",
                "lightsail:GetInstanceMetricData",
                "lightsail:StopInstance"
            ],
            "Resource": "*"
        },
        {
            "Sid": "SNSPublishAlert",
            "Effect": "Allow",
            "Action": [
                "sns:Publish"
            ],
            "Resource": "arn:aws:sns:*:*:lightsailmonitor"
        }
    ]
}

4.点击 下一步 (Next)，策略名称填入 LightsailGlobalMonitorPolicy ，点击 创建策略 (Create policy)。

第四步：部署核心监控代码

1. 回到 Lambda 函数的 代码 (Code) 选项卡。
2. 将下方完整代码粘贴到 lambda_function.py 中（注意核对第 28 行的 SNS_TOPIC_ARN 是否与你第一步创建的一致）：

import json
import boto3
import calendar
import os
from datetime import datetime, date, time, timedelta
from concurrent.futures import ThreadPoolExecutor

# 内置全球所有支持 Lightsail 的 AWS 区域（免去 GetRegions 权限要求）
# 注：如果某个默认未启用的区域（如雅加达 ap-southeast-3 ）报错烦人，可直接从列表中注释掉
LIGHTSAIL_REGIONS = [
    'ap-northeast-1', # 日本 (东京)
    'ap-northeast-2', # 韩国 (首尔)
    'ap-southeast-1', # 新加坡
    'ap-southeast-2', # 澳大利亚 (悉尼)
    'ap-southeast-3', # 印度尼西亚 (雅加达)
    'ap-south-1',     # 印度 (孟买)
    'us-east-1',      # 美国东部 (弗吉尼亚北部)
    'us-east-2',      # 美国东部 (俄亥俄)
    'us-west-2',      # 美国西部 (俄勒冈)
    'ca-central-1',   # 加拿大 (中部)
    'eu-central-1',   # 欧洲 (法兰克福)
    'eu-west-1',      # 欧洲 (爱尔兰)
    'eu-west-2',      # 欧洲 (伦敦)
    'eu-west-3',      # 欧洲 (巴黎)
    'eu-north-1'      # 欧洲 (斯德哥尔摩)
]

# 请填写你的 SNS 主题 ARN （注意：到主题名称为止，不要包含末尾的订阅 ID ）
SNS_TOPIC_ARN = 'arn:aws:sns:us-east-1:xxxxxx:lightsailmonitor'

# 触发强制关机的流量阈值比例（ 0.95 代表 95%）
QUOTA_THRESHOLD = 0.95

def push_notification(arn, msg):
    """发送 SNS 告警通知（自动从 ARN 提取对应区域）"""
    sns_region = arn.split(':')[3]
    sns_client = boto3.client('sns', region_name=sns_region)
    sns_client.publish(
        TopicArn=arn,
        Message=msg,
        Subject='[AWS Alert] Lightsail Region Network Quota Exceeded'
    )

def get_current_month_first_day_zero_time():
    today = date.today()
    first_day = today.replace(day=1)
    return datetime.combine(first_day, time.min)
    
def get_current_month_last_day_last_time():
    today = date.today()
    last_day = today.replace(day=calendar.monthrange(today.year, today.month)[1])
    return datetime.combine(last_day, time(23, 59, 59))
    
def stop_instance(client, instance_name):
    client.stop_instance(
        instanceName=instance_name,
        force=True
    )
    
def list_instances(client):
    instances_list = []
    paginator = client.get_paginator('get_instances')
    for page in paginator.paginate():
        for instance in page['instances']:
            instances_list.append(instance['name'])
    return instances_list
 
def get_month_dto_quota(client, instance_name):
    """获取单台实例当月可用配额（当月新建实例按剩余时间比例折算）"""
    response = client.get_instance(instanceName=instance_name)
    dto_quota = response['instance']['networking']['monthlyTransfer']['gbPerMonthAllocated']
    current_datetime = datetime.now()
    instance_created_datetime = response['instance']['createdAt']
    
    if (instance_created_datetime.year == current_datetime.year) and (instance_created_datetime.month == current_datetime.month):
        month_ts = get_current_month_last_day_last_time().timestamp() - get_current_month_first_day_zero_time().timestamp()
        instance_valide_ts = get_current_month_last_day_last_time().timestamp() - instance_created_datetime.timestamp()
        dto_quota = (instance_valide_ts / month_ts) * dto_quota
        print(f"[{instance_name}] created in current month, prorated quota: {dto_quota:.2f}GB")
    else:
        print(f"[{instance_name}] created in previous month, full quota: {dto_quota}GB")
    
    return dto_quota
    
def get_instance_data_usage(client, instance_name, data_type):
    """获取单台实例当月累计流量（ Bytes ）"""
    start_time_str = get_current_month_first_day_zero_time().strftime('%Y-%m-%dT%H:%M:%SZ')
    end_time_str = get_current_month_last_day_last_time().strftime('%Y-%m-%dT%H:%M:%SZ')
 
    response = client.get_instance_metric_data(
        instanceName=instance_name,
        metricName=data_type,
        period=6 * 600 * 24,
        unit='Bytes',
        statistics=['Sum'],
        startTime=start_time_str,
        endTime=end_time_str 
    )
 
    data_points = response['metricData']
    total_data_usage = sum([data_point['sum'] for data_point in data_points])
    print(f"[{instance_name}] total {data_type} usage: {total_data_usage}")
    return total_data_usage

def check_single_region(region):
    """按区域汇总所有 Lightsail 实例的共享流量配额与总使用量"""
    try:
        client = boto3.client('lightsail', region_name=region)
        instances = list_instances(client)
        if not instances:
            return
            
        print(f"=== Found {len(instances)} instance(s) in {region}: {instances} ===")
        
        region_total_quota = 0
        region_total_usage = 0
        instance_details = []

        # 1. 遍历区域内所有实例，累加总配额和总使用量
        for i in instances:
            quota = get_month_dto_quota(client, i) * 1000 * 1000 * 1000
            usage = get_instance_data_usage(client, i, "NetworkOut") + get_instance_data_usage(client, i, "NetworkIn")
            
            region_total_quota += quota
            region_total_usage += usage
            
            detail_str = f"  - {i}: usage {usage / (1000**3):.2f} GB / quota {quota / (1000**3):.2f} GB ({(usage/quota)*100:.2f}%)"
            instance_details.append(detail_str)

        # 2. 计算整个区域的共享流量使用百分比
        usage_percent = (region_total_usage / region_total_quota) * 100 if region_total_quota > 0 else 0
        
        summary_msg = (
            f"Region: {region}\n"
            f"Instances Count: {len(instances)}\n"
            f"Region Total Usage: {region_total_usage} Bytes ({region_total_usage / (1000**3):.2f} GB)\n"
            f"Region Total Quota: {region_total_quota} Bytes ({region_total_quota / (1000**3):.2f} GB)\n"
            f"Region Shared Usage Percent: {usage_percent:.2f}%\n"
            f"Instance Breakdown:\n" + "\n".join(instance_details)
        )
        print(summary_msg)
        
        # 3. 如果区域总使用量超过区域总配额的阈值（默认 95%），关停该区域所有实例并发送告警
        if (region_total_quota * QUOTA_THRESHOLD) < region_total_usage:
            print(f"[{region}] Shared quota exceeded {QUOTA_THRESHOLD*100}% ({usage_percent:.2f}%), force closing all instances in {region}: {instances}")
            for i in instances:
                try:
                    stop_instance(client, i)
                except Exception as stop_err:
                    print(f"[{region}] Failed to stop instance {i}: {stop_err}")
            
            push_notification(SNS_TOPIC_ARN, summary_msg)
            
    except Exception as e:
        print(f"Skipping region {region} due to error: {e}")

def lambda_handler(event, context):
    print(f"Scanning Lightsail regions: {LIGHTSAIL_REGIONS}")
    
    # 使用多线程并发扫描所有区域
    with ThreadPoolExecutor(max_workers=10) as executor:
        executor.map(check_single_region, LIGHTSAIL_REGIONS)
             
    return {
        'statusCode': 200,
        'body': json.dumps('Global total_data_usage check completed!')
    }

1. 点击代码框上方的 Deploy （部署） 按钮保存代码。
2. 点击 Test （测试） 按钮运行一次，确认日志中能正常输出实例名称与流量百分比。典型运行结果输出如下:

=== Found 2 instance(s) in ap-northeast-1: ['AIToBoxJP', 'AIToBoxJP-2'] ===
Region: ap-northeast-1
Instances Count: 2
Region Total Usage: 1558625396515.0 Bytes (1558.63 GB)
Region Total Quota: 4096000000000 Bytes (4096.00 GB)
Region Shared Usage Percent: 38.05%
Instance Breakdown:
  - AIToBoxJP: usage 1158.63 GB / quota 2048.00 GB (56.57%)
  - AIToBoxJP-2: usage 400.00 GB / quota 2048.00 GB (19.53%)

第五步：配置定时自动触发器 (EventBridge)

为了让脚本全天候自动巡检，需要为其绑定一个定时触发器：

1. 在 Lambda 函数页面顶部“函数概览 (Function overview)”区域，点击 + 添加触发器 (+ Add trigger)。
2. 在下拉菜单中选择 EventBridge (CloudWatch Events)。
3. 规则 (Rule)：选择 创建新规则 (Create a new rule)。
4. 规则名称：输入 LightsailHourlyCheck 。
5. 规则类型：选择 计划表达式 (Schedule expression)。
6. 计划表达式：

   • 若希望每小时检查一次（推荐）：输入 rate(1 hour)

   • 若希望每 15 分钟检查一次（适合高带宽防突发流量）：输入 rate(15 minutes)

7. 点击右下角 添加 (Add) 即可生效。

第六步：可选优化与运维注意事项

1. 控制 CloudWatch 日志保留期（节省存储空间）：

   • 进入 CloudWatch -> 日志组 (Log groups) -> 点击 /aws/lambda/lightsailmonitor 。

   • 在“保留 (Retention)”设置中，建议将其设置为 1 个月 (30 days) 或 7 天，避免长期运行积累过多历史日志。

2. 关于删除实例的计费陷阱（重要）：

   • 在 AWS Lightsail 中，停止（ Stopped ） 状态的实例依然会保留其月度流量配额供同区域其他机器共享。

   • 但如果在月中直接 删除（ Deleted ） 了某台跑过流量的实例，该实例贡献的配额会从区域流量池中扣除，而已消耗的流量可能会引起剩余机器超额。因此，若某台机器不再使用且当月跑了较多流量，建议先关机（ Stop ）保留到月底，下个月初再删除。

   • AWS lightsail 的流量，是双向计算的，出和入是都计算的；比如你一台 VPS 流量是 2000GB ；这个月出 500GB ，入 1600GB ，那么总流量是 2100GB ，就超出限额了；

[garvinchen1015]

收藏了，感谢整理。