跨账号CloudWatch查询：如何用AWS Lambda从账号A查询账号B日志？

可行。只需配置跨账号权限并在Lambda中调用对应AWS API即可实现，具体操作步骤如下：

1. 配置账号B的CloudWatch日志组资源策略 #

在账号B中，为目标日志组添加资源策略，允许账号A的Lambda执行角色访问该日志组并执行Insights查询：

1. 登录账号B的AWS控制台，进入CloudWatch服务
2. 找到目标日志组，点击「Actions」→「Add resource policy」
3. 输入如下策略（替换占位符为实际值）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::账号A-ID:role/Lambda执行角色名称"
      },
      "Action": [
        "logs:StartQuery",
        "logs:GetQueryResults",
        "logs:DescribeLogGroups"
      ],
      "Resource": "arn:aws:logs:区域ID:账号B-ID:log-group:目标日志组名称:*"
    }
  ]
}

2. 配置账号A的Lambda执行角色权限 #

在账号A中，为Lambda的执行角色添加权限，允许其调用CloudWatch Insights相关API：

1. 登录账号A的AWS控制台，进入IAM服务
2. 找到Lambda对应的执行角色，点击「Add permissions」→「Create inline policy」
3. 选择「JSON」模式，输入如下策略（替换占位符）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:StartQuery",
        "logs:GetQueryResults",
        "logs:DescribeLogGroups"
      ],
      "Resource": "arn:aws:logs:区域ID:账号B-ID:log-group:目标日志组名称:*"
    }
  ]
}

3. Lambda代码实现跨账号查询 #

以Python为例，编写Lambda代码执行跨账号CloudWatch Insights查询：

import boto3
import time

def lambda_handler(event, context):
    # 初始化CloudWatch客户端
    logs_client = boto3.client('logs')
    
    # 账号B的日志组ARN
    target_log_group = "arn:aws:logs:us-east-1:账号B-ID:log-group:目标日志组名称"
    
    # 定义Insights查询语句（替换为你的查询逻辑）
    query = """
        fields @timestamp, @message
        | filter @message like /ERROR/
        | sort @timestamp desc
        | limit 100
    """
    
    # 启动查询
    response = logs_client.start_query(
        logGroupName=target_log_group,
        startTime=int(time.time()) - 3600,  # 查询最近1小时的日志
        endTime=int(time.time()),
        queryString=query
    )
    
    query_id = response['queryId']
    
    # 轮询等待查询完成
    while True:
        query_result = logs_client.get_query_results(queryId=query_id)
        if query_result['status'] in ['Complete', 'Failed', 'Cancelled']:
            break
        time.sleep(2)
    
    # 处理查询结果
    if query_result['status'] == 'Complete':
        print("查询结果：")
        for result in query_result['results']:
            print({field['field']: field['value'] for field in result})
        return {
            'statusCode': 200,
            'body': query_result['results']
        }
    else:
        print(f"查询失败，状态：{query_result['status']}")
        return {
            'statusCode': 500,
            'body': f"Query failed with status: {query_result['status']}"
        }

4. 配置Lambda定时触发（可选） #

如果需要定期执行查询，可通过CloudWatch Events（EventBridge）配置定时触发：

• 进入账号A的Lambda控制台，找到目标函数
• 点击「Add trigger」，选择「EventBridge (CloudWatch Events)」
• 创建新规则，设置定时表达式（比如rate(1 hour)表示每小时执行一次）

注意事项 #

• 确保Lambda执行角色和日志组资源策略中的ARN、区域、账号ID完全匹配
• 调整查询的startTime和endTime参数以匹配你的时间范围需求
• 若查询结果较大，需额外处理分页逻辑（本例为简化仅获取基础结果）

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