Selenium × ECS × APMツールで作る管理画面ログインhealthcheck

サーバーサイドエンジニアの鬼木です！

今回はYappliのCMSにログインできるかどうかを確認、通知するhealthcheck機構を導入した記事になります。

とあるサービス障害で一時的にCMSのログイン認証機構に障害が発生し、標準的なTCPレベルのhealthcheckでは検知出来ず初動が遅れた背景から今回の実装に至りました。

上記のようなインシデントの対策として、サービスの入り口であるログイン画面の表示からログインし、ログイン後の最初のページを表示までを確認するhealthcheckをSeleniumで実装しました。 こちらのhealthcheckは10分おきに実行され、その実行ログをDatadogで収集、異常を検知した際にSlackに通知する仕組みになっています。全体構成は以下のようになります。

application

appicationは言語はPythonでSeleniumを使用してログインの確認を行いました。実装は以下のようになります。

import os
import ast
import time
import boto3
import base64
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from botocore.exceptions import ClientError

sleepSec = 10
retry = 2

secret_name = os.environ['ENV'] + "-cms-healthcheck"
region_name = "ap-northeast-1"

def getChromeOption():
    chrome_options = Options()
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--headless")
    chrome_options.add_argument("--disable-dev-shm-usage")
    chrome_options.add_argument("start-maximized")
    chrome_options.add_argument("enable-automation")
    chrome_options.add_argument("--disable-infobars")
    chrome_options.add_argument('--disable-extensions')
    chrome_options.add_argument("--disable-browser-side-navigation")
    chrome_options.add_argument("--disable-gpu")
    chrome_options.add_argument('--ignore-certificate-errors')
    chrome_options.add_argument('--ignore-ssl-errors')
    prefs = {"profile.default_content_setting_values.notifications" : 2}
    chrome_options.add_experimental_option("prefs",prefs)
    return chrome_options

for num in range(retry):
    try:
        session = boto3.session.Session()
        client = session.client(
            service_name='secretsmanager',
            region_name=region_name
        )

        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
        if 'SecretString' not in get_secret_value_response:
            raise Exception("SecretString is not contined in get_secret_value_response")

        secret = ast.literal_eval(get_secret_value_response['SecretString'])

        chrome_options = getChromeOption()
        driver = webdriver.Chrome(options=chrome_options)

        driver.delete_all_cookies()
        driver.get('https://xxxxxxxx') # 管理画面URL
        driver.find_element_by_name('email').send_keys(secret['healthcheck_user_email'])
        driver.find_element_by_name('password').send_keys(secret['healthcheck_user_password'])
        driver.find_element_by_id('gtm-login-button').click()

        time.sleep(sleepSec)
        cur_url = driver.current_url
        # url遷移チェック
        if cur_url != 'https://xxxxxxxx': # ログイン後URL
            raise Exception("transition to dashboard failed")

        time.sleep(sleepSec)
        # 表示チェック
        appTitle = driver.find_element_by_class_name('CmsAppHeaderAppList-linkName')
        if appTitle.text != 'SRE検証アプリ':
            raise Exception("check app title failed")
        driver.quit()
        break
    except Exception as e:
        if num < retry - 1:
            continue
        raise e

print("True")

以下でコードの解説をしていきたいと思います！

解説

secret_name = os.environ['ENV'] + "-cms-healthcheck"
region_name = "ap-northeast-1"

# 略

for num in range(retry):
    try:
        session = boto3.session.Session()
        client = session.client(
            service_name='secretsmanager',
            region_name=region_name
        )

        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
        if 'SecretString' not in get_secret_value_response:
            raise Exception("SecretString is not contined in get_secret_value_response")

        secret = ast.literal_eval(get_secret_value_response['SecretString'])

        # 略

今回はログインの確認をするためのtest用のアカウント情報をAWS Secrets Managerを使用して管理、取得を行っています。 機密情報の取り扱い方に関しては色々なやり方があると思いますが、扱いやすいのと同時にこのようなサービスが用意されていることでcredential情報をケースによってS3に置いたり他のサービスに置いたりと分散しないなど管理方法についてのコストも下げられるメリットがあるなと思いました。

        driver.delete_all_cookies()
        driver.get('https://xxxxxxxx') # 管理画面URL
        driver.find_element_by_name('email').send_keys(secret['healthcheck_user_email'])
        driver.find_element_by_name('password').send_keys(secret['healthcheck_user_password'])
        driver.find_element_by_id('gtm-login-button').click()

        time.sleep(sleepSec)
        cur_url = driver.current_url
        # url遷移チェック
        if cur_url != 'https://xxxxxxxx': # ログイン後URL
            raise Exception("transition to dashboard failed")

        time.sleep(sleepSec)
        # 表示チェック
        appTitle = driver.find_element_by_class_name('CmsAppHeaderAppList-linkName')
        if appTitle.text != 'SRE検証アプリ':
            raise Exception("check app title failed")
        driver.quit()
        break
    except Exception as e:
        if num < retry - 1:
            continue
        raise e

print("True")

retry時にfreshな状態でSeleniumを実行したいため、delete_all_cookiesでcookie情報の初期化を行っています。また、ブラウザを介したE2Eテストなどを行うときの注意点でもありますが、遷移時にはsleepを入れて遷移を待つようにしています。 またこのapplicationは高頻度で実行されるためネットワークやその他の要因などで処理がエラーになって誤検知になってしまう可能性を考慮し、全体の処理を2回までretryするようにしています。

ecosystem

上記のコードをECSのScheduling Taskで10分おきに実行し、applicationから出力されたログをDatadogが収集し、それをもとにSlackに通知するようにしています。

Yappliでは多くのapplicationのログをDatadog、New Relicで収集しています（現在DatadogからNew Relicに移行中）。今回は簡易的なhealthcheckなこともあって初めはapplicationからslackへのnotifyを行うことを考えていましたが、その責務をDatadogで受けることでもう少し複雑なapplicationでは閾値の変更などの検知のルールをapplicationのdeploy無しに柔軟に対応することができますし、処理のfailだけでなく監視の仕組み自体が止まっていることなどの検知も可能になります。

まとめ

SeleniumとECSとDatadogを使ったhealthcheck機構について紹介しました。

一度こういった仕組みを作れば移行は簡易的に実装、追加できますし、特に実際の処理を書くSeleniumは扱いやすく、真新しい技術ではありませんがその分ナレッジも多いのでこういった単純なhealthcheckでは確認できないケースのアラートも億劫になることなく開発、導入できるのが便利だと感じました！エコシステムの面からもサービスの品質向上を図っていきたいです。