Elasticsearch nodes run into OOM during sustained ThreadPoolRejections

[mahdibh]

Elasticsearch version: 1.7.3
JVM version: Java HotSpot(TM) 64-Bit Server VM (build 25.66-b17, mixed mode)
OS version: Linux ip-XXXXXXX.ec2.internal 4.1.13-19.30.amzn1.x86_64 #1 SMP Fri Dec 11 03:42:10 -- UTC 2015 x86_64 x86_64 x86_64 GNU/Linux
Description of the problem including expected versus actual behavior:
3 data nodes (7GB heap), 3 dedicated master nodes.
1 index with 128 primary shards, 2 replicas. Index has 600k documents.
Default configuration, with reduced search thread pool size (from default of 13 to 5 threads). We also changed the search queue size from 1000 to 999 (to verify that we could issue cluster level settings updates during the test).

Steps to reproduce:
We were hitting this issue in production and were able to successfully reproduce it in a staging environment by doing the following:

1. spin off 1000 threads running the following query concurrently (same query across all threads) in a tight loop

POST index_name/object/_search { "size" : 5000, "query" : { "filtered" : { "query" : { "filtered" : { "query" : { "query_string" : { "query" : "(field1:value1 AND NOT field2:_PREFIX_*)", "lowercase_expanded_terms" : false } }, "filter" : { "and" : { "filters" : [ { "term" : { "field3" : "value2" } }, { "term" : { "type" : "MTS" } }, { "or" : { "filters" : [ { "term" : { "active" : true } }, { "and" : { "filters" : [ { "term" : { "active" : false } }, { "range" : { "lastActive" : { "from" : 0, "to" : null, "include_lower" : true, "include_upper" : true }, "_cache" : false } } ] } } ] } } ] } } } }, "filter" : { "missing" : { "field" : "_deletedOnMs", "null_value" : true, "existence" : true } } } }, "version" : false, "_source" : { "includes" : [ ], "excludes" : [ "_props" ] }, "sort" : [ { "createdOnMs" : { "order" : "asc" } } ] }

