はじめに

私は今年度入社したのですが、数か月、業務でAWSに触れた中で大変だったCloudWatch LogsのS3連携の構築作業をご紹介いたします。(このようなブログの執筆自体、初めてです。)

S3に転送する理由

ログはCloudWatch Logsで保管、管理できるのですが、S3に保管した方がコストを低く抑えることができます。また、転送したログを他のサービスを用いて、各種データの分析等に利用することもできます。

構成

作成するアーキテクチャはこのような図になります。

EventBridgeで指定した時間にLambdaを起動させ、ログをS3に転送するという仕組みです。Lambdaを使用することで任意のタイミングでのログを転送することができます。

作業手順

リソースの作成

まずはCloudFormationを用いず、リソースを構築していきます。(CloudFormationを用いる場合はこれらの作業は不要です。)

IAMロール(ポリシー)

IAMポリシーを以下のように設定します。LambdaにS3バケット、CouudWatch Logsの操作権限を付与しています。こちらをIAMロールにアタッチし、Lambdaに紐づけます。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateExportTask",
                "s3:GetBucketAcl",
                "s3:PutObject"
            ],
            "Resource": "*"
        }
    ]
}

S3パケットポリシー

転送するS3のパケットポリシーを以下のように設定します。
region_name,backet_name,account_idを適宜変更してください。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "logs.region_name.amazonaws.com"
            },
            "Action": "s3:GetBucketAcl",
            "Resource": "arn:aws:s3:::backet_name",
            "Condition": {
                "StringEquals": {
                    "AWS:SourceAccount": "account_id"
                }
            }
        },
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "logs.region_name.amazonaws.com"
            },
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::backet_name/*",
            "Condition": {
                "StringEquals": {
                    "s3:x-amz-acl": "bucket-owner-full-control",
                    "AWS:SourceAccount": "account_id"
                }
            }
        }
    ]
}

Lambdaコード

Lambdaコードは以下の通りです。CreateExportAPIを叩くことで、S3に転送しています。S3バケットには指定したロググループが日ごとに保管されます。
log_group_name,backet_nameは適宜変更してください。

lambda_sample.py

from datetime import datetime,date,time,timedelta
import boto3
import os

def lambda_handler(event, context):

    #当日、前日の日付を取得
    today = datetime.combine(date.today(),time())
    yesterday = datetime.combine(date.today()-timedelta(1),time())

    #Unix timeを設定
    unix_start = datetime(1970,1,1)

    #S3バケットへログを転送
    #時刻をミリ秒にしint型にキャスト
    client = boto3.client('logs')
    response = client.create_export_task(
        logGroupName      = "log_group_name",
        fromTime          = int((yesterday-unix_start).total_seconds() * 1000),
        to                = int((today-unix_start).total_seconds() * 1000),
        destination       = "backet_name", 
        destinationPrefix = yesterday.strftime("%Y-%m-%d") 
    )
    return response

EventBridge

左側のナビゲーションペインで「ルール」を選択します。
下記のように転送したい時間を設定します。ここでは日本時間の午前0時に設定しています。

ターゲットはLambdaの「Invoke」を選択します。作成したLambda関数に紐づけます。

これでリソース構築完了です。

CloudFormationでIaC化する

次にCloudFormationを活用した方法をご紹介いたします。このような作業を何度も繰り返すのは面倒なのでIaC化しました。

テンプレート

まず、テンプレートを作成していきます。
冒頭で指定するロググループ、S3バケットのパラメータを定義しています。

S3のライフサイクルルールを用いてS3ストレージクラスをGlacier Deep Archiveに変更しています。
LambdaはRuntime: pyhthon3.10、タイムアウト: 1分で設定しています。また、コードのアップロードは後で行うため、ここではダミーのzipファイルを指定しています。

テンプレート内のregion_name, Lambda_name, IAM_roles_name, IAM_policy_name, EventBridge_nameは適宜変更してください。

sample.yaml

AWSTemplateFormatVersion: '2010-09-09'

Parameters: 
  LogsGroupsToTransfer: 
    Type: String
    Description: "Name of log group to transfer"
  S3Bucket:
    Type: String
    Description: "Name of the S3 bucket"

Resources:
  S3BucketLogTransfer:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Ref S3Bucket
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      LifecycleConfiguration:
        Rules:
          - Id: !Ref S3Bucket
            Status: Enabled
            Transitions:
              - TransitionInDays: 0
                StorageClass: DEEP_ARCHIVE 
  BucketPolicy:
    Type: AWS::S3::BucketPolicy
    Properties:
      Bucket: !Ref S3BucketLogTransfer
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Sid: GetBucketAclPolicy
            Effect: Allow
            Principal:
              Service: logs.region_name.amazonaws.com
            Action: 
              - s3:GetBucketAcl
            Resource: 
              - !Sub arn:aws:s3:::${S3Bucket}
          - Sid: PutObjectPolicy
            Effect: Allow
            Principal:
              Service: logs.region_name.amazonaws.com
            Action: 
              - s3:PutObject
            Resource:
              - !Sub arn:aws:s3:::${S3Bucket}/*
            Condition:
              StringEquals:
                s3:x-amz-acl: bucket-owner-full-control

  LambdaLogTransferToS3Bucket:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: "Lambda_name"
      Handler: "lambda_function.lambda_handler"
      Code:
        ZipFile: !Sub |
          def lambda_handler(event, context):
            print('Created Lambda') 
      Runtime: "python3.10"
      Role: !GetAtt IAMRoleLambdaLogTransferToS3Bucket.Arn
      Environment:
        Variables:
          LOGS_GROUP_NAME: !Ref LogsGroupsToTransfer
          S3_BUCKET_NAME: !Ref S3BucketLogTransfer
      Timeout: 60

  IAMRoleLambdaLogTransferToS3Bucket:
    Type: AWS::IAM::Role
    Properties:
      RoleName: "IAM_roles_name"
      Description: "Role for IAMroles"
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
      - !Ref CustomIAMPolicyLogTransferToS3Bucket

  CustomIAMPolicyLogTransferToS3Bucket: 
    Type: AWS::IAM::ManagedPolicy
    Properties: 
      ManagedPolicyName: "IAM_policy_name"
      Description: "Customer ManagedPolicy for IAMpolicy"
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
              - logs:CreateExportTask
              - logs:DescribeExportTasks
              - s3:GetBucketAcl
              - s3:PutObject
            Resource: "*"

  EventBridgeRuleLogTransferToS3Bucket:
    Type: AWS::Events::Rule
    Properties: 
      EventBusName: default
      Name: "EventBridge_name"
      ScheduleExpression: cron(00 15 * * ? *)
      State: ENABLED
      Targets: 
        - Arn: !GetAtt LambdaLogTransferToS3Bucket.Arn
          Id: !Ref LambdaLogTransferToS3Bucket

  LambdaPermission:
    Type: AWS::Lambda::Permission
    Properties:
      Action: lambda:InvokeFunction
      FunctionName: !GetAtt LambdaLogTransferToS3Bucket.Arn
      Principal: events.amazonaws.com
      SourceArn: !GetAtt EventBridgeRuleLogTransferToS3Bucket.Arn

スタック

次に、スタックを作成していきます。
テンプレートファイルのアップロードから作成したテンプレートを選びます。

続いて、下記のように任意のロググループ名とS3バケット名を設定します。ロググループはカンマで区切ると複数設定できます。

Lambdaコード

最後に、作成したLambdaに以下のコードをアップロードします。複数のロググループを指定できるようにするため、先ほど出てきたコードを少し変えています。
ここだけは手動で作業するので、S3からコードを参照する形にするとより自動化できます！

lambda_sample.py


from datetime import datetime,date,time,timedelta
import boto3
import os

def lambda_handler(event, context):

    #環境変数を設定
    log_group_name = os.environ['LOGS_GROUP_NAME'].split(',')
    s3_bucket_name = os.environ['S3_BUCKET_NAME']
    
    #当日、前日の日付を取得
    today = datetime.combine(date.today(),time())
    yesterday = datetime.combine(date.today()-timedelta(1),time())
    
    #Unix timeを設定
    unix_start = datetime(1970,1,1)

    #ロググループに保存されているログをS3バケットへエクスポート
    #時刻をミリ秒にしint型にキャスト
    client = boto3.client('logs')
    for item in log_group_name:
        print(item)
        response = client.create_export_task(
            logGroupName      = item,
            fromTime          = int((yesterday-unix_start).total_seconds() * 1000),
            to                = int((today-unix_start).total_seconds() * 1000),
            destination       = s3_bucket_name, 
            destinationPrefix = item + '/%s' % yesterday.strftime("%Y-%m-%d") 
        )
        taskId = (response['taskId'])
        status = 'RUNNING'
        #ログエクスポートが「完了」するまで待機
        while status in ['RUNNING','PENDING']:
            response_desc = client.describe_export_tasks(
                taskId=taskId
            )
            status = response_desc['exportTasks'][0]['status']['code']

苦労したこと

Lambda、S3などの各リソースを構築したあと、CloudFomationでテンプレ化したのですが、ポリシー設定の変更が大変でした。個人的にポリシー設定のエラーはあまり細かくないと思っており、原因分析に時間がかかりました... もっとAWS公式ドキュメントをうまく活用できるようになりたいです。

付録：Lambdaを使わない方式もある！

EventBeidgeSchedulerでCreateExportAPIを叩くだけの実装もできます。Lambdaのコード管理すら不要になります。
ただ、こちらの方式だとプレフィックスに日付を入れるなどのカスタマイズ性は落ちてしまいます。今回はロググループを複数設定することができず断念しました。

明日は @tx_matsu-trさん の「映像伝送業界のこれまでの流れと現在のトレンドについて」です。