如何避免Lambda误删停止ECS服务关联的旧ECR镜像？

方案一：检查镜像是否被ECS任务定义引用 #

核心逻辑是：删除镜像前先收集所有ECS任务定义中关联的镜像摘要（digest），只要镜像被任务定义引用，无论对应ECS服务是否运行，都跳过删除操作。

修改后的Lambda代码 #

import boto3
from datetime import datetime, timedelta

def get_ecs_referenced_image_digests():
    ecs_client = boto3.client("ecs")
    ecr_client = boto3.client("ecr")
    referenced_digests = set()
    
    # 遍历所有ECS任务定义（处理分页）
    next_token = None
    while True:
        task_defs_params = {"nextToken": next_token} if next_token else {}
        task_defs_response = ecs_client.list_task_definitions(**task_defs_params)
        
        # 获取每个任务定义的详细信息
        for task_def_arn in task_defs_response["taskDefinitionArns"]:
            task_def = ecs_client.describe_task_definitions(taskDefinition=task_def_arn)["taskDefinitions"][0]
            # 解析容器镜像引用
            for container in task_def["containerDefinitions"]:
                image_uri = container["image"]
                if "@sha256:" in image_uri:
                    # 直接提取镜像digest
                    digest = image_uri.split("@")[1]
                    referenced_digests.add(digest)
                else:
                    # 处理标签形式的镜像引用，查询ECR获取对应digest
                    try:
                        repo_name, tag = image_uri.split("/")[-1].split(":")
                        image_details = ecr_client.describe_images(
                            repositoryName=repo_name,
                            imageIds=[{"imageTag": tag}]
                        )["imageDetails"][0]
                        referenced_digests.add(image_details["imageDigest"])
                    except (ecr_client.exceptions.ImageNotFoundException, ValueError):
                        # 镜像不存在或格式异常，跳过
                        continue
        
        next_token = task_defs_response.get("nextToken")
        if not next_token:
            break
    return referenced_digests

def lambda_handler(event, context):
    ecr_client = boto3.client("ecr")
    repo_name = "test"
    # 获取所有需要保留的镜像digest
    protected_digests = get_ecs_referenced_image_digests()
    
    cutoff_date = datetime.now() - timedelta(days=30)
    todo_delete = []
    next_token = None
    
    # 遍历ECR仓库所有镜像（处理分页）
    while True:
        image_params = {"repositoryName": repo_name, "nextToken": next_token} if next_token else {"repositoryName": repo_name}
        images_response = ecr_client.describe_images(**image_params)
        
        for image_detail in images_response["imageDetails"]:
            image_digest = image_detail["imageDigest"]
            # 仅删除超过30天且未被任务定义引用的镜像
            if image_detail["imagePushedAt"] < cutoff_date and image_digest not in protected_digests:
                todo_delete.append(image_digest)
        
        next_token = images_response.get("nextToken")
        if not next_token:
            break
    
    # 批量删除镜像（ECR单次最多删100个）
    if todo_delete:
        for batch_idx in range(0, len(todo_delete), 100):
            batch = todo_delete[batch_idx:batch_idx+100]
            delete_response = ecr_client.batch_delete_image(
                repositoryName=repo_name,
                imageIds=[{"imageDigest": digest} for digest in batch]
            )
            print(f"Deleted batch {batch_idx//100 +1}: {delete_response}")

代码说明 #

• get_ecs_referenced_image_digests：遍历所有ECS任务定义，自动识别镜像的digest引用或标签引用，将所有关联的镜像digest存入集合，作为保护名单。
• 主函数逻辑：先获取保护名单，再筛选出推送时间超过30天且不在保护名单内的镜像执行删除，同时处理了任务定义和ECR镜像列表的分页问题，避免遗漏数据。

额外优化建议 #

• 强制任务定义使用镜像digest：让ECS任务定义直接引用镜像仓库地址@sha256:xxx格式的镜像，而非标签，这样可以彻底避免标签覆盖、镜像删除导致的服务启动失败问题。
• 添加删除前告警：给Lambda配置SNS通知，在执行删除前推送待删除镜像列表，适合需要人工确认的高敏感场景。

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