寻求用Python CDK在ECS Fargate运行单容器的最简示例及镜像拉取故障解决

需求背景 #

• 运行单个容器，选用AWS ECS而非EKS集群
• 容器无需HTTP入站暴露，但需具备出站互联网访问能力
• 仅需单实例运行，基于健康检查实现故障自动重启

遇到的错误 #

镜像拉取失败，错误信息如下：

CannotPullContainerError: ref pull has been retried 5 time(s): failed to copy: httpReadSeeker: failed open: failed to do request: Get "https://prod-eu-central-1-starport-layer-bucket.s3.eu-central-1.amazonaws.com/..."

原代码实现 #

class MyStack(Stack):
    def __init__(self, scope: Construct, id: str, props, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        repository = props["ecr-repository"]

        vpc = aws_ec2.Vpc(self, "VPC", vpc_name="vpc", max_azs=2, nat_gateways=0)

        vpc.add_interface_endpoint(
            "S3Endpoint",
            service=aws_ec2.InterfaceVpcEndpointAwsService.S3,
            private_dns_enabled=False,
        )
        vpc.add_interface_endpoint(
            "EcrDockerEndpoint",
            service=aws_ec2.InterfaceVpcEndpointAwsService.ECR_DOCKER,
        )
        vpc.add_interface_endpoint(
            "EcrEndpoint", service=aws_ec2.InterfaceVpcEndpointAwsService.ECR
        )
        vpc.add_interface_endpoint(
            "CloudWatchLogsEndpoint",
            service=aws_ec2.InterfaceVpcEndpointAwsService.CLOUDWATCH_LOGS,
        )

        cluster = aws_ecs.Cluster(
            self, "EcsCluster", cluster_name="ecs-cluster", vpc=vpc
        )

        execution_role = iam.Role(
            self,
            "ecs-devops-execution-role",
            assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
            role_name="ecs-devops-execution-role",
        )
        execution_role.add_to_policy(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                resources=["*"],
                actions=[
                    "ecr:GetAuthorizationToken",
                    "ecr:BatchCheckLayerAvailability",
                    "ecr:GetDownloadUrlForLayer",
                    "ecr:BatchGetImage",
                    "logs:CreateLogStream",
                    "logs:PutLogEvents",
                ],
            )
        )

        task_definition = aws_ecs.FargateTaskDefinition(
            self,
            "EcsFargateTaskDefinition",
            family="EcsFargateFamily",
            execution_role=execution_role,
        )

        image = aws_ecs.ContainerImage.from_ecr_repository(
            repository=repository, tag="1.0.0"
        )

        container = task_definition.add_container(
            "app",
            image=image,
            environment=dict(
                WEBSOCKET_CHANNELS="channel-name", WEBSOCKET_TIMEOUT_DURATION="PT15M"
            ),
        )

        log_group = aws_logs.LogGroup(
            self,
            "ecs-devops-service-logs-groups",
            log_group_name="ecs-devops-service-logs",
        )

        container.add_port_mappings(aws_ecs.PortMapping(container_port=8080))

        service = aws_ecs_patterns.ApplicationLoadBalancedFargateService(
            self,
            "EcsFargateService",
            service_name="fargate-service",
            cluster=cluster,
            desired_count=1,
            task_definition=task_definition,
        )

        service.target_group.configure_health_check(path="/actuator/health")

        self.output_props = props.copy()

修复后的代码 #

class ComputeStack(Stack):
    def __init__(self, scope: Construct, id: str, props, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        repository = props["ecr-repository"]
        secret = props["sm-secret"]

        vpc = aws_ec2.Vpc(self, "VPC", vpc_name="vpc", max_azs=2, nat_gateways=0)

        cluster = aws_ecs.Cluster(
            self, "EcsCluster", cluster_name="ecs-cluster", vpc=vpc
        )

        task_definition = aws_ecs.FargateTaskDefinition(
            self,
            "EcsFargateTaskDefinition",
            memory_limit_mib=2048,
            family="EcsFargateFamily",
        )

        image = aws_ecs.ContainerImage.from_ecr_repository(
            repository=repository, tag="1.0.0"
        )

        container = task_definition.add_container(
            "app",
            image=image,
            logging=aws_ecs.LogDrivers.aws_logs(stream_prefix="ecs-fargate"),
            memory_limit_mib=2048,
            environment=dict(
                WEBSOCKET_CHANNELS="channel-name",
                WEBSOCKET_TIMEOUT_DURATION="PT15M",
            ),
        )

        container.add_port_mappings(aws_ecs.PortMapping(container_port=8080))

        service = aws_ecs_patterns.ApplicationLoadBalancedFargateService(
            self,
            "EcsFargateService",
            service_name="fargate-service",
            cluster=cluster,
            desired_count=1,
            assign_public_ip=True,
            task_definition=task_definition,
        )

        service.target_group.configure_health_check(path="/actuator/health")

        self.output_props = props.copy()

关键修复说明 #

• 启用assign_public_ip=True，让Fargate任务通过公网直接访问ECR镜像存储和S3层桶，无需依赖VPC接口端点
• 移除手动创建的执行角色：CDK会自动生成具备镜像拉取、日志推送所需权限的执行角色，避免权限配置冗余
• 明确配置任务和容器的内存限制（2048MiB），避免因资源不足导致的运行异常
• 直接绑定CloudWatch Logs驱动，无需手动创建LogGroup，简化日志配置流程

非常感谢@gshpychka的帮助！

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