elasticsearchでlogstashのupdateで起きる、conflictによるデータの欠落と解消

# 起きた状況
以前書いた
[elasticsearchのlogstashで同一IDのデータを更新/マージする方法（updateとdoc_as_upsert）](https://qiita.com/henatyokotraveler/items/a8b1b8ce01c5b63a7f1f)
でデータ処理をしていたところ、
タグを付与する更新処理がされていないデータが出てきた。
更新処理用のCSVはlogstashで読み込まれているが、一部が反映されていない。
また、Grafanaでタグの付与されないデータでグラフ化し、１０秒で表示更新して、
タグの付与状況をリアルタイムでずっと眺めていると、付与自体も結構遅いデータがある。


# 最初はelasticsearchのキューのrejectを疑った
[fluentd -> Elasticsearch 大量データ転送でトラブル](http://blog.kakipo.com/trouble-with-fluentd-and-elasticsearch/)
の「原因2: 処理キューが閾値を超えた場合にリクエストを破棄する」とか

>Elasticsearch はデフォルトで閾値を超えたキューを破棄する設定となっています。 例えば index の作成は 50 個まで、bulk 処理は 20 個までといった具合。
メモリの消費を抑え適切に管理するための設定ではあるのですが、 この設定により作られるハズの index が作成されなかったり、 投入されるハズの record が作成されなかったりします。

[es_rejected_execution_exceptionが出たのでthread_pool.bulk.queue_sizeを増やした](https://wyukawa.hatenablog.com/entry/20170403/1491204203)
>es_rejected_execution_exceptionはElasticsearchが持っているスレッドプールのキューサイズを超えるbulk requestがきたので、それをrejectした、つまりログがElasticsearchに一部入らなかったということです。

とかの情報にあるような、キューのrejectを疑った。
でも、ログ見てもrejectはないし、
[Elasticsearchのcat thread pool](https://sakura818uuu.hatenadiary.com/entry/2017/11/03/155954)
に出てるキューの確認コマンド打ってもrejectの痕跡はない

```
curl -XGET 'localhost:9200/_cat/thread_pool?pretty'
```
# versionのconflict
logstashの方にはなかったけど、elasticsearchの方を見ていたら、
retry_on_conflict=>1
というのが目についた。
ちょっと他の部分を覚えていないのがよくないのだが、
elasticのディスカッションの下記にあったのような感じだった
[Version conflict, document already exists (current version [1])](https://discuss.elastic.co/t/version-conflict-document-already-exists-current-version-1/139230)

>[2018-07-09T15:10:44.971-0400][WARN ][logstash.outputs.elasticsearch] Failed action. {:status=>409, :action=>["update", {:_id=>"f4:4d:30:60:8a:31", :_index=>"state_mac", :_type=>"state", :_routing=>nil, :_retry_on_conflict=>1}, 2018-07-09T19:09:45.000Z %{host} %{message}], :response=>{"update"=>{"_index"=>"state_mac", "_type"=>"state", "_id"=>"f4:4d:30:60:8a:31", "status"=>409, "error"=>{"type"=>"version_conflict_engine_exception", "reason"=>"[state][f4:4d:30:60:8a:31]: version conflict, document already exists (current version [1])", "index_uuid"=>"huFaDcR5RgeG92F5S8F9kw", "shard"=>"2", "index"=>"state_mac"}}}}

# API
最初はconflictを無視して強制上書きするオプションを探した。
APIだと、下記のような情報が見つかる

[Update By Query API](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html#docs-update-by-query)に記述がある。

>If you want to simply count version conflicts, and not cause the _update_by_query to abort, you can set conflicts=proceed on the url or "conflicts": "proceed" in the request body. 

[Reindexが登場!](https://www.elastic.co/jp/blog/reindex-is-coming)とelasticのブログで日本語でもあった

>先ほどと同じように、この処理中にドキュメントを更新しようとすると、失敗しますが、そのような場合には、失敗した場所からリトライするような機能もあります。もしあなたが、bananasタグがあった場合にchocolateタグを追加し、同時にアップデートするようなアプリケーションを作っている場合、_update_by_queryでバージョンの矛盾を安全に無視できます。conflicts=proceedと設定します。これはバージョンの矛盾を数え、アップデートを継続します。

ただ、logstashのelasticsearchのoutputプラグインでは上記を行う方法がないようだった。

# logstashの場合の対処法
logstashのelasticsearchのプラグインには、retry_on_conflictというオプションがある。
デフォルトが１なので、
これを増やすと、conflict時にリトライが行われ、ちゃんと更新される確率が上がる。
[retry_on_conflict](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-retry_on_conflict)
>retry_on_conflict
・Value type is number
・Default value is 1
 The number of times Elasticsearch should internally retry an update/upserted document See the partial updates for more info

上記に関連して、下記２つの設定もある。
デフォルトは、リトライは2秒、4秒、8秒と倍々していき、64秒を上限にretry_on_conflictの回数を繰り返す。

>retry_initial_interval
・Value type is number
・Default value is 2
Set initial interval in seconds between bulk retries. Doubled on each retry up to retry_max_interval

>retry_max_interval
・Value type is number
・Default value is 64
Set max interval in seconds between bulk retries.

**logstashの書き方サンプル**
[elasticsearchのlogstashで同一IDのデータを更新/マージする方法（updateとdoc_as_upsert）](https://qiita.com/henatyokotraveler/items/a8b1b8ce01c5b63a7f1f)のUpdate設定にretry_on_conflictで5回のリトライを追加したサンプル

```
output {
    elasticsearch {
        hosts => ["testserver:9200"]
        index => "testindex"
        document_id => "%{hogeid}"
        doc_as_upsert => true
        action => "update"
        retry_on_conflict => 5
    }
}
```

*参考：elasticのドキュメントでこんなのもあった*
[Updates and Conflicts](https://www.elastic.co/guide/en/elasticsearch/guide/current/partial-updates.html#_updates_and_conflicts)

# その他
logstashのretry_on_conflictに加えて、下記も行った。
元データの更新を行うタグ付与の対象が、
過去5分の追加データ全てで、複数回タグ付与が行われ、conflictが起きやすい状況だった。
タグ付与の対象を、まだタグの付与されていないデータに限定するようにクエリ変更も行った。

logstashのリトライと更新のconflictが起きやすいデータ更新を見直したことで、conflictによるデータ欠落は亡くなった。