Logstashを使ってApacheログを読み込んでみた（２）

# はじめに

- [Logstashを使ってApacheログを読み込んでみた（１）](https://qiita.com/shotakaha/items/baf1ac9de6455c67f234)の続き
- 自分の使うアクセスログを読み込ませてみる
- ``Logstash``に直接読み込ませてテストしてから、``Filebeat``で監視することにする

# まとめ

- ``Logstash``にログを直接読み込ませて、フィルタの調整を行った
- ``Filebeat``を使って、ログを同期したときに、``Elasticsearch``に取り込めるようにした

# 関連記事

- [Logstashを使ってApacheログを読み込んでみた（１）](https://qiita.com/shotakaha/items/baf1ac9de6455c67f234)
- [Logstashを使ってApacheログを読み込んでみた（２）](#)

---

# バックグラウンドで起動

- ``Elasticsearch``と``Kibana``はバックグラウンドで起動しておく

```bash
$ brew services start elasticsearch
==> Successfully started `elasticsearch` (label: homebrew.mxcl.elasticsearch)

$ brew services start kibana
==> Successfully started `kibana` (label: homebrew.mxcl.kibana)
```

# ``Logstash``に直接読み込ませてテストした

- 以下の内容で``import-apache.conf``を作成した（``first-pipeline.conf``をコピーして編集）
  1. ``input``  : 標準入力
  - ``filter`` : Apacheのコンバイン形式、``GeoIP``、+α
  - ``output`` : 標準出力 + ``Elasticsearch``


```bash
$ cp first-pipeline.conf import-apache.conf
$ emacs import-apache.conf
##（編集内容は下記参照）
$ logstash -f import-apache.conf < logstash-tutorial.log
```

```conf:import-apache.confの内容
# The # character at the beginning of a line indicates a comment.
# Use comments to describe your configuration.
input {
#    beats {
#        port => "5044"
#    }
      stdin { }
}
# The filter part of this file is commented out to indicate that
# it is optional.
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}" }
        break_on_match => false
        tag_on_failure => ["_message_parse_failure"]
    }
    geoip {
        source => "clientip"
    }
    date {
        match => [ "timestamp", "dd/MMM/yyyy:HH:mm::ss Z"]
        locale => "en"
        target => "@timestamp"
    }
    useragent {
        source => "agent"
        target => "useragent"
    }
}
output {
    stdout { codec => rubydebug }
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}
```

# フィルタについて

- [Logstash - Filter plugins](https://www.elastic.co/guide/en/logstash/current/filter-plugins.html)


## ``grok``フィルタ

- [Grok filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html)
- ``break_on_match => false`
- ``tag_on_failure`` : ``match``に失敗したときにタグに書き出す内容。デフォルトは``_grokparsefailure``


## ``geiop``フィルタ

- [GeoIP filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-geoip.html)
- ``tag_on_failure`` : デフォルト値　``_geoip_lookup_failure``

```conf:GeoIPフィルタ
    geoip {
        source => "clientip"
    }
```

```json:GeoIPフィルタの結果
...（省略）...
        "clientip" => "86.1.76.62",
        "geoip" => {
              "latitude" => 51.4434,
           "region_name" => "Lambeth",
              "timezone" => "Europe/London",
          "country_name" => "United Kingdom",
                    "ip" => "86.1.76.62",
        "continent_code" => "EU",
         "country_code3" => "GB",
             "city_name" => "Balham",
         "country_code2" => "GB",
              "location" => {
            "lon" => -0.1468,
            "lat" => 51.4434
        },
...（省略）...
```


## ``date``フィルタ

- [Date filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html)
- ``tag_on_failure`` : デフォルト値　``_dateparsefailure``

```conf:dateフィルタ
    date {
        match => [ "timestamp", "dd/MMM/yyyy:HH:mm::ss Z"]
        locale => "en"
        target => "@timestamp"
    }
```
```json:dateフィルタ適用前

"timestamp" => "04/Jan/2015:05:30:37 +0000",  ## dd/MMM/yyyy:HH:mm:ss Z のフォーマット（Mが3つ）
"@timestamp" => 2018-08-23T02:15:58.786Z,     ## データを読み込んだ時刻になっている   
```

```json:dateフィルタ適用後
"timestamp" => "04/Jan/2015:05:30:37 +0000",
"@timestamp" => 2015-01-04T05:30:37.000Z,    ## ログが書き込まれた時刻に書き換わっている
````

 
```json:dateフィルタに失敗した例
...（省略）...
    "tags" => [
        [0] "_dateparsefailure"
    ],
...（省略）...
```


## ``useragent``フィルタ

- [Useragent filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-useragent.html)
- ``source`` : UA情報を含んでいる文字列を指定する。必須。
- ``target`` : 解析したUA情報を出力する先の変数名

```conf:Useragentフィルタ
    useragent {
        source => "agent"
        target => "useragent"
    }
```

```json:Useragentフィルタの出力
## source（これを読み込んで）
    "agent" => "\"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
## target（こういう情報が出てくる）
    "useragent" => {
          "major" => "24",
           "name" => "Iceweasel",
          "minor" => "3",
             "os" => "Linux",
          "patch" => "0",
         "device" => "Other",
          "build" => "",
        "os_name" => "Linux"
    },
```


# 本番ファイルとの比較

- サンプルログ（100行）と、読み込ませたいログ（約41万行）の実行時間を計ってみた
- 行数が多いので、標準出力には出さず、``Elasticsearch``だけに突っ込むことにした
- 4分かからないくらいで読み込めてしまった。はやい！！すごい！！

```bash
$ wc -l logstash-tutorial.log
100 logstash-tutorial.log
$ time logstash -f import-apache.conf < logstash-tutorial.log
38.46 real       102.74 user         3.12 sys

$ wc -l weblogs/ccprod01/access_log
408488 weblogs/ccprod01/access_log
## 行数が多いので、標準出力はなしにして実行
$ time logstash -f import-apache.conf < $MY_APACHE_LOG
133.23 real       216.88 user         4.73 sys
```

# 読み込ませたデータの削除

- サンプルをいろいろ読み込ませてるので、``Elasticsearch``のインデックスが汚れているので一度クリアしたくなってきた
- [Delete index](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-delete-index.html)
- ``Kibana`` → ``Dev Tools`` の ``Console``を使って、上記コマンドを実行する
- インデックス名は``logstash-*``に変更した

```console
DELETE logstash-*
```

# ログの再読込

```bash:行数の確認
$ wc -l weblogs/prod0*/*access_log

  408488 weblogs/prod01/access_log
  414190 weblogs/prod02/access_log
  409775 weblogs/prod03/access_log
  436533 weblogs/prod04/access_log

 1062117 weblogs/prod01/ssl_access_log
 1011087 weblogs/prod02/ssl_access_log
  856647 weblogs/prod03/ssl_access_log
  884293 weblogs/prod04/ssl_access_log
```

```bash:読み込み時間
$ time logstash -f import-apache.conf < weblogs/prod01/access_log
113.15 real       225.74 user         4.73 sys

$ time logstash -f import-apache.conf < weblogs/prod02/access_log
126.99 real       219.36 user         4.59 sys

$ time logstash -f import-apache.conf < weblogs/prod03/access_log
128.44 real       232.02 user         4.87 sys

$ time logstash -f import-apache.conf < weblogs/prod04/access_log
129.59 real       235.75 user         5.04 sys

$ time logstash -f import-apache.conf < weblogs/prod01/ssl_access_log
240.79 real       385.14 user         7.03 sys

$ time logstash -f import-apache.conf < weblogs/prod02/ssl_access_log
228.33 real       362.32 user         6.48 sys

$ time logstash -f import-apache.conf < weblogs/prod03/ssl_access_log
207.41 real       333.99 user         6.19 sys

$ time logstash -f import-apache.conf < weblogs/prod04/ssl_access_log
200.58 real       344.14 user         6.13 sys
```

# ``Filebeat``と組み合わせる

- ``/usr/local/etc/filebeat/filebeat.yml``の入力ファイルの``path``を修正する
- ``import-apache.conf``の``input``を``beat``にする

```conf:filebeat.yml
- type: log

  # Change to true to enable this prospector configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    # - /Users/shotakaha/repos/kuma/es-tutorial/logstash-tutorial.log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod01/access_log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod02/access_log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod03/access_log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod04/access_log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod01/ssl_access_log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod02/ssl_access_log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod03/ssl_access_log
    - /Users/shotakaha/repos/kuma/es-tutorial/prod04/ssl_access_log
```

```conf:import-apache.confを修正
# The # character at the beginning of a line indicates a comment.
# Use comments to describe your configuration.
input {
    beats {
        port => "5044"
    }
}
# The filter part of this file is commented out to indicate that
# it is optional.
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
        match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
        locale => "en"
        target => "@timestamp"
    }
    geoip {
        source => "clientip"
    }
    useragent {
        source => "agent"
        target => "useragent"
    }
}
output {
    #stdout { codec => rubydebug }
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}
```

```bash
$ logstash -f import-apache.conf
$ filebeat -c filebeat.yml -d "publish"
$ tail -f /usr/local/var/log/filebeat/filebeat
## たまにtailして終わるのを確認する
```

- 最初の読み込みが終わったことを確認した
- ``Filebeat``のログを表示（``tail -f filebeat``）しながら、Apacheログを更新（``rsync``で同期）した
- ``Filebeat``のログが追加されるのを確認した
- ``Filebeat``、``Logstash``をバックグラウンド（デーモン？）で起動する方法を調べる