AWS Step Functions 調査メモ

## はじめに

AWS Step Functions の調査メモ。深堀りはせず概要を把握することが目標です。

## Step Functions で Lambda 実行

Step FunctionsでCloudWatch logsにログを書き込むLambdaを実行します。

以下はログを書き込むLambda(Python3.8)です。
Lambda実行時に受け取ったJSONデータをログに記録します。
また、return で受け取ったJSONデータを返しています。

```python:test001
import json

def lambda_handler(event, context):
    print("test desu {0}".format(event) )
    return (event)
```

- 必要なIAMロールは、CluodWatch Logs の以下3つ
    - CreateLogGroup
    - CreateLogStream
    - PutLogEvents


作成したStepFunctionsのjsonコードです。

```json:MyStateMachine001
{
  "Comment": "This is your state machine",
  "StartAt": "Lambda Test001",
  "States": {
    "Lambda Test001": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:test001:$LATEST",
        "Payload.$": "$"
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "End": true,
      "Comment": "あいうえお"
    }
  }
}
```

Step FunctionsのWorkflow Studioです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/a856450a-1414-3cf0-3f1c-e11e04c9f15c.png)

実行時のペイロードです

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/82ea949b-88c4-f7f9-29ab-ed3aaea3ec5e.png)

実行ステータスが成功と表示されればStep Functionsとしては成功しています

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/91f600cd-5183-3f9e-29e9-6a11eaa3853c.png)

Lambda `test001` の実行ログにStep Functions実行時のJSONデータが記録されています（期待した挙動）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/77331b6d-f12e-4848-c3c0-ea5f93d6b43b.png)

## ペイロードとは

ペイロードとは、StepFunctions 実行時に渡すJSON形式のデータのこと。
どうやら引数みたいなものらしい。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/0e5901ae-bf4b-1805-c4ef-54a42d5943a3.png" width="500">

ペイロードには3つの設定があります。

|設定|概要|
|----|-----|
|Use state input as payload|StepFuncitons 実行時に設定|
|Enter payload|StepFuncitonsコード内に設定|
|No payload|ペイロード不要|

## フローとは

フローとは、Step Functions に用意された条件分岐などのプログラム処理のことです。
2021年8月時点では7種類のフローがあります。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/6484a562-1944-b63f-610e-1f2e87dde199.png" width = 300>

|フロー|概要|
|---|---|
|Choice|条件分岐 If文みたいなもの。|
|Parallel|並行処理|
|Map|ループ処理 For文に似てる。同時実行数を設定できる。|
|Pass|情報を渡すだけ。入力値をもとに出力値を作成できる。デバッグする際に便利|
|Wait|指定した時間(秒)処理と止める。指定した日時まで処理を止めることもできる。|
|Success,Fail|Step Functions の実行結果を設定できる。|

## フロー使ってみる

7種類のフローを使ってみます

---

### 【Choice】使ってみる

`Choice`は条件分岐の処理です。実行結果によって、実行する処理を変えることができます。

![図1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/de612d57-09ae-ffee-97f7-4bfb4e7b37d6.png)

作成したStep Functionsのjsonデータです。

```json
{
  "Comment": "This is your state machine",
  "StartAt": "Lambda Test001",
  "States": {
    "Lambda Test001": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:test001:$LATEST",
        "Payload.$": "$"
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "Comment": "あいうえお",
      "Next": "Choice"
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.HOGE",
          "StringEquals": "HOGE 200",
          "Next": "Lambda HOGE 200"
        }
      ],
      "Default": "Lambda Not HOGE 200"
    },
    "Lambda HOGE 200": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:test001:$LATEST",
        "Payload": {
          "FUGA": "FUGA 100"
        }
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "End": true
    },
    "Lambda Not HOGE 200": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:test001:$LATEST",
        "Payload": {
          "FUGA": "FUGA 200"
        }
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "End": true
    }
  }
}
```

実行時のペイロード①

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/5612d589-5d8e-7af3-9c34-70152f5ca5b5.png)

＜実行結果①＞

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/6393d922-36f2-1172-6ab7-ab3935187489.png)

実行時のペイロード②

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/31262c95-245e-c9b5-d9c6-91994f078771.png)

＜実行結果②＞

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/6f72928d-9117-b04f-8dbb-c0c3fc861221.png)

---

### 【Parallel】使ってみる

平行処理ができる様子

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/a4b69a64-5a8b-1a93-a95e-4564253e7bd8.png)

---

### 【Map】【Wait】使ってみる

lambda `test001` を指定した回数実行する処理を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/50b01287-cae5-536c-12e1-ee57e55750eb.png)

`項目配列へのパス`に、JSONデータの中の配列情報をもつ箇所を指定します。
`最大同時実行`を 1 に設定し、1回のループで1つ処理を行うようにしました。

lambda `test001`に S3バケットにファイル出力する仕組みを追加

