本番リクエストを開発環境に投げる→エラーを検知→修正するというサイクルで開発をすると品質が上がっていくというのを最近実感しました

この記事は一休.comアドベントカレンダー2018の23日目です。

一休.comの開発基盤をやっています akasakas です。

長いタイトルですいません。

本日のお話

本番リクエストを開発環境に投げて、エラーを検知し、修正するというサイクルで開発をすると品質が上がっていくというのを最近実感しました

という話です。

図にするとこんな感じのイメージです

f:id:akasakas:20181215135503p:plain

やっていること

本番リクエストログから抽出
開発環境に対して、リクエストを投げる
バグみつかる
なおす

を 細かく 繰り返す

これを繰り返せば、 ~~ちゃんと試験考えなくても、勝手に~~ 品質上がっていくようになるのかなと思いました。

安心してリリースができるのはエンジニアの心理的に非常にいいことなのかなと思います。

具体的にどうやっているのか？

一休ではLogentriesでアクセスログを管理しています。

LogentriesのAPIを使い

本番リクエストを取得
開発環境にリクエストを投げる
HTTPステータスコードを確認

というところまでやりたいと思います。

参考実装の前に諸注意

LogentriesのAPIは呼び出し制限があるので、用法用量を守って、正しくお使いください。詳しくはこちらになります。

docs.logentries.com

参考実装

参考実装がこちらになります。

import logging
import datetime
import requests
import time
import re
import json

settings = {
    'dev_host': 'dev.hostname.com',
    'sampling_seconds': 180,
    'logentries': {
        'endpoint': 'https://rest.logentries.com/',
        'apikey': 'logentries api key', 
        'logs_query': 'query/logs',
        'query_api': 'query',
        'query_statement': 'where(' \
        + '   path = /query_statement(regular_expression)/ ' \
        + ')',
        'logs': [ 
            'logenrties log set key id'
        ]
    }
}

logging.basicConfig(format='[%(asctime)s] %(message)s')
logger = logging.getLogger('real-request-to-development')
logger.setLevel(logging.DEBUG)

def calc_query_time():
    now = datetime.datetime.now()
    from_time = datetime.datetime.now() - datetime.timedelta(seconds=(settings['sampling_seconds']))
    return int(time.mktime(from_time.timetuple())) * 1000, int(time.mktime(now.timetuple())) * 1000


def le_query(from_unixtime, to_unixtime):
    headers = {'x-api-key': settings['logentries']['apikey']}
    body = {
        "logs": settings['logentries']['logs'],
        "leql": {"during": {"from": from_unixtime, "to": to_unixtime}, "statement": settings['logentries']['query_statement']}
    }
    res = requests.post(f'{settings["logentries"]["endpoint"]}{settings["logentries"]["logs_query"]}', headers=headers, json=body)
    
    return res.json()["id"]


def le_longtime_query(id):
    headers = {'x-api-key': settings['logentries']['apikey']}
    res = requests.get(f'{settings["logentries"]["endpoint"]}{settings["logentries"]["query_api"]}/{id}', headers=headers)

    return res.json()


def get_request_url(json_data):
    dev_url_list = []
    for log_row in json_data["events"]:
        log_data = re.search(r'{.*}' , log_row["message"])
        log_json = json.loads(log_data.group(0))

        path = log_json["path"]
        query = log_json["query"]
        useragent = log_json["useragent"]

        dev_url = f'https://{settings["dev_host"]}{path}{query}'

        dev_url_dict = {"dev_url": dev_url, "ua": useragent}
        dev_url_list.append(dev_url_dict)
    return dev_url_list


def send_request(dev_url_list):
    for dev_url_dict in dev_url_list:
        headers = {
            'User-Agent': dev_url_dict["ua"]
        }
        dev_response = requests.get(dev_url_dict["dev_url"], headers=headers)

        if dev_response.status_code >= 500:
            logger.error(f'development url is {dev_url_dict["dev_url"]}')
            logger.error(f'development response status is {str(dev_response.status_code)}')


#########################################
def start():
    from_unixtime, to_unixtime = calc_query_time()
    le_query_id = le_query(from_unixtime, to_unixtime)
    json_data = le_longtime_query(le_query_id)
    dev_url_list = get_request_url(json_data)
    send_request(dev_url_list)


if __name__ == "__main__":
    start()

1つずつ解説していきたいと思います

流れとしては

Logentries API で検索用のクエリIDを取得&実際のリクエストを取得
実際のリクエストのホストを開発環境用のホストに書き換える
リクエストを投げる&500エラーをログ出力

2.3は難しいことはしていないので、説明は省略します。

Logentries API で検索用のクエリIDを取得&実際のリクエストを取得

Logentries API で検索用のクエリIDを取得

下記で、検索用のクエリIDを取得しています。このクエリIDをベースにして、実際のリクエストを取得します。

必要なパラメータは

logs(どこのログか？ここではアクセスログ)
statement(どんな条件か？正規表現でpathを記述するのが一般的？)
from,to(unixtimeなので、事前にfrom,toをunixtimeにする必要がある)

です。

詳細はこちらをご覧ください。 docs.logentries.com

def le_query(from_unixtime, to_unixtime):
    headers = {'x-api-key': settings['logentries']['apikey']}
    body = {
        "logs": settings['logentries']['logs'],
        "leql": {"during": {"from": from_unixtime, "to": to_unixtime}, "statement": settings['logentries']['query_statement']}
    }
    res = requests.post(f'{settings["logentries"]["endpoint"]}{settings["logentries"]["logs_query"]}', headers=headers, json=body)
    
    return res.json()["id"]

実際のリクエストを取得

先ほど取得したクエリIDをベースから、実際のリクエストを取得しています。

詳細はこちらをご覧ください。

docs.logentries.com

def le_longtime_query(id):
    headers = {'x-api-key': settings['logentries']['apikey']}
    res = requests.get(f'{settings["logentries"]["endpoint"]}{settings["logentries"]["query_api"]}/{id}', headers=headers)

    return res.json()

実際のリクエストを開発環境に投げることでエラーを検知というところまでできました。ここから、試験と修正を繰り返せば、品質は上がり、安心してリリースできることになると思います。