Basic認証付きWebページを定期的に確認し、エラーログ出力時、Slackに通知するLambdaを作成

はじめに

管理しているwebページにエラーが発生した場合、URL:https:hoge.com/Log/error_20210920.logにログが出力されます。Status Codeは200です。
対してエラーが発生していない場合、URL:https:hoge.com/Log/error_20210920.logにアクセスすると、Status Codeは、404 Not Foundが返されます。
一般の方には、https:hoge.com/Log/error_20210920.logにアクセスできないようBasic認証をします。
Lambdaは、https:hoge.com/Log/error_20210920.logにアクセスしたときのStatus Codeの確認を実行します。

Basic認証付きWebページに対して、CloudWatch EventでLambdaを定期的にエラーログが出力されていないか確認し、出力されていればSlackに通知する仕組みを作成していきます。
ちなみにURL末尾のerror_20210920.logは、毎日、当日の日付に変わります。

流れ

①Lambda作成
②SlackのWebhookURL発行
③CloudWatch Eventの設定

Lambda作成

1.下記の設定でLambda作成
・タイプ：一から作成
・関数名：confirm-StatusCode-to-Slack
・ランタイム：python3.8

2.設定→環境変数を設定します
キー：WebhookURL、値：webhookのURLにします。
コード内でos.environ['WebhookURL']とすると値を参照できます。
また、Basic認証用にキーと値を設定します。
キー：TZ、値：Asia/Tokyoを設定しています。
理由として、Lambdaでは東京リージョンでもタイムゾーンはUTCであるため、日時もUTC時刻になってしまうことにありましたので、その対策として設定しています。

Confirm-StatusCode-to-Slack

import datetime
import urllib.request
import base64
import json
import os

def lambda_handler(event, context):
    #今日の日付のdateオブジェクトを取得
    day = datetime.date.today()

    #format関数で日付をYYYYMMDD形式に変換
    today = format(day, '%Y%m%d')

    url = "https:hoge.com/Log/error_" + today + ".log"

    try:
        # Basic認証の情報
        user = os.environ['user']
        password = os.environ['password']

        # Basic認証用の文字列を作成.
        basic_user_and_pasword = base64.b64encode('{}:{}'.format(user, password).encode('utf-8'))

        # Basic認証付きの、GETリクエストを作成する.
        request = urllib.request.Request(url, 
            headers={"Authorization": "Basic " + basic_user_and_pasword.decode('utf-8')})

        # 送信して、サイトにアクセスする
        response = urllib.request.urlopen(request, timeout = 10)

        #ステータスコードを取得する。
        status_code = response.getcode()

        #ステータスコードが正常(200)な時の処理。
        if status_code == 200:

            # slack通知
            post_slack()

        else:
            print("[failure] %s is unhealthy. \n" % url)
            print("[status-code] : %s. \n" % status-code)

    except (urllib.error.HTTPError) as error:
        #エラーログがない場合、ステータス404が返される
        status_code = error.code
        print("エラーログなし %s\n URL: %s" %(status_code,url))

    except (urllib.error.URLError) as error:
        #ステータスコードが取得できない
        status_code = "HTTP通信の不可"
        print(status_code)

def post_slack():
    day = datetime.date.today()
    today = format(day, '%Y%m%d')

    url = "https:hoge.com/Log/error_" + today + ".log"

    message = " <!channel> エラーログが出力されました。" + url

    send_data = {
        "text": message,
    }
    send_text = json.dumps(send_data)

    request = urllib.request.Request(
        os.environ['WebhookURL'], 
        data=send_text.encode('utf-8'), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode('utf-8')

SlackのWebhookURL発行

WebhookURL発行を参考に、Lambdaで使用するWebhookURLを発行してください

CloudWatch Eventの設定

イベント条件は下記になります。
・月曜日から日曜日、毎日
・8:00から18:00まで
・2時間ごとにLambdaを実行

1.CloudWatch Eventのルール作成をクリック

2.イベントソースはスケジュールのCron式を選択し、ターゲットは先程作成したLambdaを選択し、設定詳細をクリックする

Cron式は、「* * * * *」の5箇所を指定して、起動時間を設定します。左から「分」「時」「日」「月」「曜日」を指定します。
Cron式:0 23-9/2 * * ? *と設定している場合、「00分」「8~18時,2時間ごと」「毎日」「毎月」「毎日(月〜日)」という起動時間が設定されます。
注意事項として、時間はUTCに設定されていますので、考慮してください。

下記はサンプルになりますので、参考にしてみてください
・cronの日時指定を、基礎から学ぶ
・cron式サンプル

3.ルール名と説明を記入し、作成をクリックする

参考

・urllib.request --- URL を開くための拡張可能なライブラリ
・[Python] 3系でBasic認証付きのHTTPリクエストを発行する（urllib.request利用）
・例外を処理する
・CloudWatch Eventのルールの式をスケジュールする