CloudWatch LogsのログをS3へ転送してみた

.
├── bin
│   └── CloudWatchLogsToS3.ts
├── cdk.json
├── lambda
│   └── src
│        ├── hello-world
│        │   └── index.py
│        └── log-export
│            └── index.py
├── lib
│   └── CloudWatchLogsToS3Stack.ts
├── node_modules
...
├── package-lock.json
├── package.json
├── .env
└── tsconfig.json

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as events from 'aws-cdk-lib/aws-events';
import * as targets from 'aws-cdk-lib/aws-events-targets';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as logs from 'aws-cdk-lib/aws-logs';
import { RemovalPolicy, Duration } from 'aws-cdk-lib';

export class CloudWatchLogsToS3Stack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const BUCKET_NAME = 'log-export-bucket-20240825'
    const LOG_GROUP_NAME = '/aws/lambda/sample-function'

    // ログ出力用のLambda関数の作成
    const sampleFunc = new lambda.Function(this, 'SampleFunction', {
      functionName: 'sample-function',
      runtime: lambda.Runtime.PYTHON_3_10,
      code: lambda.Code.fromAsset('../lambda/src/hello-world'),
      handler: 'index.handler',
    });

    // ログ出力用Lambdaのロググループの作成
    const logGroup = new logs.LogGroup(this, 'LogGroup', {
      logGroupName: LOG_GROUP_NAME,
      retention: logs.RetentionDays.ONE_MONTH,
      logGroupClass: logs.LogGroupClass.STANDARD,
      removalPolicy: RemovalPolicy.DESTROY
    });

    // ログエクスポート用のLambda関数の作成
    const logExportFunc = new lambda.Function(this, 'LogExportFunction', {
      functionName: 'log-export-function',
      runtime: lambda.Runtime.PYTHON_3_10,
      code: lambda.Code.fromAsset('../lambda/src/log-export'),
      handler: 'index.handler',
      timeout: Duration.minutes(15),
      environment: {
        BUCKET_NAME: BUCKET_NAME,
        LOG_GROUP_NAME: LOG_GROUP_NAME
      }
    });
    // ログエクスポートのために権限の追加
    logExportFunc.addToRolePolicy(new iam.PolicyStatement({
      actions: [
        'logs:DescribeLogStreams',
        'logs:GetLogEvents',
        'logs:CreateExportTask'
      ],
      resources: [logGroup.logGroupArn]
    }));

    // S3バケットの作成
    const bucket = new s3.Bucket(this, 'LogsBucket', {
      bucketName: BUCKET_NAME,
      versioned: true,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      autoDeleteObjects: true,
    });
    // ログエクスポートのための権限を追加
    bucket.grantWrite(logExportFunc);
    bucket.addToResourcePolicy(new iam.PolicyStatement({
      effect: iam.Effect.ALLOW,
      principals: [new iam.ServicePrincipal('logs.ap-northeast-1.amazonaws.com')],
      actions: [
        's3:GetBucketLocation',
        's3:ListBucket',
        's3:PutObject',
        's3:GetBucketAcl'
      ],
      resources: [
        bucket.bucketArn,
        `${bucket.bucketArn}/*`,
      ]
    }))

    // SampleFunction用EventBridgeルールの作成
    const sampleFuncRule = new events.Rule(this, 'EventBridgeRuleForSampleFunc', {
      ruleName: 'sample-func-rule',
      schedule: events.Schedule.rate(cdk.Duration.hours(1))
    });
    sampleFuncRule.addTarget(new targets.LambdaFunction(sampleFunc))

    // ログエクスポート用EventBridgeルールの作成
    const logExportRule = new events.Rule(this, 'EventBridgeRuleForLogExportFunc', {
      ruleName: 'log-export-rule',
      schedule: events.Schedule.cron({
        minute: '0',
        hour: '15',
        day: '*',
        month: '*',
        // weekDay: '*',
        year: '*'
      })
    });
    logExportRule.addTarget(new targets.LambdaFunction(logExportFunc))
  }
}

import logging
from datetime import datetime, timedelta, timezone

# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# JSTのタイムゾーンを設定
JST = timezone(timedelta(hours=9))

def handler(event, context):
    # 現在の日時をJSTで取得
    current_time_jst = datetime.now(JST).strftime('%Y-%m-%d %H:%M:%S')

    # ログ出力
    logger.info('Hello World!')
    logger.info(f'実行日時 (JST): {current_time_jst}')

    return {
        'statusCode': 200,
        'body': 'Log generated successfully!'
    }

import boto3
import os
from datetime import datetime, timedelta, timezone

# AWS クライアントの作成
cloudwatchlogs = boto3.client('logs', region_name='ap-northeast-1')
s3 = boto3.client('s3', region_name='ap-northeast-1')

# 環境変数からS3バケット名とロググループ名を取得
bucket_name = os.environ['BUCKET_NAME']
log_group_name = os.environ['LOG_GROUP_NAME']

def handler(event, context):
  # エクスポートする時間範囲（前日の00:00から23:59まで）
  from_time, to_time = get_previous_day_time_range()

  # JSTのタイムゾーンを設定
  JST = timezone(timedelta(hours=9))

  # ログにエクスポートするロググループと期間の情報をJSTで出力
  print(f"Starting export task for log group '{log_group_name}' from {datetime.fromtimestamp(from_time / 1000, JST)} to {datetime.fromtimestamp(to_time / 1000, JST)}.")

  try:
    # エクスポートタスクを作成
    create_export_task(log_group_name, from_time, to_time)
    print("Export task created successfully.")
    return {
      'statusCode': 200,
      'body': "Export task created successfully!"
    }

  except Exception as e:
    print(f"An error occurred while creating export task: {e}")
    return {
      'statusCode': 500,
      'body': f'Error exporting logs: {e}'
    }

def get_previous_day_time_range():
  # JSTのタイムゾーンを設定
  JST = timezone(timedelta(hours=9))

  # 現在の時刻を取得し、前日の00:00と23:59:59.999の範囲を設定
  now = datetime.now(JST)
  start_time = datetime(now.year, now.month, now.day, tzinfo=JST) - timedelta(days=1)
  end_time = start_time + timedelta(hours=23, minutes=59, seconds=59, microseconds=999999)

  # タイムスタンプに変換
  from_time = int(start_time.timestamp() * 1000)
  to_time = int(end_time.timestamp() * 1000)

  return from_time, to_time

def create_export_task(log_group_name, from_time, to_time):
  # エクスポートタスク名を設定
  export_task_name = f"export-{log_group_name}-{datetime.now().strftime('%Y-%m-%d')}"

  # '/aws/lambda/'のプレフィックスを削除したロググループ名を使用
  simplified_log_group_name = log_group_name.replace('/aws/lambda/', '')
  
  # 年月日をYYYYMMDD形式で取得
  date_str = datetime.now().strftime('%Y%m%d')
    
  # エクスポートタスクの作成
  response = cloudwatchlogs.create_export_task(
      taskName=export_task_name,
      logGroupName=log_group_name,
      fromTime=from_time,
      to=to_time,
      destination=bucket_name,
      destinationPrefix=f"logs/{simplified_log_group_name}/{date_str}"
  )

  # エクスポートタスク名を返す
  return export_task_name