Introduction
In this blog post, I am going to describe a new feature of Elasticsearch called Adaptive Replica Selection (ARS). Before describing what ARS does, I’ll first describe the problem that it tries to solve. Imagine that you have a cluster with three nodes. And across those three nodes, you have a shard and its replicas.
이 게시물에서는 Elasticsearch의 새로운 기능인 Adaptive Replica Selection (ARS)를 설명하려 한다. 먼저 해결하려는 문제에 대해 설명하겠다. 3개의 node를 가진 cluster를 가정하자. 3개의 node에 shard와 그 replica가 있다.
Then imagine that one node starts to experience distress.
그런 다음, 한 node에 문제가 있다고 가정하자.
When a search request is received, Elasticsearch first determines which shards need to be queried, and then determines which nodes contain copies of that shard. Once it’s done that, it then sends the request to a single copy of the shard via round-robin selection. So if Elasticsearch sends the request to shard copy 2, we will see degraded performance compared to requests to the other copies:
search request가 들어오면, 먼저 Elasticsearch는 query에 어떤 shard가 필요한지를 결정하고, 해당 shard의 복사본을 가지고 있는 node가 어떤 것인가를 결정한다. 이것이 완료되면, round-robin 방식으로 shard의 단일 복사본에 request를 전송한다. 따라서, Elasticsearch가 shard 복사본 2에 request를 전송하면, 다른 복사본에 request하는 것에 비해 성능이 저하될 것이다.
- shard copy 1: 100ms
- shard copy 2 (degraded): 1350ms
- shard copy 3: 150ms
The degradation of the node with the second copy of the data can be caused by any number of things, like long garbage collection cycles, high disk IO, network saturation, or heterogeneous node types. This causes tail latency of requests to rise, not to mention it frustrates whoever is unlucky enough to be waiting for the query when it’s run against shard copy 2.
data의 두번째 복사본이 있는 node의 성능 저하는 긴 GC 주기, 높은 disk IO, network 포화, 이기종 node 유형 같은 여러 가지 원인으로 발생할 수 있다. 이로 인해, request에 대한 대기 시간이 길어지고, shard 복사본 2에 대해 실행한 query를 기다리는 불운한 이를 좌절케 한다는 것은 말할 것도 없다.
So what can be done? We would like Elasticsearch to be smart enough to route requests to the other copies of data until this node has recovered enough to handle more search requests. In Elasticsearch, we call this process of selecting the best available copy of the shard Adaptive Replica Selection.
그렇다면, 무엇을 할 수 있을까? Elasticsearch는 이 node가 더 많은 search request를 처리할 수 있도록 복구될 때까지 data의 다른 복사본으로 request를 route할 수 있을 만큼 충분히 스마트하다고 여긴다. Elasticsearch에서, shard의 가장 활용 가능한 복사본을 선택하는 이 process를 Adaptive Replica Selection라 한다.
Adaptive Replica Selection
ARS is an implementation of an academic paper called C3: Cutting Tail Latency in Cloud Data Stores via Adaptive Replica Selection. Originally, the paper was written for Cassandra, which means we had to adapt some things to fit Elasticsearch due to the differences in behavior between the two.
ARS는 C3: Adaptive Replica Selection을 통한 Cloud Data 저장소에서의 대기 시간 단축이라는 학술 논문을 구현한 것이다. 원래 이 논문은 Cassandra를 위해 작성되어, 둘 사이의 동작 방식의 차이 때문에 몇 가지를 Elasticsearch에 맞도록 수정해야 했다.
Our ARS implementation is based on a formula where, for each search request, Elasticsearch ranks each copy of the shard to determine which is likeliest to be the "best" copy to send the request to. Instead of sending requests in a round-robin fashion to each copy of the shard, Elasticsearch selects the "best" copy and routes the request there.
ARS 구현은 각 search request에 대하여, ELasticsearch가 request를 전송할 때 "최상"의 복사본이 될 가능성이 가장 높은 복사본을 결정하기 위하여, 각 shard 복사본의 순위를 결정하는 수식을 기반으로 한다. shard의 각 복사본에 round-robin 방식으로 request를 전송하는 대신, Elasticsearch는 "최상"의 복사본을 선택하고 거기에 request를 route한다.
The ARS formula initially seems complex, but let's break it down:
ARS 수식은 처음에는 복잡해 보인다. 하지만 그것을 나눠보면
Ψ(s) = R(s) - 1/µ̄(s) + (q̂(s))^3 / µ̄(s)
Where q̂(s) is:
q̂(s) = 1 + (os(s) * n) + q(s)
And looking at the individual pieces:
개별 부분을 보면
- os(s) is the number of outstanding search requests to a node
os(s) 는 node에서 아직 처리하지 못한 search request의 수이다. - n is the number of data nodes in the system
n 은 system에서 data node의 수이다. - R(s) is the EWMA of the response time (as seen from the node coordinating the search) in milliseconds
R(s) 는 response 시간(coordinating node에서 본)의 EWMA이다. ms 단위 - q(s) is the EWMA of the number of events in the search thread pool queue
q(s) 는 search thread pool queue에서 event 수의 EWMA이다. - µ̄(s) is the EWMA of service time of search events on the data node in milliseconds
µ̄(s)̄ 는 data node에서의 search event의 service time의 EWMA이다.
Response time is the time measured for a request from the coordinating node, whereas service time is the time measured on the node executing the request (not including any time the request may have spent in the queue). Roughly speaking, this means that as the difference between the response time and service time grows — as well as the number of searches in the queue and the outstanding requests for a node — the higher a score the node will receive (lower scored nodes are considered the “best” ranked). For example, the (q̂(s))^3 part of the formula means that as the number of connections to a data node, and the number of searches in the node’s queue increases, the score will go up in a cubic exponential manner. By factoring in the number of outstanding search requests in the formula, we also make sure that nodes with identical load and response times still distribute requests among shard copies.
response time은 coordinating node에서 request에 대해 측정한시간이고, 반면에 service time은 request를 실행한 node에서 축정한 시간(request가 queue에서 소비한 시간은 포함하지 않음)이다. 대략적으로 말하면, 이는 response time과 service time(뿐만 아니라 queue의 search 수와 node에서 아직 처리하지 못한 request의 수)의 차이가 증가하면, node가 받을 score가 더 높아진다(더 낮은 score를 가진 node가 "최상"의 순위로 간주된다)는 것을 의미한다. 예를 들어, 수식의 (q̂(s))^3 부분은 data node에 대한 연결 수와 node의 queue에 있는 search의 수가 증가함에 따라, score가 3제곱의 형식으로 증가하는 것을 의미한다. 수식에서, 아직 처리하지 못한 search request의 수를 고려하여, 동일한 loadd와 response time을 가진 node가 여전히 shard 복사본간에 request를 배포하는지를 확인한다.
Elasticsearch gathers most of these values using a specialized thread pool for search actions on each of the nodes. For each search response, these values are piggybacked on the results back to the coordinating node, which stores them so the node can be ranked for subsequent requests. When the next search request comes in, the node with the lowest score will serve the request. If the node encounters an exception, the node with the next best score serves the search request. Since we don't want loaded nodes to have searches routed away forever, every time a node is not the best ranked node, we slightly adjust the score to make it a better candidate for handling future requests.
Elasticsearch는 각 node의 search 작업을 위하여 특별한 thread pool을 사용하여 이들 값의 대부분을 수집한다. 각 search response에 대해, 이들 값은 coordinating node로 반환되는 결과에 덧붙여지고, coordinating node는 그것을 저장하여, node는 이어지는 request에에 대해 순위를 매길 수 있다. 다음 search request가 들어오면, 가장 낮은 score를 가진 node가 request를 처리한다. 만약 node에서 exception이 발생하면, 다음 최상의 score를 가진 node가 request를 처리한다. 부하를 가진 node에 계속해서 search가 route되지 않도록, node가 최상위 node가 아닐 때 마다 score를 약간 조정하여, 향후 request를 처리할 수 있는 더 나은 후보가 되도록 한다.
Going back to our original example, the selected node for the request would be either shard copy 1 or 3, avoiding the overloaded shard copy 2.
원래의 예제로 되돌아가서, request에 의해 선택된 node는 과부하가 걸린 shard copy 2를 피하여, copy 1이나 3이 될 것이다.
ARS is available in Elasticsearch 6.1 and later, but is turned off by default in all 6.x releases. It can be turned on dynamically by changing the cluster.routing.use_adaptive_replica_selection setting:
ARS는 available in Elasticsearch 6.1 이후 버전에서 이용할 수 있다. 그러나, 모든 6.x 에서 기본적으로 비활성화되어 있다. cluster.routing.use_adaptive_replica_selection 설정을 변경하여, 동적으로 활성화할 수 있다.
PUT /_cluster/settings { "transient": { "cluster.routing.use_adaptive_replica_selection": true } }
In Elasticsearch 7.0 and later, ARS will be turned on by default.
Elasticsearch 7.0 이후에는 기본적으로 활성화된다.
Improvements with ARS
How much do we expect this to help? We ran benchmarks with Rally for many scenarios, both with and without simulating load on one of the nodes containing a copy of the data. In our case, we used a single coordinating node connected to a cluster of five nodes, searching an index with five primary shards each with a single replica. In each benchmark, the search requests were sent as quickly as possible from 100 Rally connections.
이것이 얼마나 도움이 될가? data의 복사본을 가지고 있는 node 중 하나에 부하를 주는 시뮬레이션을 하거나 하지 않고, 많은 시나리오에 대해 Rally로 벤치마크했다. 우리의 경우, 5개의 node를 가진 cluster에 연결된 단일 coordinating node를 사용하여, 각각 하나의 replica를 가진 5개의 primary shard가 있는 index를 검색했다. 각 벤치마크에서, search request는 100개의 Rally connection에서 가능한 한 빨리 전송하였다.
Non-loaded case:
| Metric | No ARS | ARS | Change % |
| Median Throughput (queries/s) | 95.7866 | 98.537 | 2.8 |
| Median latency (ms) | 1003.29 | 970.15 | -3.3 |
| 90th percentile latency (ms) | 1339.69 | 1326.79 | -0.9 |
| 99th percentile latency (ms) | 1648.34 | 1648.8 | 0.027 |
The non-loaded case makes sense in a system where none of the nodes are under undue stress (high GCs, disk issues, etc). We expected throughput and latency to remain largely the same, which it did. Since a coordinating node is being used for queries, we avoided hotspots from a node having to coordinate any of the search responses.
부하가 없는 사례는 node가 과도한 스트레스(높은 GC, disk 문제 등)를 받지 않는 싯스템에 적당하다. 처리량과 대기시간이 거의 동일하게 유지되기를 기대했고 그러했다. coordinating node는 query에 대해 사용되므로, search response를 조정해야 하는 node의 hotspot을 피했다.
Single node under load:
| Metric | No ARS | ARS | Change % |
| Median Throughput (queries/s) | 41.1558 | 87.8231 | 113.4 |
| Median latency (ms) | 411.721 | 1007.22 | 144.6 |
| 90th percentile latency (ms) | 5215.34 | 1839.46 | -64.7 |
| 99th percentile latency (ms) | 6181.48 | 2433.55 | -60.6 |
In this scenario we simulated load on a node by using the stress command with the parameters stress -i 8 -c 8 -m 8 -d 8 (which runs stress with 8 CPU workers, 8 IO workers, 8 memory workers, and 8 hard drive workers). With ARS there is a large improvement in throughput for the loaded case, as well as a trade-off of 50th percentile latency for a large improvement in 90th and 99th percentile latency. So we were able to route around the degraded node quite well. The 50th percentile latency increase is expected since instead of requests having a "luck-of-the-draw" for whether they go to an unstressed or stressed node, they now avoid the stressed node, slightly increasing load for the unstressed nodes (and thus latency).
이 시나리오에서 stress -i 8 -c 8 -m 8 -d 8 (각각 8개의 CPU worker, IO worker, memory worker, hard drive worker를 가진 stress를 실행한다)라는 매개변수를 가진 stress command를 사용하여, node애 부하를 주는 테스트를 했다. ARS를 사용하면, 부하를 가진 경우에 대해 처리량이 크게 개선되고, 90~99번째 백분위 대기시간이 크게 증가하여 50번째 백분위와 절충된다. 그래서, 성능이 저하된 node 주변으로 잘 route할 수 있다. request가 stress를 받는 node로 갈지 받지 않는 node로 갈지를 운에 맡기는 대신, stress를 받는 node를 피하고 stress를 받지 않는 node에 대한 부하(따라서 대기시간도)를 약간 증가시키기 때문에, 50번째 백분위 대기시간 증가가 예상된다.
Single replica, round-robin requests:
| Metric | No ARS | ARS | Change % |
| Median Throughput (queries/s) | 89.6289 | 95.9452 | 7.0 |
| 50th percentile latency (ms) | 1088.81 | 1013.61 | -6.9 |
| 90th percentile latency (ms) | 1706.07 | 1423.83 | -16.5 |
| 99th percentile latency (ms) | 2481.1 | 1783.73 | -28.1 |
Finally we tested without a coordinating node, sending the requests in a round-robin manner to each node in the cluster without any stress induced. An improvement can be seen in both throughput and latency. So even if the cluster is experiencing even load, ARS can still improve throughput and latency.
마지막으로, coordinating node없이, stress를 유발하지 않고 cluster의 각 node에 round-robin 방식으로 request를 전송하여 테스트했다. 처리량과 대기시간 모두 개선되었다. 따라서, cluster에 부하가 발생하더라도, ARS는 여전히 처리량과 대기시간을 개선할 수 있다.
Conclusion
As you can see, we expect Adaptive Replica Selection to help in many situations. ARS allows the coordinating node to be aware of the load on the data nodes and allows it to choose the best shard copies for executing a search, improving search throughput as well as latency. If you are using Elasticsearch 6.1 or later, please try it out and let us know any feedback you have!
보다시피, Adaptive Replica Selection는 많은 상황에서 도움이 될 것으로 기대한다. ARS를 사용하면, coordinating node는 data node의 부하를 인식할 수 있고, search 실행을 위한 최상의 shard 복사본을 선택하여, 처리량과 대기시간을 향상시킬 수 있다. Elasticsearch 6.1 이후를 사용하고 있다면, 적용해 보자.
원문 : Improving Response Latency in Elasticsearch with Adaptive Replica Selection