lambdaを確実に１回だけ起動させる方法〜起動失敗と、重複起動を乗り越えて〜

lambdaの起動方法

lambdaの起動方法には同期呼び出しと、非同期呼び出しがあります。
lambdaを呼び出すトリガーが何かによって、どちらの呼び出し方か変わります。

同期呼び出し例

• API Gateway(非同期にすることも可)
• Cognito
• Alexa
  など・・・

非同期呼び出し例

• CloudWatch Events
• CloudWatch Logs
• CodeCommit
• S3
• SNS
• SES
• KinesisFirehose
• CloudFormation
• CodeCommit
• AWS Config
  など・・・

非同期呼び出しの際、lambdaは内部的にキューイングされ実行されます。

lambdaのエラーパターン

非同期呼び出しの際は以下の２パターンの実行失敗の可能性があります

• キューイングからの実行に失敗し、起動自体が出来ない
• 重複起動される

それぞれどのように対応すればいいか、見ていきましょう

起動失敗時

lambdaは起動が失敗した際に、起動に失敗したイベントをSQSかSNSのデッドレターキュー(DLQ)に書き込むことが出来ます。
デフォルトでは書き込む設定になってないので、lambdaの設定画面から書き込み設定をすることが必要です。

DLQに書き込む設定を行ったあとは、DLQを監視するlambdaを作りましょう。
CloudWatchEventsで5分おきなどに、DLQを監視するlambdaを起動させ、DLQにeventがある場合には、再度lambdaにeventを渡して実行させれば、起動失敗時にも自動的に再実行できます。

SQSを使用した場合のlambdaのサンプルコードです(python3.6)

from __future__ import print_function

import boto3
import logging
import os

# set retry lambda function name.
LAMBDA_NAME=os.environ['lambdaFunction']

# dead letter que name.
QUE_URL=os.environ['queUrl']

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    
    logger.info("Event: " + str(event))
    
    sqs_client = boto3.client('sqs')
    
    sqs_message = sqs_client.receive_message(
        QueueUrl = QUE_URL,
        MaxNumberOfMessages = 1
    )
    
    logger.info("queue: " + str(sqs_message))
    
    if 'Messages' in sqs_message:
        retry_event = sqs_message['Messages'][0]['Body']
        receipt_handle = sqs_message['Messages'][0]['ReceiptHandle']
        
        try:
            lambda_client = boto3.client('lambda')
            
            logger.info("retry lambda:" + LAMBDA_NAME)
            
            response = lambda_client.invoke(
                FunctionName=LAMBDA_NAME,
                InvocationType='RequestResponse',
                Payload= retry_event
            )
             
            logger.info("lambda response:" + str(response))

            if response['StatusCode'] == 200:
                sqs_client.delete_message(
                    QueueUrl=QUE_URL,
                    ReceiptHandle=receipt_handle
                )
            else:
                logger.error("lambda invoke error:" + str(response))
                raise Exception(srt(response))
        except Exception as e:
            logger.error(e)
            raise e
    else:
        logger.info("sqs not contain Messages:" + str(sqs_message))

SQSに登録されるイベント情報には、起動に失敗したlambdaの名前が入ってません。
なので、環境変数などで再実行されるlambdaを渡す必要があります。

重複起動時

重複起動時の対応方法は複数ありますが、いくつか組み合わせるのがいいかなと思います。
まず、重複起動しても後続処理が問題無いように冪等に設計するのが良いと思います。
その上で、重複起動しないような仕組みを入れるとより確実かなと思います。

また、lambdaの重複起動は10分後など、タイムラグがある場合があるところにも注意が必要です。

例えばS3にファイルが置かれたトリガでlambdaを起動する際は、以下のような重複起動の抑制方法があります。

lambdaが起動されるタイミングでDynamoDBにトリガとなったファイル名を登録、２重起動された場合、すでにファイル名が登録されているので、lambda内で後続処理を起動しないようにする。

上記の通り、起動失敗時と、重複起動時のエラー処理入れることで確実に１回のみ(重複起動時には実際は起動してるが後続は動かない)lambdaを実行することが出来ます。