ei blog

クロススタック構成のLamda LayerをCDKでの運用の仕方を検討する

#AWS
#CDK
#Python
#Lambda Layer
#クロススタック

CDKを使ってLambda Layerを親スタックに置き、子スタックのlambda達に連携させる構成を検討したい

CloudFromationのExport/Inportを使用して子lambdaに連携させることは可能ですが、そのままでは以下の課題があります

  • 一度目のデプロイは通るが、Lambda Layerを更新使用とするとLayerのarnが変わることになるので依存関係のエラーが起こり失敗します
  • SSMパラメータストアへLayerのarnを保存し、子スタックでSSM経由で連携させる方法がありますが、こちらはこちらでLayerのみを更新した場合、子スタック側へ更新が検知されない問題があります

  • SSMパラメータストア経由でのLaryerArn渡しはそのまま採用する

  • Layerのソースコードのディレクトリをhash化する変更検知処理を作成し、hash値を子lambdaのdescriptionに埋め込むことで変更を反映させる

root/
  ├ cdk/
  │  ├ .venv
  │  ├ cdk_stacks/
  │  │    ├ modules.py // CDK用にまとめられるmaduleをまとめている
  │  │    │    ├ cdk_tools.py
  │  │    │    ├ lambda_module.py
  │  │    │    └ lambda_layer_module.py
  │  │    ├ __init__.py
  │  │    ├ common_stack.py // 共通で使用したい親スタック
  │  │    └ child_stack.py // 子スタック
  │  └ app.py
  └ functions/src/
               ├ child_hoge_lambda/app.py
               ├ child_huga_lambda/app.py
               └ lambda_layers/common_modules
                  /python // python配下に設置したmoduleを子lambdaで呼び出すことができるようになる
                     └common_modules 
                            └ module_a.py
                            └ module_b.py         
from aws_cdk import (
    Stack,
    aws_ssm as ssm,
)
from cdk_stacks.modules.lambda_layer_module import LambdaLayerModule


class CommonStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        project_prefix = self.node.try_get_context('project_prefix')
        
        # layer
        common_modules_layer_module = LambdaLayerModule(
            stack=self,
            name=f"{project_prefix}CommonModules",
            path="functions/src/lambda_layers/common_modules"
        )
        common_modules_layer = common_modules_layer_module.create_lambda_layer()
        self.common_modules_layer_hash = common_modules_layer_module.create_layer_hash()

        ssm_parameters = [
            {
                "key": "CommonModulesLayerVersionArn",
                "value": common_modules_layer.layer_version_arn
            }
        ]
        for parameter in ssm_parameters:
            ssm_key = parameter["key"]
            ssm.StringParameter(
                self,
                id=f"{ssm_key}Parameter",
                parameter_name=ssm_key,
                string_value=parameter["value"]
            )
import os
from aws_cdk import (
    Stack,
    aws_ssm as ssm,
    aws_lambda as lambda_,
)
from constructs import Construct
from cdk_stacks.modules.lambda_module import LambdaModule

dirname = os.path.dirname(__file__)


class ChildStack(Stack):
    """
    子スタック
    子lambdaをここで定義、親スタックからssm経由でLayerVersionArnとhashを取得し、
    子lambdaのdescriptionに埋め込む
    """

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        common = kwargs['common']
        del kwargs['common']
        super().__init__(scope, construct_id, **kwargs)

        common_modules_layer_hash = common.common_modules_layer_hash

        # SSMパラメータストア経由でLayerVersionArnを取得する
        common_modules_layer_version_arn = ssm.StringParameter.from_string_parameter_name(
            self, "CommonModulesLayerVersionArn",
            string_parameter_name="CommonModulesLayerVersionArn"
        ).string_value
        common_modules_layer = lambda_.LayerVersion.from_layer_version_arn(
            self, "CommonModulesLayer",
            layer_version_arn=common_modules_layer_version_arn
        )

        lambda_params = [
            {
                "name": "ChildHogeLambda",
                "path": "functions/src/child_hoge_lambda",
                "role_props": {
                    "tables": []
                }
            },
            {
                "name": "ChildHugaLambda",
                "path": "functions/src/child_huga_lambda",
                "role_props": {
                    "tables": []
                }
            }
        ]

        for lambda_param in lambda_params:
            lambda_name = lambda_param["name"]
            lambda_module = LambdaModule(
                stack=self,
                name=lambda_name,
                path=lambda_param["path"],
                handler="app.lambda_handler",
                environment={},
                timeout=29,
                memory_size=512,
                layers=[common_modules_layer], # layerを渡す
                layer_hashes=[common_modules_layer_hash], # hashデータを渡す
                role_props=lambda_param["role_props"],
            )
            lambda_function_alias = lambda_module.lambda_function_alias
from aws_cdk import (
    AssetHashType,
    BundlingOptions,
    aws_lambda as lambda_,
)
from cdk_stacks.modules.cdk_tools import CDKTools


