ALBのアクセスログに対してElasticStackのMachine Learningを試してみた


# はじめに

Advent Calendar25日目ということでクリスマスですね！:santa:
でも、いつもと変わらないですね！

ということで、
今回は、異常検知してくれる？ElasticStackのX-Packの中のMachine Learningについて試してみましたので書きたいと思います( ﾟДﾟ)ゞﾋﾞｼｯ[^1]

[^1]: Machine Learningは、X-Packという有償プラグインですが、期間限定で試すことが可能です

# こんな流れで話すよ

1. ログ取込フロー
2. 環境について
3. ALBのログを出力して、LogstashからElasticsearchにストア
4. Machine Learning使うよ！

# ログ取込フロー

AWSでWebサービスを提供している環境で、WebAPサーバの前段にはALBを配置してます。
以下の流れでログを取り込みます。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/152721/bd9cfa0a-cae8-24b2-7604-b51a7f87d627.png)

1. ALBログを指定バケットに出力
2. LogstashがS3からALBのログを収集
3. LogstashがElasticsearchにストア

# 環境について

* ALB
* Amazon Linux AMI 2017.09.1 (HVM)
* Logstash 6.0
* logstash-input-s3
* Elasticsearch 6.0
* Kibana 6.0
* X-Pack 6.0

ElasticStackは、事前にインストールしていることを前提にしてます。
インストールされていない環境は、以下を参考にして頂ければと思います。

