如何将AWS Lambda报错信息上报至PHP应用？

针对你的Lambda函数可能出现的限流失败、执行中断等问题，以下是三种可靠的上报方案，可直接获取org_id、图片名称等关键上下文信息：

方案1：在原Lambda中添加异常捕获与主动上报 #

这是最直接的方案，能在错误发生时立即获取处理上下文的所有关键信息，避免依赖外部日志解析。

修改原Lambda代码 #

添加全局异常捕获逻辑，捕获所有错误后构造上报数据，发送至PHP应用的接口：

import boto3
from PIL import Image, ImageOps
import os
import io
import traceback
import json
from urllib.request import Request, urlopen
from urllib.error import URLError

s3 = boto3.client("s3")
output_bucket = "laravo-cloud"
# 替换为你的PHP应用上报接口地址
WEBHOOK_URL = "https://你的PHP应用接口地址"

Image.MAX_IMAGE_PIXELS = 500000000

def lambda_handler(event, context):
    # 初始化关键变量，避免异常时未定义
    org_id = None
    image_name = None
    original_image_key = None
    bucket_name = None
    
    try:
        bucket_name = event["Records"][0]["s3"]["bucket"]["name"]
        original_image_key = event["Records"][0]["s3"]["object"]["key"]

        if "product_images/original/" not in original_image_key:
            print(f"Skipping file {original_image_key} as it does not match the required path.")
            return

        parts = original_image_key.split("/")
        org_id = parts[0]
        image_name = parts[-1]
        image_extension = image_name.split(".")[-1]

        # 原有的下载、处理、上传逻辑（保持不变）
        original_image_obj = s3.get_object(Bucket=bucket_name, Key=original_image_key)
        original_image = Image.open(original_image_obj["Body"])

        sizes_1x1 = [(size, size) for size in range(100, 901, 50)]
        sizes_5x8 = [(size, int(size * 8 / 5)) for size in range(100, 901, 50)]

        original_image.thumbnail((1000, 1000), Image.LANCZOS)

        s3.copy_object(
            Bucket=output_bucket,
            CopySource={'Bucket': bucket_name, 'Key': original_image_key},
            Key=f"{org_id}/product_images/original/{image_name}",
            ACL='public-read'
        )

        for size in sizes_1x1:
            generate_thumbnail_and_upload(
                original_image, size, org_id, "1x1", bucket_name, image_name, image_extension
            )

        for size in sizes_5x8:
            generate_thumbnail_and_upload(
                original_image, size, org_id, "5x8", bucket_name, image_name, image_extension
            )

        s3.delete_object(Bucket=bucket_name, Key=original_image_key)

    except Exception as e:
        # 构造详细报错信息
        error_type = type(e).__name__
        error_msg = str(e)
        error_trace = traceback.format_exc()
        s3_object_url = f"https://s3.console.aws.amazon.com/s3/object/{bucket_name}/{original_image_key}" if bucket_name and original_image_key else "无法获取S3链接"
        
        # 构造上报Payload
        payload = {
            "error_type": error_type,
            "error_message": error_msg,
            "error_trace": error_trace,
            "org_id": org_id,
            "image_name": image_name,
            "s3_object_url": s3_object_url,
            "lambda_request_id": context.aws_request_id
        }
        
        # 发送至PHP应用（使用标准库urllib，无需额外依赖）
        req = Request(
            WEBHOOK_URL,
            data=json.dumps(payload).encode('utf-8'),
            headers={'Content-Type': 'application/json'}
        )
        try:
            with urlopen(req, timeout=10) as response:
                print(f"错误上报成功，状态码：{response.getcode()}")
        except URLError as req_e:
            print(f"上报至PHP应用失败：{str(req_e)}")
        
        # 重新抛出异常，确保CloudWatch记录完整错误
        raise

# 原有的generate_thumbnail_and_upload函数保持不变
def generate_thumbnail_and_upload(
    original_image, target_size, org_id, ratio_folder, bucket_name, image_name, image_extension
):
    aspect_ratio = original_image.width / original_image.height
    target_aspect = target_size[0] / target_size[1]

    if aspect_ratio > target_aspect:
        new_width = target_size[0]
        new_height = int(new_width / aspect_ratio)
    else:
        new_height = target_size[1]
        new_width = int(new_height * aspect_ratio)

    img_resized = original_image.resize((new_width, new_height), Image.LANCZOS)
    thumb_image = Image.new("RGB", target_size, (255, 255, 255))
    paste_x = (target_size[0] - new_width) // 2
    paste_y = (target_size[1] - new_height) // 2
    thumb_image.paste(img_resized, (paste_x, paste_y))

    buffer = io.BytesIO()
    save_format = "JPEG" if image_extension == "jpg" else image_extension.upper()
    thumb_image.save(buffer, format=save_format)
    buffer.seek(0)

    size_folder = str(target_size[0])
    thumb_image_key = f"{org_id}/product_images/{ratio_folder}/{size_folder}/{image_name}"

    s3.put_object(
        Bucket=output_bucket,
        Key=thumb_image_key,
        Body=buffer,
        ContentType=f"image/{image_extension.lower()}",
        ACL='public-read'
    )

