CloudFrontのログをAthenaで読むためにリネームするLambda用Pythonスクリプト書いてみた

こんにちは、臼田です。

皆さんLambdaしていますか？

さて、Amazon AthenaでS3に格納されているデータを分析できるようになってから久しいですが、CloudFrontのログを直接参照しようとすると少し問題になります。

CloudFrontのログはディストリビューション毎に一括して保存されていて、Athenaはそのディストリビューションのログのレンジ(日時)を絞ってクエリできません。

その為、CloudFrontのログを「年/月/日」のようにプレフィックスを変更することで、Athenaで利用しやすい状態となります。

この方法については以前下記ブログで紹介されていたのですが、ログが保存されるたびに実行しなくても日時バッチ的な方法でもいいかなーといったことや、エラーによる処理漏れとかがあっても簡単に再試行出来るようにと思い別のスクリプトを作成しました。

LambdaでCloudFrontのログをリネームしてみた

コンセプト

今回のスクリプトのコンセプトは以下のとおりです。

• 指定したバケットの特定のプレフィックスにあるCloudFrontのログを、指定したバケットの特定の日時を切ったプレフィックスに移動
• 移動元、移動先は環境変数で指定する
• 何回実行しても大丈夫(冪等性)

実装

基本的には上記に上げたブログを参考にしてください。

違いがある部分についてここで説明していきます。

Lambda

Lambda用のコードは下記のとおりです。Python3.6で実装しています。

# -*- coding: utf-8 -*-

import os
import boto3


ORIGIN_BUCKET_NAME = os.environ['ORIGIN_BUCKET_NAME']
ORIGIN_BUCKET_KEY = os.environ['ORIGIN_BUCKET_KEY']
TARGET_BUCKET_NAME = os.environ['TARGET_BUCKET_NAME']
TARGET_BUCKET_KEY = os.environ['TARGET_BUCKET_KEY']


def lambda_handler(event, context):
	s3 = boto3.resource('s3')
	ob = s3.Bucket(ORIGIN_BUCKET_NAME)
	tb = s3.Bucket(TARGET_BUCKET_NAME)

	try:
		# get origin files list
		files = ob.objects.filter(Prefix=ORIGIN_BUCKET_KEY)
		for f in files:
			# ignore folder
			if f.key[-1] == "/": continue
			# get object name
			object_name = f.key.split('/')[-1]
			# create new object's prefix
			timestamp = f.key.split('.')[-3]
			# the following prefix is 'target/yyyy/mm/dd/'
			prefix = TARGET_BUCKET_KEY + timestamp[:-3].replace('-', '/') + '/'
			# the following prefix is 'target/yyyy/mm/dd/hh/'
			# prefix = TARGET_BUCKET_KEY + timestamp.replace('-', '/') + '/'
			# the following prefix is 'target/yyyy-mm-dd/'
			# prefix = TARGET_BUCKET_KEY + timestamp[:-3] + '/'
			# the following prefix is 'target/yyyy-mm-dd-hh/'
			# prefix = TARGET_BUCKET_KEY + timestamp + '/'
			target_object = prefix + object_name

			# create origin source
			source = {'Bucket': ORIGIN_BUCKET_NAME, 'Key': f.key}
			# copy to new object
			new_obj = tb.Object(target_object)
			new_obj.copy(source)
			# delete origin object
			r = f.delete()

	except Exception as e:
		print(e)
		raise e


# call lambda_handler
if __name__ == "__main__":
	lambda_handler({}, {})

環境変数でバケット名とプレフィックスを指定します。

これは、移動先に同じバケットを指定することもできますし、別のバケットを指定することもできます。

今回は、CloudFrontで生成されるプレフィックスから、archive/へリネーム(移動)するように設定しています。

リネームするパターンですが、複数パターン用意しました。

• 'target/yyyy/mm/dd/'
• 'target/yyyy/mm/dd/hh/'
• 'target/yyyy-mm-dd/'
• 'target/yyyy-mm-dd-hh/'

この辺は好みになると思いますが、大体は「'target/yyyy/mm/dd/'」で大丈夫だと思います。

IAMロール

boto3にてバケットを参照しているため、参照元のブログの設定に追加してActionに"s3:ListBucket"、Resourceに"arn:aws:s3:::バケット名"("/*"を含まない)を追加する必要があります。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:GetObject",
                "s3:PutObject",
                "s3:DeleteObject"
            ],
            "Resource": [
                "arn:aws:s3:::cf-log-test-bucket-brrrcx",
                "arn:aws:s3:::cf-log-test-bucket-brrrcx/*"
            ]
        }
    ]
}

トリガー

Lambdaのトリガーはコンセプト通りなんでも大丈夫です。

cronでデイリーで実行するなどがいいと思います。

まとめ

イベントドリブンで引数を取って実行するのもいいですが、エラー設計とか、エラー時の対処が面倒になることもあります。

冪等性を意識して実装しておくことも一つの方法としていいかと思います。