Libertyのログをjson形式で出力してFilebeatでELKに送る

[LibertyからLogstash CollectorでELKへのログ出力を試す](https://qiita.com/sotoiwa/items/da89f095a0644dae966c)
の続きです。

バージョン18.0.0.1からLibertyのログがjson形式で出力できるようになったので、Logstash Collectorを使わず、json形式のログを直接FilebeatでELKに送れるのか試してみます。

# Logstash Collectorの出力

そもそもLogstash Collectorがどのようなデータを送っていたのかを確認します。

[liberty_logstash.conf](https://github.com/WASdev/sample.logstash.collector/blob/master/version-1.1/Kibana5/Beta/20170516/liberty_logstash.conf)を参考にして`logstash_debug.conf`を作成します。filterを無効にし、Elasticsearchではなく標準出力に出力するように設定します。

```ruby:logstash_debug.conf
input {
    beats {
        port => "5043"
        ssl => true
        ssl_certificate => "config/logstash.crt"
        ssl_key => "config/logstash.key"
    }
}
filter {
}
output {
    stdout {}
}
```

この構成ファイルを指定してlogstatshを起動します。

```shell-session
sotoiwa@soto-no-air:~/opt/logstash/logstash-5.3.3
$ bin/logstash -f config/logstash_debug.conf
Sending Logstash's logs to /Users/sotoiwa/opt/logstash/logstash-5.3.3/logs which is now configured via log4j2.properties
[2018-06-19T16:09:54,797][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500}
log4j:WARN No appenders could be found for logger (io.netty.util.internal.logging.InternalLoggerFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
[2018-06-19T16:09:55,925][INFO ][logstash.inputs.beats    ] Beats inputs: Starting input listener {:address=>"0.0.0.0:5043"}
[2018-06-19T16:09:56,039][INFO ][logstash.pipeline        ] Pipeline main started
[2018-06-19T16:09:56,163][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
```

Logstash Collectorを設定したLibertyを起動してログを送ると、以下のような内容を受け取ることが確認できます。

```
2018-06-19T07:14:14.057Z %{host} {"type":"liberty_message","host":"soto-no-air.hogehoge.com","ibm_userDir":"\/Users\/sotoiwa\/opt\/ibm\/wlp\/usr\/","ibm_serverName":"defaultServer","message":"CWWKF0011I: サーバー defaultServer は、Smarter Planet に対応する準備ができました。","ibm_threadId":"00000018","ibm_datetime":"2018-06-19T16:14:08.723+0900","ibm_messageId":"CWWKF0011I","module":"com.ibm.ws.kernel.feature.internal.FeatureManager","loglevel":"AUDIT","ibm_sequence":"1529392448723_000000000001C","ibm_tags":["serverRackA5","billingAppTeam"]}
```

形式は

```
時刻 ホスト名 json
```

となっており、時刻とホスト名はLogstashが出しているもので、json部分は以下のような内容です。

```json
{
	"type": "liberty_message",
	"host": "soto-no-air.hogehoge.com",
	"ibm_userDir": "/Users/sotoiwa/opt/ibm/wlp/usr/",
	"ibm_serverName": "defaultServer",
	"message": "CWWKF0011I: サーバー defaultServer は、Smarter Planet に対応する準備ができました。",
	"ibm_threadId": "00000018",
	"ibm_datetime": "2018-06-19T16:14:08.723+0900",
	"ibm_messageId": "CWWKF0011I",
	"module": "com.ibm.ws.kernel.feature.internal.FeatureManager",
	"loglevel": "AUDIT",
	"ibm_sequence": "1529392448723_000000000001C",
	"ibm_tags": [
		"serverRackA5",
		"billingAppTeam"
	]
}
```

# Libertyのjson形式のログをFilebeatでELKに送る

Libertyのログをjson形式に変え、このログをFilebeatでELK Stackに送るようにしてみます。

## Libertyのログをjson形式にする

Libertyのログ出力に関するKnowledge Centerのトピックは以下です。

[ロギングおよびトレース](https://www.ibm.com/support/knowledgecenter/en/SSEQTP_liberty/com.ibm.websphere.wlp.doc/ae/rwlp_logging.html)

`messages.log`とコンソール出力はそれぞれ設定項目があります。また`server.xml`で設定する以外にも、環境変数で設定する方法や`bootstrap.properties`で設定する方法があります。

なお、Logstash Collectorではソースとして`message,trace,garbageCollection,ffdc,accessLog`がありましたが、jsonでの出力の場合は今のところ`garbageCollection`はないようです。

Kubernetes環境では[Twelve-Factor App](https://12factor.net/ja/)で述べられているように、コンソール出力（標準出力）するのがよさそうですが、今回は`messages.log`のほうをjsonに変えてローカルファイルに出力させてみます。

`server.xml`に下記を追加します。

```xml
  <logging messageFormat="json" messageSource="message,trace,accessLog,ffdc"></logging>
```

Logstash Collectorが設定されている場合はコメントアウトします。

この設定をするとmessages.logが以下のようなjsonになります。

```
{"type":"liberty_message","host":"soto-no-air.dhcp.hakozaki.ibm.com","ibm_userDir":"\/Users\/sotoiwa\/opt\/ibm\/wlp\/usr\/","ibm_serverName":"defaultServer","message":"CWWKF0011I: サーバー defaultServer は、Smarter Planet に対応する準備ができました。","ibm_threadId":"00000018","ibm_datetime":"2018-06-19T18:37:39.890+0900","ibm_messageId":"CWWKF0011I","module":"com.ibm.ws.kernel.feature.internal.FeatureManager","loglevel":"AUDIT","ibm_sequence":"1529401059890_000000000001A"}
```

整形すると次の通りです。

```json
{
	"type": "liberty_message",
	"host": "soto-no-air.dhcp.hakozaki.ibm.com",
	"ibm_userDir": "/Users/sotoiwa/opt/ibm/wlp/usr/",
	"ibm_serverName": "defaultServer",
	"message": "CWWKF0011I: サーバー defaultServer は、Smarter Planet に対応する準備ができました。",
	"ibm_threadId": "00000018",
	"ibm_datetime": "2018-06-19T18:37:39.890+0900",
	"ibm_messageId": "CWWKF0011I",
	"module": "com.ibm.ws.kernel.feature.internal.FeatureManager",
	"loglevel": "AUDIT",
	"ibm_sequence": "1529401059890_000000000001A"
}
```

Logstash Collectorの場合は設定でタグを付けていたので、その部分を除くとLogstash Collectorの送信内容と同じです。

内容が同じなので、json形式のログをFilebeatでLogstashに送れば後は同じことができそうです。

## Filebeatの準備

下記からダウンロードして展開します。

https://www.elastic.co/jp/downloads/beats/filebeat

今回も5.3.3を使用します。

設定ファイルfilebeat.ymlを修正します。

監視対象のログを指定します。

```yaml
  paths:
    #- /var/log/*.log
    #- c:\programdata\elasticsearch\logs\*
    - /Users/sotoiwa/opt/ibm/wlp/usr/servers/defaultServer/logs/messages.log
```

Elasticsearchに直接送られる設定になっているのでコメントアウトします。

```yaml
#-------------------------- Elasticsearch output ------------------------------
#output.elasticsearch:
  # Array of hosts to connect to.
  #hosts: ["localhost:9200"]
```

Logstashに送るようにします。Logstash Collectorフィーチャーと同じくSSLを使用するので、証明書を指定します。

```yaml
#----------------------------- Logstash output --------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5043"]

  # Optional SSL. By default is off.
  # List of root certificates for HTTPS server verifications
  ssl.certificate_authorities: ["/Users/sotoiwa/opt/logstash/logstash-5.3.3/config/logstash.crt"]
  ssl.verification_mode: none
```

## Filebeatの起動

Logstashのはデバッグ用にstdoutにそのまま出力するように設定して起動しておきます。

```ruby:logstash_debug.conf
input {
    beats {
        port => "5043"
        ssl => true
        ssl_certificate => "config/logstash.crt"
        ssl_key => "config/logstash.key"
    }
}
filter {
}
output {
    stdout {}
}
```

設定ファイルを指定してFilebeatを起動します。テストなので、パーミッションチェックを厳密に行わないようにオプションをつけています。

```shell-session
$ sudo ./filebeat -e -c filebeat.yml -strict.perms=false
2018/05/18 10:14:27.627617 beat.go:285: INFO Home path: [/Users/sotoiwa/opt/filebeat/filebeat-5.3.3-darwin-x86_64] Config path: [/Users/sotoiwa/opt/filebeat/filebeat-5.3.3-darwin-x86_64] Data path: [/Users/sotoiwa/opt/filebeat/filebeat-5.3.3-darwin-x86_64/data] Logs path: [/Users/sotoiwa/opt/filebeat/filebeat-5.3.3-darwin-x86_64/logs]
2018/05/18 10:14:27.627685 beat.go:186: INFO Setup Beat: filebeat; Version: 5.3.3
2018/05/18 10:14:27.627692 metrics.go:23: INFO Metrics logging every 30s
2018/05/18 10:14:27.630309 logstash.go:90: INFO Max Retries set to: 3
2018/05/18 10:14:27.630414 outputs.go:108: INFO Activated logstash as output plugin.
2018/05/18 10:14:27.630569 publish.go:295: INFO Publisher name: soto-no-air.dhcp.makuhari.japan.ibm.com
2018/05/18 10:14:27.630764 async.go:63: INFO Flush Interval set to: 1s
2018/05/18 10:14:27.630783 async.go:64: INFO Max Bulk Size set to: 2048
2018/05/18 10:14:27.631287 beat.go:221: INFO filebeat start running.
2018/05/18 10:14:27.631373 registrar.go:85: INFO Registry file set to: /Users/sotoiwa/opt/filebeat/filebeat-5.3.3-darwin-x86_64/data/registry
2018/05/18 10:14:27.631434 registrar.go:106: INFO Loading registrar data from /Users/sotoiwa/opt/filebeat/filebeat-5.3.3-darwin-x86_64/data/registry
2018/05/18 10:14:27.632698 registrar.go:123: INFO States Loaded from registrar: 1
2018/05/18 10:14:27.632816 registrar.go:236: INFO Starting Registrar
2018/05/18 10:14:27.632966 spooler.go:63: INFO Starting spooler: spool_size: 2048; idle_timeout: 5s
2018/05/18 10:14:27.632810 crawler.go:38: INFO Loading Prospectors: 1
2018/05/18 10:14:27.632911 sync.go:41: INFO Start sending events to output
2018/05/18 10:14:27.634280 prospector_log.go:65: INFO Prospector with previous states loaded: 1
2018/05/18 10:14:27.634740 prospector.go:124: INFO Starting prospector of type: log; id: 11279772214848098663
2018/05/18 10:14:27.634770 crawler.go:58: INFO Loading and starting Prospectors completed. Enabled prospectors: 1
```

## Filebeatが送信した内容の確認

FilebeatからLogstashに出力された内容は以下です。

```
2018-06-19T09:47:23.055Z soto-no-air.dhcp.hakozaki.ibm.com {"type":"liberty_message","host":"soto-no-air.dhcp.hakozaki.ibm.com","ibm_userDir":"\/Users\/sotoiwa\/opt\/ibm\/wlp\/usr\/","ibm_serverName":"defaultServer","message":"CWWKF0011I: サーバー defaultServer は、Smarter Planet に対応する準備ができました。","ibm_threadId":"00000018","ibm_datetime":"2018-06-19T18:42:40.117+0900","ibm_messageId":"CWWKF0011I","module":"com.ibm.ws.kernel.feature.internal.FeatureManager","loglevel":"AUDIT","ibm_sequence":"1529401360117_000000000001A"}
```

Logstash Collectorフィーチャーのときと同じく、

```
時刻 ホスト名 json
```

になっています。

json部分を整形すると、次のようになっています。

```json
{
	"type": "liberty_message",
	"host": "soto-no-air.dhcp.hakozaki.ibm.com",
	"ibm_userDir": "/Users/sotoiwa/opt/ibm/wlp/usr/",
	"ibm_serverName": "defaultServer",
	"message": "CWWKF0011I: サーバー defaultServer は、Smarter Planet に対応する準備ができました。",
	"ibm_threadId": "00000018",
	"ibm_datetime": "2018-06-19T18:42:40.117+0900",
	"ibm_messageId": "CWWKF0011I",
	"module": "com.ibm.ws.kernel.feature.internal.FeatureManager",
	"loglevel": "AUDIT",
	"ibm_sequence": "1529401360117_000000000001A"
}
```

Logstashの設定を変えて、

```ruby:logstash_debug.conf
input {
    beats {
        port => "5043"
        ssl => true
        ssl_certificate => "config/logstash.crt"
        ssl_key => "config/logstash.key"
    }
}
filter {
}
output {
    #stdout {}
    stdout { codec => json }
}

```

としてみると、出力は以下となります。

```json
{
	"@timestamp": "2018-06-19T09:56:17.095Z",
	"offset": 13182,
	"@version": "1",
	"beat": {
		"hostname": "soto-no-air.dhcp.hakozaki.ibm.com",
		"name": "soto-no-air.dhcp.hakozaki.ibm.com",
		"version": "5.3.3"
	},
	"input_type": "log",
	"host": "soto-no-air.dhcp.hakozaki.ibm.com",
	"source": "/Users/sotoiwa/opt/ibm/wlp/usr/servers/defaultServer/logs/messages.log",
	"message": "{\"type\":\"liberty_message\",\"host\":\"soto-no-air.dhcp.hakozaki.ibm.com\",\"ibm_userDir\":\"\\/Users\\/sotoiwa\\/opt\\/ibm\\/wlp\\/usr\\/\",\"ibm_serverName\":\"defaultServer\",\"message\":\"CWWKF0011I: サーバー defaultServer は、Smarter Planet に対応する準備ができました。\",\"ibm_threadId\":\"00000018\",\"ibm_datetime\":\"2018-06-19T18:56:16.541+0900\",\"ibm_messageId\":\"CWWKF0011I\",\"module\":\"com.ibm.ws.kernel.feature.internal.FeatureManager\",\"loglevel\":\"AUDIT\",\"ibm_sequence\":\"1529402176541_0000000000019\"}",
	"type": "log",
	"tags": [
		"beats_input_codec_plain_applied"
	]
}
```

`message`フィールドのjsonの中に`"type": "liberty_message"`があるものの、Logstash Collectorの場合と異なり、`type`は`log`なので、これでは[liberty_logstash.conf](https://github.com/WASdev/sample.logstash.collector/blob/master/version-1.1/Kibana5/Beta/20170516/liberty_logstash.conf)のfilter設定の条件分岐から漏れてしまい、jsonフィルタープラグインに処理されません。

## typeによる分岐

もともと、[liberty_logstash.conf](https://github.com/WASdev/sample.logstash.collector/blob/master/version-1.1/Kibana5/Beta/20170516/liberty_logstash.conf)のfilterの条件分岐では、`message,trace`ではjsonフィルターのみを行い、`garbageCollection,ffdc,accessLog`ではjsonフィルターに加えてmessageフィールドを削除していました。

`liberty_message`と`liberty_trace`の場合は`message`フィールドのjsonのなかに入れ子で`message`フィールドがあるので、jsonプラグインによって、jsonのなかの`message`フィールドが最終的な`message`フィールドに設定されていました。

`liberty_gc`、`liberty_ffdc`、`liberty_accesslog`の場合は`message`フィールドのjsonのなかに入れ子の`message`フィールドはないので、jsonが`message`フィールドに設定されたままです。この`message`フィールドを不要として削除しているいるのが分岐内の`remove_field`の処理です。

残していてもよさそうですが、先にjsonフィルターをかけてから、フィルター後のtypeフィールドで処理を分岐し、`liberty_gc`、`liberty_ffdc`、`liberty_accesslog`の場合は`message`フィールドを削除するように書くと次のようになります。

```ruby:liberty_logstash.conf
input {
    beats {
        port => "5043"
    ssl => true
        ssl_certificate => "config/logstash.crt" 
        ssl_key => "config/logstash.key" 
    }
}
filter {
    if [type] == "log" {
        json {
            source => "message"
        }
        if [type] == "liberty_accesslog" {
            mutate {
                remove_field => [ "message" ]
            }
        }
        else if [type] == "liberty_gc" {
            mutate {
                remove_field => [ "message" ]
            }
        }
        else if [type] == "liberty_ffdc" {
            mutate {
                remove_field => [ "message" ]
            }
        }
    }
}
output {
    elasticsearch {
        hosts => "localhost:9200"
        template => "config/liberty_logstash_template_ibm.json"
        template_overwrite => "true"
    }
}

```

## ElasticsearchとKibanaからの確認

ここまでLogstashだけを起動していましたが、Logstashの設定を上記の通りに変更してElasticsearchに送るように設定し、ElasticsearchとKibanaを起動すれば、あとはLogstash Collectorでやっていたときと同じです。

# IBM Cloud Privateの場合

IBM Cloud Privateの場合は、Libertyから標準出力にログを出力すると、デフォルトでDaemonSetとしてデプロイされているFilebeatがログを収集してELKログを送るようになっています。

# 参考リンク

[ロギングおよびトレース](https://www.ibm.com/support/knowledgecenter/en/SSEQTP_liberty/com.ibm.websphere.wlp.doc/ae/rwlp_logging.html)

[IBM Cloud Private での Liberty メッセージの分析](https://www.ibm.com/support/knowledgecenter/ja/SSEQTP_liberty/com.ibm.websphere.wlp.doc/ae/twlp_icp_json_logging.html)