Amazon CloudWatch Logs を Amazon Data Firehose のみ利用して、解凍処理して Amazon S3 に転送する [Amazon Data Firehose + AWS Lambda + Amazon CloudWatch + Amazon S3 + AWS CloudFormation]

こんにちは。SCSKのふくちーぬです。

2/9にて Amazon Kinesis Data Firehose の名称が、Amazon Data Firehose に変更されましたね。もともと取っつきにくい名前だったのに、さらに混乱しそうなところではあります。

Introducing Amazon Data Firehose, formerly known as Amazon Kinesis Data Firehose
aws.amazon.com

今回は、Amazon CloudWatch Logs のログを Amazon Data Firehose を利用して解凍済みのログとして Amazon S3 に転送する方法を紹介します。

Amazon Data Firehose にてログの解凍処理ができるようになりました

以前までは、CloudWatch LogsのログをS3に転送する際にgzip形式のまま転送されていました。運用管理者がログの中身を見るためには、Lambdaにて解凍処理を実施させておくか、運用管理者自身が解凍ツール（7zip等）を使うひと手間が必要でした。

2023/12/15のアップデートにより、Amazon Data Firehose 側で解凍処理を任せることが可能になりました。

Amazon Kinesis Data Firehose が、解凍された CloudWatch Logs の送信先への配信をサポート
aws.amazon.com

Amazon Data Firehose 解凍処理には、0.00403USD/1GBの料金がかかることは注意しましょう。また、解凍後のデータとなるためS3のストレージ料金もわずかながら増大します。やはり大量のログデータを分析する際には、圧縮された状態で（gzip形式等）Athenaで分析する手段を取ることには変わりはないです。しかし、S3でログをすぐに確認する用途で利用できそうなので検証してみます。

検証

CloudWatch Logs のデータを Data Firehose にて、解凍処理を実施した上でS3に転送します。またリソースの構築には、全て CloudFormation を利用します。

アーキテクチャー

  • Lambda実行ログをCloudWatch Logsに保存する
  • CloudWatch Logsにログが入ったことを検知するサブスクリプションフィルターにて、Data Firehose に流す
  • Data Firehose にてログの解凍処理を実施の上、S3に転送する

CloudFormationテンプレート

以下のテンプレートをデプロイしてください。

AWSTemplateFormatVersion: "2010-09-09"
Description: "kinesis stack"
Parameters: 
  ResourceName:
    Type: String 
  BUCKETNAME: 
    Type: "String"

