使用Pulumi AWS API Gateway Python模块配置CloudWatch日志遇错求助

尝试用pulumi_aws_apigateway模块创建Lambda与API Gateway，并启用CloudWatch的「Full Request and Response Logs」时，触发以下错误：

./__main__.py", line 60, in <module>
        stage_log = apigateway.StageLog('alert-stage-log',
    AttributeError: module 'pulumi_aws_apigateway' has no attribute 'StageLog'

pulumi_aws_apigateway高层封装模块并未提供StageLog属性，以下是替代方案：

用户原代码 #

import json
import pulumi
import pulumi_aws as aws
from pulumi import export
import pulumi_aws_apigateway as apigateway

IDENTITY_NAME = "alert"

role = aws.iam.Role(
    f"{IDENTITY_NAME}-lambda-role",
    assume_role_policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": { "Service": "lambda.amazonaws.com" },
            "Action": "sts:AssumeRole"
        }]
    })
)

policy = aws.iam.RolePolicy(
    f"{IDENTITY_NAME}-sqs-role-policy",
    role=role.id,
    policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Action": ["logs:*", "cloudwatch:*"],
            "Resource": "*",
            "Effect": "Allow",
        },
        {
            "Action": ["sqs:SendMessage"],
            "Effect": "Allow",
            "Resource": "*"
        }
      ],
    })
    )

f = aws.lambda_.Function(
    "alertlambda",
    runtime="python3.9",
    code=pulumi.AssetArchive({
        ".": pulumi.FileArchive("./handler"),
    }),
    timeout=300,
    handler="handler.handler",
    role=role.arn,
    opts=pulumi.ResourceOptions(depends_on=[policy]),
)

api = apigateway.RestAPI(f"{IDENTITY_NAME}api", stage_name="alert", routes=[
    apigateway.RouteArgs(path="/{proxy+}", method="ANY", event_handler=f),
])

# Create a CloudWatch Log Group
log_group = aws.cloudwatch.LogGroup('alert-log-group', name='/aws/api-gateway/alert-api')

# Connect the API Gateway stage to the CloudWatch Log Group
stage_log = apigateway.StageLog('alert-stage-log',
                                rest_api_id=api.id,
                                stage_name="alert",
                                cloudwatch_log_group_arn=log_group.arn)

pulumi.export('url', api.url)

替代方案：使用AWS原生Stage资源配置日志 #

pulumi_aws_apigateway是简化封装，若要配置日志，直接使用pulumi_aws.apigateway.Stage原生资源即可，同时需要确保API Gateway拥有写入CloudWatch的权限。

修改后的完整代码 #

import json
import pulumi
import pulumi_aws as aws
from pulumi import export
import pulumi_aws_apigateway as apigateway

IDENTITY_NAME = "alert"

# Lambda角色（原代码保留）
role = aws.iam.Role(
    f"{IDENTITY_NAME}-lambda-role",
    assume_role_policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": { "Service": "lambda.amazonaws.com" },
            "Action": "sts:AssumeRole"
        }]
    })
)

policy = aws.iam.RolePolicy(
    f"{IDENTITY_NAME}-sqs-role-policy",
    role=role.id,
    policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Action": ["logs:*", "cloudwatch:*"],
            "Resource": "*",
            "Effect": "Allow",
        },
        {
            "Action": ["sqs:SendMessage"],
            "Effect": "Allow",
            "Resource": "*"
        }
      ],
    })
    )

f = aws.lambda_.Function(
    "alertlambda",
    runtime="python3.9",
    code=pulumi.AssetArchive({
        ".": pulumi.FileArchive("./handler"),
    }),
    timeout=300,
    handler="handler.handler",
    role=role.arn,
    opts=pulumi.ResourceOptions(depends_on=[policy]),
)

# 创建API（移除stage_name参数，后续手动创建Stage）
api = apigateway.RestAPI(f"{IDENTITY_NAME}api", routes=[
    apigateway.RouteArgs(path="/{proxy+}", method="ANY", event_handler=f),
])

# 创建CloudWatch日志组
log_group = aws.cloudwatch.LogGroup('alert-log-group', name='/aws/api-gateway/alert-api')

# 创建API Gateway执行角色，用于写入CloudWatch日志
api_gateway_role = aws.iam.Role(
    f"{IDENTITY_NAME}-api-gateway-role",
    assume_role_policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": { "Service": "apigateway.amazonaws.com" },
            "Action": "sts:AssumeRole"
        }]
    })
)

# 给API Gateway角色添加写入CloudWatch的权限
api_gateway_policy = aws.iam.RolePolicy(
    f"{IDENTITY_NAME}-api-gateway-log-policy",
    role=api_gateway_role.id,
    policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": log_group.arn
        }]
    })
)

# 手动创建Stage并配置全请求响应日志
api_stage = aws.apigateway.Stage(
    f"{IDENTITY_NAME}-api-stage",
    rest_api=api.id,
    stage_name="alert",
    deployment=api.deployment,
    access_log_settings=aws.apigateway.StageAccessLogSettingsArgs(
        destination_arn=log_group.arn,
        # 全请求响应日志格式，可根据需求调整
        format=json.dumps({
            "requestId": "$context.requestId",
            "sourceIp": "$context.identity.sourceIp",
            "method": "$context.httpMethod",
            "path": "$context.path",
            "status": "$context.status",
            "responseLength": "$context.responseLength",
            "requestTime": "$context.requestTime",
            "requestBody": "$context.request.body",
            "responseBody": "$context.response.body"
        })
    ),
    opts=pulumi.ResourceOptions(depends_on=[api_gateway_policy])
)

pulumi.export('url', api.url)

关键说明 #

1. 移除RestAPI的stage_name参数：避免自动创建无日志配置的Stage，改用手动创建的aws.apigateway.Stage
2. 新增API Gateway执行角色：API Gateway需要独立角色来写入CloudWatch日志，不能复用Lambda角色
3. 配置access_log_settings：通过该参数关联日志组，并定义日志格式，包含请求/响应体等全量信息
4. 依赖关系：确保Stage在权限策略创建完成后再部署，避免权限不足

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