```python:(Lambda)test001
import json
import boto3
from datetime import datetime

s3 = boto3.resource('s3')

def lambda_handler(event, context):
    bucket = 'tmp-hokdvhvwzw'    # S3バケット名
    key = 'test_' + datetime.now().strftime('%Y-%m-%d-%H-%M-%S') + '.txt'
    file_contents = str(event)  # ファイルの内容
    
    obj = s3.Object(bucket,key)     # S3バケット名とパスを指定
    obj.put( Body=file_contents )   # S3バケットにファイル出力

    print("key = {0}, event = {1}".format(key, event) )
    return (event)
```

- 必要なIAMロール
    - CluodWatch Logs
        - CreateLogGroup
        - CreateLogStream
        - PutLogEvents
    - S3バケット
        - AmazonS3FullAccess
        - ↑ ホントはここまで強い権限は必要ない

実行時に設定するJSONデータ

```json
{
  "detail": {
    "shipped": [
      { "prod": "E01", "dest-code": 1001, "note": "AAA" },
      { "prod": "E02", "dest-code": 1002, "note": "BBB" },
      { "prod": "E03", "dest-code": 1003, "note": "CCC" },
      { "prod": "E04", "dest-code": 1004, "note": "DDD" },
      { "prod": "E05", "dest-code": 1005, "note": "EEE" }
    ]
  }
}
```

S3バケットに出力されたファイル

```powershell
PS C:\Users\user01> cat .\test_2021-08-08-23-18-06.txt
{'prod': 'E01', 'dest-code': 1001, 'note': 'AAA'}
PS C:\Users\user01> cat .\test_2021-08-08-23-18-36.txt
{'prod': 'E02', 'dest-code': 1002, 'note': 'BBB'}
PS C:\Users\user01> cat .\test_2021-08-08-23-19-06.txt
{'prod': 'E03', 'dest-code': 1003, 'note': 'CCC'}
PS C:\Users\user01> cat .\test_2021-08-08-23-19-36.txt
{'prod': 'E04', 'dest-code': 1004, 'note': 'DDD'}
PS C:\Users\user01> cat .\test_2021-08-08-23-20-07.txt
{'prod': 'E05', 'dest-code': 1005, 'note': 'EEE'}
PS C:\Users\user01>
```

---

### 【Pass】使ってみる

`Pass`の次に実行するlambda `test001` に渡すJSONデータを加工します。
lambdaはこれ ↓ を使用しました。

```python:test001
import json

def lambda_handler(event, context):
    print("test desu {0}".format(event) )
    return (event)
```

`Pass`の設定で、HOGE,FUGA の値をもつJSONデータを作成します。FUGAの値は、ペイロードのJSONデータから取得します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/1a3fac4d-383b-be94-4103-534cdda32b5f.png)

実行時に設定するJSONデータ

```json
{
  "FUGA": "FUGA 123"
}
```

Step Functions 実行後のLambda test001 のログ

```
2021-08-09T09:50:24.212+09:00	test desu {'HOGE': 'HOGE-1234567890', 'FUGA': 'FUGA 123'}
```

---

### 【Success】【Fail】使ってみる

ペイロードの中のHOGEの値が 100 の場合にSuccessとなり、それ以外はFailになる処理を作りました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/47bdb0c3-35ea-825c-1eaf-50d8ece1bf43.png)

＜実行結果＞

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/72e8d3a4-8473-4f75-f4e7-444faac11e37.png)

---

## 作ってみた（１）：S3バケット上のJSONデータをもとにループ処理する

S3バケットに格納したJSONデータを取得し、そのJSONデータの情報をもとにループ処理を実行します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/b405c251-17d1-1936-d6ea-1d3db2c18da8.png)

Lambda `testLoadJSONfroms3`

```python:testLoadJSONfroms3
import boto3
import json

def lambda_handler(event, context):
    print(event)
    s3 = boto3.resource("s3")
    bucket = s3.Bucket( event["bucketName"] )
    jsonFile = bucket.Object(event["prefix"] + "/" + event["json"])
    jsonData = jsonFile.get()
    jsonInfo = jsonData['Body'].read()

    return jsonInfo
```

```python:test001
import json

def lambda_handler(event, context):
    print("test desu {0}".format(event) )
    return (event)
```

※IAMロールの見直し※
作成したStep Functionsに2種類（`testLoadJSONfroms3`,`test001`）のLambdaを実行する権限があるかIAMロールを要確認

```json:StepFunctions
{
  "Comment": "This is your state machine",
  "StartAt": "Lambda LoadJSONfromS3",
  "States": {
    "Lambda LoadJSONfromS3": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "Payload.$": "$",
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:testLoadJSONfromS3:$LATEST"
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "Next": "Pass (Debug)"
    },
    "Pass (Debug)": {
      "Type": "Pass",
      "Next": "Pass (Filter)"
    },
    "Pass (Filter)": {
      "Type": "Pass",
      "InputPath": "$.Payload",
      "Next": "Map"
    },
    "Map": {
      "Type": "Map",
      "End": true,
      "Iterator": {
        "StartAt": "Lambda TEST001",
        "States": {
          "Lambda TEST001": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "OutputPath": "$.Payload",
            "Parameters": {
              "Payload.$": "$",
              "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:test001:$LATEST"
            },
            "Retry": [
              {
                "ErrorEquals": [
                  "Lambda.ServiceException",
                  "Lambda.AWSLambdaException",
                  "Lambda.SdkClientException"
                ],
                "IntervalSeconds": 2,
                "MaxAttempts": 6,
                "BackoffRate": 2
              }
            ],
            "Next": "Wait"
          },
          "Wait": {
            "Type": "Wait",
            "Seconds": 5,
            "End": true
          }
        }
      },
      "MaxConcurrency": 1
    }
  }
}
```

