Auto Queue Resizing & Adaptive Replica Selection

For the automatic queue resizing:

• https://github.com/elastic/elasticsearch/issues/3890
• https://github.com/elastic/elasticsearch/pull/23884

For adaptive replica selection:

• https://github.com/elastic/elasticsearch/issues/24915
• https://github.com/elastic/elasticsearch/pull/24989
• https://github.com/elastic/elasticsearch/pull/25430
• https://github.com/elastic/elasticsearch/pull/26128

The implementation is based on Little's Law, a simple formula

L = λW

Where L is the long term average number of requests, λ is the arrival rate of new requests, and W is the average time a request spends being executed.

In Elasticsearch, there's a new threadpool type called fixed_auto_queue_size. To use it, the use configures a target "response rate" they want the average response time be. So for instance, if a user configures thread_pool.search.target_response_rate: 4s to a targeted response rate of four seconds.

Now, assume the node is serving search requests at a rate of 15 reqs/s, so there can be 15 * 4 = 60 total requests in the queue at any particular time in order for . Elasticsearch can then adjust the queue size towards this limit. If requests start to be served at a faster rate (the node is recovering) then the queue can be increased, or if the node cannot serve them at 15 requests a second, the queue can be lowered more. The minimum and maximum queue size can also be configured. In Elasticsearch 6.0 the minimum and maximum are both configured to the initial size of 1000.

By default Elasticsearch can adjust the thread pool every thread_pool.search.auto_queue_frame_size tasks, which defaults to 2000.

With a single node stressed¹ in a 5-node cluster, this is the measured impact for average response time in milliseconds.

This in particular helps with a particular customer who had a cluster where one node would constantly be "hot" due to unevenly distributed data, or degraded hardware. In addition, rather than having a node keel over due to being inundated with requests, a node can start rejecting them earlier.

Full benchmarks and write-up

The ARS formula is a little more complicated:

Ψ(s) = R(s) - 1/µ̄(s) + (q̂(s))^b / µ̄(s)

Where q̂(s) is:

q̂(s) = 1 + (os(s) * n) + q(s)

• os(s) is the number of outstanding search requests to a node
• n is the number of clients (data nodes) in the system
• R(s) is the EWMA² of the response time (as seen from the coordinating node) in milliseconds
• q(s) is the EWMA of the number of events in the thread pool queue
• µ̄(s) is the EWMA of service time of search events on the data node in milliseconds

For the moving averages, we use an α of 0.3 and for the formula we use a b value of 3 as recommended in the C3 paper.

This was implemented in multiple PRs since it was a little more invasive, what happens is that basically with every query, we piggy-back information back to the coordinating node about the current queue size and EWMA service time for tasks.

Assume you perform a query for an index that has data distributed on three different nodes. Elasticsearch scatters the query out to the data nodes. In this case, because we don't have any ranking information, all nodes are considered equal and we fall back to the regular round-robin of requests.

And along with the results, each node will return its current queue size and EWMA service time:

In addition, the coordinating node measures search requests for the EWMA of the response time of each node, keeping track of it locally. These numbers are then used for the next request, ranking each copy of the data to determine the best place to send the request. For instance, for three copies of the data on nodes with scores of 128, 108, and 113 respectively, the shard copy on node "data2" is chosen to serve the request.

After the routing is determined, the adaptive replica selection also adjusts the non-winning nodes' stats to gradually bring them towards the winning node's stats³. In doing this, we prevent a situation where a long GC that temporarily drastically increases a node's score causes the node not to receive any more requests ever.

This was tested with three "main" scenarios, a non-loaded and loaded setup with a client node handling all coordination, and a non-loaded cluster with requests sent in a round-robin manner.

Single replica, non-loaded case:

So again, not a huge latency difference, as expected for the unloaded cluster.

Single replica, es3 under load:

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. When a node is loaded we were able to route around the degraded node quite well.

Single replica, round robin requests:

Again a nice improvement in both throughput and latency for the non-stressed round-robin test case. So even if the cluster is experiencing even load, it can still help with throughput and latency.

Full benchmarks