AWS Lightsail 流量监控与超标自动开关机实战（基于 Serverless 架构）

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 成本运行）：

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

核心设计亮点#

零受控端侵入（Agentless）：完全基于 AWS IAM 鉴权通过云端 API 交互，不在目标 Lightsail 机器内部安装任何软件，机器重装、改系统亦不受任何影响；
当月新建机器配额智能折算：对于月中新建的机器，脚本会自动计算当月剩余有效天数比例并折算实际可用配额，彻底杜绝月中新建机器误触发超标；
出入双向流量精准聚合：自动按天聚合查询 NetworkIn 与 NetworkOut 数据点求和，精准比对当月总配额；
防重复打扰与幂等关机：关机后将实例写入 AWS SSM Parameter Store，对于已经处于停止状态的实例保持静默，不反复推送骚扰通知；
每月 1 号自动重置与唤醒：北京时间每月 1 号 00:00，自动清空历史告警记录；对于上月因超标被系统关机、且当月新流量 ≤ 1MB 的实例，自动执行开机恢复服务；
每月首尾日全量用量报表：在月初与月末自动汇总账号下所有 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 秒即按天聚合）
`SSM_PARAM_STOPPED`	String	`/lightsail/stopped-by-quota`	SSM 中记录被停机实例列表的参数路径
`SSM_PARAM_ALERTED`	String	`/lightsail/alerted-instances`	SSM 中记录已触发告警实例字典的参数路径

AWS 资源配置与部署流程#

第一步：创建 IAM 执行角色与权限策略#

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

1
{
2
  "Version": "2012-10-17",
3
  "Statement": [
4
    {
5
      "Sid": "LightsailOperations",
6
      "Effect": "Allow",
7
      "Action": [
8
        "lightsail:GetInstances",
9
        "lightsail:GetInstance",
10
        "lightsail:GetInstanceMetricData",
11
        "lightsail:StartInstance",
12
        "lightsail:StopInstance"
13
      ],
14
      "Resource": "*"
15
    },
16
    {
17
      "Sid": "SSMParameterAccess",
18
      "Effect": "Allow",
19
      "Action": [
20
        "ssm:GetParameter",
21
        "ssm:PutParameter",
22
        "ssm:DeleteParameter"
23
      ],
24
      "Resource": "arn:aws:ssm:*:*:parameter/lightsail/*"
25
    },
26
    {
27
      "Sid": "SNSPublish",
28
      "Effect": "Allow",
29
      "Action": [
30
        "sns:Publish"
31
      ],
32
      "Resource": "*"
33
    },
34
    {
35
      "Sid": "CloudWatchLogs",
36
      "Effect": "Allow",
37
      "Action": [
38
        "logs:CreateLogGroup",
39
        "logs:CreateLogStream",
40
        "logs:PutLogEvents"
41
      ],
42
      "Resource": "arn:aws:logs:*:*:*"
43
    }
44
  ]
45
}

将该策略命名为 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: {}

第三步：创建并配置 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：

1
import json
2
import boto3
3
import calendar
4
import os
5
import time
6
from datetime import datetime, date, time as dt_time, timedelta
7
from urllib.request import Request, urlopen
8
from urllib.error import HTTPError, URLError
9

10
# ------------------------------ 全局常量定义（纯环境变量驱动，无硬编码）------------------------------
11
# AWS 区域（与 Lightsail / SSM / Lambda 同区域）
12
AWS_REGION = os.getenv('AWS_REGION', os.getenv('AWS_DEFAULT_REGION', 'ap-southeast-1'))
13

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

20
# SSM 参数名称（持久化存储被停 / 已告警实例）
21
SSM_PARAM_STOPPED = os.getenv('SSM_PARAM_STOPPED', '/lightsail/stopped-by-quota')
22
SSM_PARAM_ALERTED = os.getenv('SSM_PARAM_ALERTED', '/lightsail/alerted-instances')
23

24
# 单位转换常量
25
GB_TO_BYTE = 10**9
26
MB_TO_BYTE = 1024 * 1024
27

28
# 实例状态常量
29
INSTANCE_STATE_STOPPED = 'stopped'
30
INSTANCE_STATE_RUNNING = 'running'
31
INSTANCE_STATE_STARTING = 'starting'
32
INSTANCE_STATE_STOPPING = 'stopping'
33

34
# 告警状态常量
35
ALERT_STATUS_TRIGGERED = 'triggered'
36
NOTIFY_RETRY_TIMES = 2
37

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

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

