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

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

核心特性

- 全球区域自适应：内置全球 15 个 Lightsail 可用区并发扫描，后续跨区迁移、新增或删除 VPS 均无需修改任何代码。

- 区域流量池共享计算：严格对齐 AWS 计费规则，自动累加同区域内所有 VPS 的流量配额与实际使用量（入站 + 出站）。

- 首月配额按天折算：自动识别当月新建的实例，按当月剩余时间比例精准折算可用免费配额。

- 防天价账单保护：当某区域总使用量超过该区域总配额的 95% 时，自动强制关停该区域所有实例，并通过 SNS 发送汇总邮件告警。

- lambda 在一定额度内是免费的；这个小监控脚本消耗非常少，是 100%免费额度内的

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

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

- 创建主题 (Topic)：

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

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

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

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

- 创建邮件订阅 (Subscription)：

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

- 协议 (Protocol)：选择 Email 。

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

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

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

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

- 基础设置：

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

- 函数名称：lightsailmonitor

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

- 点击右下角 创建函数。

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

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

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

dit)。

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

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

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

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

- 在 Lambda 函数的 配置 (Configuration) -> 权限 (Permissions) 页面，点击“执行角色 (Execution role)”下方的角色名称链接，进入 IAM 控制台。

- 点击 添加权限 (Add permissions) -> 创建内联策略 (Create inline policy)。

• 点击右上角切换到 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)。

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

- 回到 Lambda 函数的 代码 (Code) 选项卡。

• 将下方完整代码粘贴到 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 / (10003):.2f} GB / quota {quota / (10003):.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!')
}
```- 点击代码框上方的 Deploy （部署） 按钮保存代码。

• 点击 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)

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

- 在 Lambda 函数页面顶部“函数概览 (Function overview)”区域，点击 + 添加触发器 (+ Add trigger)。

- 在下拉菜单中选择 EventBridge (CloudWatch Events)。

- 规则 (Rule)：选择 创建新规则 (Create a new rule)。

- 规则名称：输入 LightsailHourlyCheck 。

- 规则类型：选择 计划表达式 (Schedule expression)。

- 计划表达式：

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

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

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

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

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

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

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

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

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

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

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