class LambdaLayerModule:
    def __init__(
        self,
        scope,
        name: str,
        path: str
    ):
        self.scope = scope
        self.name = name
        self.path = path

    def create_lambda_layer(self):
        """
        pip install -r requirements.txt -t /asset-output/python
        のコマンドでpythonディレクトリにパッケージをインストールすることで、
        子lambdaでパッケージを使えるようになります
        """
        lambda_layer = lambda_.LayerVersion(
            self.scope, f"{self.name}Layer",
            code=lambda_.Code.from_asset(
                self.path,
                asset_hash_type=AssetHashType.SOURCE,
                bundling=BundlingOptions(
                    image=lambda_.Runtime.PYTHON_3_10.bundling_image,
                    command=[
                        "bash",
                        "-c",
                        " && ".join(
                            [
                                "cp -au . /asset-output",
                                "pip install -r requirements.txt -t /asset-output/python",
                            ]
                        ),
                    ],
                    user="root:root"
                ),
            ),
            compatible_runtimes=[
                lambda_.Runtime.PYTHON_3_10],
        )
        return lambda_layer

    def create_layer_hash(self):
        layer_hash = CDKTools.get_dir_hash(self.path)
        return layer_hash
import hashlib
import os


class CDKTools:

    @classmethod
    def get_dir_hash(cls, directory_path: str):
        """
        directory_pathで指定したディレクトリのhashを取得
        CI/CDでpytestを実行すると"__pycache__"などのディレクトリが生成されてしまうので、
        find_except_dirで回避する処理を入れています
        """
        def find_except_dir(root, except_dirs):
            for dir in except_dirs:
                if dir in root:
                    return True
            return False

        if not os.path.exists(directory_path):
            return -1
        hash = hashlib.sha1()
        for root, dirs, files in os.walk(directory_path):
            except_dirs = ["__pycache__"] # 無視したいディレクトリを指定
            if find_except_dir(root, except_dirs):
                continue
            for file in files:
                with open(os.path.join(root, file), "rb") as f:
                    while True:
                        buf = f.read(hash.block_size * 0x800)
                        if not buf:
                            break
                        hash.update(buf)
        return hash.hexdigest()
from aws_cdk import (
    AssetHashType,
    BundlingOptions,
    Duration,
    RemovalPolicy,
    aws_lambda as lambda_,
    aws_logs as logs,
    aws_dynamodb as dynamodb,
    aws_s3 as s3
)
from typing import TypedDict


class RolePropsType(TypedDict, total=False):
    tables: list[dynamodb.Table]
    buckets: list[s3.Bucket]


class LambdaModule:
    def __init__(
        self,
        scope,
        name: str,
        path: str,
        handler: str,
        environment,
        timeout: int,
        memory_size: int,
        layers,
        layer_hashes=None,
        role_props: RolePropsType = {}
    ):
        self.scope = scope
        self.name = name
        self.path = path
        self.handler = handler
        self.environment = environment
        self.timeout = timeout
        self.memory_size = memory_size
        self.layers = layers
        self.layer_hashes = layer_hashes
        self.role_props = role_props

    def joined_hash(self):
        """
        複数のlayerに対応できるようjoinしてまとめる
        """
        if self.layer_hashes is None:
            return ""
        return (", ").join(self.layer_hashes)

    def create_lambda_function(self):
        lambda_function = lambda_.Function(
            self.scope, f"{self.name}Function",
            code=lambda_.Code.from_asset(
                self.path,
                asset_hash_type=AssetHashType.SOURCE,
                bundling=BundlingOptions(
                    image=lambda_.Runtime.PYTHON_3_10.bundling_image,
                    command=[
                        "bash",
                        "-c",
                        " && ".join(
                            [
                                "pip install -r requirements.txt -t /asset-output",
                                "cp -au . /asset-output",
                            ]
                        ),
                    ],
                    user="root:root"
                ),
            ),
            handler=self.handler,
            runtime=lambda_.Runtime.PYTHON_3_10,
            environment=self.environment,
            description='Depend on: {hash}'.format(
                hash=self.joined_hash()),  # ここでdescriptionにhashを埋め込む
            tracing=lambda_.Tracing.ACTIVE,
            timeout=Duration.seconds(
                self.timeout),
            memory_size=self.memory_size,
            current_version_options={
                "removal_policy": RemovalPolicy.RETAIN
            },
            layers=self.layers,
            log_retention=logs.RetentionDays.TWO_MONTHS,
            architecture=lambda_.Architecture.ARM_64,
        )
        return lambda_function

    def create_role_lambda(self):
        role_props = self.role_props
        lambda_function = self.lambda_function

        if role_props.get("tables"):
            for table in role_props["tables"]:
                table.grant_read_write_data(lambda_function)
        
        if role_props.get("buckets"):
            for bucket in role_props["buckets"]:
                bucket.grant_read_write(lambda_function)

    def create_lambda_alias(self):
        self.create_lambda_function()
        self.create_role_lambda()

        self.lambda_function_alias: lambda_.Alias = self.lambda_function.add_alias(
            "Live")
← Back to home