AWS LambdaとSlack incoming webhookでサイトの生存確認チェッカーを作る #AWS

主要サービスにはいろいろ監視ソフト入れて、トラフィックの異常増大や、生存確認、リソースの利用状況なんかを可視化しているケースが多いと思います。ですが、例えば自社サイト(wordpressとかが多い)はじめ、管理コストをあまりかけられないサービスなりサイトって有りますよね。そんなサイトたちをAWS LambdaとSlackのincoming webhookを使って落ちているときに通知してくれるスクリプトを書いてみました。

Slackのincoming webhookのURLを発行する

ワークスペースの設定　→　APP管理　→　カスタムインテグレーションへ移動
直リンクだと以下のページ
https://##yourdomain##.slack.com/apps/manage/custom-integrations

Incoming Webhookの項目を選択　→　Slackに追加を選択　→　通知を送りたいSlackチャネルを選択　→　発行されたURLをコピーしておく

https://hooks.slack.com/services/xxxxxxxx/xxxxxxx/xxxxxxxxxxxxこんな感じのURLが発行されるはずである。これをメモって置きましょう。

Lambdaのクローリングコードを作成

AWS Lambdaから関数の作成を行い、python3.7で以下のコードを書く

from __future__ import print_function

import datetime
import urllib.request
import json


SITES = [
    'https://www.example.com/',
    'https://ここにチェックしたいサイトを入れてね'
    ]
DESCRIPTION = "サイトの生存確認用チェッカーからの連絡"
SLACK_WEBHOOK = 'ここにSlackのincoming webhookのURLをコピペ'
DIFF_JST_FROM_UTC = 9

def lambda_handler(event, context):
    for site in SITES:
        request = urllib.request.Request(site)
        now = datetime.datetime.now() + datetime.timedelta(hours=DIFF_JST_FROM_UTC)
        nowstr = now.strftime("%Y/%m/%d %H:%M:%S")
        try:
            with urllib.request.urlopen(request) as res:
                url = res.geturl()
                print(url)
                if not url == site :
                    raise ValueError("error!")
        except urllib.error.HTTPError as e:
            print('Check failed!')
            post_slack("<!here>\n{0}\n{1}\n※※※※※※※※\n{2}\nHTTP通信エラー\n※※※※※※※※".format(nowstr,site,DESCRIPTION))
            raise
        except urllib.error.URLError as e:
            print('Check failed!')
            post_slack("<!here>\n{0}\n{1}\n※※※※※※※※\n{2}\n URLエラー\n※※※※※※※※".format(nowstr,DESCRIPTION))
            raise
        except ValueError as e:
            print('Check failed!')
            post_slack("<!here>\n{0}\n※※※※※※※※\n{1}\n ルーティングエラー：リダイレクトされている\n※※※※※※※※".format(nowstr,DESCRIPTION))
            raise
        else:
            print('Check passed!')
            # post_slack("{0}\n{1} 成功".format(nowstr,DESCRIPTION))
            return
        finally:
            print('Check complete at {}'.format(str(datetime.datetime.now())))

def post_slack(message):

    send_data = {
        "username": "AWS Lambda",
        "text": message,
    }
    send_text = "payload=" + json.dumps(send_data)
    request = urllib.request.Request(
        SLACK_WEBHOOK, 
        data=send_text.encode("utf-8"), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode("utf-8")

上記のスクリプトをクローリングさせる

関数作成ページの上の方にあるトリガーの追加をクリック

CloudWatch Events/EventBridgeを選択して、いい具合にルールを作る。
今回は15分に1回テスト通知させる。もし、時間間隔を変更したいときにはスケジュール式をよしなに書いてみてください。
書き方の公式ドキュメントはこちら

実際に試してみる

SITESの配列のURLに存在しないページやドメインを記載してみてください。
incoming webhookを設定したSlackチャネルにエラーメッセージが飛んできたら成功です。
なお、このスクリプト、問題ないときには何も発生しません。