※本記事は、AWS CDK Advent Calendar 2022の9日目の記事となります

この記事の目的

今回は、大きく2つの話を書こうと思います。

CDKでStep Functionsのステートマシンをデプロイする
ステートマシンを環境に依存しないようにしてあげる

ステートマシンのデプロイのあれこれ

Step Functionsのステートマシンをデプロイする方法はいくつかあります。

マネコンのビジュアルワークフローで作る
ステートマシンのJSON定義をコピペする
- 開発機でビジュアルワークフローで作ったものを本番機にコピペとか
CloudFormationでデプロイ
SAMでデプロイ

何気に多いのは、マネコンのビジュアルワークフローで作る、そしてそれを別の環境にコピペする、ではないでしょうか。
これ、ステートマシンをコードで管理できてないですよね。したとしても、手でコピペする時点でイケてないと思います。

あと、CloudFormationでデプロイできますが、問題が一つ。
ステートマシンをIaCの中に直接含めてしまうことへの是非です。
公式ドキュメントにもありますが、こんな感じで記述することは可能です。

  Type: "AWS::StepFunctions::StateMachine"
  Properties:
    DefinitionString:
      !Sub
        - |-
          {
            "Comment": "A Hello World example using an AWS Lambda function",
            "StartAt": "HelloWorld",
            "States": {
              "HelloWorld": {
                "Type": "Task",
                "Resource": "${lambdaArn}",
                "End": true
              }
            }
          }
        - {lambdaArn: !GetAtt [ MyLambdaFunction, Arn ]}
    RoleArn: !GetAtt [ StatesExecutionRole, Arn ]

ステートマシンは、インフラのコードなのか、アプリケーション（のシェルのような位置付け）なのか、といと個人的には後者なイメージです。だから、すごく違和感なんです。
ステートマシン（つまりフロー）を修正するたびにスタックを更新するって、何か気持ち悪いですよね。
もちろん、他のスタックとは独立したスタックにすることになるんでしょうけど。

あと、最後のSAM。
これは、ステートマシンの定義をJSONで管理できます。だから便利ですね。
ただ、SAMなのでサーバーレスアプリが前提、という制約があります。
ステートマシンだけを管理する、ということであれば違和感ないですが、今回はCDKのことも書きたいので、あえてサーバーレスしかできないしね、ということにさせてください。

CDKでステートマシンをデプロイ

ということで、少し強引ですが、CDKでデプロイっていう選択肢について書いていきたいと思います。

前提

言語はTypeScript
ステートマシンからECSタスクをバッチのジョブとして実行する想定
事前に、ECSクラスター、タスク定義が作成されていること

CDKのコード作成

プロジェクト作成

mkdir sfn
cd sfn
cdk init --language=typescript

スタックの記述

