CloudWatch Logs のログイベントのみ抽出するよう CloudFormation で定義してみる

Data Firehose のログイベント抽出機能を利用した CloudFormation テンプレートを作ってみました
2024.08.06
この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
こんにちは、AWS 事業本部コンサルティング部のたかくに（@takakuni_）です。
今回はタイトルの通り、 CloudWatch Logs のログイベントのみ抽出するよう CloudFormation で定義してみようと思います。
 前置きData Firehose では、デフォルトで CloudWatch Logs から送信されるログイベントに付加情報を加え、 JSON データを 1 行で配信する仕様となっています。
具体的には以下のような形式で配信され、 CloudWatch Logs の中身は logEvents に記録かつ 1 行で S3 バケットへ保管される状況になります。（長いですね...）
配信サンプル
{"messageType":"DATA_MESSAGE","owner":"123456789012","logGroup":"/log/tkkn-123456789012-group","logStream":"tkkn-123456789012-stream","subscriptionFilters":["tkkn-123456789012-filter"],"logEvents":[{"id":"38422339286007929498451156720408396114833289959594065920","timestamp":1722917281192,"message":"{   \"timestamp\": \"2023-05-15T14:37:22.456Z\",   \"level\": \"ERROR\",   \"source\": \"payment_service\",   \"event_type\": \"exception\",   \"message\": \"Failed to process payment\",   \"error_code\": \"PAY_ERR_001\",   \"stack_trace\": \"java.lang.NullPointerException: at com.example.PaymentProcessor.processPayment(PaymentProcessor.java:125)\",   \"user_id\": \"user_789xyz\",   \"transaction_id\": \"tx_987654321\",   \"amount\": 99.99,   \"currency\": \"USD\" }"}]}{"messageType":"DATA_MESSAGE","owner":"123456789012","logGroup":"/log/tkkn-123456789012-group","logStream":"tkkn-123456789012-stream","subscriptionFilters":["tkkn-123456789012-filter"],"logEvents":[{"id":"38422339449182482116099726255845264023101920974913339392","timestamp":1722917288509,"message":"{   \"timestamp\": \"2023-05-15T19:12:03.789Z\",   \"level\": \"WARN\",   \"source\": \"auth_service\",   \"event_type\": \"security_audit\",   \"action\": \"login_attempt\",   \"status\": \"failed\",   \"reason\": \"invalid_credentials\",   \"username\": \"john.doe@example.com\",   \"ip_address\": \"203.0.113.42\",   \"geo_location\": {     \"country\": \"United States\",     \"city\": \"New York\"   },   \"device_type\": \"mobile\",   \"attempt_count\": 3 }"}]}
S3 バケットへ配信されたログを Athena で分析したいケースにおいて、 logEvents（CloudWatch Logs の中身）だけ取り出し、JSON Lines の形式で配信するケースをよく見かけます。
 logEvents だけ抜き取るlogEvents を抜き取る方法はいくつかあり、 Data Firehose の処理の中に変換用の Lambda を組み込むパターンと、ネイティブで logEvents の中身だけ取り出し配信する機能の 2 つがあります。
https://dev.classmethod.jp/articles/cloudwatch-logs-kinesis-origin-fireshose-s3-format/
とくに後者のネイティブ機能は、とても素晴らしいアップデートですね。
https://dev.classmethod.jp/articles/data-firehose-message-extraction-cloudwatch-logs/
また、ネイティブの機能の場合、改行されて格納されるようです。
というわけで、今回はネイティブの機能を利用して、 CloudWatch Logs, Data Firehose, S3 までの一連の流れを Lambda 無しで構築していこうと思います。
 構成図今回作成する構成は以下のとおりです。
至ってシンプルで面白みがないですが、一連の Cfn が周りになかったためブログにしてみます。
全文log.yaml
AWSTemplateFormatVersion: '2010-09-09'
Description: 'CloudWatch Logs to S3 via Data Firehose template'
Parameters:
  Environment:
    Type: String
  LogExpirationInDays:
    Type: Number
    Default: 90
