Could not push logs to Elasticsearch cluster

[masa]

Elasticsearch 6.6 と td-agent 3.3.0 を使用しています。

td-agent から大量のデータを elasticsearch へ送信した際に、途中で elasticsearch にデータが登録できなくなりました。 td-agent のログには、以下のようなワーニングが出力されています。
elasticsearch から接続がリセットされているようです。 どのようにチューニングをしたら正常にデータを受信できるようになるでしょうか？ ご存知の方がおりましたらご教示頂きますようお願い致します。

2019-02-20 14:12:31 +0900 [warn]: #1 failed to flush the buffer. retry_time=7 next_retry_seconds=2019-02-20 14:13:37 +0900 chunk="5824c6165400b5abb0eb9b70cc611261" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>"localhost", :port=>9200, :scheme=>"http"}): Broken pipe (Errno::EPIPE)"
2019-02-20 14:12:31 +0900 [warn]: #1 suppressed same stacktrace
2019-02-20 14:12:31 +0900 [warn]: #2 failed to flush the buffer. retry_time=6 next_retry_seconds=2019-02-20 14:13:02 +0900 chunk="5824c640a1a3968be2e30360ac72a20c" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>"localhost", :port=>9200, :scheme=>"http"}): Connection reset by peer (Errno::ECONNRESET)"
2019-02-20 14:12:31 +0900 [warn]: #2 suppressed same stacktrace
2019-02-20 14:12:47 +0900 [warn]: #0 failed to flush the buffer. retry_time=0 next_retry_seconds=2019-02-20 14:12:48 +0900 chunk="5824c66c015477bae2ebb910ba210d1c" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>"localhost", :port=>9200, :scheme=>"http"}): Connection reset by peer (Errno::ECONNRESET)"
2019-02-20 14:12:47 +0900 [warn]: #0 suppressed same stacktrace
2019-02-20 14:12:50 +0900 [warn]: #0 failed to flush the buffer. retry_time=1 next_retry_seconds=2019-02-20 14:12:51 +0900 chunk="5824c66c015477bae2ebb910ba210d1c" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>"localhost", :port=>9200, :scheme=>"http"}): Connection reset by peer (Errno::ECONNRESET)"

[r4-keisuke (Sunggyu Kei Rhie)]

こんにちはー

td-agentとelasticsearchが同一サーバ上に存在してるように見えますがあってますか？
そしたらネットワーク系の問題はなさそうなので、

・curlでlocalhost:9200で何が見えるか確認する→そもそも見れてない
・_cat/healthや_cat/indicesなどで今Elasticsearchは健康なのか確認する→Elasticsearchが受けれる状態にない
・td-agentの再起動で解消するか確認する→Elasticsearchの問題ではない
・ディスクの容量は十分か？→Elasticsearchの問題ではない

などで問題の切り分けを行ったほうがいいかと思いますー

ログを見てどんな状態のことをこういうログで吐くのかはよくわからないですね。。。

[mukai]

こんにちは。

td-agent から大量のデータを elasticsearch へ送信した際に、途中で elasticsearch にデータが登録できなくなりました。

Elasticsearchはマシンスペックによって、時間あたりの受け取れるデータ量(データ件数)に上限があります。
上限以上のデータが来た場合、データを受け取ることができずに(恐らく)破棄されます。

ひょっとしたら下記のページが参考になるかもしれません。

fluentd + elasticsearch ではまったところ - notebook
fluentd + elasticsearch ではまったところ - notebook
fluentdの設定で、buffer_chunk_limitを多めに変更したのでelasticsearch側で受け付けきれなくなったみたいでした
なのでelasticsearch側で受け付けられるサイズをfluentdのバッファーサイズと同じにしました。

多種多様なログをFluentd-Elasticsearch-Kibanaしたメモ - Tech Notes
多種多様なログをFluentd-Elasticsearch-Kibanaしたメモ - Tech Notes
これに出会ったのは大分前なのだけど、流量が多い場合にバッファのサイズが大きくなると、Elasticsearchへの投入でタイムアウトが発生したりした.
対応として、fluent-plugin-elasticsearchの設定でrequest_timeoutを長めに設定している.

多少板違いな内容で失礼。

[mnozawa (Makoto Nozawa)]

「大量のログ」「途中で」というワードからの想像ですが、Elasticsearch側のキューが溢れた可能せいがあります。
GET _cat/thread_pool?v とやってrejectedが1以上になっているものがあれば途中で処理が追いつかなくなって接続を拒否したかもしれません
この数値はElasticsearchの再起動を行うと0に戻ると思うので、まだであれば確認してみてはどうでしょう。

