新年あけましておめでとうございます、技術推進室とサービス開発部を兼務している宇津井です。

昨年は当社のエンジニアブログがクリエイターブログに変わり、デザイナーが記事投稿頑張ってていい感じです。エンジニアもGMOペパボ Advent Calendar 2018の26日目という謎の記事投稿があったりしますが全体的に少なめ。ちょっと寂しいので半年ほど前に実施したCDNの監視について書いてみようと思います。

今回はAmazon CloudFront(以下CF)を利用していて落とし穴になりがちなのがエラー率の監視です。

盲点になりがちなCDNのエラー監視

サーバーサイドプログラムに関してはエラーの発生をなんらかの方法で監視しているケースが殆どだと思いますが、CDNはその性質上静的ファイルを扱うことが多く、例え404エラーが出ていたとしても、ブラウザ等の見た目に支障がないことも多く無視されがちです。

きっかけは忘れましたが、とあるサービスで利用しているCFのエラー率をマネージドコンソールで見ていたら信じられないことに50%を超えるエラーが発生している時間帯がある事に気付きました。

CFでエラーが発生している場合、そのエラー内容をコンソールから確認できるケースは少ないです。コンソール上で確認しようとすると、Top50リクエストの中でエラーが含まれていれば良いのですが、これには通常リクエストも含まれるため、エラーとなっているリクエストが含まれないケースが多くなります。

その場合どうするかというと、CloudFrontのアクセスログをS3に出して、アクセスログの解析をする事になります。この設定自体はすぐに行えます。(社内ではTerraformのPrivate Registryを利用してCloudFrontを構成しているので自動的にこの設定が行われるようになっています)
https://docs.aws.amazon.com/ja_jp/AmazonCloudFront/latest/DeveloperGuide/AccessLogs.html

ログファイル配信のタイミング
CloudFront は、ディストリビューションのアクセスログを 1 時間に最大で数回配信します。一般的に、ログファイルには、一定期間内に CloudFront が受信したリクエストに関する情報が含まれています。CloudFront は通常、その期間のログファイルを、ログに書き込まれたイベントの発生から 1 時間以内に Amazon S3 バケットに配信します。ただし、ある期間のログファイルエントリの一部またはすべてが、最大で 24 時間遅れることもあります。ログエントリが遅れた場合、CloudFront はこれらをログファイルに保存します。そのファイル名には、ファイルが配信された日時ではなく、リクエストが発生した期間の日時が含められます。
CloudFront は、ログファイルを作成する場合、ログファイルに対応する期間中にオブジェクトについてリクエストを受信したすべてのエッジロケーションから、ディストリビューションの情報を集約します。
CloudFront は、ディストリビューションに関連付けられているオブジェクトに対して CloudFront が受信したリクエストの数によって、1 つの期間に対して複数のファイルを保存することもできます。
CloudFront は、ロギングが有効化し 4 時間後ほどから確実にアクセスログを書き出し始めます。この時間以前にも少しのアクセスログを取得できる場合もあります。

上記引用の通り、S3には数分〜数十分おきにgzipで圧縮されたアクセスログがどんどん作られていきます。一定の確率でエラーがで続けているときは最新のアクセスログをダウンロードして、手元で解凍して、grepして、uniqで集計して、sortして、、、みたいな事を行えば原因が解決する場合があります。

自動化

原因が１箇所の場合は、この作業も１回で終わるので手作業で集計してもいいのですが、これ実は自動化できるんです。自動化に当たって利用する事になるAWSサービスは、CloudWatch, SNS, Lambda, Athena辺りです。それらを組み合わせると、一定の期間内にエラー率が任意の閾値に達した場合にSlackにエラー明細のTop10を添付して、アラートをSlackのチャンネルに投げるなんてことができます。(こんなイメージです)

届いたアラートメッセージ

最終的にこんなメッセージがSlackに届きます。
最新版ではReferrerも記載されているので、この通知だけで、どのリクエストがどのくらいエラー発生していて、そのリクエストがどこで発生しているのかがざっくり把握できてしまいます。

ざっくりした設定方法

CloudFrontのLogging設定は済んでいる事を前提とします。
以下は全てGlobal Region(N. Virginia: us-east-1)での操作とします。

CloudWatch Alarm

まずCloudWatchのAlarmにCloudFrontのTotalErrorRateの閾値を登録します。

Metric

Threshold

Athena

ログ集計をサクッとやりたいのでAthenaを利用します。

以下のようなテーブルを予め作っておきます。