Resources:
  ##############################################################
  # S3 Bucket
  ##############################################################
  LogBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub '${Environment}-log-bucket-${AWS::AccountId}'
      OwnershipControls:
        Rules:
          - ObjectOwnership: BucketOwnerEnforced
      LifecycleConfiguration:
        Rules:
          - Id: !Sub 'ExpirationIn-${LogExpirationInDays}Days'
            ExpirationInDays: !Ref 'LogExpirationInDays'
            Status: Enabled
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      Tags:
        - Key: Name
          Value: !Sub '${Environment}-log-bucket-${AWS::AccountId}'

  ##############################################################
  # Data Firehose
  ##############################################################
  DataFirehoseLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/aws/kinesisfirehose/${Environment}-data-firehose-${AWS::AccountId}'
      RetentionInDays: !Ref LogExpirationInDays
      Tags:
        - Key: Name
          Value: !Sub '/aws/kinesisfirehose/${Environment}-data-firehose-${AWS::AccountId}'

  DataFirehoseLogStream:
    Type: AWS::Logs::LogStream
    Properties:
      LogGroupName: !Ref DataFirehoseLogGroup
      LogStreamName: !Sub '${Environment}-data-firehose-${AWS::AccountId}'

  DataFirehoseRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub '${Environment}-data-firehose-role-${AWS::AccountId}'
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: firehose.amazonaws.com
            Action: sts:AssumeRole
            Condition:
              StringEquals:
                sts:ExternalId: !Ref 'AWS::AccountId'

  DataFirehoseRolePoliy:
    Type: AWS::IAM::RolePolicy
    Properties:
      RoleName: !Ref DataFirehoseRole
      PolicyName: !Sub '${Environment}-data-firehose-policy-${AWS::AccountId}'
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          # S3 バケットへの送信
          - Effect: 'Allow'
            Action:
              - 's3:AbortMultipartUpload'
              - 's3:GetBucketLocation'
              - 's3:GetObject'
              - 's3:ListBucket'
              - 's3:ListBucketMultipartUploads'
              - 's3:PutObject'
            Resource:
              - !Sub 'arn:aws:s3:::${LogBucket}'
              - !Sub 'arn:aws:s3:::${LogBucket}/*'
          # Firehose のログ記録
          - Effect: 'Allow'
            Action:
              - 'logs:PutLogEvents'
            Resource:
              - !GetAtt DataFirehoseLogGroup.Arn

  FirehoseDeliveryStream:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: !Sub '${Environment}-data-firehose-${AWS::AccountId}'
      DeliveryStreamType: 'DirectPut'
      DeliveryStreamEncryptionConfigurationInput:
        KeyType: 'AWS_OWNED_CMK'
      ExtendedS3DestinationConfiguration:
        RoleARN: !GetAtt DataFirehoseRole.Arn
        BucketARN: !Sub 'arn:aws:s3:::${LogBucket}'
        BufferingHints:
          IntervalInSeconds: 300
          SizeInMBs: 5
        CloudWatchLoggingOptions: # Amazon CloudWatch エラーのログ記録
          Enabled: 'true'
          LogGroupName: !Ref DataFirehoseLogGroup
          LogStreamName: !Sub '${Environment}-data-firehose-${AWS::AccountId}'
        CompressionFormat: GZIP # データレコードの圧縮 → S3 へ送信する際の圧縮方式
        ProcessingConfiguration:
          Enabled: true
          Processors:
            - Type: Decompression # Amazon CloudWatch Logs からソースレコードを解凍する
              Parameters:
                - ParameterName: CompressionFormat
                  ParameterValue: GZIP
            - Type: CloudWatchLogProcessing # ログイベントからのみメッセージフィールドを抽出
              Parameters:
                - ParameterName: DataMessageExtraction
                  ParameterValue: true
      Tags:
        - Key: Name
          Value: !Sub '${Environment}-data-firehose-${AWS::AccountId}'

  ##############################################################
  # CloudWatch Logs
  ##############################################################
  CwlRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub '${Environment}-cwl-role-${AWS::AccountId}'
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: !Sub 'logs.${AWS::Region}.amazonaws.com'
            Action: sts:AssumeRole

  CwlRolePolicy:
    Type: AWS::IAM::RolePolicy
    Properties:
      RoleName: !Ref CwlRole
      PolicyName: !Sub '${Environment}-cwl-policy-${AWS::AccountId}'
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Action:
              - 'firehose:PutRecord'
            Resource:
              - !GetAtt FirehoseDeliveryStream.Arn

  LogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub '/log/${Environment}-${AWS::AccountId}-group'
      RetentionInDays: !Ref LogExpirationInDays
      Tags:
        - Key: Name
          Value: !Sub '/log/${Environment}-${AWS::AccountId}-group'

  LogStream:
    Type: AWS::Logs::LogStream
    Properties:
      LogGroupName: !Ref LogGroup
      LogStreamName: !Sub '${Environment}-${AWS::AccountId}-stream'

  SubscriptionFilter:
    Type: AWS::Logs::SubscriptionFilter
    Properties:
      FilterName: !Sub '${Environment}-${AWS::AccountId}-filter'
      DestinationArn: !GetAtt FirehoseDeliveryStream.Arn
      FilterPattern: ''
      LogGroupName: !Ref LogGroup
      RoleArn: !GetAtt CwlRole.Arn
    DependsOn:
      - CwlRolePolicy
 Data Firehose肝となる部分は Data Firehose の部分であるため、ここだけ切り取って解説します。
