Step FunctionsでCloudWatch LogsのロググループをS3へエクスポートする

 1{
 2    "Version": "2012-10-17",
 3    "Statement": [
 4        {
 5            "Sid": "1",
 6            "Effect": "Allow",
 7            "Principal": {
 8                "Service": "logs.${リージョン名}.amazonaws.com"
 9            },
10            "Action": "s3:PutObject",
11            "Resource": "arn:aws:s3:::${バケット名}/*",
12            "Condition": {
13                "StringEquals": {
14                    "s3:x-amz-acl": "bucket-owner-full-control"
15                }
16            }
17        },
18        {
19            "Sid": "2",
20            "Effect": "Allow",
21            "Principal": {
22                "Service": "logs.${リージョン名}.amazonaws.com"
23            },
24            "Action": "s3:GetBucketAcl",
25            "Resource": "arn:aws:s3:::${バケット名}"
26        }
27    ]
28}

 1#!/usr/bin/env python
 2# -*- coding:utf-8 -*-
 3
 4import logging
 5import boto3
 6from datetime import datetime
 7from datetime import timedelta
 8from dateutil.parser import parse
 9import pytz
10import time
11import os
12
13def _is_executing_export_tasks():
14    '''
15    export taskが実行中かどうかチェック
16    '''
17    client = boto3.client('logs')
18    for status in ['PENDING', 'PENDING_CANCEL', 'RUNNING']:
19        response = client.describe_export_tasks(limit = 50, statusCode=status)
20        if 'exportTasks' in response and len(response['exportTasks']) > 0:
21            return True
22    return False
23
24def _get_target_date(event):
25    '''
26    CloudWatch Eventsの(実行日時 - 1)日をエクスポート対象にする
27    '''
28    target = None
29    tokyo_timezone = pytz.timezone('Asia/Tokyo')
30
31    utc_dt = datetime.strptime(event['time'], '%Y-%m-%dT%H:%M:%SZ')
32    tokyo_time = utc_dt.astimezone(tokyo_timezone)
33    target = tokyo_time - timedelta(days=1)
34    t = target.replace(hour=0, minute=0, second=0, microsecond=0)
35
36    target_date = t.strftime('%Y%m%d')
37    from_time = int(t.timestamp() * 1000)
38    to_time = int((t + timedelta(days=1) - timedelta(milliseconds=1)).timestamp() * 1000)
39    return from_time, to_time, target_date
40
41def _get_log_group(next_token):
42    '''
43    ロググループを取得する
44    '''
45    client = boto3.client('logs')
46    if next_token is not None and next_token != '':
47        response = client.describe_log_groups(limit = 50, nextToken = next_token)
48    else:
49        # nextTokenは空文字を受け付けない
50        response = client.describe_log_groups(limit = 50)
51
52    if 'logGroups' in response:
53        yield from response['logGroups']
54    # ロググループが多くて50件(最大)を超えるようなら再帰呼出し
55    if 'nextToken' in response:
56        yield from _get_log_group(next_token = response['nextToken'])
57
58def _is_bucket_object_exists(bucket_name, bucket_dir):
59    client = boto3.client('s3')
60    response = client.list_objects_v2(Bucket = bucket_name, Prefix = (bucket_dir))
61    return 'Contents' in response and len(response['Contents']) > 0
62
63def _export_logs_to_s3(bucket_name, bucket_dir, from_time, to_time, log_group_name):
64    client = boto3.client('logs')
65    response = client.create_export_task(taskName = bucket_dir, logGroupName = log_group_name, fromTime = from_time, to = to_time, destination = bucket_name, destinationPrefix = bucket_dir)
66
67def lambda_handler(event, context):
68    bucket_name = os.environ['BUCKET_NAME']
69    from_time, to_time, target_date = _get_target_date(event=event)
70
71    if _is_executing_export_tasks():
72        return {
73            "status": "running",
74            "time": event['time']
75        }
76
77    for log_group in _get_log_group(next_token=None):
78        bucket_dir = log_group['logGroupName'] + '/' +target_date
79        if log_group['logGroupName'].find('/') == 0:
80            bucket_dir = log_group['logGroupName'][1:]
81
82        if _is_bucket_object_exists(bucket_name = bucket_name, bucket_dir = bucket_dir):
83            continue
84        _export_logs_to_s3(bucket_name = bucket_name, log_group_name = log_group['logGroupName'], from_time= from_time, to_time = to_time, bucket_dir = bucket_dir)
85        return {
86            "status": "running",
87            "time": event['time']
88        }
89
90    return {
91        "status": "completed",
92        "time": event['time']
93    }

 1{
 2  "Comment": "Export Cloudwatch LogStream recursively",
 3  "StartAt": "Export Awslogs to S3",
 4  "TimeoutSeconds": 86400,
 5  "States": {
 6    "Export Awslogs to S3": {
 7      "Type": "Task",
 8      "Resource": "${Lambdaのarn}",
 9      "Next": "Finished exporting?"
10    },
11    "Finished exporting?": {
12      "Type": "Choice",
13      "Choices":[{
14        "Variable": "$.status",
15        "StringEquals": "running",
16        "Next": "Wait a minute"
17      },{
18        "Variable": "$.status",
19        "StringEquals": "completed",
20        "Next": "Success"
21      }]
22    },
23    "Wait a minute": {
24      "Type": "Wait",
25      "Seconds": 5,
26      "Next": "Export Awslogs to S3"
27    },
28    "Success": {
29      "Type": "Succeed"
30    }
31  }
32}

 1{
 2    "Version": "2012-10-17",
 3    "Statement": [
 4        {
 5            "Effect": "Allow",
 6            "Action": [
 7                "states:StartExecution"
 8            ],
 9            "Resource": [
10                "${Step Functionsのarn}"
11            ]
12        }
13    ]
14}