* [ElasticStackインストール手順](https://qiita.com/micci184/items/0310abecb558d2fb63a9)
* [Install Elasticsearch](https://www.elastic.co/guide/en/elasticsearch/reference/current/rpm.html)
* [Install Kibana](https://www.elastic.co/guide/en/kibana/current/rpm.html)
* [Install Logstash](https://www.elastic.co/guide/en/logstash/current/installing-logstash.html)
* [Install X-Pack](https://www.elastic.co/guide/en/x-pack/current/installing-xpack.html)

# ALBのログを出力して、LogstashからElasticsearchにストア

ALBのロギングを有効化については以下を参考にしてください。

* [ALBロギング有効化](http://docs.aws.amazon.com/ja_jp/elasticloadbalancing/latest/application/load-balancer-access-logs.html)

ここからは、S3に格納されているALBログをLogstashで取得するための方法について書きます。

## S3 input pluginをインストール

S3からログを取得するには、Logstashのプラグインである"S3 input plugin"をインストールする必要があります。

* [S3 input plugin](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-s3.html)

インストール方法を以下に書きます。

```bash
$ cd /usr/share/logstash
$ bin/logstash-plugin install logstash-input-s3
```

## ALBのログを取り込むための準備

ALBのログは、以下の形式で出力されます。

* [ALBアクセスログ](http://docs.aws.amazon.com/ja_jp/elasticloadbalancing/latest/application/load-balancer-access-logs.html)

```bash
### Log Format
type timestamp elb client:port target:port request_processing_time target_processing_time response_processing_time elb_status_code target_status_code received_bytes sent_bytes "request" "user_agent" ssl_cipher ssl_protocol target_group_arn trace_id

### Sample Log
http 2016-08-10T22:08:42.945958Z app/my-loadbalancer/50dc6c495c0c9188 
192.168.131.39:2817 10.0.0.1:80 0.000 0.001 0.000 200 200 34 366 
"GET http://www.example.com:80/ HTTP/1.1" "curl/7.46.0" - - 
arn:aws:elasticloadbalancing:us-east-2:123456789012:targetgroup/my-targets/73e2d6bc24d8a067
"Root=1-58337262-36d228ad5d99923122bbe354"
```

この出力されたログを取り込むために、Grok Patternを作成したり、型変換をLogstashで施す必要があります。

### フィールド定義

それでは、ログフォーマットからフィールド名と型の定義を行います。
ALBが受け付けてからクライアントに返すまでの処理時間を表すフィールドをfloatにします。
対象は、"request_processing_time"、"target_processing_time"、"response_processing_time"です。

|Log                      |Field                    |Type    |
|-------------------------|-------------------------|--------|
|type                     |type                     |string  |
|timestamp                |date                     |date    |
|elb                      |elb                      |string  |
|client:port              |client_port              |string  |
|target:port              |target_port              |string  |
|request_processing_time  |request_processing_time  |float   |
|target_processing_time   |target_processing_time   |float   |
|response_processing_time |response_processing_time |float   |
|elb_status_code          |elb_status_code          |string  |
|target_status_code       |target_status_code       |string  |
|received_bytes           |received_bytes           |long    |
|sent_bytes               |sent_bytes               |long    |
|request                  |ELB_REQUEST_LINE         |string  |
|user_agent               |user_agent               |string  |
|ssl_cipher               |ssl_cipher               |string  |
|ssl_protocol             |ssl_protocol             |string  |
|target_group_arn         |target_group_arn         |string  |
|trace_id                 |trace_id                 |string  |

### Grok Pattern

ALBのログに正規表現をかけて、構造化するため、Grokする必要があります。
Grokを直接Logstashのconfファイルに書くことも可能ですが、可読性を重視するため、パターンファイルとして外出しします。

以下が、ALBのログをよしなに取り込むためのGrok Patternファイルです。

```bash:alb_patterns
### Application Load Balancing
ALB_ACCESS_LOG %{NOTSPACE:type} %{TIMESTAMP_ISO8601:date} %{NOTSPACE:elb} (?:%{IP:client_ip}:%{INT:client_port}|-) (?:%{IP:backend_ip}:%{INT:backend_port}|-) (:?%{NUMBER:request_processing_time}|-1) (?:%{NUMBER:target_processing_time}|-1) (?:%{NUMBER:response_processing_time}|-1) (?:%{INT:elb_status_code}|-) (?:%{INT:backend_status_code}|-) %{INT:received_bytes} %{INT:sent_bytes} \"%{ELB_REQUEST_LINE}\" \"(?:%{DATA:user_agent}|-)\" (?:%{NOTSPACE:ssl_cipher}|-) (?:%{NOTSPACE:ssl_protocol}|-) %{NOTSPACE:target_group_arn} \"%{NOTSPACE:trace_id}\"
```

### Multiple Pipelines

Logstashのconfファイルを呼び出すためにMultiple Pipelinesを使用します。
Pipeline.ymlの定義は、以下です。

```yaml:pipelines.yml
- pipeline.id: alb
  pipeline.batch.size: 125
  path.config: "/etc/logstash/conf.d/alb.cfg"
  pipeline.workers: 1
```

Pipelineの設定については、以下を参考にして頂ければと思います。

* [絶対的に使った方がいいLogstashのMultiple Pipelinesについて書いてみた](https://qiita.com/micci184/items/24e197a168891f089b3d)


### Logstashのディレクトリ構成

ここまで準備しましたが、以下の様に各ファイルを配置します。

```bash
/etc/logstash
  ├ logstash.yml
  ├ conf.d
  │　└ alb.cfg
  ├ jvm.options
  ├ log4j2.properties
  ├ alb_patterns
  └ pipelines.yml
```

* ALBのalb.cfgをconf.d配下に配置
* ALBのGrok PatternファイルをLogstash配下に配置
* alb.cfgを呼び出すpipeline.ymlをLogstash配下に配置

### ALBのconfファイルを作成するよ

* Input: S3からALBのログを取得する[^2]

[^2]: S3バケットへのアクセスするためのIAM Roleが適用されていることを前提とする

* Filter: 取得したログに対してフィルタをかける
* Output: Elasticsearchにストア

```bash:alb.cfg
input {
  s3 {
    region => "ap-northeast-1"
    bucket => "hoge"
    prefix => "hoge"
    ### S3へのポーリング間隔を指定
    interval => "60"
    sincedb_path => "/var/lib/logstash/sincedb_alb"
  }
}
filter {
  ### 読み込むGrok Patternファイルを"patterns_dir"で指定
  grok {
    patterns_dir => ["/etc/logstash/alb_patterns"]
    match => { "message" => "%{ALB_ACCESS_LOG}" }
  }
  date {
    match => [ "date", "ISO8601" ]
    timezone => "Asia/Tokyo"
    target => "@timestamp"
  }
  ### グローバルIPから国などをマッピングするために指定
  geoip {
    source => "client_ip"
  }
  ### デフォルトの型がstringのため、フィールド定義で定義した型に変換
  mutate {
    convert => [ "request_processing_time", "float" ]
    convert => [ "response_processing_time", "float" ]
    convert => [ "target_processing_time", "float" ]
    convert => [ "received_bytes", "int" ]
    convert => [ "sent_bytes", "int" ]
    ### 不要なフィールドを削除
    remove_field => [ "date", 'message' ]
  }
}
output {
  elasticsearch {
    hosts => [ "localhost:9200" ]
  }
}
```

ここまで整ったらLogstashを起動

```bash
$ initctl start logstash
logstash start/running, process 3121
```

# Machine Learning使うよ！

KibanaでアクセスしてMachine Learingをクリックします。
新しくジョブを作成するので、"Create new job"をクリックします。

以下の画面で"Single metric"をクリックします。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/152721/c038a676-b7a7-b668-5f7d-885b918c73b1.png)

"Aggregation"と"Filed"を指定します。
Filedは、"target_processing_time"にすることで、バックエンドのサーバへリクエストしてから返ってくるまでの時間をみます。
そうすることで、サーバ内の処理遅延が発生しているかを確認できます。
画面の右上から期間を指定します。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/152721/05329a38-c547-5c86-73d2-4943946c9d3b.png)

"Name"と"Description"を入力し、"Create job"をクリックします。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/152721/ff93e2ec-9891-4ee2-89e1-b8995442670c.png)

ジョブが実行され、結果を表示するため"View Results"をクリックします。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/152721/7815cd8b-f5fe-5ff9-2d6f-e1da4470263b.png)

指定期間の中で普段より乖離している部分を赤やオレンジでマークされます。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/152721/72b4db8e-fa51-c8c8-d2f1-48b063f6f5e2.png)

今回は、Single metricでやりましたが、Multi metricを使うことで、
グローバルIPアドレスやUserAgentと合わせて検出することも可能です。


# さいごに

いかがでしたか？ログを取り込んでジョブを作成するだけで簡単に実行できました。
今までは、人の目で判断していたところをMachine Learningを用いることで、普段の傾向から乖離している部分を自動で検出してくれるのは便利ですね。

Advent Calendar最終日ですが、これにておしまいです。
ありがとうございました！