サンプルで覚えるAWS Step Functions

ワークフローサービスとして発表されたStep Functions。
他のワークフローサービスにSimple Workflow(SWF)もありますが、新規アプリケーションの場合はStep Functionsの方を検討することが推奨されています。ニーズを満たさない場合はSWFを使用してみてよいそうです。

Step functionsには現在6種類のblueprintがサンプルとして用意されています。
こちらのコードを眺めるだけでも仕様が最低限必要なことが把握できるように思われます。

最初に用意されたBlueprintを眺めて、最後に定義された全種類のStateを用いてステートマシンを作成します。

Blueprint

Hello World

{
  "Comment": "A Hello World example of the Amazon States Language using a Pass state",
  "StartAt": "HelloWorld",
  "States": {
    "HelloWorld": {
      "Type": "Pass",
      "Result": "Hello World!",
      "End": true
    }
  }
}

Wait State

{
  "Comment": "An example of the Amazon States Language using wait states",
  "StartAt": "FirstState",
  "States": {
    "FirstState": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME",
      "Next": "wait_using_seconds"
    },
    "wait_using_seconds": {
      "Type": "Wait",
      "Seconds": 10,
      "Next": "wait_using_timestamp"
    },
    "wait_using_timestamp": {
      "Type": "Wait",
      "Timestamp": "2015-09-04T01:59:00Z",
      "Next": "wait_using_timestamp_path"
    },
    "wait_using_timestamp_path": {
      "Type": "Wait",
      "TimestampPath": "$.expirydate",
      "Next": "wait_using_seconds_path"
    },
    "wait_using_seconds_path": {
      "Type": "Wait",
      "SecondsPath": "$.expiryseconds",
      "Next": "FinalState"
    },
    "FinalState": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME",
      "End": true
    }
  }
}

Retry Failure

{
  "Comment": "A Retry example of the Amazon States Language using an AWS Lambda Function",
  "StartAt": "HelloWorld",
  "States": {
    "HelloWorld": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME",
      "Retry": [
        {
          "ErrorEquals": ["HandledError"],
          "IntervalSeconds": 1,
          "MaxAttempts": 2,
          "BackoffRate": 2.0
        },
        {
          "ErrorEquals": ["States.TaskFailed"],
          "IntervalSeconds": 30,
          "MaxAttempts": 2,
          "BackoffRate": 2.0
        },
        {
          "ErrorEquals": ["States.ALL"],
          "IntervalSeconds": 5,
          "MaxAttempts": 5,
          "BackoffRate": 2.0
        }
      ],
      "End": true
    }
  }
}

Parallel

{
  "Comment": "An example of the Amazon States Language using a parallel state to execute two branches at the same time.",
  "StartAt": "Parallel",
  "States": {
    "Parallel": {
      "Type": "Parallel",
      "Next": "Final State",
      "Branches": [
        {
          "StartAt": "Wait 20s",
          "States": {
            "Wait 20s": {
              "Type": "Wait",
              "Seconds": 20,
              "End": true
            }
          }
        },
        {
          "StartAt": "Pass",
          "States": {
            "Pass": {
              "Type": "Pass",
              "Next": "Wait 10s"
            },
            "Wait 10s": {
              "Type": "Wait",
              "Seconds": 10,
              "End": true
            }
          }
        }
      ]
    },
    "Final State": {
      "Type": "Pass",
      "End": true
    }
  }
}

Catch Failure

{
  "Comment": "A Catch example of the Amazon States Language using an AWS Lambda Function",
  "StartAt": "HelloWorld",
  "States": {
    "HelloWorld": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME",
      "Catch": [
        {
          "ErrorEquals": ["HandledError"],
          "Next": "CustomErrorFallback"
        },
        {
          "ErrorEquals": ["States.TaskFailed"],
          "Next": "ReservedTypeFallback"
        },
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "CatchAllFallback"
        }
      ],
      "End": true
    },
    "CustomErrorFallback": {
      "Type": "Pass",
      "Result": "This is a fallback from a custom lambda function exception",
      "End": true
    },
    "ReservedTypeFallback": {
      "Type": "Pass",
      "Result": "This is a fallback from a reserved error code",
      "End": true
    },
    "CatchAllFallback": {
      "Type": "Pass",
      "Result": "This is a fallback from a reserved error code",
      "End": true
    }
  }
}

Choice State

{
  "Comment": "An example of the Amazon States Language using a choice state.",
  "StartAt": "FirstState",
  "States": {
    "FirstState": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME",
      "Next": "ChoiceState"
    },
    "ChoiceState": {
      "Type" : "Choice",
      "Choices": [
        {
          "Variable": "$.foo",
          "NumericEquals": 1,
          "Next": "FirstMatchState"
        },
        {
          "Variable": "$.foo",
          "NumericEquals": 2,
          "Next": "SecondMatchState"
        }
      ],
      "Default": "DefaultState"
    },

    "FirstMatchState": {
      "Type" : "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:OnFirstMatch",
      "Next": "NextState"
    },

    "SecondMatchState": {
      "Type" : "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:OnSecondMatch",
      "Next": "NextState"
    },

    "DefaultState": {
      "Type": "Fail",
      "Cause": "No Matches!"
    },

    "NextState": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME",
      "End": true
    }
  }
}