47
def update_ssm_instance_list(instance_name, operation='add', ssm_param=SSM_PARAM_STOPPED):
48
    """通用 SSM 列表更新函数，支持并发重试与状态清空"""
49
    try:
50
        # 每月 1 号清空整个 SSM 参数
51
        if instance_name == '' and operation == 'remove':
52
            ssm_client.put_parameter(
53
                Name=ssm_param,
54
                Value='{}' if ssm_param == SSM_PARAM_ALERTED else '[]',
55
                Type='StringList' if ssm_param == SSM_PARAM_STOPPED else 'String',
56
                Overwrite=True
57
            )
58
            print(f"SSM 参数 {ssm_param} 已全部清空（配额重置）")
59
            return True
60

61
        for _ in range(3):
62
            try:
63
                response = ssm_client.get_parameter(Name=ssm_param, WithDecryption=False)
64
                ssm_data = json.loads(response['Parameter']['Value'])
65
            except ssm_client.exceptions.ParameterNotFound:
66
                ssm_data = {} if ssm_param == SSM_PARAM_ALERTED else []
67
            except json.JSONDecodeError:
68
                ssm_data = {} if ssm_param == SSM_PARAM_ALERTED else []
69

70
            origin_data = ssm_data.copy()
71
            if ssm_param == SSM_PARAM_STOPPED:
72
                if operation == 'add' and instance_name not in ssm_data:
73
                    ssm_data.append(instance_name)
74
                    print(f"实例 {instance_name} 已添加到 SSM 被停列表")
75
                elif operation == 'remove' and instance_name in ssm_data:
76
                    ssm_data.remove(instance_name)
77
                    print(f"实例 {instance_name} 已从 SSM 被停列表移除")
78
                else:
79
                    return True
80
            elif ssm_param == SSM_PARAM_ALERTED:
81
                if operation == 'add':
82
                    ssm_data[instance_name] = ALERT_STATUS_TRIGGERED
83
                    print(f"实例 {instance_name} 已标记为 SSM 已告警")
84
                elif operation == 'remove' and instance_name in ssm_data:
85
                    del ssm_data[instance_name]
86
                    print(f"实例 {instance_name} 已从 SSM 告警列表移除")
87
                else:
88
                    return True
89

90
            if origin_data == ssm_data:
91
                return True
92

93
            ssm_client.put_parameter(
94
                Name=ssm_param,
95
                Value=json_serialize(ssm_data),
96
                Type='StringList' if ssm_param == SSM_PARAM_STOPPED else 'String',
97
                Overwrite=True
98
            )
99
            return True
100
        return False
101
    except Exception as e:
102
        print(f"更新 SSM 列表失败：{str(e)}")
103
        return False
104

105
def get_general_instance_usage(instance_name, start_time, end_time, data_types=None):
106
    """通用流量查询函数，支持出入双向指标聚合"""
107
    if data_types is None:
108
        data_types = ['NetworkIn', 'NetworkOut']
109
    start_time_str = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
110
    end_time_str = end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
111
    total_usage = 0.0
112

113
    for data_type in data_types:
114
        try:
115
            response = lightsail_client.get_instance_metric_data(
116
                instanceName=instance_name,
117
                metricName=data_type,
118
                period=METRIC_PERIOD,
119
                unit='Bytes',
120
                statistics=['Sum'],
121
                startTime=start_time_str,
122
                endTime=end_time_str
123
            )
124
            metric_data = response.get('metricData', [])
125
            if not metric_data:
126
                continue
127
            metric_sum = sum(dp.get('sum', 0.0) for dp in metric_data)
128
            total_usage += metric_sum
129
        except Exception as e:
130
            print(f"获取实例 {instance_name} {data_type} 指标失败：{str(e)}")
131

132
    print(f"实例 {instance_name} 流量使用量：{total_usage:.2f} Byte（{total_usage/GB_TO_BYTE:.2f} GB）")
133
    return total_usage
134

135
def get_alerted_instances():
136
    """获取已告警实例列表，用于防止重复打扰"""
137
    try:
138
        response = ssm_client.get_parameter(Name=SSM_PARAM_ALERTED, WithDecryption=False)
139
        alerted_instances = json.loads(response['Parameter']['Value'])
140
        return alerted_instances if isinstance(alerted_instances, dict) else {}
141
    except (ssm_client.exceptions.ParameterNotFound, json.JSONDecodeError):
142
        return {}
