Serverless FrameworkでStep Function (Map + エラーハンドル)

Step FunctionでMapを使った動的なタスク実行ができるようになりました。そしてそうなってくると、そのステップ内でエラーが起きた時にリトライさせたくなってきます。

ということで、リトライのあるMapのタスクを作ってみました。

タスクの流れ

• 数字の配列（1 ~ 5）をreturnするLambdaを実行
• 各数字それぞれをMapで同時に処理する
• 偶数のタスクは失敗する
• 失敗したタスクは数字の1を渡してリトライさせる
• 奇数のタスクはさらに+1する
• 最後に処理された結果をCloudWatch Logsへ吐き出す

Lambda１つでやれよという案件ですが、Step Functionであえてやってみます。

各Lambda Function

先に関数を実装しておきましょう。

import { Handler } from 'aws-lambda';
import 'source-map-support/register';

// 配列のセットアップ
export const map: Handler = async (event) => {
  console.log(event)
  return [1,2,3,4,5]
}

// 偶数奇数判定
export const mapTask: Handler = async (event) => {
  console.log(event)
  if (event % 2 === 0) throw new Error(`${event} is not odd`)
  return {
    event,
    times: event * 2
  }
}

export const mapTask2: Handler = async (event) => {
  return {
    ...event,
    // 特に意味はないけど、+1した値を足してみる
    add: event.times + 1
  }
}

export const errorHandler: Handler = async (event) => {
  console.log(event)
  // oddは弾かれるので1を渡して調整してやる
  return 1
}

export const end: Handler = async (event) => {
  // 結果のログ出力
  console.log(JSON.stringify(event))
  return {
    statusCode: 200,
    body: JSON.stringify(event)
  };
}

Functionレベルで分割するので、それぞれの見通しはかなりよくなりました。

Serverless Frameworkでスタックを作る

あとは実際に実行するスタックを定義すればOKです。

service:
  name: sls-playground
plugins:
  - serverless-webpack
  - serverless-step-functions

provider:
  name: aws
  runtime: nodejs10.x
  logRetentionInDays: 14
  region: us-east-1
  stage: development

functions:
  hello:
    handler: handler.hello
  map:
    handler: handler.map
  mapTask:
    handler: handler.mapTask
  mapTask2:
    handler: handler.mapTask2
  end:
    handler: handler.end
  errorHandler:
    handler: handler.errorHandler

stepFunctions:
  stateMachines:
    yourCatchMachine:
      definition:
        Comment: "A Catch example of the Amazon States Language using an AWS Lambda Function"
        StartAt: HelloWorld
        States:
          # 配列をセットアップ
          HelloWorld:
            Type: Task
            Resource:
              Fn::GetAtt: [map, Arn]
            Next: mapped_task
          # Mapで１つずつ処理する
          mapped_task:
            Type: Map
            Iterator:
              StartAt: MapTask
              States:
                # 偶数奇数判定
                MapTask:
                  Type: Task
                  Resource:
                    Fn::GetAtt: [mapTask, Arn]
                  Catch:
                    - ErrorEquals: ["States.ALL"]
                      Next: MapError
                  Next:  MapTask2
                # +1する処理
                MapTask2:
                  Type: Task
                  Resource:
                    Fn::GetAtt: [mapTask2, Arn]
                  End: true
                # 1を渡して強制的に奇数にする
                MapError:
                  Type: Task
                  Resource:
                    Fn::GetAtt: [errorHandler, Arn]
                  Next: MapTask
            Next: end_task
          # 結果をログ出力
          end_task:
            Type: Task
            Resource:
              Fn::GetAtt: [end, Arn]
            End: true

これを実行すると、このような結果がend_taskのLambdaログに残ります。

[
  {
    "event": 1,
    "times": 2,
    "add": 3
  },
  {
    "event": 1,
    "times": 2,
    "add": 3
  },
  {
    "event": 3,
    "times": 6,
    "add": 7
  },
  {
    "event": 1,
    "times": 2,
    "add": 3
  },
  {
    "event": 5,
    "times": 10,
    "add": 11
  }
]

偶数の値（２と４）が１に丸められて処理されていることがわかります。

注意点

Map Taskがエラーを投げる-> Map Errorで値を修正 -> Map Taskを呼び出しという形になっています。つまり無限ループの可能性がここには潜んでいます。

今回は簡略化するためにこの形にしていますが、実際にはリトライ回数をみて強制的に落とすような仕組みがあった方がよさそうです。