CREATE EXTERNAL TABLE `cloudfront_logs_cdn_example_com`(
  `date` date, 
  `time` string, 
  `location` string, 
  `bytes` bigint, 
  `requestip` string, 
  `method` string, 
  `host` string, 
  `uri` string, 
  `status` int, 
  `referrer` string, 
  `useragent` string, 
  `querystring` string, 
  `cookie` string, 
  `resulttype` string, 
  `requestid` string, 
  `hostheader` string, 
  `requestprotocol` string, 
  `requestbytes` bigint, 
  `timetaken` float, 
  `xforwardedfor` string, 
  `sslprotocol` string, 
  `sslcipher` string, 
  `responseresulttype` string, 
  `httpversion` string, 
  `filestatus` string, 
  `encryptedfields` int)
ROW FORMAT DELIMITED 
  FIELDS TERMINATED BY '\t' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  's3://cf-logs-cdn_example-com/'
TBLPROPERTIES (
  'skip.header.line.count'='2')

通知を自動化しなくても、このようなクエリーを登録しておくだけでも便利です。

SELECT uri,
         status,
         referrer,
         count(*)
FROM cloudfront_logs_cdn_example_com
WHERE responseresulttype = 'Error'
        AND date > now() - interval '1' day
GROUP BY  uri, status, referrer
ORDER BY  4 DESC

Lambda

上記アラームの通知先にSNSを設定して、SNSからLambdaを呼び出すので、まずLambdaを作成します。

コード例:

import boto3
import json
import logging
import os
import time

from base64 import b64decode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError

SLACK_CHANNEL = os.environ['slackChannel']

HOOK_URL = os.environ['HookUrl']

logger = logging.getLogger()
logger.setLevel(logging.INFO)

athena = boto3.client('athena')

def send_slack_message(message):
    alarm_name = message['AlarmName']
    new_state = message['NewStateValue']
    reason = message['NewStateReason']

    table_name = ""
    if alarm_name == "CF-TotalError-img-example-com":
        table_name = "cloudfront_logs_img_example_com"
    elif alarm_name == "CF-TotalError-galaxy-example-com":
        table_name = "cloudfront_logs_galaxy_example_com"
    elif alarm_name == "CF-TotalError-static-example-com":
        table_name = "cloudfront_logs_static_example_com"

    if table_name != "":
        response = athena.start_query_execution(
            QueryString='SELECT uri, status, referrer, count(*) AS error_count \
                         FROM %s \
                         WHERE responseresulttype = \'Error\' \
                         AND date > now() - interval \'1\' day \
                         GROUP BY  uri, status, referrer \
                         ORDER BY  4 DESC  \
                         limit 10' % table_name,
            QueryExecutionContext={
                'Database': 'default'
            },
            ResultConfiguration={
                'OutputLocation': 's3://aws-athena-query-results-307184118939-us-east-1/athena-cloudfront-alarm'
            }
        )
        # レスポンスからクエリの ID をゲット
        query_id = response['QueryExecutionId']

        # クエリ ID を引数に渡して、進捗状況を確認
        # ステータスが SUCCEEDED or FAILED or CANCELLED になるまでループ
        while True:
            response = athena.get_query_execution(QueryExecutionId=query_id)
            state = response['QueryExecution']['Status']['State']
            if state in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
                break
            else:
                print('wait...')
                time.sleep(10)

        # クエリ ID を引数に渡して、実行結果を取得
        response = athena.get_query_results(QueryExecutionId=query_id)

        resultset_rows = ""
        for rows in response['ResultSet']['Rows']:
            resultset_rows = resultset_rows + '\t'.join([data['VarCharValue'] for data in rows['Data']]) + "\n"
        # resultset_rows = ",".join(map(str, response['ResultSet']['Rows']))

    slack_message = {
        'channel': SLACK_CHANNEL,
        'pretext': "%s state is now %s: %s" % (alarm_name, new_state, reason),
        'color': "#D00000",
        'fields':[
            {
               'title':'Popular Errors',
               'value': resultset_rows
            }
        ]
    }

    req = Request(HOOK_URL, json.dumps(slack_message).encode('utf-8'))
    try:
        response = urlopen(req)
        response.read()
        logger.info("Message posted to %s", slack_message['channel'])
    except HTTPError as e:
        logger.error("Request failed: %d %s", e.code, e.reason)
    except URLError as e:
        logger.error("Server connection failed: %s", e.reason)

def lambda_handler(event, context):
    logger.info("Event: " + str(event))
    message = json.loads(event['Records'][0]['Sns']['Message'])
    logger.info("Message: " + str(message))

    send_slack_message(message)

SNS

SNS Topicを作成します。

受け取り方としてLambdaを登録します。

もう一度CloudWatch Alarm

最後に、CloudWatch AlarmのAction欄に上記のSNS Topicを指定して完了。

Infrastructure as a code

Lambda以外はある程度terraform化できると思うので、いつかやっとこうと思ってます。
と社内向けに書いたのが2018/7/24のようですが、この記事を書いている今日(2018/12/28)現在やってませんね・・・(汗)