Lambda 函数编程模型：事件、上下文与日志
写 Lambda 函数其实就三件事：接住事件、用好上下文、打好日志。这三个东西拧在一起，才构成一个可以稳定运行的巡检任务。我刚开始接触 Lambda 时，觉得 event 不就是个 JSON 嘛，context 用不上就直接忽略了，日志用 print 随便打打。结果线上出了几次问题，排查起来那个费劲……嗯，今天咱们就把这三个点彻底捋清楚。
1. 事件对象（event）——Lambda 的「入口参数」
每次调用 Lambda，AWS 都会把一个事件对象扔到你的 handler 里。说白了，它就是触发源给你的一堆数据。
举个最常见的场景：CloudWatch Events 定时触发巡检。事件结构大概长这样：
{
"version": "0",
"id": "12345678-1234-1234-1234-123456789012",
"detail-type": "Scheduled Event",
"source": "aws.events",
"time": "2025-04-01T10:00:00Z",
"region": "us-east-1",
"resources": ["arn:aws:events:us-east-1:123456789012:rule/MyScheduledRule"],
"detail": {}
}你可能会问：detail 是空的，那我怎么知道要巡检哪个资源？
我的做法是在 CloudWatch Events 的 输入转换器（Input Transformer）里自己塞参数。比如把目标安全组 ID 或者 S3 桶名放到 detail 里，这样 event 就能带着业务上下文进来了。
事件来源不同，结构天差地别。S3 事件、SQS 消息、API Gateway 请求 …… 每个都有自己的 schema。我建议一开始就写个辅助函数，把 event 里最常用的字段抽出来。比如：
def extract_event_info(event):
source = event.get('source', 'unknown')
# 根据 source 做不同解析
if source == 'aws.events':
return {'trigger_type': 'scheduled', 'time': event['time']}
elif source == 'aws.s3':
record = event['Records'][0]
return {'bucket': record['s3']['bucket']['name'],
'object': record['s3']['object']['key']}
# 其他来源...
return eventif not event 的 guard 判断。
2. 上下文对象（context）——Lambda 的「运行体检表」
context 里装着当前函数执行环境的各种元数据。很多新手忽略它，但在我看来，它在生产环境里反而比 event 更有用。
下面这张表列出我常用的 context 属性：
| 属性 | 类型 | 说明 |
|---|---|---|
function_name | string | 当前函数名 |
function_version | string | 当前版本号（$LATEST 或数字版本） |
memory_limit_in_mb | int | 分配的内存大小 |
log_group_name | string | 对应的 CloudWatch Log Group |
log_stream_name | string | 当前的日志流 |
aws_request_id | string | 每次调用的唯一 ID — 排查问题利器 |
remaining_time_in_millis() | int | 剩余执行时间（毫秒） |
get_remaining_time_in_millis() | 方法 | 同上，动态获取 |
我常用的几个场景：
- 在日志里带上
context.aws_request_id，这样从日志流里能直接定位到某次调用。 - 用
context.get_remaining_time_in_millis()做超时保护。特别是巡检大批量资源时，如果只剩 5 秒，就该提前保存状态，避免丢数据。 - 把
log_group_name写进异常消息，方便之后去对应日志流里翻锅。
if context.get_remaining_time_in_millis() < 10000: break，太实用了。
3. 日志输出——用 print 还是 logging？
Lambda 的日志会自动流向 CloudWatch Logs。你用 print() 打出的内容，会被 Lambda 运行时捕获，然后写进日志流。听起来挺方便对吧？但问题来了：print 是异步的，而且没有级别控制。
我的建议很简单：项目里永远用 Python 标准库的 logging。下面是我常用的配置：
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info("巡检开始 | RequestID: %s", context.aws_request_id)
# ... 业务逻辑 ...
logger.warning("发现安全组规则未授权: sg-xxx")
logger.error("S3 桶访问失败, bucket=%s", bucket_name)
return {"statusCode": 200}你想想看，如果只用 print，线上日志会变成一团乱麻——你分不清哪些是正常信息、哪些是错误、哪些只是调试用的。而 logging 模块天然支持：
- 日志级别：DEBUG/INFO/WARNING/ERROR/CRITICAL
- 格式定制：加上时间戳、RequestID、函数名
- 行为控制：可以随时切换 DEBUG 模式，不影响生产
实际项目中我还会加 结构化日志，就是把日志打成 JSON 格式，方便后续分析工具解析。示例：
import json
def log_json(level, message, extra=None):
log_entry = {
"time": "2025-04-01T10:00:00Z", # 实际用 datetime
"level": level,
"message": message,
"request_id": context.aws_request_id
}
if extra:
log_entry.update(extra)
print(json.dumps(log_entry)) # 这里用 print 输出 JSON 字符串print 输出 JSON 时，可能会因为线程安全导致日志顺序错乱。我建议还是用 logger.info 配合自定义 Formatter，更靠谱。
4. 三者如何协同？一张图说清楚
事件、上下文、日志不是孤立的，它们在 Lambda 执行流程里是一体的。我画了张简图，你可以看到每一次调用的完整脉络：
从图里能看出：事件和上下文是 handler 的两个输入，handler 在处理过程中产生日志，最终返回结果。日志会异步写入 CloudWatch Logs。我特别喜欢把 context.aws_request_id 作为每条日志的标签，这样你就能在 CloudWatch 里按 Request ID 过滤出某次调用的完整日志，排查问题快多了。
5. 实战小技巧
- 给日志打时间戳：Lambda 环境时区默认 UTC，我习惯在日志格式化字符串里加上
%(asctime)s，并设置datefmt='%Y-%m-%dT%H:%M:%S'，这样日志和 CloudWatch 时间线对齐。 - 结构化日志：用 JSON 格式输出，配合 CloudWatch Logs Insights 的
parse命令，能直接提取字段做聚合分析。我的 Elasticsearch 就是靠这个自动从日志里抽数据的。 - 利用 context 调试：如果函数版本是 $LATEST，可以加个
if context.function_version == '$LATEST': logger.setLevel(logging.DEBUG)，这样开发调试时能看到更多信息，生产版本自动切回 INFO。
- event 是触发数据，要适配不同来源；
- context 是环境信息，能帮你做超时保护、追踪调用；
- 日志用 logging 模块，加上 RequestID 和时间戳，最好结构化输出。