仕事で開発しているプロダクトのアーキテクチャをわかりやすく制御するため、StepFunctionsを導入しようとして調査しました。
Lambdaを主体としたマイクロサービスという形で作成しているのですが、Lambdaを細分化する方針なので、細分化された機能群をStepFunctionsで統合して機能として動かそうかと。
この記事ははオフィシャルで出ているチュートリアルを実行して概要を理解し、StepFunctionsをCDKで作成してデプロイできるところまで進みます。
StepFunctionsの概要
ドキュメントを見ると複雑な用語が並んでいますが、今回は目的に鑑みてStateMachineに特化して解説していこうと思います。
かなりざっくり説明すると、複数のtipsとして作成した機能体を統合して処理とデータの流れを規定、管理するマネジメントツールです。それゆえスケーラビリティの調整も容易であり、より調整しやすく運用しやすいアーキテクチャを実現できます。単一責任原則に従ってコードを細分化して管理する手法という感じで、Lambdaのような関数でなくてもちょっとした調整を行うPassとかChoiceなどの機能体、またそれらの機能体をMap処理に乗せるようなことも簡単なコードでさっくりと実現できます。
このようなツールなので、使い方次第で山ほど価値を生んでくれます。ドキュメントを見ながらご自身の用途を探してみてください。
個人的には結果として仕上がるフロー図と、その処理の進行状況をその図の中で示してくれるUIが気に入っています。
ローカルテストのためのSAMというシステム上でも稼働させられるので、ローカル検証も非常に簡単です。
触って理解を深めよう
では早速公式のチュートリアルを触ってみましょう！いっぱいあるので、興味あるやつを触ってみてください。
ここでは基本的にAmazon States Languageを使って処理の流れを規定しています。ここでAmazon States Languageに慣れておけば、CDKデプロイで作成したStateMachineをリモート上で詳細に検証できるので、やり込んでみてください！
CDKでループのチュートリアルを再現しよう
Githubにアクセスしてmainブランチをクローンして開いてみてください。
このコードが実現するのは、このチュートリアルで作成するものと全く同様のものです。制作物をAmazon States Languageに変換したjsonコードは下記になります。
{
"StartAt": "ConfigureCount",
"States": {
"ConfigureCount": {
"Type": "Pass",
"Result": {
"count": 10,
"index": 0,
"step": 1
},
"ResultPath": "$.iterator",
"Next": "Iterator"
},
"Iterator": {
"Next": "IsCountReached",
"Type": "Task",
"ResultPath": "$.iterator",
"Resource": "arn:aws:lambda:ap-northeast-1:333005747499:function:IteratorLambda"
},
"ExampleWork": {
"Type": "Pass",
"Result": {
"success": true
},
"ResultPath": "$.result",
"Next": "Iterator"
},
"IsCountReached": {
"Type": "Choice",
"Choices": [
{
"Variable": "$.iterator.continue",
"BooleanEquals": true,
"Next": "ExampleWork"
}
],
"Default": "Done"
},
"Done": {
"Type": "Pass",
"End": true
}
}
}
さて、ではリポジトリの内容を見ていきましょう。
Lambdaのスクリプトコードはsrcの中を見ていただければわかるはずですので飛ばし、今回はCDKに特化して説明します。
下準備としてルートディレクトリのターミナルで下記を実行しておいてください。これにより、.distディレクトリにデプロイするLambdaのコードが準備され、CDK周りのモジュールもインストールされます。
npm install
npm run dist
cd aws-cdk
npm install
CDKにおいて、エントリポイントはaws-cdk/cdk.jsonにて宣言します。ここではaws-cdk/bin/index.tsを指定しています。
ここでもしJSファイルを指定する感じにすると、tscでコンパイルしてからでなければデプロイできなくなります。
従って、まずはaws_cdk/bin/index.tsの内容から見ていきます。
import 'source-map-support/register';
import * as cdk from '@aws-cdk/core';
import { AwsStack } from '../lib'; // デプロイするパッケージの中身は../lib/aws-stackにて定義されている
const account = 'AWS_ID';
const app = new cdk.App();
new AwsStack(app, 'CdkLoopDev', {tags: {stage: 'dev'}, env: {region: 'ap-northeast-1', account}}); // stg環境をデプロイする際のCloudFormationパッケージ
このファイルにはあまりごちゃごちゃ書かずに何をデプロイするのかだけ大づかみに記述するのが通例です。ここでは'AWS_ID'となっているaccount定数にAWSのIDとして持っているアカウントID（数字の文字列）を代入してください。ここで大事なのは、デプロイするまとまりのことを僕がCdkLoopDevと呼ぶと定義づけている点のみです。
ここから本番です。コードは下記になります。
import * as sfn from '@aws-cdk/aws-stepfunctions';
import * as tasks from '@aws-cdk/aws-stepfunctions-tasks';
import * as lambda from '@aws-cdk/aws-lambda';
import * as iam from '@aws-cdk/aws-iam';
import * as cdk from '@aws-cdk/core';
export class AwsStack extends cdk.Stack {
constructor(scope: cdk.Construct, id: string, props?: cdk.StackProps) {
super(scope, id, props);
const configCount = new sfn.Pass(this, 'ConfigureCount', {
result: sfn.Result.fromObject({
count: 10,
index: 0,
step: 1,
}),
resultPath: '$.iterator'
});
const role = new iam.Role(this, `LambdaExecuteRole`, {
assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
});
role.addManagedPolicy(iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSLambdaBasicExecutionRole'));
let iterlambda = new lambda.Function(this, 'IteratorLambda', {
functionName: 'IteratorLambda',
code: lambda.Code.fromAsset('../.dist/iterator'),
handler: 'index.handler',
runtime: lambda.Runtime.NODEJS_12_X,
tracing: lambda.Tracing.ACTIVE,
role
});
const iterator = new tasks.LambdaInvoke(this, 'Iterator', {
lambdaFunction: iterlambda,
resultPath: '$.iterator',
payloadResponseOnly: true,
});
const isCountReaced = new sfn.Choice(this, 'IsCountReached');
const exampleWork = new sfn.Pass(this, 'ExampleWork', {
result: sfn.Result.fromObject({
success: true
}),
resultPath: '$.result'
});
const done = new sfn.Pass(this, 'Done', {});
const definition = configCount
.next(iterator)
.next(isCountReaced
.when(sfn.Condition.booleanEquals('$.iterator.continue', true), exampleWork.next(iterator))
.otherwise(done)
);
const stateMachine = new sfn.StateMachine(this, 'CdkLoop', {
definition,
});
}
}
私がまず最初に思ったのが、Amazon States Languageじゃないんだ。。ということでした。
せっかく覚えたのに…というところですが、ご安心ください。どのみちCDKが結果としてどのようなJSONになるかを確認することになるので、努力は無駄になりません。
さて、ざっくりとコードを解説していきます。
● インストールするモジュールは@aws-cdk/aws-stepfunctions @aws-cdk/aws-stepfunctions-tasks @aws-cdk/aws-lambda @aws-cdk/aws-iam @aws-cdk/coreになります。
● 次に下記のコードです。
const configCount = new sfn.Pass(this, 'ConfigureCount', {
result: sfn.Result.fromObject({
count: 10,
index: 0,
step: 1,
}),
resultPath: '$.iterator'
});
sfn.Passという構造体は、AWS States Languageでいうところの"Type": "Pass"に対応しています。今回のコードでは上のJSONでいう"ConfigureCount"というPassに対応しています。ループ処理で徐々にカウントアップさせていく $.iterator内のオブジェクトの初期宣言になります。
● そして次のコードです。
const role = new iam.Role(this, `LambdaExecuteRole`, {
assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
});
role.addManagedPolicy(iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSLambdaBasicExecutionRole'));
let iterlambda = new lambda.Function(this, 'IteratorLambda', {
functionName: 'IteratorLambda',
code: lambda.Code.fromAsset('../.dist/iterator'),
handler: 'index.handler',
runtime: lambda.Runtime.NODEJS_12_X,
tracing: lambda.Tracing.ACTIVE,
role
});
const iterator = new tasks.LambdaInvoke(this, 'Iterator', {
lambdaFunction: iterlambda,
resultPath: '$.iterator',
payloadResponseOnly: true,
});
これはJSON内の"Iterator"に対応するコードになります。まずrole及びiterlambdaに関して。これはlambdaを生成するためのコードになります。必要な権限を付与したIAMロールを定義し、lambdaの宣言と同時にそのロールをアタッチしています。
ここで重要なのはiteratorです。Taskステートに関しては、cdk上ではモジュールごと切り分けられています。このタスクはiterlambdaをinvokeし、返却されるデータを'$.iterator'というresultPathで受け取ります。ここでpayloadResponseOnly: true,が必要な理由ですが、これがないとresultPathを'$.iterator'としていても、そのパスの中にPayloadなどのプロパティが複数生成されてしまうので、このStateMachineで変数的に利用する'$.iterator'をシンプルに保つためにこのプロパティを指定するわけです。このおかげで、lambdaの返り値がそのまま'$.iterator'に保存されます。
● そして次。
const isCountReached = new sfn.Choice(this, 'IsCountReached');
const exampleWork = new sfn.Pass(this, 'ExampleWork', {
result: sfn.Result.fromObject({
success: true
}),
resultPath: '$.result'
});
const done = new sfn.Pass(this, 'Done', {});
const definition = configCount
.next(iterator)
.next(isCountReached
.when(sfn.Condition.booleanEquals('$.iterator.continue', true), exampleWork.next(iterator))
.otherwise(done)
);
まずChoiceステートisCountReachedですが、これは条件に従って分岐させるためのステートになります。'$.iterator.continue'のboolean値がtrueならexampleWorkPass、falseならdonePassに続きますが、宣言の時点では分岐を記述せず、後続のPassを宣言した後で記述します。（ここではdefinition定数内で記述しています）
exampleWorkは'$.result'に{success: true}を格納してiteratorに接続するためのPassであり、doneはEndに続くためだけのPassです。
また、ここまではJSONでいうところの"Next":"〇〇"みたいな定義文を書いていませんでしたが、これをdefinitionで定義しています。これは後続のStateMachineの宣言にて使用するdefinitionプロパティとして使用するものです。
● 最後の宣言文はご覧の通り。
さて解説は以上になりますが、もっと試したいことがある場合にはぜひCDKのリファレンスを覗いてみて試してください。
CDKデプロイ
では早速デプロイしてみましょう！まずはpackage.jsonを編集します。
８行目の、スクリプト定義の"bootstrap"をご覧ください。文字列の中にAWS_IDとあると思いますので、ご自身のAWSアカウントIDと書き換えてください。
で、下記のコマンドを実行します。
npm run build
npm run bootstrap
npm run deploy
これでデプロイは完了です。
awsのマネジメントコンソールにログインし、AWS StepFunctionsのコンソールにアクセスしてください。
ステートマシンの中にCdkLoopで始まる名称のものがあると思いますので、それを開いて実行してみます。
成功すると思いますが、もし失敗したら定義のタブからJSONをチェックし、エラー文が出ていればトラブルシュートしてみてください。
最後に
とまあ手元でのテストの一部を公開してみましたが、StepFunctionsの雰囲気を把握できてきたんじゃないかと思います。StepFunctionsにはアクティビティなどの便利機能がたくさんあるので、やりたいことに従って色々と拡張してみてください。トリガーやMap、Choiceなどを状況に合わせてうまく使えば相当強力な武器になると思います。