Athenaを使ってAWSのログを眺める

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

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

## 今回対象のログ
1. CloudTrail
2. ALBアクセスログ
3. CloudFrontウェブディストリビューションアクセスログ
4. 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というデータベース上にできるはずです

#### クエリサンプル
```ruby:EC2インスタンスを起動したログを抽出
SELECT
  *
FROM
  "default".table_name
WHERE
  eventName='RunInstances';
```

```ruby:特定ユーザーのログを抽出
SELECT
  *
FROM
  "default".table_name
WHERE
  useridentity.username = 'xxxxx';
```

### ALBアクセスログ
* ApplicationLoadBalancerのアクセスログ
* アクセスログの設定が終わっていることを前提とします
    * 設定方法：https://docs.aws.amazon.com/ja_jp/elasticloadbalancing/latest/application/load-balancer-access-logs.html

#### テーブルの作成
以下のクエリをAthenaで実行する

```ruby: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/というところに置かれるので、パーティションを設定しています

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

```ruby:パーティションを設定するクエリ
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'にパーティションが当てられる

```ruby:パーティションを設定するクエリ(同時に複数)
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を用いてシェルなどでループさせるのが良い


#### クエリサンプル
```ruby:ロードバランサもしくはターゲットグループが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 を参照

```ruby: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と同じなので省略

#### クエリサンプル
```ruby:キャッシュからレスポンスを返したリクエスト
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

#### テーブルの作成
```ruby: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というパーティションをつけることもできます
    * パーティションを設定しないとファイル数が膨大になり、スキャン量が増え、課金されまくるのでご注意を

#### サンプルクエリ
```ruby: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などで構築するよりも安価に構築することができます
* ただ課金にはお気をつけを・・・