Python3.9のサポート期限が迫り、Lambdaランタイムを更新する必要があります。
自分の環境では複数のAWSアカウントを所有しており、マネジメントコンソールによる各Lambdaランタイムの更新は非常に手間がかかると感じました。
そこで今回は、IAMRole,Lambda,StepFunctios,という要素を使い、
クロスアカウントロールで複数のLambdaランタイムを一括更新してみました。
システム構成図
今回のシステム構成図は以下になります。
基本は下記記事を参考にさせていただいており、IAM権限設定やlambdaRunTimeUpdate関数処理内容をカスタマイズし、クロスアカウントでのLambdaランタイム更新をしていきます。
https://qiita.com/hirosys-biz/items/58465b5221e3e60b576a
IAMの設定
MaintRole（アカウントA用）,MaintRole（アカウントB用）を作成
まずは実際に各Lambda関数ランタイムバージョンをアップデートする際に、lambdaRuntimeUpdate関数がスイッチロールする、保守用IAMロールを作成します。
MaintRole(アカウントA用）
信頼されたエンティティタイプ：AWSアカウント（AccountA)
許可ポリシー：AdministratorAccess
MaintRole（アカウントB用）
信頼されたエンティティタイプ：AWSアカウント（AccountA)
許可ポリシー：AdministratorAccess
lambdaRuntimeUpdateRoleを作成
lambdaRuntimeUpdate関数の許可アクション権限定義する、lambdaRuntimeUpdateRoleを作成します。
lambdaRuntimeUpdatePolicy
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": "arn:aws:logs:*:*:*"
},
{
"Effect": "Allow",
"Action": [
"lambda:ListFunctions",
"lambda:UpdateFunctionConfiguration"
],
"Resource": "*"
}
]
}
maintRole_AssumePolicy
{
"Version": "2012-10-17",
"Statement": {
"Effect": "Allow",
"Action": "sts:AssumeRole",
"Resource": [
"arn:aws:iam::<AccountA_ID>:role/MaintRole",
"arn:aws:iam::<AccountB_ID>:role/MaintRole"
]
}
}
lambdaRuntimeUpdateRole
信頼されたエンティティ：AWSのサービス>Lambda
許可を追加：lambdaRuntimeUpdatePolicy,maintRole_AssumePolicy
Lambda関数の設定
lambdaRuntimeUpdate関数を作成
関数名：lambdaRuntimeUpdate
ランタイム：Python3.12
実行ロール：lambdaRuntimeUpdateRole
lambdaRuntimeUpdate関数のPythonコード修正
import traceback
import logging
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def get_cross_account_lambda_client(role_arn):
sts_client = boto3.client('sts')
assumed_role = sts_client.assume_role(
RoleArn=role_arn,
RoleSessionName="CrossAccountLambdaUpdate"
)
credentials = assumed_role['Credentials']
lambda_client = boto3.client(
'lambda',
aws_access_key_id=credentials['AccessKeyId'],
aws_secret_access_key=credentials['SecretAccessKey'],
aws_session_token=credentials['SessionToken']
)
return lambda_client
def lambda_handler(event, context):
# 入力を検証
if event:
if 'current' in event:
current = event['current'].lower()
else:
return {"Message": "current が存在しないか、現在のランタイムが指定されていません"}
if 'target' in event:
target = event['target'].lower()
else:
return {"Message": "target が存在しないか、新しいランタイムが指定されていません"}
if 'account_ids' in event:
#isinstance は Python の組み込み関数で、指定したオブジェクトが特定のデータ型（クラス）に属しているかを確認するための関数
#event['account_ids'] がリストであるかどうかを確認して、もしevent['account_ids'] が単一の文字列等であっても、リストに変換
account_ids = event['account_ids'] if isinstance(event['account_ids'], list) else [event['account_ids']]
else:
return {"Message": "account_ids が存在しないか、アカウントIDが指定されていません"}
else:
return {"Message": "StepFunctionsから次のように入力してください。{\"current\": <現在のランタイム>, \"target\": <新しいランタイム>, \"account_ids\": [<アカウントID>]}"}
# 各アカウントでランタイムの更新処理を行う
update_results = []
for account_id in account_ids:
try:
role_arn = f"arn:aws:iam::{account_id}:role/MaintRole"
lambda_client = get_cross_account_lambda_client(role_arn)
paginator = lambda_client.get_paginator('list_functions')
# Lambdaランタイム更新処理
target_functions = []
for lambdafunctions in paginator.paginate():
for lambdafunction in lambdafunctions['Functions']:
if current == lambdafunction['Runtime']:
try:
target_functions.append(lambdafunction['FunctionName'])
lambda_client.update_function_configuration(
FunctionName=lambdafunction['FunctionName'],
Runtime=target
)
except:
logger.error(f"アカウント {account_id} の関数 {lambdafunction['FunctionName']} の更新中にエラーが発生しました: {traceback.format_exc()}")
# 現在のアカウントの結果を追加
update_results.append({
"AccountId": account_id,
"CurrentRuntime": current,
"TargetRuntime": target,
"UpdatedFunctions": target_functions
})
except Exception as e:
logger.error(f"アカウント {account_id} の処理中にエラーが発生しました: {traceback.format_exc()}")
update_results.append({
"AccountId": account_id,
"Error": str(e)
})
# すべてのアカウントの結果を返す
return update_results
StepFunctionsの設定
LambdaRuntimeUpdateStateMachine（StepFunctionのステートマシン）の作成
{
"Comment": "state machine of LambdaRuntimeUpdate",
"StartAt": "LambdaRuntimeUpdate",
"States": {
"LambdaRuntimeUpdate": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"Parameters": {
"Payload.$": "$",
"FunctionName": "arn:aws:lambda:us-east-1:<AccountA_ID>:function:lambdaRuntimeUpdate:$LATEST"
},
"Retry": [
{
"ErrorEquals": [
"Lambda.ServiceException",
"Lambda.AWSLambdaException",
"Lambda.SdkClientException",
"Lambda.TooManyRequestsException"
],
"IntervalSeconds": 1,
"MaxAttempts": 3,
"BackoffRate": 2
}
],
"End": true,
"OutputPath": "$.Payload"
}
}
}
LambdaRuntimeUpdateStateMachineの実行
すべての設定が完了しました。
下記入力値でLambdaRuntimeUpdateStateMachineを実行してみましょう。
{
"current": "python3.9",
"target": "python3.12",
"account_ids": [
"AccountA_ID",
"AccountB_ID"
]
}
LambdaRuntimeUpdateStateMachine上で、下記のような出力が得られれば成功です！
[
{
"AccountId": "AccountA_ID",
"CurrentRuntime": "python3.9",
"TargetRuntime": "python3.12",
"UpdatedFunctions": [
"test01_Lambda"
]
},
{
"AccountId": "AccountB_ID",
"CurrentRuntime": "python3.9",
"TargetRuntime": "python3.12",
"UpdatedFunctions": [
"test01_Lambda",
"test02_Lambda"
]
}
]
まとめ
以上、「クロスアカウントロールで複数のLambdaランタイムを一括更新」してみました。
セキュリティ設定やlambdaRuntimeUpdate関数は環境に応じてカスタマイズする必要があるかもしれないですが、上記をもとに運用作業の負荷を減らし、開発業務に専念していきましょう。