概要

AWSで最低限設定したほうがいいログと、それをAthenaで閲覧するためのDDLを紹介します

AmazonAthena

https://docs.aws.amazon.com/ja_jp/athena/latest/ug/what-is.html
S3にあるファイルにクエリを実行できる
事前にテーブルを作成しておく必要がある
スキャン量(対象のバイト数)に対して課金
- ファイルをgz圧縮などしておく方がよい
- 列指向ファイルフォーマットに変換しておくのも効果的
- 必要な情報は事前にカラムとして定義しておくと、スキャンする量が格段に減る
- 「スキャン量」なので、LIMITで制限しても課金に対しては意味がない

今回対象のログ

CloudTrail
ALBアクセスログ
CloudFrontウェブディストリビューションアクセスログ
WAF(WebACL)
他にも必要あれば追記します

CloudTrail

https://aws.amazon.com/jp/cloudtrail/
AWS上の操作ログを記録するサービス
CloudTrailは設定済みなのを前提とします
- 設定方法：https://docs.aws.amazon.com/ja_jp/awscloudtrail/latest/userguide/cloudtrail-create-a-trail-using-the-console-first-time.html

テーブルの作成

「マネジメントコンソール」→「CloudTrail」→「イベント履歴」→画面上部の「Amazon Athena で高度なクエリを実行します」をクリック
ログを保存してあるS3バケットを選択後、テーブルの作成でテーブル作成ができます
Athenaのdefaultというデータベース上にできるはずです

クエリサンプル

EC2インスタンスを起動したログを抽出

SELECT
  *
FROM
  "default".table_name
WHERE
  eventName='RunInstances';

特定ユーザーのログを抽出

SELECT
  *
FROM
  "default".table_name
WHERE
  useridentity.username = 'xxxxx';

ALBアクセスログ

テーブルの作成

以下のクエリをAthenaで実行する

ALBアクセスログのテーブル定義

CREATE EXTERNAL TABLE [DB名].[テーブル名](
  request_protocol STRING,
  request_timestamp STRING,
  elb_name STRING,
  request_ip STRING,
  request_port INT,
  backend_ip STRING,
  backend_port INT,
  request_processing_time DOUBLE,
  backend_processing_time DOUBLE,
  client_response_time DOUBLE,
  elb_response_code STRING,
  backend_response_code STRING,
  received_bytes BIGINT,
  sent_bytes BIGINT,
  request_verb STRING,
  url STRING,
  protocol STRING,
  user_agent STRING,
  ssl_cipher STRING,
  ssl_protocol STRING,
  arn STRING,
  extra STRING
)
PARTITIONED BY (yyyy STRING, mm STRING, dd STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
  'serialization.format' = '1',
  'input.regex' = '([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*):([0-9]*) (-1|[.0-9]*) (-1|[.0-9]*) (-1|[.0-9]*) (-|[0-9]*) (-|[0-9]*) ([-0-9]*) ([-0-9]*) \"([^ ]*) ([^ ]*) (- |[^ ]*)\" (\"[^\"]*\") ([^ ]*) ([^ ]*) ([^ ]*) (.*)$'
)
LOCATION 's3://[ログ保管バケット名]/[プレフィックス]/AWSLogs/123456789012/elasticloadbalancing/ap-northeast-1';

request_timestampはStringにしてます
ALBアクセスログはs3://[ログ保管バケット名]/[プレフィックス]/AWSLogs/123456789012/elasticloadbalancing/ap-northeast-1/yyyy/mm/dd/というところに置かれるので、パーティションを設定しています

パーティションの設定

パーティションの設定にはパーティションごとのクエリ発行が必要

パーティションを設定するクエリ

ALTER TABLE [DB名].[テーブル名] ADD IF NOT EXISTS
PARTITION (yyyy='2018', mm='12', dd='08')
LOCATION 's3://[ログ保管バケット名]/[プレフィックス]/AWSLogs/123456789012/elasticloadbalancing/ap-northeast-1/2018/12/08';

これでyyyy='2018', mm='12', dd='08'にパーティションが当てられる

パーティションを設定するクエリ(同時に複数)

ALTER TABLE [DB名].[テーブル名] ADD IF NOT EXISTS
PARTITION (yyyy='2018', mm='12', dd='08')
LOCATION 's3://[ログ保管バケット名]/[プレフィックス]/AWSLogs/123456789012/elasticloadbalancing/ap-northeast-1/2018/12/08'
ALTER TABLE [DB名].[テーブル名] ADD IF NOT EXISTS
PARTITION (yyyy='2018', mm='12', dd='09')
LOCATION 's3://[ログ保管バケット名]/[プレフィックス]/AWSLogs/123456789012/elasticloadbalancing/ap-northeast-1/2018/12/09'
ALTER TABLE [DB名].[テーブル名] ADD IF NOT EXISTS
PARTITION (yyyy='2018', mm='12', dd='10')
LOCATION 's3://[ログ保管バケット名]/[プレフィックス]/AWSLogs/123456789012/elasticloadbalancing/ap-northeast-1/2018/12/10'
;

これで同時に3つ分パーティションが貼られる
AWSCLIを用いてシェルなどでループさせるのが良い

クエリサンプル

