Elasticsearch supports a wide range of use-cases across our user base, and more and more of these rely on fast indexing to quickly get large amounts of data into Elasticsearch. Even though Elasticsearch is fast and index performance is continually improved, it is still possible to overwhelm it. At that point you typically see parts of bulk requests getting rejected. In this blog post we will look at the causes and how to avoid it.
Elasticichearch는 사용자 기반 전체에서 광범위한 사용 사례를 지원하며, 점점 더 많은 사용자들이 빠른 indexing을 통해 대량의 data를 Elasticsearch로 빠르게 가져오고 있다. 비록 Elasticsearch가 빠르고, index 성능이 지속적으로 개선되고 있지만, 이것을 압도하는 것은 여전히 가능한 일이다. 이 시점에서 일반적으로 bulk request의 일부가 거부되는 것을 볼 수 있다. 이 블로그 포스트에서 우리는 그 원인과 그것을 피하는 방법을 살펴볼 것이다.
This is the second installment in a series of blog posts where we look at and discuss your common questions. The first installment discussed and provided guidelines around "How many shards one should aim to have in an Elasticsearch cluster?"
이는 우리가 여러분들의 흔한 질문들을 보고 토론하는 블로그 포스트들의 두번째 부분이다. 첫번째는 "How many shards one should aim to have in an Elasticsearch cluster?" 에서 토론하여 가이드라인을 제공했다.
What happens when a bulk indexing request is sent to Elasticsearch?
Let’s start at the beginning and look at what happens behind the scenes when a bulk indexing request is sent to Elasticsearch.
먼저 bulk indexing request가 Elasticsearch에 전송되면, 배후에서 어떤 일이 발생하는지 살펴보자.
When a bulk request arrives at a node in the cluster, it is, in its entirety, put on the bulk queue and processed by the threads in the bulk thread pool. The node that receives the request is referred to as the coordinating node as it manages the life of the request and assembles the response. This can be a node dedicated to just coordinating requests or one of the data nodes in the cluster.
bulk request가 cluster의 node에 도달하면, 전체적으로, 그것을 bulk queue에 넣고, bulk thread pool의 thread에 의해 처리된다. request를 받은 node는, request의 수명을 관리하고 reponse을 구성하기 때문에 coordinating node라고 한다. 이는 단지 request를 coordinating하는 전용 node이거나 cluster의 data node 중 하나일 수 있다.
A bulk request can contain documents destined for multiple indices and shards. The first processing step is therefore to split it up based on which shards the documents need to be routed to. Once this is done, each bulk sub-request is forwarded to the data node that holds the corresponding primary shard, and it is there enqueued on that node’s bulk queue. If there is no more space available on the queue, the coordinating node will be notified that the bulk sub-request has been rejected.
bulk request는 복수의 indices 및 shard를 위한 document가 포함될 수 있다. 그러므로 첫번째 처리 단계는 document를 routing해야 하는 shard에 따라 data를 분할하는 것이다. 이 작업이 완료되면, 각 bulk sub-request이 해당 primary shard를 저장하고 있는 data node로 전달되며, 이 node의 bulk queue에 추가된다. queue에 사용할 수 있는 공간이 더 이상 없으면, coordinating node에 bulk sub-request가 거부되었음을 알리게 된다.
The bulk thread pool processes requests from the queue and documents are forwarded to replica shards as part of this processing. Once the sub-request has completed, a response is sent to the coordinating node.
bulk thread pool은 queue에서 request를 처리하고, document는 이 처리 과정의 일환으로 replica shard로 전달된다. sub-request가 완료되면 coordinating node로 reponse가 전송된다.
Once all sub-requests have completed or been rejected, a response is created and returned to the client. It is possible, and even likely, that only a portion of the documents within a bulk request might have been rejected.
모든 sub-request가 완료되거나 거부되면, response가 생성되어 client로 반환된다. bulk request에 포함된 document 중 일부만 거부될 수도 있다.
The reason Elasticsearch is designed with request queues of limited size is to protect the cluster from being overloaded, which increases stability and reliability. If there were no limits in place, clients could very easily bring a whole cluster down through bad or malicious behaviour. The limits that are in place have been set based on our extensive experience supporting Elasticsearch for different types of use-cases.
Elasticsearch가 제한된 크기의 request queue로 설계된 이유는 안정성과 신뢰성을 높여, cluster가 과부하가 걸리는 것을 방지하기 위함입니다. 제한이 없는 경우, client는 좋지 않은 혹은 악의적인 동작을 통해, 전체 cluster를 아주 쉽게 중단시킬 수 있다. 현재의 제한 사항은 다양한 유형의 사용 사례에 대해 Elasticsearch를 지원하는 우리의 광범위한 경험에 기초하여 설정되었다.
When using the HTTP interface, requests that results in at least a partial rejection will return with response code 429, 'Too many requests'. The principle also applies when the transport protocol is used, although the protocol and interface naturally is different. Applications and clients may report these errors back to the user in different ways, and some may even attempt to handle this automatically by retrying any rejected documents.
HTTP interface를 사용하는 경우, 최소한의 부분적인 거부를 return하는 request는 response code 429, 'Too many requests' 와 함께 return된다. 비록 protocol과 interface는 당연히 다르지만, transport protocol이 사용될 경우에도 이 원칙이 적용된다. application과 client는 이러한 오류를 사용자에게 다른 방식으로 보고할 수 있으며, 어떤 사용자는 거부된 document를 다시 시도하여, 이 오류를 자동으로 처리하려고 할 수도 있다.
How can we test this in practice?
In order to illustrate the practical impact of this behaviour, we devised a simple test where we use our benchmarking tool Rally to run bulk indexing requests against a couple of Elastic Cloud clusters with varying number of data nodes. Configuration and instructions on how to run Rally is available in this gist.
이러한 동작이 실제 상황에서 미치는 영향을 설명하기 위해, 우리는 benchmarking tool Rally를 사용하여, data node 수가 다양한 2 개의 Elastic Cloud clusters에 대해,bulk indexing request을 실행하는 간단한 테스트를 실시했다. Rally를 운영하는 방법에 대한 구성 및 지침은 여기에서 확인할 수 있다.
The same indexing workload was run against three different Elastic Cloud clusters. We have been indexing with one replica shard configured wherever possible. The clusters consisted of one, two and three data nodes respectively, with each data node having 8GB RAM (4GB heap for Elasticsearch, 4GB native memory). Invoking the GET /_nodes/thread_pool API we could see that each data node by default had a fixed bulk thread pool size of two with a queue size of 200:
3개의 서로 다른 Elastic Cloud cluster에 대해서도 동일한 indexing 작업을 실행했다. 우리는 가능한 한, 항상 하나의 replica shard를 설정하고 indexing 작업을 수행했다. cluster는 각각 1개, 2개 및 3개의 data node로 구성되었으며, 각 data node는 8 GB RAM(Elasticsearch에 4GB heap, 4 GB 기본 momory)을 가진다. GET /_nodes/thread_pool API를 호출하면 ,기본적으로 각 data node의 queue의 크기가 200이고, 고정된 bulk thread pool 크기가 2라는 것을 알수 있다.
%> curl -XGET http://<es_url>:<es_port>/_nodes/thread_pool</es_port></es_url>
"bulk": {
"type": "fixed",
"min": 2,
"max": 2,
"queue_size": 200
}
During the test we indexed into a varying number of shards (2, 4, 8, 16, and 32) using a varying number of concurrent clients (8, 16, 24, 32, 48, and 64) for each cluster. For every combination of shard and client count we indexed 6.4 million documents with a batch size of 100 documents and another 6.4 million documents with a batch size of 200 documents. This means that in total we attempted to index 384 million documents per cluster.
test를 하면서, 각 cluster에 대해 다양한 수의 동시(concurrent) client(8, 16, 24, 32, 48, 64)를 사용하여 다양한 수의 shard(2, 4, 8, 16, 16, 32)로 index했다. shard와 client 수를 조합할 때마다, 640만개의 document는 batch 크기를 100으로, 다른 640만개의 document는 batch 크기를 200으로 index했다. 즉, cluster별로 384만개의 document를 indexing하려고 했다.
For this test we treat the clusters as a black box, and perform the analysis from the client’s perspective. To limit the scope we will also not look at the impact of various configurations on performance as that is a quite large topic on its own.
이 테스트에서는 cluster를 black box로 취급하고, client의 관점에서 분석을 수행했다. 범위를 제한하기 위해, 다양한 구성이 성능에 미치는 영향은 다루지 않을 것이다. 이는 상당히 큰 주제이기 때문이다.
All the generated, detailed metrics were sent to a separate Elastic Cloud instance for analysis using Kibana. For each request Rally measures how many the documents in the bulk request were rejected and successful. Based on this data we can classify each request as successful, partially rejected, and fully rejected. A few requests also timed out, and these have also been included for completeness.
생성된 모든 상세 측정값은 Kibana를 사용하여 분석할 수 있도록 별도의 Elastic Cloud instance로 전송되었다. 각 request에 대해, bulk request에서 거부되고 성공한 문서 수를 Rally로 측정했다. 이러한 data를 토대로, 각 request를 성공, 부분 거부, 완전한 거부로 분류하였다. 또한 약간의 time out된 request도, 완벽을 기하기 위해, 포함되었다.
Unlike Beats and Logstash, Rally does not retry failed indexing requests, so each has the same number of requests executed but the final number of documents indexed varied from run to run depending on the volume of rejections.
Beats및 Logstash와 달리, Rally는 실패한 indexing request를 다시 시도하지 않기 때문에, 실행된 request의 수는 동일하지만, 최종적으로 index된 document의 수는 거부의 양에 따라 달라진다.
How bulk rejection frequency depend on shard count, clients count, and data node count?
Bulk rejections occur when the bulk queues fill up. The number of queue slots that get used depends both on the number of concurrent requests, and the number of shards being indexed into. To measure this correlation we have added a calculated metric, client shard concurrency, to each run. This is defined as the number of shards being indexed into, multiplied by the number of concurrent indexing threads, and indicates how many queue slots would be needed to hold all bulk sub-requests.
bulk 거부는 bulk queue이 가득 차면 발생한다. 사용되는 queue slot의 수는 동시 request 수와 indexing 되는 shard 수에 따라 달라진다. 이러한 상관 관계를 측정하기 위해, 각각의 실행에 계산된 지표인 client shard concurrency이 추가되었다. 이는 index될 shard의 수와 동시 indexing thread 수를 곱한 것으로 정의되며, 모든 bulk sub-request을 보관하는 데 필요한 queue slot 수를 나타낸다.
In the graph below, we show how the percentage of requests that result in partial or full rejections, depends on the client shard concurrency for the three different clusters.
아래 그래프는, 서로 다른 세개의 cluster에 대해, client shard concurrency에 따라, 부분 또는 완전한 거부가 발생하는 request의 비율이 어떻게 달라지는지 보여준다.
For clusters with one or two nodes we can see that appearance of bulk rejections start when the client shard concurrency level is somewhere between 192 and 256. This makes sense as each node has a bulk queue size of 200. For the cluster with 3 nodes we can see that it is able to handle even higher level of client shard concurrency without any bulk rejections appearing.
node가 하나 또는 두개인 cluster의 경우, client shard concurrency 수준이 192개에서 256개 사이가 되면, bulk 거부가 시작되는 것을 볼 수 있다. 각 node의 bulk queue 크기가 200개이므로 이는 타당합니다. node가 3개인 cluster의 경우, bulk 거부 반응 없이 더 높은 수준의 client shard concurrency을 처리할 수 있다는 것을 알 수 있다.
Once we get over this limit, we start seeing partial bulk rejections, where at least one sub-request has managed to get queued and processed. A relatively small portion of requests also result on full rejections as the concurrency level increases, especially for the single node cluster.
이 한계를 넘으면 부분적인 bulk 거부가 발생하기 시작하며, 여기서 하나 이상의 sub-request가 queue에 들어가 처리된다. 또한, 상대적으로 request의 작은 부분이, concurrency 수준이 증가함에 따라, 특히 단일 node cluster에 대해, 완전한 거부를 야기한다.
When we compare the single and two node clusters, we can see that the percentage of fully successful requests increases slightly and that there are fewer full rejections. This is expected, as the total bulk queue across the cluster is twice as large and requests are sent to all data nodes. Even though the total bulk queue size is twice as large across the cluster, the 2 node cluster does not appear able to handle twice the client shard concurrency of the single node cluster. This is likely due to the fact that distribution is not perfect and that the introduction of replica shards have resulted in each indexing operation requiring more work and being slower as a result. An important thing to note is also that all partial rejections are treated as equals in this graph. The number of rejected documents is not shown and does indeed vary depending on the cluster size, but we will shortly look at that in greater detail.
단일 node cluster와 2 node cluster를 비교하면, 완전히 성공한 request의 비율이 약간 증가하고 완전한 거부 횟수가 줄어드는 것을 알 수 있다. 이는 cluster의 총 bulk queue 크기가 2배가 되고 모든 data node에 request가 전송되기 때문에 예상되는 일다. 전체 bulk queue 크기가 cluster에서 2배로 되지만, 2 node cluster가 단일 node cluster의 client shard concurrency의 2배를 처리할 수 없을 것으로 보이지는 않는다. 이는 분포가 완벽하지 않고, replica shard를 도입하면 각각의 indexing 연산에 더 많은 작업이 필요하고, 그 결과 작업 속도가 느려지기 때문일 수 있다. 주목해야 할 중요한 점은 이 그래프에서 모든 부분적 거부가 동등하게 취급된다는 것이다. 거부된 문서의 수는 표시되지 않으며 cluster 크기에 따라 실제로 달라질 수 있다. 하지만 이에 대해서는 곧 자세히 살펴보겠습니다.
When we go to three data nodes, we see a more marked improvement, and receive requests without any rejections at high levels of concurrency. We also only see full rejections for the highest concurrency levels.
3개의 data node가 되면 더 큰 향상을 볼 수 있으며, 높은 수준의 concurrency로 어떤 거부 없이 request를 받는다. 또한 가장 높은 concurrenc 수준에 대한 완전한 거부만을 볼 수 있다.
If we instead plot the average portion of rejected documents per request as a function of shard and client count for the three clusters, we get the following graphs.
대신에 3개의 cluster에 대한 shard의 기능과 client 수에 따라, request 시 거부된 document의 평균 부분을 표시해 보면, 다음 그래프를 볼 수 있습니다.
Here we can see that the percentage of rejected events grows with increased concurrency levels for all cluster sizes. We can also see that the rejection levels drop across the board with the more data nodes we add, which is expected.
여기서 모든 cluster의 크기에 대해 향상된 concurrency 수준과 함께 거부되는 event의 비율이 증가함을 알 수 있다. 또한 data node를 더 많이 추가하면, 거부 수준이 전반적으로 저하될 것으로 예상된다.
Earlier we saw that partial rejections started at approximately the same time for both one and two node clusters. If we now look at these graphs, we can see that the portion of rejected documents grows faster for the single node cluster compared to the one with two data nodes. This means that even though we saw a similar level of partially rejected requests, the larger cluster had more documents indexed per request.
앞에서 단일 node cluster와 2 node cluster 모두에서, 부분 거부가 거의 동시에 시작되는 것을 보았다. 이제 이러한 그래프를 살펴보면, data node가 2개인 cluster에 비해 단일 node cluster에서 거부된 document 부분이 더 빨리 증가함을 알 수 있다. 즉, 유사한 수준의 부분적으로 거부된 request가 있었음에도 불구하고, 더 큰 cluster가 request당 더 많은 document를 indexing했다.
Can’t I just get around this by increasing the bulk queue size?
One of the most common reactions when faced with bulk rejections is to increase the size of the bulk queue. Why not set it to a really large value so you do not have to worry about this again?
bulk 거부에 직면했을 때 가장 일반적인 반응 중 하나는 bulk queue의 크기를 늘리는 것입니다. 이 값을 정말 큰 값으로 설정하면 더 이상 걱정할 필요가 없을까?
Increasing the size of the queue is not likely to improve the indexing performance or throughput of your cluster. Instead it would just make the cluster queue up more data in memory, which is likely to result in bulk requests taking longer to complete. The more bulk requests there are in the queue, the more precious heap space will be consumed. If the pressure on the heap gets too large, it can cause a lot of other performance problems and even cluster instability.
queue 크기를 늘린다고 해서 cluster의 indexing 성능이나 처리량이 향상되지는 않는다. 대신 cluster는 memory에서 더 많은 data가 queue에 추가되도록 하는 것뿐이라서, bulk request를 완료하는 더 많은 시간이 소요될 수 있다. queue에 bulk request가 많을수록 더 많은 귀중한 heap 공간이 사용된다. heap에 대한 압박이 너무 커지면 다른 많은 성능 문제가 발생하고 cluster가 불안정해질 수도 있다.
Adjusting the queue sizes is therefore strongly discouraged, as it is like putting a temporary band-aid on the problem rather than actually fixing the underlying issue. So what else can we do improve the situation?
따라서 queue 크기를 조정하는 것은, 실제로 근본적인 문제를 해결하는 것이 아니라, 문제를 일시적으로 해결하는 것과 마찬가지로, 매우 바람직하지 않습니다. 그렇다면 우리가 이 상황을 개선할 수 있는 다른 방법은 무엇일까?
Can coordinating only nodes help?
By introducing coordinating only nodes, the data nodes will be able to focus on processing sub-requests, as the request itself will not take up a slot on their bulk queue. This is generally good, but the actual benefit of this arrangement is likely to vary from use-case to use-case. In many use cases it does relatively little difference, and we see lots of successful indexing heavy use cases that do not use dedicated coordinating nodes.
node만 coordinating하는 방식을 도입하면, data node가 sub-request 처리에 집중할 수 있다. 즉, request 자체가 bulk queue의 slot을 차지하지 않기 때문이다. 이는 일반적으로 좋지만, 이 방식의 실질적인 이점은 사용 사례에 따라 달라질 수 있다. 많은 사용 사례에서 비교적 차이가 거의 없으며, 전용 coordinating node를 사용하지 않는 많은 성공적인 사용 사례를 볼 수 있다.
What conclusions can we draw?
As always, there is not necessarily any good one-size-fits-all solution, and the way to address bulk rejections will vary from use-case to use-case. If you see bulk rejections, try to understand why they are taking place and whether it is a single node or the whole cluster that is affected.
언제나 그렇듯이, 반드시 좋은 만능 솔루션이 있는 것은 아니며, bulk 거부를 처리하는 방법은 사용 사례에 따라 달라질 수 있다. bulk 거부가 보이면, 왜 이러한 거부가 발생하고 있으며 단일 node인지 또는 영향을 받는 전체 cluster인지를 이해하려고 해 보자.
If the cluster is unable to cope with the load, ensure that all nodes are sharing the load evenly. If this does not help, it may be necessary to scale the cluster out or up. This will increase capacity and make it less likely that queues are filled up. Increasing the bulk queue size is only likely to postpone the problems, and may actually make them worse.
cluster가 작업을 처리할 수 없는 경우, 모든 node가 작업을 균등하게 공유하고 있는지 확인하자. 이 방법이 도움이 되지 않는 경우, cluster의 scale-out 또는 scale-up이 필요할 수도 있다. 이렇게 하면 용량이 증가하고 queue이 채워질 가능성이 줄어든다. bulk queue 크기를 늘리는 것은 문제를 지연시킬 뿐이며 실제로 문제를 더 악화시킬 수도 있다.
Also remember that rejected requests do not always mean that all documents were unsuccessful. Make sure you inspect the full response and retry the appropriate documents. Logstash and Beats already do this by default.
또한 거부된 request가 항상 모든 document가 성공하지 못했음을 의미하는 것은 아니다. 전체 response를 확인하고 해당 document를 다시 시도하자. Logstash와 Beats는 이미 기본적으로 이렇게 하고 있다.
We hope this has given you a better understand of how it works. If you have any further questions, there are many ways to engage with us, including through our forum.
이 블로그 게시물이 bulk의 동작 방법을 이해하는데 있어 더 많은 도움이 되기를 바란다. 질문이 있다면, forum등의 다양한 방법으로 우리에게 연락하기 바란다.
원문 : Why am I seeing bulk rejections in my Elasticsearch cluster?