如何在 AWS 区域中列出预留和预调配了并发的 Lambda 函数？

3 分钟阅读

我想列出我的 AWS 区域中预留和预调配了并发的 AWS Lambda 函数。

解决方法

Lambda 控制台不会显示哪些 Lambda 函数配置了预留和预调配的并发。使用 Lambda Python 脚本，列出在特定 AWS 区域中将预调配和预留的并发作为目标的函数。

先决条件

确保您的 Lambda 函数的 AWS Identity and Access Management (IAM) 角色具有以下权限：

lambda:GetFunctionConcurrency
lambda:ListFunctions
lambda:ListProvisionedConcurrencyConfigs

为 Lambda 函数创建 IAM 角色

完成以下步骤：

请确保 IAM 角色具有下列权限：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iam:CreateRole",
        "iam:PassRole"
      ],
      "Resource": "*"
    }
  ]
}

登录 IAM 控制台。
在左侧导航窗格中，选择 Roles（用户）。
选择 Create role（创建角色）按钮。
在 Select type of trusted entity（选择可信实体类型）部分中，选择 AWS service（AWS 服务），然后选择 Lambda 作为用例。选择 Next: Permissions（下一步：权限）按钮。
搜索 AWSLambdaBasicExecutionRole 策略，然后将其选中。此策略会将基本运行权限授予您的 Lambda 函数。

按如下方式附加所需的自定义权限：

在新的浏览器选项卡或窗口中选择 Create policy（创建策略），以保持角色创建过程的完整。

在策略创建工具中，切换到 JSON 选项卡并输入以下策略文档：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "lambda:GetFunctionConcurrency",
        "lambda:ListFunctions",
        "lambda:ListProvisionedConcurrencyConfigs"
      ],
      "Resource": "*"
    }
  ]
}

选择 Review policy（查看策略），为策略命名（例如 LambdaConcurrencyPermissions），然后选择 Create policy（创建策略）。
返回角色创建选项卡或窗口。刷新策略列表（如有必要），然后通过搜索策略名称并将其选中来附加刚才创建的策略。
（可选）继续执行 Next: Tags（下一步：标签）。根据需要添加标签，然后选择 Next: Review（下一步：审核）。
为您的角色命名（例如，LambdaExecutionRoleWithConcurrencyPermissions）。查看角色详情，然后选择 Create role（创建角色）。

创建 Lambda 函数

完成以下步骤：

打开 AWS Lambda 控制台。
选择 Create function（创建函数）按钮。
前往 Author from scratch（从头开始创作）部分。
在 Function name（函数名称）字段中输入 Lambda 函数的名称。
选择 Python 3.10 作为运行时。
在 Permission（权限）部分的 Change default execution role（更改默认执行角色）下，选择 Use an existing role（使用现有角色）。
在 Existing role（现有角色）下拉列表中，选择您在前面的步骤中创建的 IAM 角色。
选择角色后，选择 Create Function（创建函数）。

使用以下代码替换默认的 Lambda 函数代码：

import json
import boto3

def lambda_handler(event, context):
    # Select the region you would like to list function concurrency
    region = "region"

    lambda_client = boto3.client('lambda', region_name=region)
    count = 0

    # Create a paginator for the 'list_functions' operation.
    paginator = lambda_client.get_paginator('list_functions')
    lambda_iterator = paginator.paginate()

    # For loop to iterate through listed functions and return function name and concurrency response
    for items in lambda_iterator:
        for function in items['Functions']:
            function_name = function['FunctionName']

            try:
                response = lambda_client.get_function_concurrency(
                    FunctionName=function_name
                )

                # If reserved concurrency is configured, print the details
                if 'ReservedConcurrentExecutions' in response:
                    count += 1
                    reserved_concurrency = response['ReservedConcurrentExecutions']
                    print(f"{count}. Function Name: {function_name}, Reserved Concurrency: {reserved_concurrency}")

            except lambda_client.exceptions.ResourceNotFoundException:
                # If the function is not found, skip it
                pass
            except Exception as e:
                print(f"Error retrieving concurrency for {function_name}: {e}")

            try:
                response = lambda_client.list_provisioned_concurrency_configs(
                    FunctionName=function_name
                )

                # If provisioned concurrency is configured, print the details
                if 'ProvisionedConcurrencyConfigs' in response and response['ProvisionedConcurrencyConfigs']:
                    provisioned_concurrency = response['ProvisionedConcurrencyConfigs'][0]['RequestedProvisionedConcurrentExecutions']
                    count += 1
                    print(f"{count}. Function Name: {function_name}, Provisioned Concurrency: {provisioned_concurrency}")

            except lambda_client.exceptions.ResourceNotFoundException:
                # If the function is not found, skip it
                pass
            except Exception as e:
                print(f"Error retrieving provisioned concurrency for {function_name}: {e}")

该代码会遍历每个 Lambda 函数。该函数会调用 get_function_concurrency 方法来检索有关指定 AWS 区域中每个函数的预留或预调配的并发配置的详细信息。

将函数代码更新到所需的目标 AWS 区域

在使用 Lambda 函数之前，请将区域更新到您想要列出具有预调配和预留并发函数的位置。

例如，要查看 us-east-1 AWS 区域中 Lambda 函数的预留和预调配并发，请更新 region 变量，如下所示：

region = "us-east-1"

运行 Lambda 函数

要运行您创建的 Lambda 函数，请导航到“Test”（测试）选项卡。然后，按照下图所示的步骤进行操作。

在 Lambda 函数中，选择 Test（测试）选项卡。
在 Configure test event（配置测试事件）对话框中，使用示例事件模板来运行测试。
输入测试的事件名称，例如 Test。
要创建测试事件，请选择 Save（保存）。
完成后，选择 Test（测试）来运行测试。

您可以在函数日志中查看输出，即列表。或者，您可以在 Lambda 函数的 Test（测试）选项卡中查看配置了预留或预调配并发的列表。该列表包含以下数据：

函数名称
配置的并发类型
配置的并发数

输出示例：

1. Function Name: ProvisionedConcurrencyTest, Provisioned Concurrency: 10
2. Function Name: ReservedConcurrencyTest, Reserved Concurrency: 50