143

144
def retry_notify(func):
145
    """通知重试装饰器，网络抖动时自动重试 2 次"""
146
    def wrapper(title, message):
147
        for retry in range(NOTIFY_RETRY_TIMES + 1):
148
            try:
149
                result = func(title, message)
150
                if result:
151
                    return True
152
                time.sleep(1)
153
            except Exception as e:
154
                print(f"通知重试第 {retry+1} 次异常：{str(e)}")
155
                time.sleep(1)
156
        return False
157
    return wrapper
158

159
# ------------------------------ Lightsail 实例操作 ------------------------------
160
def stop_instance(instance_name):
161
    """停止实例，带幂等性校验"""
162
    instance_state = get_instance_state(instance_name)
163
    if instance_state not in [INSTANCE_STATE_RUNNING, INSTANCE_STATE_STARTING]:
164
        print(f"实例 {instance_name} 当前状态 {instance_state}，无需停止")
165
        return True
166
    try:
167
        response = lightsail_client.stop_instance(instanceName=instance_name, force=True)
168
        print(f"实例 {instance_name} 停止成功：{json_serialize(response)}")
169
        update_ssm_instance_list(instance_name, operation='add', ssm_param=SSM_PARAM_STOPPED)
170
        return True
171
    except Exception as e:
172
        print(f"实例 {instance_name} 停止失败：{str(e)}")
173
        return False
174

175
def start_instance(instance_name):
176
    """开启实例，带幂等性校验"""
177
    instance_state = get_instance_state(instance_name)
178
    if instance_state != INSTANCE_STATE_STOPPED:
179
        print(f"实例 {instance_name} 当前状态 {instance_state}，无需开启")
180
        return True
181
    try:
182
        response = lightsail_client.start_instance(instanceName=instance_name)
183
        print(f"实例 {instance_name} 开启成功：{json_serialize(response)}")
184
        return True
185
    except Exception as e:
186
        print(f"实例 {instance_name} 开启失败：{str(e)}")
187
        return False
188

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

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

210
def get_month_dto_quota(instance_name):
211
    """计算实例当月网络配额（月中创建自动按比例折算）"""
212
    try:
213
        response = lightsail_client.get_instance(instanceName=instance_name)
214
        instance = response.get('instance', {})
215
        base_quota_gb = instance.get('networking', {}).get('monthlyTransfer', {}).get('gbPerMonthAllocated', 0.0)
216

217
        if base_quota_gb <= 0:
218
            return 0.0
219

220
        current_datetime = datetime.utcnow()
221
        instance_created_datetime = instance.get('createdAt')
222
        # 非当月创建实例，返回全额配额
223
        if not (instance_created_datetime and instance_created_datetime.year == current_datetime.year and instance_created_datetime.month == current_datetime.month):
224
            return base_quota_gb
225

226
        # 当月创建实例，按天折算可用配额
227
        first_day_zero = get_current_month_first_day_zero_time()
228
        last_day_last = get_current_month_last_day_last_time()
229
        month_ts = last_day_last.timestamp() - first_day_zero.timestamp()
230
        instance_valide_ts = last_day_last.timestamp() - instance_created_datetime.timestamp()
231

232
        if month_ts <= 0 or instance_valide_ts <= 0:
233
            return base_quota_gb
234

235
        return (instance_valide_ts / month_ts) * base_quota_gb
236
    except Exception as e:
237
        print(f"计算实例 {instance_name} 配额失败：{str(e)}")
238
        return 0.0
239

240
def get_instance_current_month_usage(instance_name):
241
    """获取实例当月累计流量使用量"""
242
    start_time = get_current_month_first_day_zero_time()
243
    end_time = datetime.utcnow()
244
    return get_general_instance_usage(instance_name, start_time, end_time)
245

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

259
@retry_notify
260
def send_wechat_notification(title, message):
261
    """消息推送通知（带重试，URL 从环境变量读取，带 User-Agent 防 WAF 拦截）"""
262
    if not WECHAT_NOTIFY_URL:
263
        print("未配置 WECHAT_NOTIFY_URL 环境变量，跳过消息通知")
264
        return False
265
    try:
266
        payload_bytes = json.dumps({
267
            "title": title,
268
            "content": message,
269
            "message": message
270
        }, ensure_ascii=False).encode('utf-8')
271
        request = Request(
272
            url=WECHAT_NOTIFY_URL,
273
            data=payload_bytes,
274
            headers={
275
                "Content-Type": "application/json",
276
                "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"
277
            },
278
            method='POST'
279
        )
