【AWS】Step Functions使ってみた。

Overview

バッチ処理をコンテナ上で動作させるための仕組みを作成しました。
コンテナ起動に失敗することが何度か発生していたので、回避するためStep Functionsを利用することにしました。

Step Functionsとは？

分散アプリケーションの構築、IT およびビジネスプロセスの自動化、AWS のサービスを利用したデータと機械学習のパイプラインの構築に使用するローコードのビジュアルワークフローサービスです。

ちょっと難しいですね。 自分なりに一言で言うならマネージドサービスを束ねてシームレスな処理を作れるサービス！ですかね。

作成したワークフロー

コンテナ起動に失敗するケース対応できるフローです。
試行回数のループはAWS公式リファレンスを参考にして作成しています。

1. 試行回数の設定
2. Fargateタスク起動
3. 実行結果をハンドリング
4. 失敗時
   1. 試行回数をカウント
   2. 試行回数到達チェック
      1. 試行回数到達時はエラー通知
   3. 2.へ戻る
5. 成功時
   1. 処理完了

ステートマシン

マネージドコンソールからステートマシンを確認できるので便利です。

参考例

ステートマシンのコードサンプルです。
States.Formatを使った埋め込みは少し分かりづらかったりするので、少し注意が必要です。

  StepFunctions:
    Type: AWS::StepFunctions::StateMachine
    Properties:
      DefinitionString: !Sub
        - |-
          {
            "Comment": "Fargate実行サンプル",
            "StartAt": "ConfigureCount",
            "States": {
              "ConfigureCount": {
                  "Type": "Pass",
                  "Result": {
                      "count": 2,
                      "index": 0,
                      "step": 1
                  },
                  "ResultPath": "$.iterator",
                  "Next": "Fargate Task"
              },
              "Fargate Task": {
                "Type": "Task",
                "Resource": "arn:aws:states:::ecs:runTask.sync",
                "Parameters": {
                    略
                  },
                  "Overrides": {
                    "ContainerOverrides": [
                      {
                        "Name": "コンテナ名",
                        "Command.$": "$.commands" # 引数を渡してFargate上でジョブを実行
                      }
                    ]
                  }
                },
                "Catch": [
                  {
                    "ErrorEquals": [
                      "States.ALL"
                    ],
                    "ResultPath": "$.result",
                    "Next": "Error Handler"
                  }
                ],
                "ResultPath": "$.result",
                "Next": "Error Handler"
              },
              "Error Handler": {
                "Type": "Task",
                "Resource": "arn:aws:lambda:region:account_id:function:function_name",
                "Catch": [
                  {
                    "ErrorEquals": [
                      "Exception" # Lambda Functionから発生させるException
                    ],
                    "ResultPath": "$.result",
                    "Next": "Iterator"
                  }
                ],
                "ResultPath": "$.result",
                "Next": "Done"
              },
              "Iterator": {
                "Type": "Task",
                "Resource": "arn:aws:lambda:region:account_id:function:function_name",
                "ResultPath": "$.iterator",
                "Next": "IsCountReached"
              },
              "IsCountReached": {
                "Type": "Choice",
                "Choices": [
                  {
                      "Variable": "$.iterator.continue",
                      "BooleanEquals": true,
                      "Next": "Fargate Task"
                  }
                ],
                "Default": "Retry Failure"
              },
              "Done": {
                "Type": "Pass",
                "End": true
              },
              "Retry Failure": {
                "Type": "Task",
                "Resource": "arn:aws:states:::sns:publish",
                "Parameters": {
                  "Message.$": "States.Format('文字列連携サンプル. Command: {}', $.commands)",
                  "TopicArn": "arn:aws:sns:region:account_id:alert"
                },
                "End": true
              }
            }
          }
        - {
            埋め込み用パラメータ定義
          }
      LoggingConfiguration:
        Destinations:
          - CloudWatchLogsLogGroup:
              LogGroupArn: "target"
        IncludeExecutionData: "true"
        Level: ALL
      RoleArn: "target"
      StateMachineName: step-functions-sample
      StateMachineType: STANDARD

最後に

ちょっと雑なまとめ方ですがうまくいかない方の参考になれば幸いです。