ロードバランサもしくはターゲットグループが5xxエラーを出したリクエストを抽出

SELECT
  request_timestamp,
  url
FROM
  [DB名].[テーブル名]
WHERE
  yyyy='2018'
  and mm='12'
  and dd='08'
  and ( elb_response_code like '5%' or backend_response_code like '5%' )
ORDER BY
  request_timestamp
;

CloudFrontウェブディストリビューションアクセスログ

アクセスログの設定が終わっていることを前提とします
- 設定方法：https://docs.aws.amazon.com/ja_jp/AmazonCloudFront/latest/DeveloperGuide/AccessLogs.html

テーブルの作成

以下のクエリをAthenaで実行する
各フィールド詳細は https://docs.aws.amazon.com/ja_jp/AmazonCloudFront/latest/DeveloperGuide/AccessLogs.html#BasicDistributionFileFormat を参照

CloudFrontアクセスログのテーブル定義

CREATE EXTERNAL TABLE IF NOT EXISTS [DB名].[テーブル名] (
  request_date STRING,
  request_time STRING,
  x_edge_location STRING,
  sc_bytes INT,
  client_ip STRING,
  cs_method STRING,
  cs_host STRING,
  cs_uri_stem STRING,
  sc_status STRING,
  cs_referer STRING,
  user_agent STRING,
  uri_query STRING,
  cookie STRING,
  x_edge_result_type STRING,
  x_edge_request_id STRING,
  x_host_header STRING,
  cs_protocol STRING,
  cs_bytes INT,
  time_taken DECIMAL(8,3),
  x_forwarded_for STRING,
  ssl_protocol STRING,
  ssl_cipher STRING,
  x_edge_response_result_type STRING,
  cs_protocol_version STRING,
  fle-status STRING,
  fle-encrypted-fields STRING
)
PARTITIONED BY (yyyy STRING, mm STRING, dd STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES (
    'serialization.format' = '\t',
    'input.regex' = '\t'
)
LOCATION 's3://[ログ保管バケット名]/[プレフィックス]'
TBLPROPERTIES (
  'skip.header.line.count'='2'
)
;

skip.header.line.countは各ファイルの先頭行を飛ばすような設定
- ログファイルにカラム名が出力されるので、これを設定しないと不要レコードがAthenaDBに入ってしまう

パーティションの設定

パーティションの設定にはパーティションごとのクエリ発行が必要
ALBと同じなので省略

クエリサンプル

キャッシュからレスポンスを返したリクエスト

SELECT
  *
FROM
  [DB名].[テーブル名]
WHERE
  yyyy='2018'
  and mm='12'
  and dd='08'
  and x_edge_result_type='Hit'
ORDER BY
  request_data,
  request_time
;

WAF(WebACL)

AWS WAFのログ
設定方法：https://docs.aws.amazon.com/ja_jp/waf/latest/developerguide/logging.html

テーブルの作成

WAFログのテーブル定義

CREATE EXTERNAL TABLE IF NOT EXISTS [DB名].[テーブル名](
  `timestamp` string,
  formatVersion string,
  webaclId string,
  terminatingRuleId string,
  terminatingRuleType string,
  action string,
  httpSourceName string,
  httpSourceId string,
  ruleGroupList array < struct <
      ruleGroupId:string,
      terminatingRule:string,
      nonTerminatingMatchingRules:array < struct <
          action:string,
          ruleId:string
      > >
    > >,
  rateBasedRuleList array < struct <
      rateBasedRuleId:string,
      limitKey:string,
      maxRateAllowed:string
  > >,
  nonTerminatingMatchingRules array < struct <
      action:string,
      ruleId:string
    > >,
  httpRequest struct <
      clientIp:string,
      country:string,
      HTTPVersion:string,
      headers:array < struct <
          name:string,
          value:string
      > >,
      uri:string,
      args:string,
      httpVersion:string,
      httpMethod:string,
      requestId:string
  >
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://[ログ保管バケット名]/[プレフィックス]'
;

WAFのログはKinesisFirehoseからの出力となります
バケット/プレフィックス直下にファイルが出力されます
- KinesisFirehoseから出力されるバケットにLambdaを設定することでyyyy/mm/ddというパーティションをつけることもできます
- パーティションを設定しないとファイル数が膨大になり、スキャン量が増え、課金されまくるのでご注意を

サンプルクエリ

Googleのクローラーのアクセスを確認する

SELECT
  `timestamp`,
  action
FROM
  [DB名].[テーブル名]
CROSS JOIN UNNEST(httprequest.headers) AS t (header)
WHERE
  header.value like '%APIs-Google%'
  and action
;

arrayとなっているところをWHERE句で検索する場合、CROSS JOINさせる必要があります

最後に

とりあえずアクセスログをAthenaで見れる状態にしておいて、なにかあった場合にすぐ調べられるようにしておいたほうがいいかと思います
S3 Selectというサービスもありますが、1ファイルしか抽出できないので、多くのファイルが出力される場合はAthenaを使うのがいいです
Apacheのアクセスログも設定できるので、ログ分析基盤の構築なども可能かと思います
- RedShiftなどで構築するよりも安価に構築することができます
ただ課金にはお気をつけを・・・