組み合わせる

以下の7種類のStateが用意されています。

• Task
• Wait
• Pass
• Succeed
• Fail
• Choice
• Parallel

7種類全てのStateを用いて以下のようなステートマシンを作成します。

実装

JSON

上記のステートマシンは以下のようなJSONで定義されます。

{
    "Comment": "Sample Step functions flow",
    "StartAt": "Process1",
    "States": {
        "Process1": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXXXX:function:Process1",
            "Next": "ChoiceStep"
        },
        "ChoiceStep": {
            "Type": "Choice",
            "Choices": [{
                "Variable": "$.processResult",
                "StringEquals": "Process2a",
                "Next": "SuccessProcess"
            }, {
                "Variable": "$.processResult",
                "StringEquals": "Process2b",
                "Next": "Process2b-1"
            }]
        },
        "SuccessProcess": {
            "Type": "Succeed"
        },
        "Process2b-1": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXXXX:function:Process2",
            "Next": "Process2b-2"
        },
        "Process2b-2": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXXXX:function:Process2",
            "Catch": [{
              "ErrorEquals": [
                "HandledError"
              ],
              "Next": "FailedProcess"
            }],
            "Next": "Process3"
        },
        "FailedProcess": {
            "Type": "Fail"      
        },
        "Process3": {
            "Type": "Wait",
            "Seconds": 30,
            "Next": "Process4"
        },
        "Process4": {
            "Type": "Parallel",
            "Next": "Process5",
            "Branches": [
                {
                    "StartAt": "Process4a",
                    "States": {
                        "Process4a": {
                            "Type": "Task",
                            "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXXXX:function:Process2",
                            "End": true
                        }
                    }
                },
                {
                    "StartAt": "Process4b-1",
                    "States": {
                         "Process4b-1": {
                            "Type": "Task",
                            "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXXXX:function:Process2",
                            "Retry": [{
                                "ErrorEquals": [
                                    "HandledError"
                                ],
                                "IntervalSeconds": 3,
                                "MaxAttempts": 5,
                                "BackoffRate": 1.5
                            }],
                            "Next": "Process4b-2"
                         },
                         "Process4b-2": {
                            "Type": "Pass",
                            "Result": {
                                "data1": 12345,
                                "data2": -12345
                            },
                            "ResultPath": "$.test-data",
                            "Next": "Process4b-3"
                         },
                         "Process4b-3": {
                            "Type": "Task",
                            "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXXXX:function:Process2",
                            "End": true
                         }
                    }
                }
            ]
        },
        "Process5": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXXXX:function:Process3",
            "End": true
        }
    }
}

Lambda関数

以下の3つのLabda関数を用意します。

• Process1

def lambda_handler(event, context):
    event["step"] = 1
    event["processResult"] = "Process2b"
    return event

• Process2

def lambda_handler(event, context):
    event["step"] += 1
    return event

• Process3

def lambda_handler(event, context):
    print event
    return event

作成手順

1.マネジメントコンソールから[Step functions]を選択。
2.[Get Started]を選択。

3.以下のように入力して[Create State Machine]を選択。
Give a name to your state machine: SampleStateMachineWithVariableOfTypes

JSONは先程のものを入力します。

Previewの読み込みボタンを押すとステートマシンの状態が表示される。

4.以下のように入力して[OK]を選択。

IAM role for your state machine executions: StatesExecutionRole-us-east-1

5.[New Execution]を選択。
6.Inputとして以下のような値を与えます。

{
    "processResult": "Process2b"
}

7.[Start Execution]を選択。

出力

Output

output: [{"processResult": "Process2b", "step": 4},{"test-data": {"data1": 12345, "data2": -12345}, "processResult": "Process2b", "step": 5}]

Tips

• statelintで構文の正しさでチェックできる。(不正確なときがある気がする。。)
• JSONの構文の正しさはjqコマンドでも確認できる。
• Previewの隣の更新ボタンを押して構文をチェックできる。
• 上の構文チェックの対象外のところは[Create State Machine]選択後にもエラーが出得るので実行前に押して確認もできる。
• Lambda作成のデバッグ 以下のようにpython sample.pyでデバッグできるようにしておくと試行錯誤がしやすい。

def lambda_handler(event, context):
    event["step"] = 1
    event["processResult"] = "Process2b"
    return event

if __name__ == "__main__":
    lambda_handler("","")

注意

• 余計なタスクを定義すると怒られる
• There is a problem with your AWL definition, please review it and try again
  • 指定するキーの綴りミスなど
• Transition references an unknown state
  • Nextでの遷移先のものが存在しない(綴りミス、スコープ内にStateが定義されていないなど)

参考

• http://docs.aws.amazon.com/step-functions/latest/dg/welcome.html
• https://aws.amazon.com/jp/swf/faqs/
• http://qiita.com/naokitomita/items/901e730a83d73bc0baa3
• http://qiita.com/ketancho/items/147a141c9f8a6de86c97