효율적인 AWS CloudTrail 검색을 위한 데이터 파이프라인 구성

AWS CloudTrail은 사용자, 역할 또는 AWS 서비스가 수행하는 작업을 이벤트로 기록하는 서비스입니다. 이벤트에는 AWS Management Console, AWS Command Line Interface 및 AWS SDK, API에서 수행되는 작업이 포함됩니다. 이벤트는 Amazon Simple Storage Service(S3)에 JSON 형식의 압축 파일로 기록됩니다. 이 파일을 직접 다운받아 조회하거나 전체 포맷을 변경하지 않고 검색하는것은 매우 어려운 일 입니다.

서버리스 데이터 통합 서비스인 AWS Glue를 사용하면 JSON 포맷의 AWS CloudTrail의 Log를 보다 직관적으로 포맷을 변경 할 수 있으며, 서버리스 대화형 분석 서비스인 Amazon Athena를 사용하여 Amazon S3에 저장된 이벤트를 조회할 수 있습니다.

이번 게시물을 통해 데이터 파이프라인을 구성하여 Amazon Athena에서 이벤트 조회 시 데이터 보정, 성능 개선 및 비용 절감하는 방법을 소개하고자 합니다.

데이터 파이프라인 필요성

데이터 보정

AWS CloudTrail 이벤트는 발생된 시간(eventtime)과 Amazon S3 Bucket에 저장되는 시간이 상이하기 때문에 Amazon Athena에서 금일 조회 시 전날 데이터가 포함됩니다. 이에 데이터 파이프라인에서 데이터를 보정하는 로직을 구현하고자 합니다.

[Amazon Athena에서 이벤트 조회 SQL]

SELECT YEAR, MONTH, DAY, MIN(EVENTTIME) as MIN_EVENTTIME
  FROM "cloudtrail_json" 
 GROUP BY YEAR, MONTH, DAY
 ORDER BY YEAR DESC, MONTH DESC, DAY DESC
 LIMIT 5;

[조회 결과]

포맷 변경

데이터 파이프라인에서 Amazon S3에 JSON 형식으로 저장된 이벤트를 Parquet 형식으로 변경합니다. Amazon Athena는 데이터 스캔된 데이터량 1TB당 $5의 요금체계를 가지고 있으며, Parquet 형식 사용 시 스캔된 데이터량을 최소화 할 수 있기 때문에 비용을 절감하고 조회 성능을 개선할 수 있습니다.

참고로 Apache Parquet 형식은 효율적인 데이터 스토리지와 검색을 지원하는 컬럼 중심의 오픈소스 데이터 파일 형식으로 효율적인 데이터 압축 및 인코딩 방식을 제공합니다.

데이터 파이프라인 개요

Amazon S3에 JSON 형식으로 수집된 AWS CloudTrail 이벤트를 Amazon Athena에서 CTAS(Create Table As Select) SQL로 Parquet 형식의 테이블을 생성하고 주기적으로 실행되는 AWS Lambda함수에서 JSON 형식의 데이터를 Parquet 형식의 테이블에 데이터를 입력하는 파이프라인을 생성합니다.

본 예제는 AWS Lambda 함수를 매일 1시에 실행되도록 구성했기 때문에 데이터 생성 시점과 조회 시점 사이에 차이가 발생할 수 있습니다, 적용할 워크로드와 상이한 경우 실행 주기와 로직을 수정해서 사용하면 됩니다.

JSON형식의 테이블은 매일 01시 이후 Amazon Athena에서 조회 가능
Parquet 형식의 테이블은 다음날 01시 이후 Amazon Athena에서 조회 가능

[아키텍처]

[흐름도]

AWS CloudTrail에서 이벤트 생성
Amazon Athena에서 AWS CloudTrail 이벤트를 위한 JSON과Parquet 형식의 테이블 생성
Amazon EventBridge에서 매일 01시에 AWS Lambda 함수를 호출
AWS Lambda 함수에서 JSON 형식 테이블에 파티션 추가 및 Parquet 형식의 테이블에 데이터 추가

데이터 파이프라인 구성하기

단계1: AWS CloudTrail 설정