280
        with urlopen(request, timeout=10) as response:
281
            res_text = response.read().decode('utf-8')
282
            print(f"消息通知发送成功，响应：{res_text}")
283
        return True
284
    except (HTTPError, URLError) as e:
285
        print(f"消息通知发送失败：{str(e)}")
286
        return False
287
    except Exception as e:
288
        print(f"消息通知发送失败（未知错误）：{str(e)}")
289
        return False
290

291
# ------------------------------ 时间工具函数 ------------------------------
292
def get_current_month_first_day_zero_time():
293
    today_utc = date.today()
294
    return datetime.combine(today_utc.replace(day=1), dt_time.min, tzinfo=None)
295

296
def get_current_month_last_day_last_time():
297
    today_utc = date.today()
298
    last_day = today_utc.replace(day=calendar.monthrange(today_utc.year, today_utc.month)[1])
299
    return datetime.combine(last_day, dt_time(23, 59, 59), tzinfo=None)
300

301
def get_beijing_current_date():
302
    return (datetime.utcnow() + timedelta(hours=8)).date()
303

304
def is_beijing_month_first_day():
305
    return get_beijing_current_date().day == 1
306

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

312
def get_all_instances_summary(instance_names):
313
    """生成所有实例当月流量汇总报表"""
314
    summary = [
315
        "=== Lightsail 实例当月网络流量汇总 ===",
316
        f"统计日期：{get_beijing_current_date().strftime('%Y-%m-%d')}（北京时间）",
317
        ""
318
    ]
319
    for instance_name in instance_names:
320
        quota_gb = get_month_dto_quota(instance_name)
321
        quota_byte = quota_gb * GB_TO_BYTE
322
        if quota_byte <= 0:
323
            summary.append(f"实例名称：{instance_name} - 无有效配额，跳过统计\n")
324
            continue
325
        total_usage_byte = get_instance_current_month_usage(instance_name)
326
        usage_percent = (total_usage_byte / quota_byte) * 100 if quota_byte > 0 else 0.0
327
        instance_info = (
328
            f"实例名称：{instance_name}\n"
329
            f"  当月用量：{total_usage_byte/GB_TO_BYTE:.2f} GB\n"
330
            f"  配额总量：{quota_gb:.2f} GB\n"
331
            f"  当前占比：{usage_percent:.2f} %\n"
332
            f"  告警阈值：{ALERT_THRESHOLD * 100:.1f} %\n"
333
        )
334
        summary.append(instance_info)
335
    return "\n".join(summary)
336

337
# ------------------------------ 每月 1 号自动开机逻辑 ------------------------------
338
def start_stopped_instances_on_month_first_day():
339
    """每月 1 号自动唤醒上月被停止且当月流量已重置（≤1MB）的实例"""
340
    if not is_beijing_month_first_day():
341
        return
342

343
    print("=== 开始执行每月 1 号实例开启逻辑 ===")
344
    update_ssm_instance_list('', operation='remove', ssm_param=SSM_PARAM_ALERTED)
345

346
    try:
347
        response = ssm_client.get_parameter(Name=SSM_PARAM_STOPPED, WithDecryption=False)
348
        stopped_instances = json.loads(response['Parameter']['Value'])
349
    except (ssm_client.exceptions.ParameterNotFound, json.JSONDecodeError):
350
        stopped_instances = []
351

352
    if not stopped_instances:
353
        send_wechat_notification("每月 1 号 Lightsail 实例开启通知", "SSM 中无被限额停止的实例，无需开启")
354
        return
355

356
    start_results = []
357
    for instance_name in stopped_instances:
358
        instance_state = get_instance_state(instance_name)
359
        if instance_state != INSTANCE_STATE_STOPPED:
360
            start_results.append(f"实例 {instance_name}：当前状态 {instance_state}，跳过")
361
            update_ssm_instance_list(instance_name, operation='remove', ssm_param=SSM_PARAM_STOPPED)
362
            continue
363

364
        total_usage = get_instance_current_month_usage(instance_name)
365
        if total_usage > MB_TO_BYTE:
366
            start_results.append(f"实例 {instance_name}：当月用量 {total_usage/MB_TO_BYTE:.2f}MB > 1MB，跳过开启")
367
            continue
368

369
        start_success = start_instance(instance_name)
370
        if start_success:
371
            update_ssm_instance_list(instance_name, operation='remove', ssm_param=SSM_PARAM_STOPPED)
