Lambda调用CloudWatch Logs FilterLogEvents遇InvalidParameterException问题排查

错误根源 #

报错Invalid character(s) in term '<'，结合你的代码排查，核心问题如下：

• 重复定义lambda_handler函数：代码开头存在一个未完成的lambda_handler，不仅会覆盖后续有效函数，还可能引发解析异常。
• 字符串未闭合：sns_topic_arn = 'arn:aws-us-yadyaxxx 缺少结尾单引号，Python解析器会错误地将后续注释、代码片段识别为字符串的一部分，进而触发非法字符报错。
• 时间戳单位错误：CloudWatch Logs要求startTime和endTime为毫秒级时间戳，原代码使用秒级时间戳会导致日志搜索范围完全错误。

修复后的完整代码 #

import boto3
import datetime

def lambda_handler(event, context):
    log_group_name = 'security'  # 日志组名称
    search_string = '4625'       # 目标搜索字符串
    sns_topic_arn = 'arn:aws-us-yadyaxxx'  # 补全SNS主题ARN的闭合引号
    custom_email = 'myemailaddy@example.com'  # 修正邮箱格式

    cloudwatch = boto3.client('logs')
    sns = boto3.client('sns')

    # 计算最近10分钟的时间范围（转换为毫秒级）
    end_time = int(datetime.datetime.now().timestamp() * 1000)
    start_time = end_time - 600 * 1000  # 600秒=10分钟

    # 使用双引号包裹搜索字符串，确保匹配日志中包含该内容的条目
    response = cloudwatch.filter_log_events(
        logGroupName=log_group_name,
        startTime=start_time,
        endTime=end_time,
        filterPattern=f'"{search_string}"'
    )

    occurrences = len(response['events'])  # 更高效的计数方式

    if occurrences >= 5:
        message = f"日志组 {log_group_name} 中，10分钟内检测到 '{search_string}' 出现 {occurrences} 次。"

        # 合并SNS通知调用，避免重复发送
        sns.publish(
            TopicArn=sns_topic_arn,
            Message=message,
            Subject='关键日志事件告警',
            MessageAttributes={
                'email': {
                    'DataType': 'String',
                    'StringValue': custom_email
                }
            }
        )

额外优化说明 #

• 过滤模式规范：用双引号包裹搜索字符串是CloudWatch Logs的标准写法，确保匹配日志中包含该字符串的条目，避免误匹配其他字段的相似值。
• SNS调用优化：原代码两次调用sns.publish会发送重复通知，合并为一次即可完成告警推送。
• 权限验证：确认Lambda角色已配置logs:FilterLogEvents（对应目标日志组）和sns:Publish（对应目标SNS主题）权限。

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