背景：使用了自定義插件之后，出現(xiàn)了數(shù)據(jù)無法寫入es的問題。

查看日志后發(fā)現(xiàn)如下信息：

[2019-xxx][INFO ][logstash.outputs.elasticsearch] retrying failed action with response code: 429 ({"type"=>"es_rejected_execution_exception", "reason"=>"rejected execution of processing of [1406096379][indices:data/write/bulk[s][p]]: request: BulkShardRequest [[logstash-xxx][0]] containing [3000] requests, target allocation id: 5nHO5QaBQ36pxxxx, primary term: 1 on EsThreadPoolExecutor[name = data-xxx/write, queue capacity = 200, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@25eb117b[Running, pool size = 24, active threads = 24, queued tasks = 200, completed tasks = 621218675]]"})

查閱相關(guān)資料，了解到出現(xiàn)"es_rejected_execution_exception"是因?yàn)閘ogstash寫入es速度小于logstash讀取數(shù)據(jù)速度，導(dǎo)致logstash頻繁給es發(fā)送bulk reuqest，從而導(dǎo)致es集群的網(wǎng)絡(luò)io過載，進(jìn)而產(chǎn)生以上的問題提示，表明elasticsearch無法繼續(xù)接收數(shù)據(jù)。
可以通過調(diào)整logstash.yml的pipeline.workers,pipeline.output.workers ,pipeline.batch.size,pipeline.batch.delay，或者output plugin的flush_size, idle_flush_time參數(shù)來進(jìn)行優(yōu)化，經(jīng)過試驗(yàn)后發(fā)現(xiàn)7.3.1版本的output plugin中flush_size,idle_flush_time已經(jīng)不可用。
調(diào)整pipeline.batch.size=3000,pipeline.batch.delay=100后，正常運(yùn)行了幾小時(shí)后又出現(xiàn)了上述問題。
查閱官方文檔后，找到一個(gè)output plugin配置項(xiàng)sniffing_delay與上述idle_flush_time類似。
懷疑是用了自定義插件導(dǎo)致處理速度過慢，但是經(jīng)過驗(yàn)證發(fā)現(xiàn)處理速度與使用自定義插件之前并沒有差距。
問題解決：將pipeline.batch.size調(diào)整為2000后，logstash能流暢處理，并未產(chǎn)生擁堵。所以，產(chǎn)生這個(gè)問題的原因還是進(jìn)入logstash的速度大于logstash輸出的速度導(dǎo)致堵塞。在進(jìn)行l(wèi)ogstash性能優(yōu)化時(shí)，不同的處理流程可能需要設(shè)置不同的控制參數(shù)才能達(dá)到較優(yōu)的性能。
說明：這個(gè)錯(cuò)誤的問題的產(chǎn)生與logstash的bulk request息息相關(guān)，es_rejected_execution_exception正是logstash像es發(fā)出請(qǐng)求后，若bulk queue已滿，es就會(huì)向logstash返回該應(yīng)答。

bulk request詳解

Why am I seeing bulk rejections in my Elasticsearch cluster?

當(dāng)我們發(fā)送一個(gè)bulk indexing request給ES集群時(shí)，ES集群做了什么？
當(dāng)一個(gè)bulk request（該請(qǐng)求中包含待處理的文檔）到達(dá)es集群時(shí)，該請(qǐng)求會(huì)被放入bulk queue中，并由bulk thread pool中的bulk thread處理，收到請(qǐng)求的節(jié)點(diǎn)為協(xié)調(diào)節(jié)點(diǎn)，因?yàn)樵摴?jié)點(diǎn)將會(huì)管理該請(qǐng)求的生命周期，并且負(fù)責(zé)響應(yīng)請(qǐng)求。
請(qǐng)求處理的第一步是將bulk request按照文檔將要被分配到的分片進(jìn)行分割。每一個(gè)sub-request就會(huì)被發(fā)送到分片所在的節(jié)點(diǎn)，并加入該節(jié)點(diǎn)的bulk queue中。如果該節(jié)點(diǎn)的bulk queue已滿，那么協(xié)調(diào)節(jié)點(diǎn)將會(huì)收到bulk sub-request被拒絕的消息。bulk queue中的請(qǐng)求將被bulk thread pool處理，并將文檔當(dāng)做處理的一部分發(fā)送到副本分片。當(dāng)bulk sub-request處理結(jié)束后，該節(jié)點(diǎn)將向協(xié)調(diào)節(jié)點(diǎn)返回響應(yīng)。
當(dāng)所有bulk sub-request被處理或被拒絕，協(xié)調(diào)節(jié)點(diǎn)將向客戶端返回響應(yīng)，這其中有可能有些文檔是被拒絕了的。
bulk queue被設(shè)置為有最大限制是因?yàn)?，如果沒有限制的話，將會(huì)很容易通過惡意的行為或者僅僅是無心之失從而讓es集群變得不穩(wěn)定或不可靠。這個(gè)限制通過大量應(yīng)用實(shí)例的經(jīng)驗(yàn)總結(jié)而來，所以通常來說是不需要修改的。
使用HTTP或其他協(xié)議的接口時(shí)，當(dāng)收到429響應(yīng)，就表示bulk request被拒絕了。這時(shí)一般的做法是重試這個(gè)請(qǐng)求，logstash和filebeat的做法就是這樣。

