求可用Cloud Custodian策略/CDK代码删除90天未调用Lambda

正确策略示例 #

policies:
  - name: lambda-delete-unused-90d
    resource: aws.lambda
    filters:
      - type: metrics
        namespace: AWS/Lambda
        metric: Invocations
        statistic: Sum
        period: 86400  # 1天的秒数
        look_back_days: 90
        comparison: eq
        value: 0
    actions:
      - type: delete

策略说明 #

• 修正了统计量参数：Lambda的Invocations指标仅支持Sum作为合法统计类型，这是之前版本3、4报错的核心原因
• 通过look_back_days:90指定检查周期，period:86400按天聚合指标数据，确保覆盖完整的90天窗口
• 过滤条件Sum=0精准匹配90天内无任何调用记录的函数
• 执行角色需具备以下权限：
  • lambda:ListFunctions
  • cloudwatch:GetMetricStatistics
  • lambda:DeleteFunction

过往版本失败原因 #

• 版本1：LastModified仅代表函数最后修改时间，与执行时间无关联，无法筛选未执行的函数
• 版本2：Lambda资源无LastExecuted原生字段，Cloud Custodian无法识别该过滤条件
• 版本3：使用了Invocations指标不支持的统计量（如Average、Maximum），触发参数校验失败
• 版本4：混合过滤条件时未遵循metrics参数格式要求，导致statistic参数异常

核心逻辑 #

通过CDK部署一个定时触发的Lambda函数，扫描所有Lambda资源，查询CloudWatch Metrics判断是否90天无调用，批量删除符合条件的函数。

CDK代码示例 #

from aws_cdk import (
    Stack,
    aws_lambda as _lambda,
    aws_events as events,
    aws_events_targets as targets,
    aws_iam as iam,
)
from constructs import Construct
from datetime import timedelta

class UnusedLambdaCleanupStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # 创建清理Lambda函数
        cleanup_lambda = _lambda.Function(
            self, "UnusedLambdaCleanup",
            runtime=_lambda.Runtime.PYTHON_3_11,
            handler="index.lambda_handler",
            code=_lambda.Code.from_inline("""
import boto3
from datetime import datetime, timedelta

lambda_client = boto3.client('lambda')
cloudwatch_client = boto3.client('cloudwatch')

def lambda_handler(event, context):
    # 分页获取所有Lambda函数
    paginator = lambda_client.get_paginator('list_functions')
    for page in paginator.paginate():
        for function in page['Functions']:
            function_name = function['FunctionName']
            function_arn = function['FunctionArn']
            
            # 查询过去90天的Invocations指标
            end_time = datetime.utcnow()
            start_time = end_time - timedelta(days=90)
            
            response = cloudwatch_client.get_metric_statistics(
                Namespace='AWS/Lambda',
                MetricName='Invocations',
                Dimensions=[{'Name': 'FunctionName', 'Value': function_name}],
                StartTime=start_time,
                EndTime=end_time,
                Period=86400,
                Statistics=['Sum'],
                Unit='Count'
            )
            
            # 判断是否无调用记录
            if not response['Datapoints'] or all(dp['Sum'] == 0 for dp in response['Datapoints']):
                print(f"Deleting unused Lambda function: {function_name}")
                lambda_client.delete_function(FunctionName=function_arn)
"""),
            timeout=timedelta(minutes=10),
        )

        # 给Lambda添加必要权限
        cleanup_lambda.add_to_role_policy(iam.PolicyStatement(
            actions=[
                "lambda:ListFunctions",
                "lambda:DeleteFunction",
                "cloudwatch:GetMetricStatistics"
            ],
            resources=["*"]
        ))

        # 设置每天凌晨2点定时触发
        rule = events.Rule(
            self, "CleanupSchedule",
            schedule=events.Schedule.cron(hour="2", minute="0")
        )
        rule.add_target(targets.LambdaFunction(cleanup_lambda))

部署说明 #

1. 将代码保存为unused_lambda_cleanup_stack.py
2. 执行CDK部署命令：

   cdk deploy UnusedLambdaCleanupStack
   

3. 部署完成后，每天凌晨2点自动执行清理任务

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