CDKでstep functionsを作ってみた #AWS

はじめに

以前にstep functionsを作る機会があり、その時はCloudFormationにに文字で定義していましたが、
CDKだとメソッドチェインで表現できるんですね
実際に作ってみての所感やコードを共有します

所感

検索しているとCDK Version1の情報が引っかかりがちです。ソースの先頭あたりを確認して、import * from 'aws-cdk-lib';であるのをまず確認
言語はTypeScriptを選択。最終的にはどの言語でもTypeScriptに変換されると聞いたので
VS Codeのコード補完は本当に助かります
困ったら検索するよりもReferenceを見た方が早く正確
Lambdaの権限までも一元管理できるのは本当に便利。デプロイに時間がかかるのはCloudFormationと同じ
ファイルをサービスやオブジェクト等で分けて管理できるのも助かる。CFnでもできるかも知れませんが(やったことなし)
step functionsの終了条件を満さない場合はスタートに戻るフローを素直に実装すると、デプロイ出来ませんでした
ポリシーの生成classが分からず。時間の関係でロール内のインラインポリシーで凌ぎました
--hotswap良いですね。Lambdaのコード修正だと、deployで1分ぐらいかかるのが、5-6秒に。開発サイクルが速く回せるのはありがたい
SFnとCI/CDを別のStackにすると、権限問題を解決できず。administrator権限のユーザで対応することに。CI/CDとアプリを同一stackで対応すればいけるらしいが未検証

ディレクトリ構成

ディレクトリ構成は、下のようになっています
(ファイル名やファイル数は変えています)

lambdaごとにデプロイ用のTypeScriptファイルを作成して,その中にLambdaの設定内容をカプセル化しています
step functions用のTypeScriptファイルを大本のファイルとして、そちらに各TypeScriptのファイルのClassをimportして利用しています

bin/
cdk.out/
lambda/
├── 00_aaaaa
│  └── app.py
├── 01_bbbbb
│  └── app.py
├── layer
│   └── python
│      ├── aaaaa.py
│      └── bbbbb.py
lib/
├── DynamoDB
│  └── aaaaa.ts
├── Lambda
│  ├── 00_aaaaa.ts
│  ├── 01_bbbbb.ts
│  ├── layer.ts
│  └── batch_aaaaa.ts
├── Policy
│  └── 00_aaaaa-policies.ts
└── StepFunctions
   └── aaaaa=sfn-stack.ts
pytest/
node_modules/
test/
cdk.json
package-lock.json
package.json
pytest.ini
README.md
tsconfig.json

コードスニペット

コードのサンプルは探せば出てくると思いますし、AWSからも公開さていますので、ポイント的なところのみ記します

Lambdaはこんな感じ
(変数や名称は変えています)
A. classのプロパティに生成したLambdaを持たせて、外部からも参照できるようにしています
B. ロールもこのファイルで管理しています
C. EventBridgeの設定も同様

lambda.ts

export class AALambda extends Construct {
    public readonly function: lambda.Function;  // A
    constructor(scope: Construct, id: string) {
        super(scope, id);
        const stage: string = this.node.tryGetContext("stage");
        let env = {
            Stage: stage,
            LOGLEVEL: "INFO",
        }
        if (stage == "dev" || stage == "stg") {
            env.LOGLEVEL = "DEBUG";
        }
        const functionName = `AALambda-${stage}`;
        this.function = new lambda.Function(this, functionName,
                {
                  functionName: functionName,
                  code: lambda.Code.fromAsset('lambda/00_aa_lambda',
                      {
                          exclude: ["__pycache__", ".DS_Store]"],
                      }),
                  runtime: lambda.Runtime.PYTHON_3_8,
                  handler: 'app.lambda_handler',
                  timeout: Duration.minutes(5),
                  memorySize: 256,
                  environment: env
                }
              );
        // B
        this.function.role?.addManagedPolicy(ManagedPolicy.fromAwsManagedPolicyName("AmazonXXX"));
        const PolicyClass = new CommonPolicies(this, "ApiSPolicies");
        this.function.role?.attachInlinePolicy(PolicyClass.aaaPolicy);
        // C
        const aaFunction = new LambdaFunction(this.function);
        const ruleName = `aa-rule-${stage}`;
        new Rule(this, ruleName, {
            ruleName: ruleName,
            schedule: Schedule.cron({minute: '0/5'}),
            targets: [aaFunction],
        });
    }
}