AWS CloudTrail콘솔에서 “Create a trail”을 선택 후 아래와 같이 필드 값을 입력하여 Trail을 생성합니다.

Amazon S3 콘솔에서 다음과 같이 AWS CloudTrail 이벤트가 저장된 것을 확인할 수 있습니다.

Amazon S3 URI 및 파일명은 다음과 같이 구성됩니다.

s3://{S3 버킷명}/AWSLogs/{Account ID}/CloudTrail/{리전명}/{YYYY}/{MM}/{DD}/{Account ID}_CloudTrail_{리전 명}_{YYYY}{MM}{DD}T{HH}{MM}Z_{임의의 문자열}.json.gz

단계2: AWS CloudTrail 이벤트용 테이블 정의

Amazon Athena에서 Amazon S3에 저장된 AWS CloudTrail 이벤트를 조회하기 위해서는 사전에 JSON 형식과 Parquet 형식의 테이블을 생성해야 합니다. 생성할 테이블들은 년, 월, 일 파티션을 가지고 있으며, 파티션 프루닝(Partition Pruning)시 사용하게 됩니다.

JSON 형식의 테이블 생성

Amazon Athena 콘솔의 Query editor 에서 아래의 SQL를 입력하여 cloudtrail_json 테이블을 생성합니다. SQL내의 location 파라미터 값은 이벤트가 저장된 Amazon S3 URI를 입력해야 합니다.

CREATE EXTERNAL TABLE cloudtrail_json (
eventversion STRING,
useridentity STRUCT<
                type:STRING,
                principalid:STRING,
                arn:STRING,
                accountid:STRING,
                invokedby:STRING,
                accesskeyid:STRING,
                userName:STRING,
                sessioncontext:STRUCT<
                                  attributes:STRUCT<
                                     mfaauthenticated:STRING,
                                     creationdate:STRING
                                  >,
                                  sessionissuer:STRUCT<  
                                     type:STRING,
                                     principalId:STRING,
                                     arn:STRING, 
                                     accountId:STRING,
                                     userName:STRING
                                  >,
                                  ec2RoleDelivery:string,
                                  webIdFederationData:map<string,string>
                >
>,
eventtime STRING,
eventsource STRING,
eventname STRING,
awsregion STRING,
sourceipaddress STRING,
useragent STRING,
errorcode STRING,
errormessage STRING,
requestparameters STRING,
responseelements STRING,
additionaleventdata STRING,
requestid STRING,
eventid STRING,
resources ARRAY<STRUCT<
                   arn:STRING,
                   accountid:STRING,
                   type:STRING
          >>,
eventtype STRING,
apiversion STRING,
readonly STRING,
recipientaccountid STRING,
serviceeventdetails STRING,
sharedeventid STRING,
vpcendpointid STRING,
tlsDetails STRUCT<
              tlsVersion:string,
              cipherSuite:string,
              clientProvidedHostHeader:string>
)
PARTITIONED BY (year string, month string, day string)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS INPUTFORMAT 'com.amazon.emr.cloudtrail.CloudTrailInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://aws-svc-logs-41007389****/AWSLogs/41007389****/CloudTrail/ap-northeast-2/';

Amazon Athena에서 생성한 테이블의 스키마 정보는 AWS Glue의 data catalog에 저장됩니다.

cloudtrail_json 테이블은 year, month, day 파티션 컬럼을 가지고 있습니다. 아래와 같이 명시적으로 파티션을 추가해야 Amazon S3에 저장된 파일을 조회할 수 있습니다.

ALTER TABLE cloudtrail_json ADD 
  PARTITION (year='2023', month='07', day='22')
  LOCATION 's3://aws-svc-logs-41007389****/AWSLogs/41007389****/CloudTrail/ap-northeast-2/2023/02/20/';

예제 AWS Lambda 함수에는 cloudtrail_json 테이블의 파티션을 생성하는 로직이 포함되어 있습니다.

Parquet 형식의 테이블 생성

Query editor 페이지에서 cloudtrail_parquet 테이블을 생성합니다. SQL내의 external_location 파라미터에는 Parquet 파일이 생성될 Amazon S3 URI를 입력해야 합니다.