ペイロードには、jsonデータがあるS3バケットとプレフィックス(ディレクトリ)を指定します。

```json:ペイロード
{
    "bucketName": {{S3バケット名}}
    "prefix": {{プレフィックス}}
    "json": {{jsonファイル名}}
}
```

S3バケットにあるJSONデータ

```json:S3バケットにあるJSONデータ
[
    {"StackName": "test001-vpc","Code": "vpc.yml", "PJPrefix": "Project1","VPCCIDR": "10.11.0.0/16"},
    {"StackName": "test001-subnet1","Code": "subnet-public.yml", "PJPrefix": "Project1","NetworkName": "Net001","PublicSubnetCIDR": "10.11.1.0/24", "AZName": "ap-northeast-1a"},
    {"StackName": "test001-sg","Code": "sg.yml", "PJPrefix": "Project1","ServiceName": "ServiceA","SGNo": "001"}
]
```



## 作ってみた（２）：RunCommandを実行し結果を取得する

EC2に対しRunCommandでOSコマンドを実行し、その結果取得するStep Functionsを作ります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/aa79c4ef-3149-72ac-2194-b2a6da24dd39.png)

Step Functions のjsonコードはこちら。

```json:StepFunctions
{
  "Comment": "This is your state machine",
  "StartAt": "Lambda RunCommand",
  "States": {
    "Lambda RunCommand": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:testRunCommand:$LATEST",
        "Payload.$": "$"
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "Next": "Wait"
    },
    "Wait": {
      "Type": "Wait",
      "Seconds": 1,
      "Next": "Lambda RunCommand2"
    },
    "Lambda RunCommand2": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "Payload.$": "$",
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:testRunCommand2:$LATEST"
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "Next": "Choice"
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.com_status",
          "StringEquals": "Success",
          "Next": "Lambda RunCommand3"
        },
        {
          "Variable": "$.com_status",
          "StringEquals": "InProgress",
          "Next": "Wait"
        }
      ],
      "Default": "Fail"
    },
    "Lambda RunCommand3": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "Payload.$": "$",
        "FunctionName": "arn:aws:lambda:ap-northeast-1:000000000000:function:testRunCommand3:$LATEST"
      },
      "Retry": [
        {
          "ErrorEquals": [
            "Lambda.ServiceException",
            "Lambda.AWSLambdaException",
            "Lambda.SdkClientException"
          ],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        }
      ],
      "End": true
    },
    "Fail": {
      "Type": "Fail"
    }
  }
}
```

Lambda `RunCommand` はRunCommandを実行します

```python:RunCommand
import json
import boto3

ssm = boto3.client('ssm')

def lambda_handler(event, context):
    command = event["command"]
    instance_id = event["instance_id"]

    r = ssm.send_command(
        InstanceIds = [instance_id],
        DocumentName = "AWS-RunShellScript",
        Parameters = {
            "commands": [ command ]    
        }
    )
    command_id = r['Command']['CommandId']
    print("command_id = {0}".format(command_id) )
    
    return { 'command_id': command_id }
```

Lambda `RunCommand2` RunCommandの実行結果を取得します。
invocations[0]['Status']は、処理中なら `InProgress` 実行成功なら `Success` 実行失敗なら `Failed`となります。

```python:RunCommand2
import json
import boto3

ssm = boto3.client('ssm')

def lambda_handler(event, context):
    command_id = event["command_id"]
    res = ssm.list_command_invocations(
              CommandId = command_id,
              Details = True
          )
    invocations = res['CommandInvocations']
    com_status = invocations[0]['Status']
    output = "output={0}".format(invocations[0]['CommandPlugins'][0]['Output'])

    return { 'command_id': command_id, 'com_status': com_status, 'output': output }
```

Lambda `RunCommand3` は特になにも処理していません。

```python:RunCommand3
import json

def lambda_handler(event, context):
    output = event["output"]
    return { 'statusCode': 200 }
```

StepFunctions実行時のペイロード
ペイロードには、実行するコマンドと実行するEC2のインスタンスIDを設定します。

```json:ペイロード(コマンド成功)
{
    "command": "ifconfig",
    "instance_id": "i-XXXXXXXXXXX"
}
```

```json:ペイロード(コマンド失敗)
{
    "command": "BADcommand",
    "instance_id": "i-XXXXXXXXXXX"
}
```

＜実行結果＞

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/274270/88a3d972-f73b-3c86-239a-bdcebbc7fb4d.png)


## さいごに

AWS Step Functionsの使い方がちょっとわかった気がします。