Resources: 
  # ------------------------------------------------------------#
  #  S3
  # ------------------------------------------------------------#   
  S3BucketForlog: 
    Type: "AWS::S3::Bucket"
    DeletionPolicy: "Delete"
    Properties: 
      BucketName: !Sub "${BUCKETNAME}-${AWS::AccountId}"
      AccessControl: "Private"
      PublicAccessBlockConfiguration: 
        BlockPublicAcls: "true"
        BlockPublicPolicy: "true"
        IgnorePublicAcls: "true"
        RestrictPublicBuckets: "true"
  # ------------------------------------------------------------#
  #  Kinesis (Kinesis Firehose,Iam Policy,Iam Role,Log Grop,Log Stream)
  # ------------------------------------------------------------#   
  Deliverystream: 
    DependsOn: 
    - "DeliveryPolicy"
    Type: "AWS::KinesisFirehose::DeliveryStream"
    Properties: 
      DeliveryStreamName: 
        !Sub "${ResourceName}-KDF-LogDeliveryStream"
      ExtendedS3DestinationConfiguration: 
        BucketARN: !Sub "arn:aws:s3:::${S3BucketForlog}"
        BufferingHints: 
          IntervalInSeconds: "60"
          SizeInMBs: "50"
        CompressionFormat: "UNCOMPRESSED"
        Prefix: !Sub "${ResourceName}-logs/"
        RoleARN: !GetAtt "DeliveryRole.Arn"
        DynamicPartitioningConfiguration: 
          Enabled: "false"
        ProcessingConfiguration: 
          Enabled: "true"
          Processors: 
            - Type: Decompression
        CloudWatchLoggingOptions: 
          Enabled: "true"
          LogGroupName: !Sub "/aws/firehose/${ResourceName}-KDF-LOG"
          LogStreamName: "error"
      Tags:
        - Key: Name
          Value: !Sub "${ResourceName}-KDF-LogDeliveryStream"
  DeliveryPolicy: 
    Type: "AWS::IAM::Policy"
    Properties: 
      PolicyName: "firehose_delivery_policy"
      PolicyDocument: 
        Version: "2012-10-17"
        Statement: 
        - Effect: "Allow"
          Action: 
          - "s3:AbortMultipartUpload"
          - "s3:GetBucketLocation"
          - "s3:GetObject"
          - "s3:ListBucket"
          - "s3:ListBucketMultipartUploads"
          - "s3:PutObject"
          Resource: 
          - !Sub "arn:aws:s3:::${S3BucketForlog}"
          - !Sub "arn:aws:s3:::${S3BucketForlog}/*"
        - Effect: "Allow"
          Action: 
          - "logs:PutLogEvents"
          Resource: 
          - !Sub "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/firehose/${ResourceName}-log-DeliveryStream:*"
      PolicyName: "IPL-KinesisFirehoseAccessPolicy"
      Roles: 
       - !Ref "DeliveryRole"
  DeliveryRole: 
    Type: "AWS::IAM::Role"
    Properties: 
      AssumeRolePolicyDocument: 
        Version: "2012-10-17"
        Statement: 
        - Sid: ""
          Effect: "Allow"
          Principal: 
            Service: "firehose.amazonaws.com"
          Action: "sts:AssumeRole"
          Condition: 
            StringEquals: 
              sts:ExternalId: 
                !Ref "AWS::AccountId"
      RoleName: "IRL-KINESISDATAFIREHOSE-LogDeliveryStream"
  LogGroupFirehose: 
    Type: "AWS::Logs::LogGroup"
    Properties: 
      LogGroupName: 
        !Sub "/aws/firehose/${ResourceName}-KDF-LOG"       
  LogStreamFirehose: 
    Type: "AWS::Logs::LogStream"
    Properties: 
      LogGroupName: !Ref "LogGroupFirehose"
      LogStreamName: "error"    
  # ------------------------------------------------------------#
  #  Lambda (Lambda,Iam Policy,Iam Role,Log Grop,Log Stream)
  # ------------------------------------------------------------#         
  HelloWorldFunction: 
    Type: "AWS::Lambda::Function"
    Properties: 
      Code:
        ZipFile: |    
          def lambda_handler(event, context):
              print("Hello,World")
              return {
                  'statusCode': 200,
                  'body': "Hello,World"
              }
      FunctionName: 
        !Sub "${ResourceName}-LMD-HelloWorld"
      Handler: "index.lambda_handler"
      Role: !GetAtt "HelloWorldFunctionRole.Arn"
      Runtime: "python3.12"
      Timeout: "3"
      Architectures: 
      - "x86_64"
  FuncLogGroup: 
    Type: "AWS::Logs::LogGroup"
    Properties: 
      LogGroupName: 
        !Sub "/aws/lambda/${HelloWorldFunction}" 
  SubscriptionFilterLambda: 
    Type: "AWS::Logs::SubscriptionFilter"
    Properties: 
      FilterName: !Sub "${ResourceName}-SFL-HelloWorld"    
      DestinationArn: !GetAtt Deliverystream.Arn
      FilterPattern: " "
      LogGroupName: 
        !Sub "${FuncLogGroup}"
      RoleArn: !GetAtt "LogsRole.Arn"        
  HelloWorldFunctionRole: 
    Type: "AWS::IAM::Role"
    Properties: 
      AssumeRolePolicyDocument: 
        Version: "2012-10-17"
        Statement: 
        -
          Action: 
          - "sts:AssumeRole"
          Effect: "Allow"
          Principal: 
            Service: 
            - "lambda.amazonaws.com"
      ManagedPolicyArns: 
      - "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
      RoleName: "IRL-LAMDA-HelloWorld"      
  LogsRole: 
    Type: "AWS::IAM::Role"
    Properties: 
      AssumeRolePolicyDocument: 
        Version: "2012-10-17"
        Statement: 
        - Effect: "Allow"
          Principal: 
            Service: 
              !Sub "logs.${AWS::Region}.amazonaws.com"
          Action: 
          - "sts:AssumeRole"
      Path: "/"
      Policies: 
      - PolicyName: "root"
        PolicyDocument: 
          Version: "2012-10-17"
          Statement: 
          - Effect: "Allow"
            Action: 
            - "firehose:*"
            Resource: !GetAtt Deliverystream.Arn
      RoleName: "IRL-CLOUDWATCHLOGS-HelloWorld"       

ログの確認

Lambdaにてテストイベントを作成して実行します。CloudWatch Logsにログが出力されていることを確認します。

問題なければFirehoseが動きS3にてログが転送されます。新たなオブジェクトが作成されていますね。

オブジェクトをダウンロードして、ファイルを開いてみます。

中身をみると、解凍されたログになっていますね。

視認性を向上させるために、少し整形します。

Lambdaの実行ログの内容を無事確認することができました。

最後に

いかがだったでしょうか。

Data Firehose の新機能である解凍処理について解説しました。

ただのストリーミング処理だけではなく、今まで Lambda が担っていた処理を Data Firehose が肩代わりしてくれる嬉しいアップデートかと思います。要件に応じて、利用の検討をいただければ幸いです。

本記事が皆様のお役にたてば幸いです。

ではサウナラ～🔥

タイトルとURLをコピーしました