CREATE TABLE cloudtrail_parquet
WITH (format = 'Parquet',
      partitioned_by = ARRAY['year', 'month', 'day'],
      external_location ='s3://aws-svc-logs-41007389****/AWSLogs-Curated/CloudTrail-Parquet/')
AS
SELECT *
FROM cloudtrail_json
WITH NO DATA;

단계3: AWS Lambda 함수 생성

AWS Lambda 콘솔에서 다음과 같이 함수를 생성합니다.

생성한 AWS Lambda 함수의 Code에 다음의 소스를 복사하고 실행환경에 맞게 수정 후 배포를 해야 합니다.

import json
import boto3
import datetime
from datetime import timedelta

DATABASE = 'aws_logs_db'
OUTPUT = "s3://aws-athena-query-results-ap-northeast-2-41007389****"

def lambda_handler(event, context):
    # TODO implement
    client = boto3.client('athena')

    now = datetime.datetime.now()
    yyyy = now.strftime('%Y')
    mm = now.strftime('%m')
    dd = now.strftime('%d')

    before_now = now + timedelta(days=-1)
    before_yyyy = before_now.strftime('%Y')
    before_mm = before_now.strftime('%m')
    before_dd = before_now.strftime('%d')

    # 1. Create cloudtrail_json partition
    cloudtrail_json_ddl = (
        f"ALTER TABLE cloudtrail_json ADD "
        f"PARTITION (year='{yyyy}', month='{mm}', day='{dd}') "
        f"LOCATION 's3://aws-svc-logs-41007389****/AWSLogs/41007389****/CloudTrail/ap-northeast-2/{yyyy}/{mm}/{dd}/'; "
    )
    print(cloudtrail_json_ddl)

    response = client.start_query_execution(
        QueryString = cloudtrail_json_ddl,
        QueryExecutionContext= {
            'Database': DATABASE
        },
        ResultConfiguration={
            'OutputLocation': OUTPUT,
        }
    )


    # 2. Append data to cloudtrail_parquet table
    cloudtrail_parquet_dml = (
        f"INSERT INTO cloudtrail_parquet "
        f"SELECT eventversion       , useridentity    , eventtime          , eventsource, eventname         , "
        f"       awsregion          , sourceipaddress , useragent          , errorcode  , errormessage      , "
        f"       requestparameters  , responseelements, additionaleventdata, requestid  , eventid           , "
        f"       resources          , eventtype       , apiversion         , readonly   , recipientaccountid, "
        f"       serviceeventdetails, sharedeventid   , vpcendpointid      , tlsdetails ,  "
        f"       '{before_yyyy}'    , '{before_mm}'   , '{before_dd}' "
        f" FROM cloudtrail_json "
        f"WHERE year = '{before_yyyy}' AND month = '{before_mm}' AND day = '{before_dd}' "
        f"  AND eventtime >= '{before_yyyy}-{before_mm}-{before_dd}T00:00:00:00Z' "
        f"UNION ALL "
        f"SELECT eventversion       , useridentity    , eventtime          , eventsource, eventname         , "
        f"       awsregion          , sourceipaddress , useragent          , errorcode  , errormessage      , "
        f"       requestparameters  , responseelements, additionaleventdata, requestid  , eventid           , "
        f"       resources          , eventtype       , apiversion         , readonly   , recipientaccountid, "
        f"       serviceeventdetails, sharedeventid   , vpcendpointid      , tlsdetails ,  "
        f"       '{before_yyyy}', '{before_mm}', '{before_dd}' "
        f" FROM cloudtrail_json "
        f"WHERE year = '{yyyy}' AND month = '{mm}' AND day = '{dd}'  "
        f"  AND eventtime <= '{before_yyyy}-{before_mm}-{before_dd}T23:59:59:00Z'; "
    )
    print(cloudtrail_parquet_dml)

    response = client.start_query_execution(
        QueryString = cloudtrail_parquet_dml,
        QueryExecutionContext= {
            'Database': DATABASE
        },
        ResultConfiguration={
            'OutputLocation': OUTPUT,
        }
    )

    return response

예제 소스에서 수정할 부분은 다음과 같습니다.