1. those queries will rapidly fill up the search queue and we get thread pool rejections.
2. after 5mn, all 3 data nodes get into a zombie state (ie, OOM'ed) and get kicked out of the cluster

Provide logs (if relevant):
thread dumps
https://gist.github.com/mahdibh/025d7a909475c43f9154e661c3ef839f
https://gist.github.com/mahdibh/b89a1d437467339a5c30eb416a96cfb5

node log
https://gist.github.com/mahdibh/320b515788400fb1560f2da1b1f2897f

Notes
All nodes in the cluster run into the same issue. We took a heap dump of one of the nodes when it was in this state. 41% of the shallow size of the heap is consumed by byte arrays. long[] comes next with 16% and then char[] with 8%. We could privately share the heap dump if it helps figuring out what's going on.

When this happens, the clusters becomes useless. Only the master nodes respond to API calls. The head plugin shows a blank list of nodes.

We can easily reproduce this internally, if there is anything we can do to provide more details, please let us know.

[clintongormley]

Hi @mahdibh

We have added eg the in-flight requests circuit breaker (#17133) which may or may not help in this situation. It depends exactly where memory is being consumed here. eg the requests circuit breaker is applied during the query phase, but not during the fetch phase and not on the coordinating node. So large _source fields might consume too much memory and result in an OOM.

Giving us access to the heap dump would be helpful. Also, would you mind trying this same test out on 5.0.0-alpha2?

We may be able to find other safeguards (see #11511) that will help us in these situations

[mahdibh]

Hi @clintongormley

We will try to repro this without requesting _source and see if that helps. Our document _source is usually 10kb. Not sure if that actually will matter here since we have only 5 search threads per node.

Would the number of shards (128) coupled with the fact that we are retrieving 5000 records (sorted) contribute to the memory bloat when the system isn't able to keep up ? My understanding is that ES will have to sort and return the top 5000 docs on each shard to the coordinating node which will have to sort across 128x5000=640k docs. Our sort field has doc values enabled and it's a long timestamp.

Also, would the ThreadPoolRejection exceptions cause extra allocations (since there is an error per shard). When the system is overloaded, it's likely all shards will reject and the response object will contain 128 shard failure messages.

I'll follow up regarding the heapdump and trying this out on 5.0.0-alpha2

[mahdibh]

I tried removing the sort in the query, reducing the number of results fetched from 5k to 100, not fetching _source and using no regex in the query but still no luck. I can get the 3 data nodes to OOM in under 3 minutes of sustained load.

I then tried to use the same index but with 8 primary shards instead of 128 (+2 replica). In that case, I couldn't get the cluster to crash (nodes disappeared from the cluster temporarily, but then came back once I stopped the load).

I've also noticed that the cluster held-up more when I disabled the logging of ThreadPoolRejectionExceptions on the ES side PUT _cluster/settings" -d'{"transient" : {"logger.action.search.type" : "ERROR"}}'). Those were logged at a peak rate of 10k/second. No wonder that would consume CPU. I'm not sure why stack traces are logged for those exceptions by default, seems like shooting oneself in the foot when ES is trying to shed itself from further load.

@clintongormley any luck finding anything suspicious in the heap dump ?

[mahdibh]

More data points:
We have upgraded to 1.7.5, but still run into the same issue.

We switched our query traffic from using the data nodes directly to just using client nodes only instead. This shifted the OOM to the client nodes (which is good, since those are way much cheaper to restart). I'll share the heap dump we got on the client nodes.

[s1monw]

Would the number of shards (128) coupled with the fact that we are retrieving 5000 records (sorted) contribute to the memory bloat when the system isn't able to keep up ? My understanding is that ES will have to sort and return the top 5000 docs on each shard to the coordinating node which will have to sort across 128x5000=640k docs. Our sort field has doc values enabled and it's a long timestamp.

may I ask why you have 128 shards on 3 nodes with 600k doc? What about 1 shard? I also wonder why on earth you are fetching 5k documents instead of using pagination or sorted scroll or search_after?

[mahdibh]

may I ask why you have 128 shards on 3 nodes with 600k doc? What about 1 shard?

This is just a load test environment. The number of shards is artificially high on so that we can determine whether that impact the memory issues we are observing. In my third update, I pointed out that we weren't able to repro this on an index with 8 shards.

I also wonder why on earth you are fetching 5k documents instead of using pagination or sorted scroll or search_after?

• Pagination would add latency, although we have never tried that. Would doing that be beneficial ? Latency aside, can you explain why that would be more efficient than retrieving 5k in one shot.
• Sorted scrolls would end up being costly due to the scroll context that has to be maintained.
• search_after isn't available in 1.7 afaik.

[s1monw]

This is just a load test environment. The number of shards is artificially high on so that we can determine whether that impact the memory issues we are observing.

I see - is having 128 shards per node a realistic test? I mean I can put a ton of shards on a node and stuff will likely go bad? Nevertheless, I would be interested in the heapdump of a client node here.

In my third update, I pointed out that we weren't able to repro this on an index with 8 shards.

sorry missed that....

Pagination would add latency, although we have never tried that. Would doing that be beneficial ? Latency aside, can you explain why that would be more efficient than retrieving 5k in one shot.

to begin with, why do you need 5k results. Elasticsearch is a top N retrieval engine in the first place where N is small like 10, 20, 30? I doubt the efficiency here, I think you are overestimating latency given that you are fetching 5k * 10kb of json latency becomes a constant factor given the amount of data you are fetching. Also keep in mind elasticsearch has to materialize all that data before sending back ie. it's not streaming. Working with chunks is the way to go.

Sorted scrolls would end up being costly due to the scroll context that has to be maintained.

what cost you are talking about here? Memory? it holds on to the index reader plus an integer per shard and scroll request. The cost is bascially an int per shard since you need the index reader anyway?

search_after isn't available in 1.7 afaik.

since you are in a testing phase why don't you go and use 2.x give that 5.x is around the corner?

[mahdibh]

One other observation that we made during this is that when the client node runs out of memory, OOM exceptions are swallowed and the node stays up in zombie state holding to its connections to other data nodes. This in turn doesn't release the netty resources on the data nodes leading to a snowball effect that causes the data nodes to also run out of memory.

2016-05-12T18:48:01.533Z WARN  [ent_worker][T#7]{New I/O worker #7}] [annel.socket.nio.AbstractNioSelector]: Unexpected exception in the selector loop.
java.lang.OutOfMemoryError: Java heap space

If you look at netty's AbstractNioSelector.java code, it's catching throwable, logging it and then moving on.
If that exception gets bubbled all the way up to the JVM, the process would exit, or at least we could add an OOM handler in the JVM options so that we can kill the process and release the buffers it's holding into the data nodes.

Note that this pattern is also present in other places in the ES code, for example IndicesTTLService.run(), RestActionListener.onFailure() swallow throwable. I'm not sure if those are done on purpose (in the hope that the OOM was momentary?). It seems to me that the best course of action when handling an OOM is either to release memory or to let the exception go up the stack but not to swallow it.