まずは、`lib/sfn-stack.ts」を下記のコードに置き換えてください。後ほど解説します。

import * as cdk from 'aws-cdk-lib';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as stepfunctions from 'aws-cdk-lib/aws-stepfunctions';
import * as fs from 'fs';
import { Construct } from 'constructs';

export class SfnStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // The code that defines your stack goes here
    const logPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    logPolicyStatement.addActions(
      "logs:CreateLogDelivery",
      "logs:GetLogDelivery",
      "logs:UpdateLogDelivery",
      "logs:DeleteLogDelivery",
      "logs:ListLogDeliveries",
      "logs:PutResourcePolicy",
      "logs:DescribeResourcePolicies",
      "logs:DescribeLogGroups"
    );
    logPolicyStatement.addResources("*");
    const ecsTaskPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    ecsTaskPolicyStatement.addActions(
      "ecs:RunTask",
      "ecs:StopTask",
      "ecs:DescribeTasks",
      "iam:PassRole"
    );
    ecsTaskPolicyStatement.addResources("*");
    const sfnPolicy : iam.Policy = new iam.Policy(this, "sfn-statemachine-policy", {
      policyName: "sfn-batch-policy",
      statements: [
        logPolicyStatement,
        ecsTaskPolicyStatement
      ]
    });
    const sfnStatemachineRole : iam.Role = new iam.Role(this, "sfn-statemachine-role", {
      roleName: "sfn-batch-role",
      assumedBy: new iam.ServicePrincipal("states.amazonaws.com"),
      path: "/"
    });
    sfnStatemachineRole.attachInlinePolicy(sfnPolicy);
    const logGroup : logs.LogGroup = new logs.LogGroup(this, 'sfn-logs-job001', {
      logGroupName: "/sfn/job001",
      retention: logs.RetentionDays.THIRTEEN_MONTHS
    });
    const file : Buffer = fs.readFileSync("statemachine/job001.json");
    const stateMachine : stepfunctions.CfnStateMachine = new stepfunctions.CfnStateMachine(this, "sfn-statemachine-job001", {
      roleArn: sfnStatemachineRole.roleArn,
      definitionString: file.toString(),
      stateMachineName: "sfn-job001",
      loggingConfiguration: {
        destinations: [{
          cloudWatchLogsLogGroup: {
            logGroupArn: logGroup.logGroupArn,
          }
        }],
        level: "ALL"
      }
    });
  }}

ステートマシンの配置

まず、ステートマシンの定義を配置するディレクトリを作成します。

mkdir statemachine
touch statemachine/job001.json

ステートマシンの記述

statemachine/job001.jsonの中身を下記のように記述します。

{
  "Comment": "Example of executing an ECS task",
  "StartAt": "Run an ECS Task",
  "States": {
    "Run an ECS Task": {
      "Type": "Task",
      "Resource": "arn:aws:states:::ecs:runTask.sync",
      "Parameters": {
        "LaunchType": "FARGATE",
        "Cluster": "arn:aws:ecs:ap-northeast-1:************:cluster/app-cluster",
        "TaskDefinition": "arn:aws:ecs:ap-northeast-1:************:task-definition/batch-app-task:1",
        "NetworkConfiguration": {
          "AwsvpcConfiguration": {
            "Subnets": [
              "subnet-*****************",
              "subnet-*****************"
            ],
            "SecurityGroups": [
              "sg-*****************"
            ],
            "AssignPublicIp": "DISABLED"
          }
        }
      },
      "Retry": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "IntervalSeconds": 1,
          "MaxAttempts": 2,
          "BackoffRate": 3
        }
      ],
      "End": true
    }
  }
}

Clusterの値は、事前に作成されているECSクラスターのARN
TaskDefinitionの値は、事前に作成されているECSタスク定義のARN
Subnetsは、タスク実行先となるVPCのサブネットID
SecurityGroupsは、タスクに割り当てるセキュリティグループID

CDKコードの解説

まず下記部分は、ステートマシンに必要なIAMロールとインラインポリシーの作成です。
主たる権限は、こんな感じです。

CloudWatch Logsへの書き込み
ECSタスク実行/停止

    const logPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    logPolicyStatement.addActions(
      "logs:CreateLogDelivery",
      "logs:GetLogDelivery",
      "logs:UpdateLogDelivery",
      "logs:DeleteLogDelivery",
      "logs:ListLogDeliveries",
      "logs:PutResourcePolicy",
      "logs:DescribeResourcePolicies",
      "logs:DescribeLogGroups"
    );
    logPolicyStatement.addResources("*");
    const ecsTaskPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    ecsTaskPolicyStatement.addActions(
      "ecs:RunTask",
      "ecs:StopTask",
      "ecs:DescribeTasks",
      "iam:PassRole"
    );
    ecsTaskPolicyStatement.addResources("*");
    const sfnPolicy : iam.Policy = new iam.Policy(this, "sfn-statemachine-policy", {
      policyName: "sfn-batch-policy",
      statements: [
        logPolicyStatement,
        ecsTaskPolicyStatement
      ]
    });
    const sfnStatemachineRole : iam.Role = new iam.Role(this, "sfn-statemachine-role", {
      roleName: "sfn-batch-role",
      assumedBy: new iam.ServicePrincipal("states.amazonaws.com"),
      path: "/"
    });
    sfnStatemachineRole.attachInlinePolicy(sfnPolicy);

ここはCloudWatch Logsのロググループの定義です。

    const logGroup : logs.LogGroup = new logs.LogGroup(this, 'sfn-logs-job001', {
      logGroupName: "/sfn/job001",
      retention: logs.RetentionDays.THIRTEEN_MONTHS
    });

そして、ここでJSONファイル化されたステートマシンを読み込みます。

    const file : Buffer = fs.readFileSync("statemachine/job001.json");

最後にステートマシンの定義ですが、ここで、事前に作成したIAMロールや、ロググループ、読み込んだJSONファイルの定義をパラメータとして与えます。
ステートマシンは、toString()で、文字列として引き渡していることがわかりますね。
あと細かいですが、ここだけL1コンストラクトを使用しています。

    const stateMachine : stepfunctions.CfnStateMachine = new stepfunctions.CfnStateMachine(this, "sfn-statemachine-job001", {
      roleArn: sfnStatemachineRole.roleArn,
      definitionString: file.toString(),
      stateMachineName: "sfn-job001",
      loggingConfiguration: {
        destinations: [{
          cloudWatchLogsLogGroup: {
            logGroupArn: logGroup.logGroupArn,
          }
        }],
        level: "ALL"
      }
    });

CDKによるデプロイ

まず初回だけ、ブートストラップを実行します。

cdk bootstrap

次に、デプロイを実行します。

cdk deploy

おそらく問題なくデプロイされ、ステートマシンも実行できるかと思います（タスク定義で指定されたタスクが正しく起動することが前提）。

次に考えないといけないこと

ステートマシンは、JSONファイルの内容を書き換えるだけで、デプロイできるようになりました。
しかし、今のままだと大きな課題が残りますね。
それは、ECSクラスターのARN、タスク定義のARN、サブネットID、セキュリティグループIDなど、環境依存の情報がステートマシンのJSONファイルに記述されてしまっていることです。
これでは、特定の環境にしかデプロイできません（厳密にはデプロイできても動かないステートマスんが構築されるだけです）。
そこで、

CDKのスタックコード
ステートマシンのJSON定義
に少しだけ手を加えるだけで、環境に依存しないステートマシンのデプロイが可能になります。

環境依存からの脱却

CDKスタックコードの修正

まずは最終系がこちらです。

import * as cdk from 'aws-cdk-lib';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as stepfunctions from 'aws-cdk-lib/aws-stepfunctions';
import * as fs from 'fs';
import { Construct } from 'constructs';

export class SfnStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // The code that defines your stack goes here
    const logPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    logPolicyStatement.addActions(
      "logs:CreateLogDelivery",
      "logs:GetLogDelivery",
      "logs:UpdateLogDelivery",
      "logs:DeleteLogDelivery",
      "logs:ListLogDeliveries",
      "logs:PutResourcePolicy",
      "logs:DescribeResourcePolicies",
      "logs:DescribeLogGroups"
    );
    logPolicyStatement.addResources("*");
    const ecsTaskPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    ecsTaskPolicyStatement.addActions(
      "ecs:RunTask",
      "ecs:StopTask",
      "ecs:DescribeTasks",
      "iam:PassRole",
      "events:PutTargets",
      "events:PutRule",
      "events:DescribeRule"
    );
    ecsTaskPolicyStatement.addResources("*");
    const ssmPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    ssmPolicyStatement.addActions(
      "ssm:GetParameter"
    );
    ssmPolicyStatement.addResources("*");
    const sfnPolicy : iam.Policy = new iam.Policy(this, "sfn-statemachine-policy", {
      policyName: "sfn-batch-policy",
      statements: [
        logPolicyStatement,
        ecsTaskPolicyStatement,
        ssmPolicyStatement
      ]
    });
    const sfnStatemachineRole : iam.Role = new iam.Role(this, "sfn-statemachine-role", {
      roleName: "sfn-batch-role",
      assumedBy: new iam.ServicePrincipal("states.amazonaws.com"),
      path: "/"
    });
    sfnStatemachineRole.attachInlinePolicy(sfnPolicy);
    const logGroup : logs.LogGroup = new logs.LogGroup(this, 'sfn-logs-job001', {
      logGroupName: "/sfn/job001",
      retention: logs.RetentionDays.THIRTEEN_MONTHS
    });
    const file : Buffer = fs.readFileSync("statemachine/job001.json");
    const stateMachine : stepfunctions.CfnStateMachine = new stepfunctions.CfnStateMachine(this, "sfn-statemachine-job001", {
      roleArn: sfnStatemachineRole.roleArn,
      definitionString: file.toString(),
      stateMachineName: "sfn-job001",
      loggingConfiguration: {
        destinations: [{
          cloudWatchLogsLogGroup: {
            logGroupArn: logGroup.logGroupArn,
          }
        }],
        level: "ALL"
      }
    });
  }}

先ほどからとの違いは、以下のみです。

まずは、Systemas Managerのパラメータストアの読み取り権限を持ったポリシーステートメント。

    const ssmPolicyStatement : iam.PolicyStatement = new iam.PolicyStatement();
    ssmPolicyStatement.addActions(
      "ssm:GetParameter"
    );
    ssmPolicyStatement.addResources("*");

上で作成したポリシーを、ポリシー定義に加えます（下記の6行目部）。

    const sfnPolicy : iam.Policy = new iam.Policy(this, "sfn-statemachine-policy", {
      policyName: "sfn-batch-policy",
      statements: [
        logPolicyStatement,
        ecsTaskPolicyStatement,
        ssmPolicyStatement
      ]
    });

つまり、ステートマシンにSystems ManagerのパラメータストアのGetParameterの権限を追加しただけ、ということです。

ステートマシンの修正

こちらもまずは最終系から。

{
  "Comment": "Example of executing an ECS task",
  "StartAt": "GetParameter Cluster",
  "States": {
    "GetParameter Cluster": {
      "Type": "Task",
      "Next": "GetParameter Task Definition",
      "Parameters": {
        "Name": "/sfn/ecs-cluster"
      },
      "Resource": "arn:aws:states:::aws-sdk:ssm:getParameter",
      "ResultSelector": {
        "value.$": "$.Parameter.Value"
      },
      "ResultPath": "$.cluster"
    },
    "GetParameter Task Definition": {
      "Type": "Task",
      "Next": "GetParameter Subnet1",
      "Parameters": {
        "Name": "/sfn/ecs-task-definition"
      },
      "Resource": "arn:aws:states:::aws-sdk:ssm:getParameter",
      "ResultSelector": {
        "value.$": "$.Parameter.Value"
      },
      "ResultPath": "$.taskdefinition"
    },
    "GetParameter Subnet1": {
      "Type": "Task",
      "Next": "GetParameter Subnet2",
      "Parameters": {
        "Name": "/sfn/subnet-1"
      },
      "Resource": "arn:aws:states:::aws-sdk:ssm:getParameter",
      "ResultSelector": {
        "value.$": "$.Parameter.Value"
      },
      "ResultPath": "$.subnet1"
    },
    "GetParameter Subnet2": {
      "Type": "Task",
      "Next": "GetParameter Security Group",
      "Parameters": {
        "Name": "/sfn/subnet-2"
      },
      "Resource": "arn:aws:states:::aws-sdk:ssm:getParameter",
      "ResultSelector": {
        "value.$": "$.Parameter.Value"
      },
      "ResultPath": "$.subnet2"
    },
    "GetParameter Security Group": {
      "Type": "Task",
      "Next": "Run an ECS Task",
      "Parameters": {
        "Name": "/sfn/security-group"
      },
      "Resource": "arn:aws:states:::aws-sdk:ssm:getParameter",
      "ResultSelector": {
        "value.$": "$.Parameter.Value"
      },
      "ResultPath": "$.securitygroup"
    },
    "Run an ECS Task": {
      "Type": "Task",
      "Resource": "arn:aws:states:::ecs:runTask.sync",
      "Parameters": {
        "LaunchType": "FARGATE",
        "Cluster.$": "$.cluster.value",
        "TaskDefinition.$": "$.taskdefinition.value",
        "NetworkConfiguration": {
          "AwsvpcConfiguration": {
            "Subnets.$": "States.Array($.subnet1.value, $.subnet2.value)",
            "SecurityGroups.$": "States.Array($.securitygroup.value)",
            "AssignPublicIp": "DISABLED"
          }
        }
      },
      "Retry": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "IntervalSeconds": 1,
          "MaxAttempts": 2,
          "BackoffRate": 3
        }
      ],
      "End": true
    }
  }
}

パラメータの取得

まず、最初に、Systems Managerのパラメータストアから値を取得する定義を追加します。

    "GetParameter Cluster": {
      "Type": "Task",
      "Next": "GetParameter Task Definition",
      "Parameters": {
        "Name": "/sfn/ecs-cluster"
      },
      "Resource": "arn:aws:states:::aws-sdk:ssm:getParameter",
      "ResultSelector": {
        "value.$": "$.Parameter.Value"
      },
      "ResultPath": "$.cluster"
    },

これは、/sfn/ecs-clusterというパラメータから値を取得し、$.clusterという変数に格納しています。
格納された値は変数名.valueで、後続の処理部で参照することができます。
同様に、

タスク定義(/sfn/ecs-task-definition)
サブネットID（/sfn/subnet-1、/sfn/subnet-2）
セキュリティグループ（/sfn/security-group）
をそれぞれ取得して、
タスク定義($.taskdefinition)
サブネットID（$.subnet1、$.subnet2）
セキュリティグループ（$.securitygroup）
に格納しています。

なお、Systems Managerの各パラメータは、事前に値が格納されていることを前提とします。
値の格納は

環境（アカウント）ごとに手動で作成
CloudFormationでVPCやECS関連のリソース作成時についでに作成
CDKでVPCやECS関連のリソース作成時についでに作成

流れ的に「CDKで」を推奨しますが、ここの手段は今回の本筋ではないのでお任せします。

パラメータ値を使用

次に、取得して変数に格納したパラメータを参照する箇所です。

      "Parameters": {
        "LaunchType": "FARGATE",
        "Cluster.$": "$.cluster.value",
        "TaskDefinition.$": "$.taskdefinition.value",
        "NetworkConfiguration": {
          "AwsvpcConfiguration": {
            "Subnets.$": "States.Array($.subnet1.value, $.subnet2.value)",
            "SecurityGroups.$": "States.Array($.securitygroup.value)",
            "AssignPublicIp": "DISABLED"
          }
        }
      },

まずECSクラスターですが、名称部がCluster.$となってます。
これはステートマシンのお作法で、値部に変数を参照する場合は「名前 .$ 」とする必要があります。
他のTaskDefinition.$、Subnets.$、SecurityGroups.$いずれも同じですね。

次に値部ですが「変数名 .value」となります。
で、単一文字列の場合は、この記述でOKですが、配列指定が必要なサブネットIDや、セキュリティグループは、States.Array()組み込み関数を使用します。
組み込み関数の詳細は、公式ドキュメントを参照してみてください。

CDKによるデプロイ

デプロイを実行します。

cdk deploy

これでステートマシンを実行した時に、パラメータストアに適切な設定がされていれば問題なく動作することでしょう。
他の環境（アカウント）でも、パラメータストアさえ定義されていれば、問題なく動作します。

応用

たとえば、ステートマシンから別のステートマシンを呼び出す、のような場合は、パラメータストアにステートマシンのARNを作成してあげればよいです。では、このパラメータ、どこで作成してあげればよいでしょうか。
勘のいい方はもうおわかりですね。
CDKスタックのステートマシン作成処理部の後に、ステートマシンのARNをパラメータストアに格納する処理を記述してあげてください。たったそれだけです。

まとめ

やっぱステートマシンはコードで管理しないとね
パラメータストアを活用すると、環境に依存しないステートマシンとして管理できるので、プログラムコードやIaCのコードと同じ感覚でコードリポジトリで管理できるよ
ステートマシンの定義ファイルの雛形は、ビジュアルワークフローで作成した時のJSON形式の定義をファイルに保存する、というやり方でもOK
- 1からJSONで作るって、結構大変だしね
複数のステートマシンがある場合は、CDKのスタックをわけたり、定義ファイルのディレクトリ構成を考えたり、管理しやすい方法で

では、CDKを使って、快適なステートマシンのコード管理ライフを！！