logEvents のみ抜き取って、ターゲットへ配信するには Amazon CloudWatch Logs からソースレコードを解凍する必要がある点に注意です。
CloudWatch Logs のサブスクリプションフィルターはログを GZIP 圧縮して Data Firehose に転送するため、解凍オプションの CompressionFormat は GZIP を指定しています。
CloudWatch Logs から Amazon Data Firehose に送信されるデータは、すでに gzip レベル 6 圧縮で圧縮されているため、Firehose 配信ストリーム内で圧縮を使用する必要はありません。その後、Firehose の解凍機能を使用して、ログを自動的に解凍できます。詳細については、 CloudWatch 「ログを使用した Kinesis Data Firehose への書き込み」を参照してください。
ロググループレベルのサブスクリプションフィルター - Amazon CloudWatch Logs
yaml
Resources:
  FirehoseDeliveryStream:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: !Sub '${Environment}-data-firehose-${AWS::AccountId}'
      DeliveryStreamType: 'DirectPut'
      DeliveryStreamEncryptionConfigurationInput:
        KeyType: 'AWS_OWNED_CMK'
      ExtendedS3DestinationConfiguration:
        RoleARN: !GetAtt DataFirehoseRole.Arn
        BucketARN: !Sub 'arn:aws:s3:::${LogBucket}'
        BufferingHints:
          IntervalInSeconds: 300
          SizeInMBs: 5
        CloudWatchLoggingOptions:
          Enabled: 'true'
          LogGroupName: !Ref DataFirehoseLogGroup
          LogStreamName: !Sub '${Environment}-data-firehose-${AWS::AccountId}'
        CompressionFormat: GZIP
        ProcessingConfiguration:
          Enabled: true
          Processors:
+            - Type: Decompression # Amazon CloudWatch Logs からソースレコードを解凍する
+              Parameters:
+                - ParameterName: CompressionFormat
+                  ParameterValue: GZIP
+            - Type: CloudWatchLogProcessing # ログイベントからのみメッセージフィールドを抽出
+              Parameters:
+                - ParameterName: DataMessageExtraction
+                  ParameterValue: true
      Tags:
        - Key: Name
          Value: !Sub '${Environment}-data-firehose-${AWS::AccountId}'
その他、設定値に関しては、お好みで変更してお使いください。
 動作確認動作確認のため、次のログを CloudWatch Logs に記録してみます。
LOG_GROUP_NAME=<ロググループ名>
LOG_STREAM_NAME=<ログストリーム名>

cat << EOF > get-html.log
[
  {
    "timestamp": $(date +%s%3N),
    "message": "{\"timestamp\": \"2023-05-15T08:30:45.123Z\", \"ip_address\": \"192.168.1.100\", \"method\": \"GET\", \"url\": \"/index.html\", \"status_code\": 200, \"user_agent\": \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.63 Safari/537.36\", \"response_time\": 0.052}"
  }
]
EOF

aws logs put-log-events --log-group-name $LOG_GROUP_NAME --log-stream-name $LOG_STREAM_NAME --log-events file://get-html.log

cat << EOF > post-api.log
[
  {
    "timestamp": $(date +%s%3N),
    "message": "{\"timestamp\": \"2023-05-15T08:31:02.456Z\", \"ip_address\": \"192.168.1.101\", \"method\": \"POST\", \"url\": \"/api/login\", \"status_code\": 401, \"user_agent\": \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.63 Safari/537.36\", \"response_time\": 0.134}"
  }
]
EOF

cat << EOF > get-logo.log
[
  {
    "timestamp": $(date +%s%3N),
    "message": "{\"timestamp\": \"2023-05-15T08:31:15.789Z\", \"ip_address\": \"192.168.1.100\", \"method\": \"GET\", \"url\": \"/images/logo.png\", \"status_code\": 200, \"user_agent\": \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.63 Safari/537.36\", \"response_time\": 0.008}"
  }
]
EOF

cat get-html.log
cat post-api.log
cat get-logo.log

aws logs put-log-events --log-group-name $LOG_GROUP_NAME --log-stream-name $LOG_STREAM_NAME --log-events file://get-html.log
aws logs put-log-events --log-group-name $LOG_GROUP_NAME --log-stream-name $LOG_STREAM_NAME --log-events file://post-api.log
aws logs put-log-events --log-group-name $LOG_GROUP_NAME --log-stream-name $LOG_STREAM_NAME --log-events file://get-logo.log
とてもシンプルなクエリですが、 Athena で解析してみました。
CREATE EXTERNAL TABLE IF NOT EXISTS web_logs (
  `timestamp` string,
  `ip_address` string,
  `method` string,
  `url` string,
  `status_code` int,
  `user_agent` string,
  `response_time` float
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
WITH SERDEPROPERTIES (
  'serialization.format' = '1'
) LOCATION 's3://<バケット名>/2024/08/06/08/' /* オブジェクトが保管されている部分を指定 */
TBLPROPERTIES ('has_encrypted_data'='false');
SELECT * FROM web_logs LIMIT 10;
無事、 logEvents だけ抜き取られてクエリができてそうです。
 まとめ以上、「CloudWatch Logs のログイベントのみ抽出するよう Cloudformation で定義してみる」でした。
非常にシンプルな内容ですが、 CloudWatch Logs から S3 へ Firehose を経由したいケースは多いのではないでしょうか。
このブログが参考になれば幸いです。 AWS 事業本部コンサルティング部のたかくに（@takakuni_）でした！