方案2：优化CloudWatch告警+SNS+Lambda流程 #

针对之前无法获取有效信息的问题，通过调整日志格式和告警规则，提取关键上下文：

1. 修改Lambda日志输出：在处理开始时打印关键信息，确保错误日志包含org_id和图片名称：

   print(f"开始处理图片：org_id={org_id}, image_name={image_name}, key={original_image_key}")
   

2. 创建CloudWatch Logs告警：基于包含ERROR或特定异常关键词（如ThrottlingException）的日志条目，配置告警触发SNS通知。

3. 编写SNS触发的上报Lambda：解析SNS消息中的日志内容，提取关键信息后上报至PHP应用：

   import json
   import re
   from urllib.request import Request, urlopen
   from urllib.error import URLError
   
   def lambda_handler(event, context):
       WEBHOOK_URL = "https://你的PHP应用接口地址"
       sns_message = json.loads(event['Records'][0]['Sns']['Message'])
       
       for log_event in sns_message.get('logEvents', []):
           log_content = log_event['message']
           
           # 正则提取关键信息（根据日志格式调整）
           org_match = re.search(r'org_id=(\w+)', log_content)
           image_match = re.search(r'image_name=(\S+)', log_content)
           error_match = re.search(r'ERROR: (.*)', log_content)
           
           org_id = org_match.group(1) if org_match else "未知"
           image_name = image_match.group(1) if image_match else "未知"
           error_msg = error_match.group(1) if error_match else log_content
           
           # 构造上报Payload并发送
           payload = {
               "error_type": "CloudWatch告警触发",
               "error_message": error_msg,
               "org_id": org_id,
               "image_name": image_name,
               "log_stream": sns_message.get('logStream', '')
           }
           
           req = Request(
               WEBHOOK_URL,
               data=json.dumps(payload).encode('utf-8'),
               headers={'Content-Type': 'application/json'}
           )
           try:
               with urlopen(req, timeout=10) as response:
                   print(f"错误上报成功：{response.getcode()}")
           except URLError as req_e:
               print(f"上报失败：{str(req_e)}")
   

方案3：处理Lambda执行前的限流失败（DLQ+上报） #

针对Lambda因并发超限被直接拒绝的情况，配置死信队列（DLQ）捕获请求并上报：

1. 配置Lambda死信队列：在Lambda控制台的「配置」→「异步调用」中，设置死信队列为SQS标准队列。

2. 编写DLQ触发的上报Lambda：解析队列中的原始请求，提取关键信息后上报：

   import json
   from urllib.request import Request, urlopen
   from urllib.error import URLError
   
   def lambda_handler(event, context):
       WEBHOOK_URL = "https://你的PHP应用接口地址"
       
       for record in event['Records']:
           message_body = json.loads(record['body'])
           request_payload = json.loads(message_body['requestPayload'])
           
           # 提取S3事件信息
           s3_event = request_payload['Records'][0]['s3']
           bucket_name = s3_event['bucket']['name']
           original_image_key = s3_event['object']['key']
           
           # 解析org_id和图片名称
           parts = original_image_key.split("/")
           org_id = parts[0] if len(parts) > 0 else "未知"
           image_name = parts[-1] if len(parts) > 0 else "未知"
           
           # 构造上报Payload
           payload = {
               "error_type": "Lambda并发限流",
               "error_message": "Lambda并发超限，请求被拒绝",
               "org_id": org_id,
               "image_name": image_name,
               "s3_object_key": original_image_key,
               "bucket_name": bucket_name
           }
           
           # 发送至PHP应用
           req = Request(
               WEBHOOK_URL,
               data=json.dumps(payload).encode('utf-8'),
               headers={'Content-Type': 'application/json'}
           )
           try:
               with urlopen(req, timeout=10) as response:
                   print(f"错误上报成功：{response.getcode()}")
           except URLError as req_e:
               print(f"上报失败：{str(req_e)}")
   

关键注意事项 #

• 确保PHP应用的上报接口支持POST请求，能接收JSON格式数据。
• Lambda执行角色需具备对应权限：访问S3、SQS、发送HTTP请求（若使用VPC，需配置NAT网关）。
• 方案1中建议设置较短的上报超时时间（如10秒），避免占用Lambda执行时长。

内容的提问来源于stack exchange，提问作者Ahmed Alaa ElDin