372
            start_results.append(f"实例 {instance_name}：开启成功")
373
        else:
374
            start_results.append(f"实例 {instance_name}：开启失败")
375

376
    summary_msg = "\n".join([
377
        "=== 每月 1 号 Lightsail 实例开启执行结果 ===",
378
        f"执行时间：{get_beijing_current_date().strftime('%Y-%m-%d')}（北京时间）",
379
        f"待处理实例数：{len(stopped_instances)}",
380
        "执行结果：",
381
    ] + start_results)
382
    send_wechat_notification("每月 1 号 Lightsail 实例开启结果通知", summary_msg)
383

384
# ------------------------------ Lambda 入口主函数 ------------------------------
385
def lambda_handler(event, context):
386
    try:
387
        # 1. 每月 1 号开机与配额重置检查
388
        start_stopped_instances_on_month_first_day()
389

390
        # 2. 获取所有实例
391
        instance_names = list_instances()
392
        if not instance_names:
393
            return {'statusCode': 200, 'body': json_serialize('No Lightsail instances found!')}
394

395
        # 3. 获取已告警实例列表
396
        alerted_instances = get_alerted_instances()
397

398
        # 4. 逐个检查实例流量
399
        for instance_name in instance_names:
400
            quota_gb = get_month_dto_quota(instance_name)
401
            quota_byte = quota_gb * GB_TO_BYTE
402
            if quota_byte <= 0:
403
                continue
404

405
            total_usage_byte = get_instance_current_month_usage(instance_name)
406
            usage_percent = (total_usage_byte / quota_byte) * 100 if quota_byte > 0 else 0.0
407
            is_over_quota = total_usage_byte > (quota_byte * ALERT_THRESHOLD)
408
            is_alerted = instance_name in alerted_instances
409
            current_state = get_instance_state(instance_name)
410
            is_running = current_state in [INSTANCE_STATE_RUNNING, INSTANCE_STATE_STARTING]
411

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

414
            # 超标停机与通知逻辑
415
            if is_over_quota:
416
                if is_running:
417
                    stop_result = stop_instance(instance_name)
418
                    action_type = "首次超标自动停机" if not is_alerted else "超标保护（检测到运行中，再次停机）"
419
                    alert_msg = (
420
                        f"实例名称：{instance_name}\n"
421
                        f"触发类型：{action_type}\n"
422
                        f"总使用量：{total_usage_byte/GB_TO_BYTE:.2f} GB\n"
423
                        f"配额总量：{quota_gb:.2f} GB\n"
424
                        f"使用占比：{usage_percent:.2f} %\n"
425
                        f"告警阈值：{ALERT_THRESHOLD * 100:.1f} %\n"
426
                        f"处理结果：{'实例已成功停止' if stop_result else '实例停止失败'}"
427
                    )
428
                    push_notification(alert_msg)
429
                    send_wechat_notification(f"Lightsail 流量超标停机通知 [{instance_name}]", alert_msg)
430

431
                    if not is_alerted:
432
                        update_ssm_instance_list(instance_name, operation='add', ssm_param=SSM_PARAM_ALERTED)
433
                else:
434
                    print(f"实例 {instance_name} 已超标且处于停止状态，静默跳过")
435
            else:
436
                if is_alerted:
437
                    print(f"实例 {instance_name} 流量在阈值内，恢复正常状态")
438
                    update_ssm_instance_list(instance_name, operation='remove', ssm_param=SSM_PARAM_ALERTED)
439

440
        # 5. 每月首尾日汇总通知
441
        if is_beijing_month_first_or_last_day():
442
            send_wechat_notification(
443
                "Lightsail 实例当月网络使用汇总",
444
                get_all_instances_summary(instance_names)
445
            )
446

447
        return {
448
            'statusCode': 200,
449
            'body': json_serialize('Lightsail quota check and auto operation completed successfully!')
450
        }
451
    except Exception as e:
452
        error_msg = f"Lambda 执行异常：{str(e)}"
453
        print(error_msg)
454
        send_wechat_notification("Lightsail 监控 Lambda 执行失败告警", error_msg)
455
        push_notification(error_msg)
456
        return {
457
            'statusCode': 500,
458
            'body': json_serialize(error_msg)
459
        }

第五步：设置 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 万秒计算时间，完全处于免费层，零额外成本。

运维避坑与经验总结#

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 函数，或在代码中循环遍历区域列表即可。