CloudFrontのアクセスログをパーティション射影を使ってコスパ良くAthenaでクエリーする仕組みをCDKで作った

import {Stack} from 'aws-cdk-lib';
import {Rule} from 'aws-cdk-lib/aws-events';
import {LambdaFunction} from 'aws-cdk-lib/aws-events-targets';
import {CfnDatabase, CfnTable} from 'aws-cdk-lib/aws-glue';
import {NodejsFunction} from 'aws-cdk-lib/aws-lambda-nodejs';
import {CfnBucket, IBucket} from 'aws-cdk-lib/aws-s3';
import {Construct} from 'constructs';
import {resolve} from 'path';

export interface AthenaTableForCloudFrontProps {
  /**
   * @default cloudfront
   */
  databaseName?: string;
  /**
   * @default access_logs
   */
  tableName?: string;
  /**
   * オリジナルのCloudFrontアクセスログが出力されるバケット
   */
  srcBucket: IBucket;
  /**
   * パーティション射影可能なオブジェクト名に変更後に出力されるバケット
   * srcBucketと同じバケットは無限ループが発生するため使用不可
   */
  destBucket: IBucket;
  /**
   * @default cloudfront/
   */
  destPrifix?: string;
}

export class AthenaTableForCloudFront extends Construct {
  public readonly cfnDatabase: CfnDatabase;
  public readonly cfnTable: CfnTable;
  public readonly handler: NodejsFunction;
  public readonly rule: Rule;

