如何配置Serverless.yml将Lambda日志分送不同CloudWatch日志组

要实现将不同级别日志发送到指定CloudWatch日志组，仅靠serverless.yml配置无法完成（CloudWatch本身不支持按日志级别自动路由），需要结合Serverless资源/权限配置与Python日志处理器逻辑来实现。

一、Serverless.yml 配置 #

1. 定义CloudWatch日志组（可选） #

如果日志组尚未创建，可以在resources模块中声明，同时设置日志保留天数：

service: your-fastapi-service

provider:
  name: aws
  runtime: python3.11
  region: us-east-1
  # 给Lambda添加写入两个日志组的权限
  iamRoleStatements:
    - Effect: Allow
      Action:
        - logs:CreateLogStream
        - logs:PutLogEvents
      Resource:
        - "arn:aws:logs:${self:provider.region}:${aws:accountId}:log-group:/aws/lambda/firstGroup:*"
        - "arn:aws:logs:${self:provider.region}:${aws:accountId}:log-group:/aws/lambda/secondGroup:*"

resources:
  Resources:
    FirstLogGroup:
      Type: AWS::Logs::LogGroup
      Properties:
        LogGroupName: /aws/lambda/firstGroup
        RetentionInDays: 30
    SecondLogGroup:
      Type: AWS::Logs::LogGroup
      Properties:
        LogGroupName: /aws/lambda/secondGroup
        RetentionInDays: 30

functions:
  api-handler:
    handler: main.handler
    events:
      - httpApi: '*'
    # 注释掉默认日志组配置，避免冲突
    # logRetention: 30

2. 依赖声明 #

在项目根目录创建requirements.txt，添加日志处理依赖：

fastapi
uvicorn
serverless-wsgi
watchtower
boto3

二、Python日志处理器配置 #

使用watchtower库（专门用于Python向CloudWatch写入日志），给不同级别日志配置独立处理器，指定对应的日志组：

import logging
import watchtower

# 初始化根Logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 避免重复添加处理器（Lambda冷启动可能重复执行）
if not logger.handlers:
    # 1. 配置INFO级别日志处理器，定向到firstGroup
    info_handler = watchtower.CloudWatchLogHandler(log_group='/aws/lambda/firstGroup')
    info_handler.setLevel(logging.INFO)
    # 过滤：仅允许INFO级别日志通过
    info_filter = logging.Filter()
    info_filter.filter = lambda record: record.levelno == logging.INFO
    info_handler.addFilter(info_filter)

    # 2. 配置ERROR级别日志处理器，定向到secondGroup
    error_handler = watchtower.CloudWatchLogHandler(log_group='/aws/lambda/secondGroup')
    error_handler.setLevel(logging.ERROR)

    # 添加处理器到Logger
    logger.addHandler(info_handler)
    logger.addHandler(error_handler)

def foo(some_data):
    logger.info(f'calling the function with the data: {some_data}') # 发送至firstGroup
    try:
        doSomething()
    except Exception:
        logger.error('ERROR! Something happened', exc_info=True) # 发送至secondGroup

关键注意事项 #

• 确保Lambda执行角色的权限覆盖两个目标日志组，serverless.yml中的iamRoleStatements已配置对应权限；
• watchtower会自动创建日志流，无需手动操作；
• 如果日志组已提前创建，可删除resources中的日志组定义，仅保留权限配置；
• 过滤逻辑可根据需求调整（比如将WARNING级别也定向到secondGroup，修改过滤条件即可）。

内容的提问来源于stack exchange，提问作者carlos de la morena