参考
https://www.elastic.co/guide/en/elasticsearch/reference/current/cat-thread-pool.html

[masa]

情報ありがとうございます。
再度、事象を発生させた状態でcat APIで状態を確認してみました。

1. _cat/thread_pool

rejected はすべて 0 となっていました。

2. _cat/health

curl 'http://localhost:9200/_cat/health?v'

epoch timestamp cluster status node.total node.data shards pri relo init unassign pending_tasks max_task_wait_time active_shards_percent
1551945208 07:53:28 elasticsearch yellow 1 1 82 82 0 0 65 0 - 55.8%

3. _cat/indices

curl 'http://localhost:9200/_cat/indices?v&s=index'

health status index uuid pri rep docs.count docs.deleted store.size pri.store.size
green open .kibana_1 JKZG6VEDQu2D5YD9RdK9tw 1 0 6 0 32.5kb 32.5kb
green open .monitoring-es-6-2019.02.28 MGLyPUVzS_y9fv98wWsUwQ 1 0 170340 1375 90.4mb 90.4mb
green open .monitoring-es-6-2019.03.01 7gqYEterTc2sxiSUG2FgBw 1 0 194518 951 104.2mb 104.2mb
green open .monitoring-es-6-2019.03.02 9iPAB3eqTJSm51K8gJIWoA 1 0 213545 1218 112.4mb 112.4mb
green open .monitoring-es-6-2019.03.03 L-UcJXBqSreo-tAQICpnJQ 1 0 230794 1099 118.7mb 118.7mb
green open .monitoring-es-6-2019.03.04 bBkTap-1Rpem8nDLPKeq3A 1 0 239833 1995 121.4mb 121.4mb
green open .monitoring-es-6-2019.03.05 sUWTSYJGS0iD9doU9NSrjQ 1 0 274145 1892 137.5mb 137.5mb
green open .monitoring-es-6-2019.03.06 xTUH6qLjR0OcGXz7yX9u7g 1 0 285648 1352 142.8mb 142.8mb
green open .monitoring-es-6-2019.03.07 0C2hpcRCSUeaFX7xW-T8Qg 1 0 97237 1470 51.3mb 51.3mb
green open .monitoring-kibana-6-2019.02.28 P6GNy-3-Simu_XV4NpTUaA 1 0 25917 0 5mb 5mb
green open .monitoring-kibana-6-2019.03.01 yD6E3TvNSSWoWgSzt-9FZw 1 0 25917 0 4.8mb 4.8mb
green open .monitoring-kibana-6-2019.03.02 Se5r3cj9Sqykxr6FYmM5ag 1 0 25917 0 4.6mb 4.6mb
green open .monitoring-kibana-6-2019.03.03 81psk98bTbKkLjM8UK36Xg 1 0 25917 0 4.4mb 4.4mb
green open .monitoring-kibana-6-2019.03.04 r4_s78wlRkW0JCoCHHiS-w 1 0 34748 0 5.8mb 5.8mb
green open .monitoring-kibana-6-2019.03.05 iefLbVdISACYvTdmdZcTKA 1 0 43195 0 6.5mb 6.5mb
green open .monitoring-kibana-6-2019.03.06 MwpI7Rl7SO2ea_vSBbiFWg 1 0 43195 0 6.4mb 6.4mb
green open .monitoring-kibana-6-2019.03.07 G7_LhdfqTD2jQY7iinSpdg 1 0 14215 0 2.4mb 2.4mb
yellow open elastalert_status EozthPCCT7-n6z8WJlr9Jw 5 1 0 0 1.2kb 1.2kb
yellow open elastalert_status_error mS5eZkXlQgO0t_FXdpdOCw 5 1 0 0 1.2kb 1.2kb
yellow open elastalert_status_past u6vLKpnFTfixRACqddsLxw 5 1 0 0 1.2kb 1.2kb
yellow open elastalert_status_silence IVMx7xgWRZiOf_0prlpwTQ 5 1 0 0 1.2kb 1.2kb
yellow open elastalert_status_status zEz2uDeMTE2K7oAzN2l54Q 5 1 0 0 1.2kb 1.2kb
yellow open logstash-2019.02.28 RivMG83OSWOte2BAxWnntQ 5 1 3243 0 1.2mb 1.2mb
yellow open logstash-2019.03.01 xUQw4iWFR2eqF1FgqjzGtg 5 1 3441 0 1.1mb 1.1mb
yellow open logstash-2019.03.02 v6Ix4iw0ScWlypkuevmpzg 5 1 597 0 391.9kb 391.9kb
yellow open logstash-2019.03.03 fJcTDSNmQduSLhjRAOkfqw 5 1 1098 0 496.7kb 496.7kb
yellow open logstash-2019.03.04 pO69tAUuS1G7CtArjX1xFg 5 1 2823 0 1mb 1mb
yellow open logstash-2019.03.05 pzAZFzJ0Spyabt9cnHYZZw 5 1 3170 0 1.1mb 1.1mb
yellow open logstash-2019.03.06 zw-dxRjFSpa12OoSKTkWIA 5 1 28697 0 6.5mb 6.5mb
yellow open logstash-2019.03.07 xSdkkJMcRCKVKFm9VmyaQQ 5 1 4550125 0 1.5gb 1.5gb