  constructor(
    scope: Construct,
    id: string,
    props: AthenaTableForCloudFrontProps
  ) {
    super(scope, id);

    /**
     * デフォルトパラメータを設定します。
     */
    const {
      databaseName = 'cloudfront',
      tableName = 'access_logs',
      srcBucket,
      destBucket,
      destPrifix = 'cloudfront/',
    } = props;

    /**
     * srcとdestバケットが同一の場合は無限ループになるためエラーを吐きます。
     * フィルター条件などを設定することで同一バケットを使うこともできますが、そこを頑張るよりもバケットをわけた方が楽かつ確実なのでここではやらないようにします。
     */
    if (srcBucket.bucketArn === destBucket.bucketArn) {
      throw new Error(
        'srcBucketとdestBucketに同じバケットを指定するとLambdaの無限実行が発生します。'
      );
    }

    /**
     * バケットにオブジェクトが配置された時のLambda発火にEventBridgeを使用します。バケットでEventBridgeを使える様にフラグを有効にします。
     */
    const cfnSrcBucket = srcBucket.node.defaultChild as CfnBucket;
    cfnSrcBucket.addPropertyOverride(
      'NotificationConfiguration.EventBridgeConfiguration.EventBridgeEnabled',
      true
    );

    /**
     * Glueデータベースを定義します。
     */
    const cfnDatabase = new CfnDatabase(this, 'Database', {
      catalogId: Stack.of(this).account,
      databaseInput: {name: databaseName},
    });
    this.cfnDatabase = cfnDatabase;

    /**
     * Glueテーブル（Athenaテーブル）を定義します。
     */
    const cfnTable = new CfnTable(this, 'Table', {
      catalogId: Stack.of(this).account,
      databaseName: cfnDatabase.ref,
      tableInput: {
        name: tableName,
        tableType: 'EXTERNAL_TABLE',
        // dateというキー名でクエリー範囲を絞り込めるように構築します。
        partitionKeys: [{name: 'date', type: 'string'}],
        parameters: {
          'skip.header.line.count': '2',
          'projection.enabled': true, // パーティション射影の有効化
          // projection.[key].type: keyはpartitionKeysのnameと一致させる
          'projection.date.type': 'date', // 型の指定
          'projection.date.range': '2021/01/01/00,NOW', // 範囲の指定
          'projection.date.format': 'yyyy/MM/dd/HH', // フォーマットの指定
          'projection.date.interval': 1,
          'projection.date.interval.unit': 'HOURS', // 最小単位が時間なのでHOURS
          'storage.location.template':
            `s3://${destBucket.bucketName}/${destPrifix}` + '${date}',　// 末尾partitionKeysのnameと一致させる
        },
        /**
         * アクセスログのフォーマットに合わせてカラムを定義する。1つ目のカラム名がdateでpartitionKeysと重複しているのでlog_dateに変更している。
         * https://docs.aws.amazon.com/ja_jp/AmazonCloudFront/latest/DeveloperGuide/AccessLogs.html#:~:text=%E3%82%B9%E3%83%9A%E3%83%BC%E3%82%B9-,%E6%A8%99%E6%BA%96%E3%83%AD%E3%82%B0%E3%83%95%E3%82%A1%E3%82%A4%E3%83%AB%E3%83%95%E3%82%A3%E3%83%BC%E3%83%AB%E3%83%89,-%E3%83%87%E3%82%A3%E3%82%B9%E3%83%88%E3%83%AA%E3%83%93%E3%83%A5
         */
        storageDescriptor: {
          columns: [
            {name: 'log_date', type: 'date'},
            {name: 'time', type: 'string'},
            {name: 'x_edge_location', type: 'string'},
            {name: 'sc_bytes', type: 'bigint'},
            {name: 'c_ip', type: 'string'},
            {name: 'cs_method', type: 'string'},
            {name: 'cs_host', type: 'string'},
            {name: 'cs_uri_stem', type: 'string'},
            {name: 'sc_status', type: 'int'},
            {name: 'cs_referer', type: 'string'},
            {name: 'cs_user_agent', type: 'string'},
            {name: 'cs_uri_query', type: 'string'},
            {name: 'cs_cookie', type: 'string'},
            {name: 'x_edge_result_type', type: 'string'},
            {name: 'x_edge_request_id', type: 'string'},
            {name: 'x_host_header', type: 'string'},
            {name: 'cs_protocol', type: 'string'},
            {name: 'cs_bytes', type: 'bigint'},
            {name: 'time_taken', type: 'float'},
            {name: 'x_forwarded_for', type: 'string'},
            {name: 'ssl_protocol', type: 'string'},
            {name: 'ssl_cipher', type: 'string'},
            {name: 'x_edge_response_result_type', type: 'string'},
            {name: 'cs_protocol_version', type: 'string'},
            {name: 'fle_status', type: 'string'},
            {name: 'fle_encrypted_fields', type: 'string'},
            {name: 'c_port', type: 'int'},
            {name: 'time_to_first_byte', type: 'float'},
            {name: 'x_edge_detailed_result_type', type: 'string'},
            {name: 'sc_content_type', type: 'string'},
            {name: 'sc_content_len', type: 'bigint'},
            {name: 'sc_range_start', type: 'bigint'},
            {name: 'sc_range_end', type: 'bigint'},
          ],
          inputFormat: 'org.apache.hadoop.mapred.TextInputFormat',
          outputFormat:
            'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat',
          serdeInfo: {
            serializationLibrary:
              'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe',
            parameters: {
              'field.delim': '\t',
              'serialization.format': '\t',
            },
          },
          location: `s3://${destBucket.bucketName}/${destPrifix}`,
        },
      },
    });
    this.cfnTable = cfnTable;

    /**
     * オリジナルのアクセスログのオブジェクトキーがパーティション射影に対応していないので対応させる。
     */
    const handler = new NodejsFunction(this, 'MoveAccessLog', {
      entry: resolve(__dirname, 'handler/move-access-log.ts'),
      bundling: {
        sourceMap: true,
        minify: true,
      },
      environment: {
        TARGET_BUCKET: destBucket.bucketName,
        TARGET_KEY_PREFIX: destPrifix,
      },
      memorySize: 256,
    });
    this.handler = handler;

    /**
     * バケットへのアクセス権限を与える。
     */
    srcBucket.grantRead(handler);
    srcBucket.grantDelete(handler);
    destBucket.grantWrite(handler);

    /**
     * バケットへオブジェクトが作成された時にLambda関数を実行する。
     */
    const rule = new Rule(this, 'CreatedEvent', {
      eventPattern: {
        source: ['aws.s3'],
        resources: [srcBucket.bucketArn],
        detailType: ['Object Created'],
      },
      targets: [new LambdaFunction(handler)],
    });
    this.rule = rule;
  }
}

// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: MIT-0

// Copyright 2022 Kato Ryo

import {
  S3Client,
  CopyObjectCommand,
  DeleteObjectCommand,
} from '@aws-sdk/client-s3';
import {EventBridgeHandler} from 'aws-lambda';
import {cleanEnv, str} from 'envalid';

const env = cleanEnv(process.env, {
  TARGET_BUCKET: str(),
  // prefix to copy partitioned data to w/o leading but w/ trailing slash
  TARGET_KEY_PREFIX: str(),
});

// regex for filenames by Amazon CloudFront access logs. Groups:
// - 1.	year
// - 2.	month
// - 3.	day
// - 4.	hour
const datePattern = '[^\\d](\\d{4})-(\\d{2})-(\\d{2})-(\\d{2})[^\\d]';
const filenamePattern = '[^/]+$';

const s3 = new S3Client({apiVersion: '2006-03-01'});

export const handler: EventBridgeHandler<
  'Object Created',
  {
    version: string;
    bucket: {
      name: string;
    };
    object: {
      key: string;
      size: number;
      etag: string;
      sequencer: string;
    };
    'request-id': string;
    requester: string; // aws account id
    'source-ip-address': string;
    reason: 'PutObject';
  },
  void
> = async event => {
  const bucket = event.detail.bucket.name;
  const sourceKey = event.detail.object.key;

  const sourceRegex = new RegExp(datePattern, 'g');
  const match = sourceRegex.exec(sourceKey);
  if (match === null) {
    console.log(
      `Object key ${sourceKey} does not look like an access log file, so it will not be moved.`
    );
    return;
  }

  const [, year, month, day, hour] = match;

  const filenameRegex = new RegExp(filenamePattern, 'g');
  const splited = filenameRegex.exec(sourceKey);
  if (splited === null) {
    `Object key ${sourceKey} does not look like an access log file, so it will not be moved.`;
    return;
  }
  const filename = splited[0];

  const targetKey = `${env.TARGET_KEY_PREFIX}${year}/${month}/${day}/${hour}/${filename}`;
  console.log(
    `Copying s3://${bucket}/${sourceKey} to s3://${env.TARGET_BUCKET}/${targetKey}.`
  );

  await s3.send(
    new CopyObjectCommand({
      CopySource: `${bucket}/${sourceKey}`,
      Bucket: env.TARGET_BUCKET,
      Key: targetKey,
    })
  );

  console.log(`Copied. Now deleting ${sourceKey}.`);
  await s3.send(new DeleteObjectCommand({Bucket: bucket, Key: sourceKey}));
  console.log(`Deleted ${sourceKey}.`);
};

import {RemovalPolicy, Stack, StackProps} from 'aws-cdk-lib';
import {Construct} from 'constructs';
import {Bucket} from 'aws-cdk-lib/aws-s3';
import {AthenaTableForCloudFront} from './athena-table-for-cloudfront';

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

    const srcBucket = new Bucket(this, 'SrcBucket', {
      removalPolicy: RemovalPolicy.DESTROY,
    });

    const destBucket = new Bucket(this, 'DestBucket', {
      removalPolicy: RemovalPolicy.DESTROY,
    });

    new AthenaTableForCloudFront(this, 'AthenaCloudFront', {
      srcBucket,
      destBucket,
    });
  }
}

SELECT * FROM "cloudfront"."access_logs"
WHERE date > "2022/03/31";

SELECT * FROM "cloudfront"."access_logs"
WHERE date > "2022/03/31"　AND date < "2022/04/02";