CDKでカスタムリソースを定義する

はじめに

cdkでカスタムリソースを実装する方法を調べて試してみたのでメモする。

カスタムリソースとは?

AWS CloudFormation のリソースタイプとして使用できないリソースを含める必要がある場合に使用できる。

テンプレートにカスタムのプロビジョニングロジックを記述してリソースを定義できる。

公式doc

CDKで使える4つのカスタムリソースプロバイダ

カスタムリソースプロバイダ自体の説明は、custom resource provider 参照。

4つのプロバイダについてはcdkドキュメントから引用。

今回はcustom-resources.Provider を試してので、この内容を記載する。

ProviderCompute TypeError HandlingSubmit to CloudFormationMax TimeoutLanguageFootprint
sns.TopicSelf-managedManualManualUnlimitedAnyDepends
lambda.FunctionAWS LambdaManualManual15minAnySmall
core.CustomResourceProviderAWS LambdaAutoAuto15minNode.jsSmall
custom-resources.ProviderAWS LambdaAutoAutoUnlimited AsyncAnyLarge

cdk(python)でカスタムリソースを定義してみる

AWS CDK Custom Resources を参考に実装してみる。

ミニフレーム的な立ち位置だと理解。

The @aws-cdk/custom-resources.Provider  construct is a “mini-framework” for implementing providers for AWS CloudFormation custom resources. The framework offers a high-level API which makes it easier to implement robust and powerful custom resources.

Provider Framework ExamplesのS3File の例をpythonで(色々省きながら)実装した。

概要としては、対象のs3バケットにオブジェクトをputするカスタムリソースである。

以下、実装したコード。

  1. cdk_custom_resource_1_stack.py: カスタムリソースを含むスタック
  2. app.py: カスタムのリソースをデプロイするlambdaコード
# cdk_custom_resource_1_stack.py

from aws_cdk import (
    Stack,
    CustomResource,
    aws_lambda as lambda_,
    aws_iam as iam,
    aws_s3 as s3,
)
from constructs import Construct
from aws_cdk.custom_resources import Provider

class CdkCustomResource1Stack(Stack):

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

        s3_bucket = s3.Bucket(self, 's3-bucket-cr1')

        provider = Provider(
            self, 's3file-provider',
            on_event_handler=lambda_.Function(
                self, 's3file-on-evnet',
                code=lambda_.Code.from_asset('lambda_function'),
                runtime=lambda_.Runtime.PYTHON_3_9,
                handler='app.on_evnet',
                initial_policy=[
                    iam.PolicyStatement(
                        resources=['*'],
                        actions=[
                            's3:GetObject*',
                            's3:GetBucket*',
                            's3:List*',
                            's3:DeleteObject*',
                            's3:PutObject*',
                            's3:Abort*',
                        ]
                    )
                ]
            )  # type: ignore
        )

        custom_resource = CustomResource(
            self, 'Resource',
            service_token=provider.service_token,
            resource_type='Custom::S3File',
            properties={
                'BucketName': s3_bucket.bucket_name,
                'ObjectKey': 'demo.txt',
                'Contents': 'hell!\\nworld! ver.5',
                'PublicRead': False,
            }
        )

        # custom_resource.node.add_dependency(s3_bucket)

        self.object_key = custom_resource.get_att_string('ObjectKey')
        self.url = custom_resource.get_att_string('URL')
        self.object_key = custom_resource.get_att_string('ETag')
# app.py
import boto3

# s3 = boto3.resource('s3')
s3 = boto3.client('s3')

def on_evnet(event, context):
    print(event)

    request_type = event['RequestType']
    if request_type == 'Create':
    elif request_type == 'Update':
        return put_object(event)
    elif request_type == 'Delete':
        return delete_object(event)

def put_object(event):
    bucket_name = event['ResourceProperties']['BucketName']
    if not (bucket_name):
        raise Exception('"BucketName" is required')

    contents = event['ResourceProperties']['Contents']
    if not (contents):
        raise Exception('"Contents" is required')

    object_key = event['ResourceProperties']['ObjectKey']
    if not (object_key):
        raise Exception('"ObjectKey" is required')

    public_read = convert_str_bool(event['ResourceProperties']['PublicRead'])
    print(f'public read = {public_read}')
    print(f'writing s3://{bucket_name}/{object_key}')

    response = s3.put_object(
        Bucket=bucket_name,
        Key=object_key,
        Body=contents,
        ACL='public-read' if public_read else 'private',
    )

    return {
        'PhysicalResourceId': object_key,
        'Data': {
            'ObjectKey': object_key,
            'ETag': response['ETag'],
            'URL': f'https://{bucket_name}.s3.amazonaws.com/{object_key}',
        }
    }

def delete_object(event):
    bucket_name = event['ResourceProperties']['BucketName']
    if not (bucket_name):
        raise Exception('"BucketName" is required')

    object_key = event['PhysicalResourceId']
    if not (object_key):
        raise Exception('PhysicalResourceId expected for DELETE events')

    s3.delete_object(
        Bucket=bucket_name,
        Key=object_key,
    )

# CustomResourceのpropertiesに定義したbool値は文字列としてlambdaで取得できるので変換する
def convert_str_bool(str_bool_value):
    if str_bool_value == 'true':
        return True
    elif str_bool_value == 'false':
        return False
    else:
        raise Exception('str bool value not expected')

補足・所感

  • CustomResourceProvider のimport元が異なり、適切に実装するのに少し苦戦
  • 抽象化されているので、あまり深い理解ができていない
    • cfnで定義する場合、lambdaがインラインかどうかによってレスポンス方法が異なるはず
  • 今回の実装だと Update のときにオブジェクトが作成され、スタックを削除したときにオブジェクトが削除されることを確認
  • cdkで生成したcfnテンプレートを確認すると、理解が捗るかもしれない

終わりに

class AwsCustomResource (construct) などもあるので、暇なときに触ってみたい。

自前でlambdaを実装せずに、特定のapiをコールするだけのカスタムリソースならば、容易に実装できそう。

AWS CDK Conference Japan 2022 のチャプター「1805 – 1835 AWS Outposts 上のリソースを CDK する 福田優真(NTT Communications イノベーションセンター)」でも簡単に紹介されているので、おすすめです。

あと、分かりやすい記事をはっておきます。

無料相談実施中
AWSを使用したサーバーレスアプリケーションの構築
サーバーレス開発内製化、AWSへの移行等
様々な課題について無料でご相談お受けいたします。
最新情報をチェックしよう!