4. _cat/nodes

curl -XGET 'http://localhost:9200/_cat/nodes?v'

ip heap.percent ram.percent cpu load_1m load_5m load_15m node.role master name
10.0.x.x 44 94 30 0.28 0.49 0.82 mdi * hoge

ディスクの容量は問題ありません。
td-agent/elasticsearch を再起動しても状況に変化はありません。

[masa]

また、td-agent に以下のログが出力される前に、

2019-03-07 16:24:25 +0900 [warn]: #0 failed to flush the buffer. retry_time=0 next_retry_seconds=2019-03-07 16:24:26 +0900 chunk="5837bfdbba07c0e583347a95680b13b4" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>"localhost", :port=>9200, :scheme=>"http"}): Connection reset by peer (Errno::ECONNRESET)"
2019-03-07 16:24:25 +0900 [warn]: #0 suppressed same stacktrace
2019-03-07 16:24:27 +0900 [warn]: #0 failed to flush the buffer. retry_time=1 next_retry_seconds=2019-03-07 16:24:28 +0900 chunk="5837bfdbba07c0e583347a95680b13b4" error_class=Fluent::Plugin::ElasticsearchOutput::RecoverableRequestFailure error="could not push logs to Elasticsearch cluster ({:host=>"localhost", :port=>9200, :scheme=>"http"}): Broken pipe (Errno::EPIPE)"

次のようなログが大量に出力されていました。

2019-03-07 16:19:14 +0900 [warn]: #0 dump an error event: error_class=Fluent::Plugin::ElasticsearchErrorHandler::ElasticsearchError error="400 - Rejected by Elasticsearch" location=nil tag="syslog.local0.info" time=2019-03-07 16:19:03.000000000 +0900 record={"host"=>"10.x.x.x", "message"=>"EvntSLog: RealSource:"host1.sample.co.jp" [INF] [Source:Service Control Manager] [Category:0] [ID:7036] [User:N\A] 2019-03-07 16:19:03 The Google Update \xA5\xB5\xA1\xBC\xA5\xD3\xA5\xB9 (gupdate) service entered the running state.", "priority"=>"info", "facility"=>"local0"}
2019-03-07 16:19:36 +0900 [warn]: #0 dump an error event: error_class=Fluent::Plugin::ElasticsearchErrorHandler::ElasticsearchError error="400 - Rejected by Elasticsearch" location=nil tag="fluent.warn" time=2019-03-07 16:19:14.055454146 +0900 record={"error"=>"#<Fluent::Plugin::ElasticsearchErrorHandler::ElasticsearchError: 400 - Rejected by Elasticsearch>", "location"=>nil, "tag"=>"syslog.local0.info", "time"=>2019-03-07 16:19:03.000000000 +0900, "record"=>{"host"=>"10.x.x.x", "message"=>"EvntSLog: RealSource:"host1.sample.co.jp" [INF] [Source:Service Control Manager] [Category:0] [ID:7036] [User:N\A] 2019-03-07 16:19:03 The Google Update \xA5\xB5\xA1\xBC\xA5\xD3\xA5\xB9 (gupdate) service entered the running state.", "priority"=>"info", "facility"=>"local0"}, "message"=>"dump an error event: error_class=Fluent::Plugin::ElasticsearchErrorHandler::ElasticsearchError error="400 - Rejected by Elasticsearch" location=nil tag="syslog.local0.info" time=2019-03-07 16:19:03.000000000 +0900 record={"host"=>"10.x.x.x", "message"=>"EvntSLog: RealSource:\"host1.sample.co.jp\" [INF] [Source:Service Control Manager] [Category:0] [ID:7036] [User:N\\A] 2019-03-07 16:19:03 The Google Update \xA5\xB5\xA1\xBC\xA5\xD3\xA5\xB9 (gupdate) service entered the running state.", "priority"=>"info", "facility"=>"local0"}"}