Layerは各Lambdaで共通なので、下のように大本のStepFunctionsのファイルで関連づけています

sfn-1.ts

    const layerClass = new CommonLayer(this, `CommonLayer-${stage}`);
    const aLambda = new ALambda(this, `AALambda-${stage}`);
    aLambda.function.addLayers(layerClass.layer);

step function部分で
A. 理由は分かりませんが、Mapからスタートするとコンパイルが通らなかったので、ダミーのステップからスタートしています
B1,2. Mapで外部APIを並列実行しています。Mapで実行してAPIを叩くのはALambdaです。外部APIの実行結果をSFn外のAPI Gateway&Lambdaが受け、BLambdaに戻しています
C. SFnのパラメータはlambdaを定義しているclassに記載しています
D. 規定回数実行したかの確認stepです。順番が前後しますが、下のEのstepがFalseの場合にこのstepを実行します。規定回数実行していない場合は5分待ってリトライします
E. 外部APIの成否判断stepです。成功していたら処理を終え、そうでなければ上記Dのstepを実行します
F1,2. シンプルにsfnChainに個々のオブジェクトを繋げていくと、永久ループになってしまいdeployできませんでした。その対応でDで処理を終了したと見せかけ、F1で1番始めに戻しています

sfn-2.ts

    // A
    const starter = new sfn.Pass(this, 'starter', {comment: 'Do nothing(for CDK compile error)'});
    
    // B1
    const parallelMap = new sfn.Map(this, `Parallel`, {
      itemsPath: sfn.JsonPath.stringAt("$"),
      parameters: {params: sfn.JsonPath.stringAt("$$.Map.Item.Value")}
    });
    // B2, C
    const paraATask = ALambdaClass.parallelTask();
    parallelMap.iterator(paraATask.next(BLambdaClass.parallelTask()));

    const CLambda = new cLambda(this, 'CLambda');

    const Success = new sfn.Succeed(this, "Success")
    const Wait5Min = new sfn.Wait(this, 'Wait5Minutes', {time: sfn.WaitTime.duration(Duration.minutes(5))});
    // D
    const AllEnd = new sfn.Choice(this, 'AllEnd')
        .when(sfn.Condition.booleanEquals('$.all_end', true), Success)
        .otherwise(Wait5Min);
    // E
    const stepSucceed = new sfn.Choice(this, 'TaskSucceed')
        .when(sfn.Condition.booleanEquals('$.task_ok', true), Success)
        .otherwise(AllEnd);
    // F1
    Wait5Min.next(starter);
    // F2
    const sfnDef = sfn.Chain
        .start(starter)
        .next(parallelMap)
        .next(CLambda.task())
        .next(stepSucceed);

    const sfnName = `SFN-${stage}`
    new sfn.StateMachine(this, sfnName, {
      definition: sfnDef,
      stateMachineName: sfnName
    });

AALambda-parallelTask.ts

    parallelTask() {
        return new tasks.LambdaInvoke(this, 'AALambdaParams', {
            lambdaFunction: this.function,
            resultPath: "$.result",
            integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
            payload: sfn.TaskInput.fromObject({
                token: sfn.JsonPath.taskToken,
                "aaaa.$": "$.params.aaaa",
                "bbbb.$": "$.params.bbbb",
            })
        });
    }

終わりに

おそらくCDKでstep functionsを書いたことがないと、良く分からない文字列の羅列にしか見えないと思われます
ですが、CDKでLoopする処理を書く場合や、纏まった数のLambda + Layerを作る際になんらかの助けになれば良いなと思っています