如何捕获AWS Lambda自身CRUD事件并实现更新时Slack告警？

要实现Lambda创建、更新、删除操作的告警，核心是利用AWS CloudTrail记录Lambda的API调用事件，再通过EventBridge筛选事件并触发通知流程，具体步骤如下：

1. 启用CloudTrail记录Lambda事件 #

确保你的AWS账户已开启CloudTrail服务，且Trail配置为记录Lambda的关键API操作：

• 必须包含的事件：CreateFunction（创建）、UpdateFunctionCode（代码更新）、UpdateFunctionConfiguration（配置更新）、DeleteFunction（删除）
• 若使用默认Trail，通常已覆盖这些事件；若为自定义Trail，需确认在"数据事件"中勾选了Lambda服务。

2. 在EventBridge创建事件规则 #

通过EventBridge筛选CloudTrail捕获的Lambda操作事件：

1. 进入EventBridge控制台，创建新规则，选择事件模式作为规则类型
2. 配置事件模式匹配Lambda的CRUD操作：

{
  "source": ["aws.lambda"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventSource": ["lambda.amazonaws.com"],
    "eventName": [
      "CreateFunction",
      "UpdateFunctionCode",
      "UpdateFunctionConfiguration",
      "DeleteFunction"
    ]
  }
}

3. 将规则的目标设置为一个专门处理Slack通知的Lambda函数。

3. 编写Slack通知处理函数 #

这个函数负责解析EventBridge传递的事件，格式化消息并发送到Slack频道，示例Python代码如下：

import json
import urllib3

def lambda_handler(event, context):
    # 提取事件关键信息
    event_detail = event["detail"]
    event_name = event_detail["eventName"]
    function_name = event_detail["requestParameters"]["functionName"]
    operator_arn = event_detail["userIdentity"]["arn"]
    event_time = event["time"]

    # 构造Slack消息
    slack_payload = {
        "text": f"Lambda函数*{function_name}*执行了*{event_name}*操作\n"
                f"操作时间: {event_time}\n"
                f"操作人ARN: {operator_arn}"
    }

    # 发送到Slack Webhook
    http = urllib3.PoolManager()
    resp = http.request(
        "POST",
        "你的Slack Incoming Webhook URL",
        body=json.dumps(slack_payload),
        headers={"Content-Type": "application/json"}
    )

    return {"statusCode": resp.status, "body": "通知已发送"}

• 替换代码中的你的Slack Incoming Webhook URL为你在Slack工作区创建的Webhook地址。

4. 配置权限与测试 #

• 若处理通知的Lambda函数部署在VPC内，需配置NAT网关以允许访问Slack公网API；公网部署的Lambda无需额外配置。
• 手动执行Lambda的更新/创建/删除操作，验证EventBridge是否触发规则，Slack频道是否收到对应通知。

内容的提问来源于stack exchange，提问作者Raja Sardar