Fluentd + Elasticsearch + Kibana を使って YAMAHA RTX1100  の syslog をそれっぽく可視化する


# 概要

fluentd + elasticsearch + kibanaを用いて、ルーターのパケットフィルタで落としたIPのログなどを可視化してみた。
fluentdのプラグイン(fluent-plugin-geoip)を使ってIPから位置情報を取得できた。

![image](https://qiita-image-store.s3.amazonaws.com/0/6379/4e8b5c38-3360-b1ab-4d04-ccd9ea21174f.png)


![貼り付けた画像_2016_02_25_2_35.jpg](https://qiita-image-store.s3.amazonaws.com/0/6379/c1104955-e053-5e37-3db2-0e128406bb1c.jpeg "貼り付けた画像_2016_02_25_2_35.jpg")



# 設定

## 前提条件

- ルータ（RTX1100）IP: 192.168.1.1
- サーバ
	- OS: Ubuntu 14.04.3 LTS Trusty
	- IP: 192.168.1.100
	- kibana, fluentd, elasticsearchを動かす
	- LANからのアクセスのみ想定故、サーバのFWはガラ空き

## パッケージインストール

公式手順参考

- [fluentd](http://docs.fluentd.org/articles/install-by-deb)
- [elasticsearch](https://www.elastic.co/downloads/elasticsearch)
- [kibana](https://www.elastic.co/downloads/kibana)
	- ただし、OS起動時に自動起動させるようにするには、init.dスクリプトをどっかから拾ってくる必要がある
		- [Kibana Beta init script](https://github.com/pbamba/kibana-beta-daemon)
		- [kibana4_init](https://github.com/akabdog/scripts/blob/master/kibana4_init)


## Apache2リバースプロキシ設定
elasticsearchとkibanaを起動すると、それぞれ5601番と9200番ポートで待ち受けるので、便宜上、適当にリバプロ設定。

```
<Location /kibana>
        Require ip 127 192.168.1
        ProxyPass http://localhost:5601
        ProxyPassReverse http://localhost:5601
</Location>

<Location /elasticsearch>
        Require ip 127 192.168.1
        ProxyPass http://localhost:9200
        ProxyPassReverse http://localhost:9200
</Location>
```

## RTX1100コンフィグ変更

サーバにルータのsyslogをぶっ飛ばす設定を追加

```
#
# SYSLOG configuration
#
syslog local address 192.168.1.1
syslog host 192.168.1.100
syslog info on
syslog notice on
```

## elasticsearch設定

elasticsearchが自動的に文字列を要素解析するらしく、
kibanaで可視化する際に「not analyzedにしないとパフォーマンス悪くなるぜ」とか仰るので、not analyzedなfieldも用意する。
あと、地理情報を入れるフィールド(```geo_location```)の型を```geo_point ```にしておく。


```
$ curl -XPUT 192.168.1.100:9200/rtx1100-*/ -d "`cat elasticsearch_default_template.json`"

# lasticsearch_default_template.json
{
    "template": "rtx1100-*",
    "mappings": {
        "_default_": {
            "dynamic_templates": [
                {
                    "string_template" : {
                        "match" : "*",
                        "mapping": {
                            "type": "string",
                            "fields": {
                                "full": {
                                    "type": "string",
                                    "index": "not_analyzed"
                                }
                            }
                        },
                        "match_mapping_type": "string"
                    }
                }
            ],
            "properties": {
                "@timestamp": { "type": "date", "index": "not_analyzed" },
                "geo_location": {"type" : "geo_point" }
            }
        }
    }
}

```


## fluentd設定
### fluentdプラグインインストール

以下のプラグインを使用。

- [fluent/fluent-plugin-rewrite-tag-filter](https://github.com/fluent/fluent-plugin-rewrite-tag-filter)
	- record内のfieldを用いてtagを書き換えられたりできるフィルター
- [uken/fluent-plugin-elasticsearch](https://github.com/uken/fluent-plugin-elasticsearch)
	- fluentdからelasticsearchに流しこめる
- [y-ken/fluent-plugin-geoip ](https://github.com/y-ken/fluent-plugin-geoip)
	- IPから地理情報を取得できるすごいやつ
- [repeatedly/fluent-plugin-multi-format-parser](https://github.com/repeatedly/fluent-plugin-multi-format-parser)
	- 複数のパーサーを順番に試してくれるパーサー
		- 通常だとパーサーは一つしか指定できない
- [tagomoris/fluent-plugin-parser](https://github.com/tagomoris/fluent-plugin-parser)
	- ```<source>```じゃなくて```<match>```の中でパースできるようにするやつ
- [hiraro/fluent-plugin-with-extra-fields-parser](https://github.com/hiraro/fluent-plugin-with-extra-fields-parser) 
	- パースした内容とは別に、指定した余計なフィールドをrecordにぶっ込めるようにするパーサー
	- いい感じのプラグインがないっぽいので作った


### fluentd設定ファイル 

- RTX1100から流れてきたsyslogを正規表現でマッチさせて分解
	- マッチしなかったそれ以外の形式のログは、分解しない
- ログの種類ごとに(elasticsearchの)indexを分けて格納


```
# 動的フィルタ => index: rtx1100-inspect
- [INSPECT] PP[01][out][105] TCP xxx.xxx.xxx.xxx:xxxxx > yyy.yyy.yyy.yyy:yyyyy (yyyy/MM/dd hh:mm:ss)

# フィルタリジェクト => index: rtx1100-reject
# リジェクト系だけはfluent-plugin-geoipで発信源の地理情報を取得する
- PP[01] Rejected at IN(xxxx) filter: TCP xxx.xxx.xxx.xxx:xxxxx > yyy.yyy.yyy.yyy:yyyyy

# コンソールログイン/ログアウト => index: rtx1100-console
- Login succeeded for TELNET: xxx.xxx.xxx.xxx
- Logout from TELNET: xxx.xxx.xxx.xxx

# VPN接続/切断 => index: rtx1100-tunnel
- [L2TP] TUNNEL[1] connected from xxx.xxx.xxx.xxx
- [L2TP] TUNNEL[1] disconnect tunnel xxxxx complete

# その他 => rtx1100-other
```

```
# /etc/td-agent/td-agent.conf

####
## Source descriptions:
##

## syslog
<source>
  type syslog
  tag raw.rtx1100
  format none
</source>

####
## Output descriptions:
##

<match raw.rtx1100.**>
  type parser
  format multi_format
  key_name message
  remove_prefix raw
  add_prefix parsed
  <pattern>
      format with_extra_fields
      base_format /^\[INSPECT\]\s+(?<target>.+)\[(?<direction>.+)\]\[(?<filter_num>\d+)\]\s+(?<proto>.+)\s+(?<src_ip>.+):(?<src_port>.+)\s+>\s+(?<dest_ip>.+):(?<dest_port>.+)\s+\((?<time>.+)\)$/
      time_format '%Y/%m/%d %H:%M:%S'
      extra_fields { "log_type": "inspect" }
  </pattern>
  <pattern>
      format with_extra_fields
      base_format /^(?<target>.+)\s+Rejected\s+at\s+(?<direction>.+)\((?<filter_num>\d+)\)\s+filter:\s+(?<proto>.+)\s+(?<src_ip>.+):(?<src_port>.+)\s+>\s+(?<dest_ip>.+):(?<dest_port>.+)$/ 
      extra_fields { "log_type": "reject" }
  </pattern>
  <pattern>
      format with_extra_fields
      base_format /^Logout\s+from\s+(?<proto>.+):\s+(?<ip>.+)$/
      extra_fields { "log_type": "console_logout" }
  </pattern>
  <pattern>
      format with_extra_fields
      base_format /^Login\s+succeeded\s+for\s+(?<proto>.+):\s+(?<ip>.+)$/ 
      extra_fields { "log_type": "console_login" }
  </pattern>
  <pattern>
      format with_extra_fields
      base_format /^\[(?<proto>.+)\]\s+(?<tunnel>.+)\s+connected\s+from\s+(?<src_ip>.+)$/ 
      extra_fields { "log_type": "tunnel_connect" }
  </pattern>
  <pattern>
      format with_extra_fields
      base_format /^\[(?<proto>.+)\]\s+(?<tunnel>.+)\s+disconnect\s+tunnel\s+\d+\s+complete$/ 
      extra_fields { "log_type": "tunnel_disconnect" }
  </pattern>  
  <pattern>
      format with_extra_fields
      base_format /^(?<msg>.+)$/
      extra_fields { "log_type": "other" }
  </pattern>  
</match>

<match parsed.rtx1100.**>
  type rewrite_tag_filter
  rewriterule1 log_type   ^inspect$       rtx1100.inspect
  rewriterule2 log_type   ^reject$        temp.rtx1100.reject
  rewriterule3 log_type   ^console_(.+)$  rtx1100.console.$1
  rewriterule4 log_type   ^tunnel_(.+)$   rtx1100.tunnel.$1
  rewriterule5 log_type   ^other$         rtx1100.other
</match>

<match rtx1100.inspect.**>
  type elasticsearch
  logstash_format true
  logstash_prefix rtx1100-inspect
  include_tag_key true
  tag_key @log_name
  hosts localhost:9200
  buffer_type memory
  num_threads 1
  flush_interval 60
  retry_wait 1.0
  retry_limit 17
</match>

<match temp.rtx1100.reject.**>
  type  geoip
  geoip_lookup_key src_ip
  <record>
    geo_location  '{ "lat" : ${latitude["src_ip"]}, "lon" : ${longitude["src_ip"]} }'
    country_code  ${country_code["src_ip"]}
  </record>
  remove_tag_prefix temp.
  skip_adding_null_record  true
  flush_interval 1s
</match>
<match rtx1100.reject.**>
  type elasticsearch
  logstash_format true
  logstash_prefix rtx1100-reject
  include_tag_key true
  tag_key @log_name
  hosts localhost:9200
  buffer_type memory
  num_threads 1
  flush_interval 60
  retry_wait 1.0
  retry_limit 17
</match>

<match rtx1100.console.**>
  type elasticsearch
  logstash_format true
  logstash_prefix rtx1100-console
  include_tag_key true
  tag_key @log_name
  hosts localhost:9200
  buffer_type memory
  num_threads 1
  flush_interval 60
  retry_wait 1.0
  retry_limit 17
</match>

<match rtx1100.tunnel.**>
  type elasticsearch
  logstash_format true
  logstash_prefix rtx1100-tunnel
  include_tag_key true
  tag_key @log_name
  hosts localhost:9200
  buffer_type memory
  num_threads 1
  flush_interval 60
  retry_wait 1.0
  retry_limit 17
</match>

<match rtx1100.other.**>
  type elasticsearch
  logstash_format true
  logstash_prefix rtx1100-other
  include_tag_key true
  tag_key @log_name
  hosts localhost:9200
  buffer_type memory
  num_threads 1
  flush_interval 60
  retry_wait 1.0
  retry_limit 17
</match>

```




# 参考文献

- [今日から始めるfluentd × Elasticsearch × kibana - カジュアルな解析・高速化 - 株式会社エウレカ](http://eure.jp/blog/fluentd_elasticsearch_kibana/)
- [Fluentd、ElasticSearch、Kibana4によるログ分析環境の構築 | hrendoh's memo](http://blog.hrendoh.com/how-to-install-fluentd-elasticsearch-and-kibana4/)
- [Kibana+Elasticsearchで文字列の完全一致と部分一致検索の両方を実現する - Qiita](http://qiita.com/harukasan/items/4ec517d8d96f557367e1)
- [プロダクション環境でElasticsearch+kibana(fluentd)でログ可視化運用をしてみてわかった事 - shnagaiのインフラ備忘録](http://kobitosan.hatenablog.com/entry/2014/12/22/131306)