elasticsearchのログにも以下のような出力があります。
"Caused by: com.fasterxml.jackson.core.JsonParseException: Invalid UTF-8 start byte 0xa5"とあり、
どうも文字コードが問題でElasticsearchでエラーになっているように思われます。

回避方法をご存知の方がおりましたら、ご教示頂けますでしょうか？

[2019-03-07T16:19:14,028][DEBUG][o.e.a.b.TransportShardBulkAction] [s0055u-logsearch] [logstash-2019.03.07][2] failed to execute bulk item (index) index {[logstash-2019.03.07][fluentd][FSEFV2kB9Q1oFsp6OQxi], source[{"host":"10.x.x.x","message":"EvntSLog: RealSource:"host1.sample.co.jp" [INF] [Source:Service Control Manager] [Category:0] [ID:7036] [User:N\A] 2019-03-07 16:19:03 The Google Update (gupdate) service entered the running state.","priority":"info","facility":"local0","@timestamp":"2019-03-07T16:19:03.000000000+09:00"}]}
org.elasticsearch.index.mapper.MapperParsingException: failed to parse field [message] of type [text]
at org.elasticsearch.index.mapper.FieldMapper.parse(FieldMapper.java:303) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.DocumentParser.parseObjectOrField(DocumentParser.java:488) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.DocumentParser.parseValue(DocumentParser.java:616) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.DocumentParser.innerParseObject(DocumentParser.java:410) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.DocumentParser.parseObjectOrNested(DocumentParser.java:384) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.DocumentParser.internalParseDocument(DocumentParser.java:96) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.DocumentParser.parseDocument(DocumentParser.java:69) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.DocumentMapper.parse(DocumentMapper.java:281) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.shard.IndexShard.prepareIndex(IndexShard.java:763) ~[elasticsearch-6.6.0.jar:6.6.0]

(～省略～)

Caused by: com.fasterxml.jackson.core.JsonParseException: Invalid UTF-8 start byte 0xa5
at [Source: org.elasticsearch.common.bytes.BytesReference$MarkSupportingStreamInputWrapper@5840cbf3; line: 1, column: 193]
at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1702) ~[jackson-core-2.8.11.jar:2.8.11]
at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:558) ~[jackson-core-2.8.11.jar:2.8.11]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidInitial(UTF8StreamJsonParser.java:3544) ~[jackson-core-2.8.11.jar:2.8.11]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidChar(UTF8StreamJsonParser.java:3538) ~[jackson-core-2.8.11.jar:2.8.11]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._finishString2(UTF8StreamJsonParser.java:2543) ~[jackson-core-2.8.11.jar:2.8.11]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._finishAndReturnString(UTF8StreamJsonParser.java:2469) ~[jackson-core-2.8.11.jar:2.8.11]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.getText(UTF8StreamJsonParser.java:315) ~[jackson-core-2.8.11.jar:2.8.11]
at org.elasticsearch.common.xcontent.json.JsonXContentParser.text(JsonXContentParser.java:83) ~[elasticsearch-x-content-6.6.0.jar:6.6.0]
at org.elasticsearch.common.xcontent.support.AbstractXContentParser.textOrNull(AbstractXContentParser.java:269) ~[elasticsearch-x-content-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.TextFieldMapper.parseCreateField(TextFieldMapper.java:719) ~[elasticsearch-6.6.0.jar:6.6.0]
at org.elasticsearch.index.mapper.FieldMapper.parse(FieldMapper.java:297) ~[elasticsearch-6.6.0.jar:6.6.0]
... 42 more

[mnozawa (Makoto Nozawa)]

おっしゃる通り文字コード問題のようですね
0xa5 から始まっているようなのでデータ自体がJIS系の文字コードでエンコードされているのではないでしょうか
例えばwindowsのイベントログなどはShift-JIS(ぽいもの)になるはずです
普段fluentdを使わないのですが、おそらくESに送るより前に文字コードを変換できるライブラリがありますので、それを咬ませるのが良いかと思います