AWS Lambda PythonでAPIヘルスチェック(Health Check)を行うシェルスクリプト(バッチ処理プログラム) 〜Route53のヘルスチェックではできないEC2インスタンス、Amazon API GatewayなどへのPOSTメソッドのパラメタ送信にも対応〜

12月 15, 20152月 23, 2017API Gateway,AWS,EC2,Lambda,Programming,Python,Shell Script

Amazon API Gateway, AWS Lambda Pythonの登場によってフルマネージドなサーバ環境でiPhone, Androidなどのスマートフォン・モバイルアプリケーションのAPIを簡単に作成できるようになりました。

一方で作成したAPIが正常に動作し続けているかの監視運用についてはAmazon API Gateway, AWS Lambda PythonについてCloudWatchで監視することが可能です。

ただ、Amazon API Gateway, AWS Lambda PythonのバックエンドにDynamoDBなどのデータベースを配置している場合、APIへのリクエストが全てのリソースを経由して正しくレスポンスが返ってくるかをチェックする必要があります。

このような場合はRoute53のヘルスチェック機能を用いてHTTP, HTTPS, TCPをチェックことが一般的ですが、Route53のヘルスチェックはPOSTメソッドを用いたパラメタ送信のヘルスチェックは現在のところ対応していません。

そのため、ヘルスチェック機能を別途用意する必要があります。今回はAWS Lambda Pythonを対象のAPIアプリケーションとは別リージョンに用意することでリージョン障害を考慮したAPIヘルスチェックプログラムを備忘録として記載しておきます。

AWS Lambda PythonでAPIヘルスチェック(Health Check)を行うシェルスクリプト(バッチ処理プログラム) 〜Route53のヘルスチェックではできないEC2インスタンス、Amazon API GatewayなどへのPOSTメソッドのパラメタ送信にも対応〜

ヘルスチェックの方式としてはPythonのライブラリに依存しないcurlコマンドを用いてヘルスチェックを行います。

curlコマンドはGETメソッドはもちろんPOSTメソッドでのパラメタ送信にも対応することができ、Content-typeなどのヘッダの内容も設定できるためAmazon API GatewayのAPIキー認証も可能な汎用的で大変便利なコマンドです。

ここではこのcurlコマンドを用いてPOSTメソッドによるパラメタ送信に対応したAPIヘルスチェックプログラム本体とAPIヘルスチェックプログラムにパラメタを渡して呼び出すプログラムの例を記載します。

APIヘルスチェックプログラム本体の例

# -*- coding: utf-8 -*-

import commands
import os

print('Loading function')

search_keyword = ""

#Linuxのコマンドを実行するためのメソッド
def _(cmd):
    return commands.getoutput(cmd)

def lambda_handler(event, context):
    print "event:", event
    
    #ヘルスチェックで問題がない場合に表示する文字列
    ok_string = event["ok_string"]
    
    #ヘルスチェックで問題がある場合に表示する文字列
    ng_string = event["ng_string"]
    
    #実行するcurlコマンド
    curl_cmd = event["curl_cmd"]
    
    #curlコマンドの実行結果のレスポンス内を検索するキーワード
    search_keyword = event["search_keyword"]
    
    #curlコマンドにステータスを付加した形でレスポンスを受ける
    curl_cmd += ' -w "%{http_code}"'
    
    #curlコマンドの実行
    response = _(curl_cmd)
    
    #レスポンスのログへの記録
    print "----------response body start.----------"
    print response
    print "----------response body end.------------"
    
    
    if search_keyword == "":
        #検索キーワードがない場合はステータス判断のみ行う。
        if response.endswith("200"):
            print "Status is 200"
            return {"result": ok_string}
        else:
            print "Status is not 200"
            return {"result": ng_string}
    else:
        #検索キーワードがある場合はステータス判断とレスポンスにキーワードが含まれるかを判断する。
        if response.endswith("200"):
            print "Status is 200"
            
            #レスポンスのステータス部分を削除する
            body = response[0:len(response)-len("200")]
            
            #レスポンスボディーの中を検索キーワードで検索する
            if search_keyword in body:
                print "Keyword is found."
                return {"result": ok_string}
            else:
                print "Keyword is not found."
                return {"result": ng_string}
        else:
            print "Status is not 200"
            return {"result": ng_string}

APIヘルスチェックプログラムにパラメタを渡して呼び出すプログラムの例

上記のAPIヘルスチェックプログラム本体をHealthCheckByCurlというLambda関数名にしたとします。
Lambda内からLambda関数を呼び出すにはlambdaのクライアントでinvokeを使用しますのでLambda RoleのポリシーにはLambda関数を実行する権限が必要です。

このようなAPIヘルスチェックプログラムにパラメタを渡して呼び出すプログラムをAWS LambdaのScheduled Eventで定期的に動作させ、NG発生時に※1の箇所にSNSやSESのメールなどでアラートを送信する仕組みを組み込めばよいかと思います。

# -*- coding: utf-8 -*-

import boto3
import base64
import json

result_ok = "healthcheck-ok"
result_ng = "healthcheck-ng"

def lambda_handler(event, context):
    try:
        client = boto3.client('lambda')
        
        #Lambda関数HealthCheckByCurlにパラメタをPayloadに渡して実行する。
        response = client.invoke(
            FunctionName='HealthCheckByCurl',
            InvocationType='RequestResponse',
            LogType='Tail',
            Payload=b'{ \
                      "curl_cmd": "curl -s -S -F \\"data=sample\\" https://www.magtranetwork.com/", \
                      "search_keyword": "magtranetwork", \
                      "ok_string": "healthcheck-ok", \
                      "ng_string": "healthcheck-ng" \
                    }'
        )
    
        #print "response:"
        #print response
        
        #レスポンスのPayloadを読み込みUTF-8でデコードする。
        response_result = response["Payload"].read().decode('utf-8')
        
        print(type(response_result))
    
        print "response_result:%s" % response_result

        #JSON形式のレスポンス内容を取得する。
        result = json.loads(response_result)
        
        if result.has_key("result"):
            if str(result["result"]) != result_ok:
                #結果がOKない場合はLogResultをbase64でデコードしてログ内容を出力する。
                #※1アラートを出す場合はここにSNSやSESでメールを送信するコードを記載する。
                print "HealthCheck is NG."
                result_log = base64.b64decode(response["LogResult"])
                print "----------invokeHealthCheck result_log Start----------"
                print result_log
                print "----------invokeHealthCheck result_log End------------"
            else:
                print "HealthCheck is OK."
    
        print "result:%s" % result
        return result
    except Exception as ex:
        print "Exception Occurred:"
        print ex
        return "Exception Occurred: %s" % ex.message
        return {"result":"Exception Occurred: %s" % ex.message}
    finally:
        print "End HealthCheck."

Reference: Tech Blog citing related sources