AWS Step FunctionsのRetryとCatchを使ってみる

AWSのStep FunctionsのRetryとCatchをServerless Frameworkから使ってみます。

Step Functionsは、CloudFormationだとState machine定義の記述が面倒です。JSONを文字列にしたものをCloudFormationのテンプレートに書くか、別ファイルにしないといけないようです。Serverless FrameworkならYAMLで直接serverless.ymlに定義できます。今回はServerless Frameworkで試しました。

Serverless FrameworkからStep Functionsを定義するにはプラグインのインストールが必要です。

$ serverless plugin install -n serverless-step-functions

ソースコード

試しに作った serverless.yml

service: sample-stepfunction

frameworkVersion: '2'

plugins:
  - serverless-step-functions

provider:
  name: aws
  region: ap-northeast-1
  lambdaHashingVersion: 20201221

resources:
  Resources:
    # Step Functions用CloudWatch LogGroup作成
    sampleStateMachineLogGroup:
      Type: AWS::Logs::LogGroup
      Properties:
        LogGroupName: /aws/states/sampleStateMachine-Logs

functions:
  # 3つのLambda作成
  samplefunc1:
    handler: handler.main1
    runtime: python3.8
  samplefunc2:
    handler: handler.main2
    runtime: python3.8
  samplefunc3:
    handler: handler.main3
    runtime: python3.8

# Step Functions作成
stepFunctions:
  stateMachines:
    sampleStateMachine:
      name: sampleStateMachine
      loggingConfig:
        level: ALL
        includeExecutionData: true
        destinations:
          - Fn::GetAtt: [sampleStateMachineLogGroup, Arn]
      definition:
        StartAt: Status1
        States:
          Status1:
            Type: Task
            Resource:
              Fn::GetAtt: [samplefunc1, Arn]
            Next: Status2
          Status2:
            Type: Task
            Resource:
              Fn::GetAtt: [samplefunc2, Arn]
            Retry:
              - ErrorEquals:
                - States.ALL
                IntervalSeconds: 1
                MaxAttempts: 2
                BackoffRate: 2
            Catch:
              - ErrorEquals:
                - States.ALL
                Next: Status3
            Next: End
          Status3:
            Type: Task
            Resource:
              Fn::GetAtt: [samplefunc3, Arn]
            Next: End
          End:
            Type: Pass
            End: true

handler.py

import time

def main1(event, context):
    print(event)
    return {"foo": "main1"}

def main2(event, context):
    print(event)
    time.sleep(10) # Lambdaをタイムアウトさせる
    return {"foo": "main2"} # タイムアウトになるのでこの値は返ってこない

def main3(event, context):
    print(event)
    return {"foo": "main3"}

3つのLambdaを定義しています。

• samplefunc1: 入力をログに残した後、正常に終了するLambda
• samplefunc2: 入力をログに残した後、必ずタイムアウトのエラーになってしまうLambda
• samplefunc3: 入力をログに残した後、正常に終了するLambda

それぞれStep FunctionsのStatus1, Status2, Status3に対応します。

実行結果

Status2にはRetryでMaxAttempts: 2の指定があるので、2回のリトライ、つまりsamplefunc2が3回実行されます。samplefunc2に渡される入力は3回とも同じです。

Status2にはCatchで'Next: Status3'の指定もあるので、samplefunc2が3回実行されたあとに、Status3に遷移します。samplefunc3への入力は次のようなオブジェクトになります。

{
  "Error": "Lambda.Unknown",
  "Cause": "The cause could not be determined because Lambda did not return an error type. Returned payload: {\"errorMessage\":\"2021-03-10T11:12:36.559Z ea381979-e25c-41e3-9fe5-39360cfa0ab6 Task timed out after 6.01 seconds\"}"
}

エラーハンドラに元の入力を引き渡す

Status2のCatchの指定に次のように ResultPath: "$.error" の指定を追加します。

            Catch:
              - ErrorEquals:
                - States.ALL
                Next: Status3
                ResultPath: "$.error"

これをデプロイして実行すると、Status3への入力は、Status2への入力にエラー情報を追加したものに変わります。元の入力を保持しておきたい場合はこの指定が必要のようです。

{
  "foo": "main1",
  "error": {
    "Error": "Lambda.Unknown",
    "Cause": "The cause could not be determined because Lambda did not return an error type. Returned payload: {\"errorMessage\":\"2021-03-10T11:25:56.258Z 42c9ade8-93e1-499d-9d60-68b0a6c7ccdf Task timed out after 6.01 seconds\"}"
  }
}