那么請(qǐng)求被拒絕的頻率與什么有關(guān)呢？
該文通過試驗(yàn)得出的結(jié)論是：

• 當(dāng)分片數(shù)與索引數(shù)越多，請(qǐng)求被拒絕的頻率就越大
• 當(dāng)集群節(jié)點(diǎn)越多，請(qǐng)求被拒絕的頻率就越小

能通過提高bulk queue size來解決這個(gè)問題嗎？

• 提高bulk queue size能做到的只是延遲問題到來的時(shí)間，并不能解決問題
• 有可能增大jvm heap占用，從而導(dǎo)致集群不穩(wěn)定

能通過設(shè)置專用的協(xié)調(diào)節(jié)點(diǎn)解決問題嗎？

• 增加專用協(xié)調(diào)節(jié)點(diǎn)只是讓數(shù)據(jù)節(jié)點(diǎn)能專心處理被分配到的子請(qǐng)求，對(duì)于有些用例可能有效，但是大多數(shù)時(shí)候是沒有什么大的效果的。

我們能做什么？
并沒有一個(gè)一勞永逸的方法，我們能做的是在問題發(fā)生時(shí)，嘗試了解原因，看看是單節(jié)點(diǎn)還是整個(gè)集群出現(xiàn)拒絕的情況。如果集群無法處理負(fù)載，請(qǐng)確認(rèn)集群負(fù)載是否均勻。如果負(fù)載均衡不能解決問題，能做的可能就是增加集群節(jié)點(diǎn)數(shù)量了，這將會(huì)增加隊(duì)列的容量，減少隊(duì)列被填滿的可能性。

關(guān)于分片

How many shards should I have in my Elasticsearch cluster？

Elasticsearch 中的數(shù)據(jù)會(huì)整理為索引。每個(gè)索引又由一個(gè)或多個(gè)分片組成。每個(gè)分片都是一個(gè) Lucene 索引實(shí)例，您可以將其視作一個(gè)獨(dú)立的搜索引擎，它能夠?qū)?Elasticsearch 集群中的數(shù)據(jù)子集進(jìn)行索引并處理相關(guān)查詢。
每個(gè)分片都有一部分?jǐn)?shù)據(jù)需要保存在內(nèi)存中，這部分?jǐn)?shù)據(jù)也會(huì)占用堆內(nèi)存空間。這包括存儲(chǔ)分片級(jí)別以及段級(jí)別信息的數(shù)據(jù)結(jié)構(gòu)，因?yàn)橹挥羞@樣才能確定數(shù)據(jù)在磁盤上的存儲(chǔ)位置。這些數(shù)據(jù)結(jié)構(gòu)的大小并不固定，不同用例之間會(huì)有很大的差別。段相關(guān)開銷有一個(gè)重要特征，那就是其并不與段的大小呈嚴(yán)格正比關(guān)系。這意味著，與較小的段相比，對(duì)于較大的段而言，其單位數(shù)據(jù)量所需的開銷要小一些。二者之間的差異可能會(huì)十分巨大。
在 Elasticsearch 中，每個(gè)查詢都是在單個(gè)分片上以單線程方式執(zhí)行的。然而，可以同時(shí)對(duì)多個(gè)分片進(jìn)行處理，正如可以針對(duì)同一分片進(jìn)行多次查詢和聚合一樣。這意味著，最低查詢延時(shí)（假設(shè)沒有緩存）將取決于數(shù)據(jù)、查詢類型，以及分片大小。盡管查詢很多個(gè)小分片會(huì)加快單個(gè)分片的處理速度，但是由于有很多任務(wù)需要進(jìn)入隊(duì)列并按順序加以處理，所以與查詢較少的大分片相比，這種方法并不一定會(huì)加快查詢速度。如果有多個(gè)并發(fā)查詢，擁有很多小分片還會(huì)降低查詢吞吐量。

ELK系統(tǒng)之logstash問題：retrying failed action with response code: 429
logstash output plugin 官方文檔
Why am I seeing bulk rejections in my Elasticsearch cluster?
How many shards should I have in my Elasticsearch cluster？