LambdaでのCloudWatch LogsからS3へのエクスポート

はじめに

CloudWatch LogsのログデータはS3へエクスポートすることができます。
CloudWatch Logsはログデータが多くなると保存料金が結構高くなるため(2019年8月現在東京リージョンで0.033USD/GB)、保存する際はS3へエクスポートすることをお勧めします。

今回、CloudWatch Eventsで1時間ごとにエクスポートを行うLambdaを作成したため、メモ。

環境

Pyton3.6を使用しました。
上記にも書きましたが、構成として
Cloudwatch Eventsで定期的(今回は1時間ごと)にLambdaを動かす。
CloudWatch Logsで１時間以内に更新があったデータをS3にエクスポートする。

事前準備

IAMロールの作成

Lambda関数がCloudWatch LogsにアクセスするためのIAMポリシー・IAMロールを作成します。

IAMポリシー

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "logs:*"
            ],
            "Resource": "*"
        }
    ]
}

Lambda用IAMロールを作成します。ポリシードキュメント(信頼関係)は以下のようにし、先ほど作成したIAMポリシーをアタッチしてLambdaにIAMロールを割り当てる。

ポリシードキュメント

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

S3バケット作成

ログをエクスポートするバケットを作成します。バケットポリシーでCloudWatch Logsからのアクセスを許可します。

バケットポリシー

    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "logs.リージョン名.amazonaws.com"
            },
            "Action": "s3:GetBucketAcl",
            "Resource": "arn:aws:s3:::バケット名"
        },
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "logs.ap-northeast-1.amazonaws.com"
            },
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::バケット名/*",
            "Condition": {
                "StringEquals": {
                    "s3:x-amz-acl": "bucket-owner-full-control"
                }
            }
        }
    ]
}

Lambda関数の作成

環境変数の設定

今回は、環境変数に以下の内容を設定しました。

• BUCKET_NAME
  S3バケット名を指定
• LOG_GROUP_NAME_PREFIX
  エクスポートするロググループ名のprefixをカンマ区切りで指定
• EXCEPT_LOG_GROUPS
  指定したprefixの中でもエクスポート対象から除外するロググループ名をカンマ区切りで指定

例えば、

LOG_GROUP_NAME_PREFIX に**/aws/lambda, /aws-glueを指定し、
/aws/lambda/cloud9xxxxxxのロググループは除外する場合、
EXCEPT_LOG_GROUPSに/aws/lambda/cloud9xxxxxx**を指定します。

Lambda関数コードの作成

export-logs.py

import datetime
import time
import boto3
import re
import sys
import os

s3_bucket_name = os.environ.get('BUCKET_NAME')
client = boto3.client('logs', region_name='ap-northeast-1')
except_log_groups = os.environ.get('EXCEPT_LOG_GROUPS').split(',')


class JST(datetime.tzinfo):
    def utcoffset(self, dt):
        return datetime.timedelta(hours=9)

    def dst(self, dt):
        return datetime.timedelta(0)

    def tzname(self, dt):
        return 'JST'


def lambda_handler(event, context):
    one_hour_ago = datetime.datetime.now() - datetime.timedelta(hours=1)
    from_datetime = one_hour_ago.replace(minute=0, second=0, microsecond=0)
    to_datetime = from_datetime + \
        datetime.timedelta(hours=1) - datetime.timedelta(milliseconds=1)
    print(f"export logs from {from_datetime} to {to_datetime}")
    for log_group_des in get_log_group_des(next_token=None):
        log_group_names = log_group_des['logGroupName']
# ロググループの除外
    for except_log_group in except_log_groups:
        log_group_names.remove(except_log_group)
    for log_group_name in log_group_names:
        try:
            res = create_export_task(
                s3_bucket_name, log_group_name, from_datetime, to_datetime)
            print(
                f"successfully created an export task for {log_group_name}. {res}")
    except Exception as e:
        print("Unexpected error:", e)
    return True


def get_log_group_des(next_token):
    if next_token is not None and next_token != '':
        response = client.describe_log_groups(
            limit=50,
            nextToken=next_token
        )
    else:
        response = client.describe_log_groups(limit=50)
    if 'logGroups' in response:
        yield from response['logGroups']['logGroupName']
    # ロググループが多くて50件(最大)を超えるようなら再帰呼出し
    if 'nextToken' in response:
        yield from get_log_group_des(next_token=response['nextToken'])


def create_export_task(s3_bucket_name, log_group_name, from_datetime, to_datetime):
    group_path_component = log_group_name[1:]
    datetime_path_component = from_datetime.astimezone(
        JST()).strftime("%Y/%m/%d/%H0000")

    s3_path_prefix = group_path_component + '/' + datetime_path_component
    client = boto3.client('logs')

    response = client.create_export_task(
        logGroupName=log_group_name,
        fromTime=int(from_datetime.timestamp() * 1000),
        to=int(to_datetime.timestamp() * 1000),
        destination=s3_bucket_name,
        destinationPrefix=s3_path_prefix
    )
    while True:
        res = client.describe_export_tasks(taskId=response['taskId'])
        status = res['exportTasks'][0]['status']['code']
        print(status)
        if status != 'RUNNING' and status != 'PENDING':
            break
        time.sleep(0.5)
    return response['taskId'

参考文献

Boto 3 Documentation