DATABASE 변수의 값은 Glue data catalog의 DATABASE 명입니다. 본 예제에서는 default를 사용하지 않고 aws_logs_db를 사용하였습니다.
OUTPUT 변수의 값은 Athena에서 쿼리 결과를 저장하기 위해 사용하는 Amazon S3 URI입니다. Athena 콘솔의 Query editor > Settings에서 설정된 값을 조회할 수 있습니다.
cloudtrail_json_ddl 변수의 값에서 LOCATION은 cloudtrail_json 테이블이 참조할 Amazon S3 URI를 입력해야 합니다.

AWS Lambda 함수를 배포한 후에 다음과 같은 추가 작업을 진행합니다.

타임아웃 변경 (Lambda > Configuration > Gerneral configuration > Edit > Timeout)
- 15분 (최대 값)
퍼미션 추가 (Lambda > Configuration > Permissions > {Role name} > Add permissions
- AmazonS3FullAccess, AmazonAthenaFullAccess, CloudWatchLogsFullAccess
trigger 설정 (Lambda > Add trigger)
- 매 01시에 실행하도록 Schedule expression 필드에 “cron(0 1 * * ? *)”를 입력
- Local timezone설정 (UTC 기준)
  - Local timezone는 ‘EventBridge > Buses > Rules > {Rule Name} > Event schedule > Edit’에서 변경

배포한 AWS Lambda 함수가 정상적으로 실행되면 Amazon S3에 Parquet 형식의 AWS CloudTrail 이벤트가 저장된 파일과 Amazon Athena에서 cloudtrail_parquet 테이블에 파티션이 추가된 것을 확인 할 수 있습니다.

AWS CloudTrail JSON vs Parquet 성능 비교

AWS CloutTrail 이벤트는 JSON과 Parquet 형식의 테이블에 다음과 같이 저장되어 있습니다.

AWS CloudTrail 이벤트 저장 기간: 23/5/29 ~ 23/07/23
AWS CloudTrail 이벤트 수: 12,321,985
테이블 특성

테이블 명	저장 형식	Total Objects	Total Size	Partition key	7/20 파일 갯수
cloudtrail_json	JSON	43,142	3.0 GiB	yyyy, mm, dd	1,348
cloudtrail_parquet	Parquet	1,703	2.4 GiB	yyyy, mm, dd	25

성능 테스트를 수행한 SQL 및 결과는 다음과 같습니다.

성능 테스트 SQL

SQL 1
- ```
select count(*) from cloudtrail_json
```

SQL 2

select year, month, day, count(*) 
  from cloudtrail_json
 group by year, month, day
 order by 1, 2, 3

SQL 3

select useridentity 
  from cloudtrail_json
 where year = '2023' and month = '07' and day = '20' 
   and eventsource = 'dynamodb.amazonaws.com'
   and eventname = 'DescribeStream';

SQL 4

select useridentity.arn as user, eventname 
  from cloudtrail_json
 where year = '2023' and month = '07' 
   and useridentity.arn is not null 
   and eventname = 'CreateBucket'

성능 테스트 결과

Table	SQL1		SQL2		SQL3		SQL4
Table	Rune time	Data scanned	Rune time	Data scanned	Rune time	Data scanned	Rune time	Data scanned
cloudtrail_json	12.1	2.99 GB	10.8	2.99 GB	3	44 MB	8.5	1.25 GB
cloudtrail_parquet	1.1	–	1.8	–	0.9	1.5 MB	1.1	540 KB

결론

AWS CloudTrail 이벤트를 JSON에서 Parquet 형식으로 변환하는 과정과 데이터 보정하는 방법을 살펴보았습니다. 또한 Amazon Athena에서 JSON 형식과 Parquet 형식의 테이블에 대해 성능 비교를 진행하여, Parquet 형식의 테이블이 적은 데이터를 스캔함으로써 조회 속도가 빠르고 비용을 절감함을 확인하였습니다.

AWS 기술 블로그

효율적인 AWS CloudTrail 검색을 위한 데이터 파이프라인 구성

데이터 파이프라인 필요성

데이터 파이프라인 개요

데이터 파이프라인 구성하기

AWS CloudTrail JSON vs Parquet 성능 비교

결론

주요 링크 모음

팔로우하기