Suboptimal shard allocation with imbalanced shards

[bobrik]

Elasticsearch version:

# elasticsearch --version
Version: 2.2.0, Build: 8ff36d1/2016-01-27T13:32:39Z, JVM: 1.8.0_72-internal

JVM version:

# java -version
openjdk version "1.8.0_72-internal"
OpenJDK Runtime Environment (build 1.8.0_72-internal-b15)
OpenJDK 64-Bit Server VM (build 25.72-b15, mixed mode)

OS version: Debian Jessie on kernel 4.1.3.

Description of the problem including expected versus actual behavior:

Elasticsearch does not try to spread shards for new indices equally if one node has less shards than others.

Elasticsearch on on of 12 machines died early in the morning (UTC) and caused several indices go red, even though each index has 1 replica. All of the red indices were today's indices. This is a separate issue, this one is about allocation. I unmounted the bad disk on unlucky node and restarted it, roughly 6 hours after the incident. Indices did not recover.

Since there wasn't much to do at this point, I've dropped red indices to let them refill from kafka. At the same time elasticsearch started recovery procedure to rebalance shards evenly. This happened for new indices:

This is how node list looked like:

It seems like elasticsearch decided to ignore the fact that it's going to rebalance shards anyway and allocated more shards to the empties node. This caused much higher load on this node and made it a bottleneck.

Another issue is that all shards disappeared. I think the following happened:

1. After the incident elasticsearch recovered available indices to have 2 copies on healthy nodes.
2. After bad node rejoined it had the same copies removed since they were redundant.
3. Since bad node has less shards now, it started recovering inactive shards from old indices that were just removed from it.

This looks suboptimal if it's right. This could be #17019 again, since I run optimize and flush every night and it flushed sync happened after the disk failure.

[ywelsch]

The balancer currently has separate steps for:

1. allocating unassigned shards (This happens when a new index is created) and
2. rebalancing shards in the cluster
   This means that if an empty node joins the cluster it will receive a large share of the new shards in step 1 while also receiving shards that come in due to rebalancing in step 2. This results in increased load on that node if the shards in 1) are hot (see also discussion in Shard Allocation Activity Balancing #12279 and solutions discussed). Can you confirm that the newly created shards are indeed hot shards (get more load due to heavy indexing etc.)?

Note that as an initial step towards integrating steps 1 and 2 better, I've recently removed these separate steps on the interface level of the balancer (which were separate methods) so that we have the flexibility to have them collaborate better on the implementation level #17028.

Our solution to the second issue is currently the use of index.unassigned.node_left.delayed_timeout (see https://www.elastic.co/guide/en/elasticsearch/reference/current/delayed-allocation.html). How long was the bad node down? If it is down for a longer time than the timeout AND we have again 2 copies on healthy nodes, we indeed remove the old shard copies from the node even though we might in the future balance one of the copies to that node. Accounting for this specific scenario would add quite some complexity to the balancer. In the presence of throttling, this could also introduce new issues (e.g. wasting space on the new node when cluster is running tight on disk space).

[bobrik]

Can you confirm that the newly created shards are indeed hot shards (get more load due to heavy indexing etc.)?

Yes, 100% ingestion happens on new indices since they are rotated daily.

index.unassigned.node_left.delayed_timeout is set to 20m, node was down for a few hours. I understand that accounting for rebalancing introduces more complexity, but copying terabytes of data when you don't need to copy anything is not something you want to see.

Not sure how you could waste space with throttling. My idea is that you can remove extra copies, but avoid doing that on only one node. Since you remove extra copies anyway, how come you waste more space? It's the opposite: you free space equally on each node instead of freeing it all on one node.

Imagine cluster with 3 nodes and completely idle indices. I kill one node and let the cluster go green again. Once it's green, I start killed node back.

Expected: equal number of extra shards removed from each node, no rebalancing.
Actual: everything is removed from rejoined node and rebalancing kicks in.

Does this make sense?

[bobrik]

To give you an idea how bad it is for indexing, all shards were moved away from the fresh node at 12:30:

Y axis is messages per second consumed from each kafka partition.

[ywelsch]

I understand your suggestion of no-copy balancing. In terms of implementation, we will have to investigate how much complexity this adds. The biggest impact on ingestion here seems to be first issue though (too many hot shards on one node). To address this in the short-term I suggest pre-allocating indices.

[nik9000]

Another work around is to set total_shards_per_node. You have to be careful with it because it is a rule rather than a suggestion, but it is useful. If the hot shards aren't all one index, well, total_shards_per_node will help some, but isn't perfect there. I have on my list to file an issue about getting it (or something like it) to work across indexes.

[DaveCTurner]

@ywelsch and I discussed this and believe a possible way forward would be to add a user-configurable (and dynamically updatable) index setting shard_weight together with an allocation decider that places an upper bound on the total weight of shards that can be allocated to each node.

The expected usage pattern with daily indices would be for new indices to carry a high shard weight, reflecting the fact that they see a lot of indexing traffic while they are "current". Once they cease being current their weight is updated to something much lower, reflecting the fact that each node can support many more non-current indices.

Note that we do not propose balancing shard weights: this is just about putting an upper bound on the total shard weight on each node. This effectively generalises the total_shards_per_node setting to account for indices with different usage patterns, such as the case with daily indices.