Slow recovery of write availability after partition of a large cluster

[djjsindy]

We have a very large cluster which have 128 nodes. This cluster have a lot of index. There are about 20,000 shards, 10000 shards is primary,the other is replica. Primary and replica locate in different racks. Write operation will always exist. In the network partition scenario the write operation will be blocked because it has to wait for replica shard failed cluster state commit. Write operation recovery time will be longer than about 10 minutes.

My opinion: Write slow recovery affected by the following three factors:

1. Each node disconnect detection is independent. In the network partition scenario, 64 nodes disconnect. Because cluster state batch processing mechanism led to the first cluster state only the first node disconnect. This cluster state's prepare and commit must be time-out,Because this cluster state sent the node contains the remaining 63 nodes.
2. Too many shard failed lead to task summary toString time is very long,20000 shard failed Calculating the task summary takes about 15 seconds.
3. Same shard, the same primary term, the same allocationId shard failed request processing did not remove the duplicate request , ShardEntry does not override equals and hashCode methods.

In my scenario, I tried to do optimization based on the above mentioned. Write recovery time reduced from 10 minutes to less than 1 minute, It seems to be working.

Please take a look at these three factors can be improved ？

Elasticsearch version (bin/elasticsearch --version):
5.3.1
Plugins installed: []

JVM version (java -version):
1.8.0_112
OS version (uname -a if on a Unix-like system):
2.6.32-220.23.2.xxxxx.el6.x86_64
Description of the problem including expected versus actual behavior:
Write operation will always exist. In the network partition scenario the write operation will be blocked because it has to wait for shard failed cluster state commit. Write operation recovery time will be longer than about 10 minutes.
Expected behavior: Recovery write time is shorter
Steps to reproduce:

my config：
cluster.routing.allocation.awareness.attributes: rack_id
node.attr.rack_id: xxxxx
discovery.zen.minimum_master_nodes: 2
discovery.zen.ping.unicast.hosts:
     - xxxxx:9300
     - xxxxx:9300
     - xxxxx:9300
cluster.routing.allocation.awareness.force.rack_id: xxxx

network partition opertion: 
sudo iptables -D INPUT 1 ;sudo iptables -D OUTPUT 1 ;sudo iptables -L -n

Provide logs (if relevant):

[spinscale]

@elastic/es-distributed can you please take a look if this is valid or might have changed with 6.0?

[DaveCTurner]

I'd expect the recovery of a partitioned 10,000 shard cluster to be faster in 6.x, but not for the reasons described here. Please could you provide the logs from a decent proportion of the nodes (≥ 5 from each side of the partition, including a selection of both master and data nodes) so we can see what was actually happening for those 10 minutes. I can provide an email address if you struggle to attach then here.

Also please describe the "optimisations" performed that reduced this from 10 minutes down to 1.

[DaveCTurner]

Also 5.3.1 is approaching a year old and missing a few fixes that might have an impact on this. Is it feasible to upgrade to a more recent version and retry your experiments? Ideally the latest possible version (currently 6.2.2) but it'd be useful to look at the latest 5.x version (currently 5.6.8) if that's not possible.

[dnhatn]

Same shard, the same primary term, the same allocationId shard failed request processing did not remove the duplicate request , ShardEntry does not override equals and hashCode methods.

@ywelsch and I discussed this issue. We will work on an improvement which avoids failing the same shard multiple times.

[djjsindy]

@spinscale @DaveCTurner @dnhatn
Could you provide your email address ?
I will send it to you some log and detailed analysis

[DaveCTurner]

@djjsindy please use david.turner@elastic.co and I will share with the rest of the team. I expect we will also want to summarise and discuss any analysis here, which might include things like index names, shard counts, and so on. If there is any information like this that you want not to be shared in public then please indicate as such in your email.

[DaveCTurner]

@djjsindy I received your email but it only contained a handful of log messages, showing nothing unexpected. I responded:

Please could we have copies of full logs for the full duration of the outage so we can see what was taking so long to recover. The lines that you quote in your email show nothing unexpectedly slow, but only cover the very start of the outage. Ideally logs from all of the master nodes and some of the data nodes. I note that you have TRACE logging enabled for the ClusterService which is great - the logs will be quite large but this extra detail will help us a lot.

Copying this here in case you didn't get my response.

Additionally, we just pushed 90bde12 (5.6), e0da114 (6.x), and 033a83b (master) which should make a difference to how fast the task summary toString() method takes. If you're able, please can you try this change out and let us know if it helps?

[DaveCTurner]

Thanks for the logs. The issue you are facing is as follows. The master detects the failure of some of the nodes (but, crucially, not all of them) and publishes a cluster state update to remove them. When committing a cluster state update, each receiving node attempts to establish connections to all the nodes that are listed in the new cluster state, which includes all the failed nodes that the master has not yet detected as failed. Each such connection attempt times out after 30 seconds because of the network partition. In 5.3.x it looks like these attempts are made in sequence; #22984 (released in v5.4.0) improves this a little so they happen up to 5-at-a-time, but never to better than 30 seconds. Once all the connection attempts have timed out, the node can finish applying the cluster state update.

Meanwhile, the master node has detected the failure of the rest of the nodes and publishes the next update. Once the other nodes have finished failing to connect to all their disconnected peers and applied the first update, the second update seems to be applied reasonably quickly.

This explains why waiting before sending the first cluster state update improves the situation dramatically: if the first cluster state update is delayed for long enough to capture all of the failed nodes then its recipients do not waste any time trying to connect to their failed peers and can apply it reasonably quickly.

However, this is unlikely to be the solution we choose: it means that Elasticsearch will block writes for the defined waiting time on the failure of just a single node. We need to think about this more deeply.