Balancing primary and replica shards in Elasticsearch

At Deepinfo, a single Elasticsearch index holds more than 400 million domain records across 1,000+ TLDs. Most days the cluster is boring, which is how it should be. When it isn't, the cause is often the same: shards are not where I expected them to be, and one node is doing more than its share.

This post is about how Elasticsearch places primary and replica shards, what it balances and what it ignores, and the handful of settings that actually move things.

Primaries and replicas do almost the same work

Every index is split into primary shards, and each primary can have replicas. The number of primaries is fixed when the index is created (short of split or shrink). The number of replicas can be changed at any time:

PUT domains/_settings
{
  "index": { "number_of_replicas": 1 }
}

A write goes to the primary first. The primary validates it, indexes it and forwards the operation to its replicas, which index it again. So replicas are not passive copies: with Elasticsearch's document replication, a replica spends roughly the same CPU on indexing as its primary. Searches can be served by any copy, and adaptive replica selection sends each one to the copy that is responding fastest.

Two things follow from that:

• A primary and its replica are never placed on the same node. A single-node cluster with number_of_replicas: 1 stays yellow forever.
• A node holding "too many primaries" is usually not the problem it looks like. A node holding too many shards of your busiest index is.

What the allocator balances

The balancer tries to give every node a similar weight. Up to 8.5, that weight came from two things: the total number of shards on the node, and the number of shards it holds from each individual index.

cluster.routing.allocation.balance.shard: 0.45    # total shards per node
cluster.routing.allocation.balance.index: 0.55    # shards of the same index per node
cluster.routing.allocation.balance.threshold: 1.0 # how uneven it may get before shards move

From 8.6 on, the desired balance allocator also weighs disk usage and forecast write load. What it still does not do is treat primaries and replicas differently. As far as rebalancing is concerned, a shard is a shard.

I rarely change these weights. When the distribution looked wrong, the cause has almost always been something else: a disk watermark, an allocation filter, or a shard count that does not divide across the nodes.

Pick shard counts that divide evenly

An index with 5 primaries and 1 replica has 10 shard copies. On 4 nodes, two nodes carry three copies and two carry two. No setting fixes that; the problem is arithmetic. When I create an index, I choose the primary count so that primaries × (replicas + 1) divides evenly by the number of data nodes, and I leave room for the cluster to grow.

For a busy index, a hard cap per node helps too:

PUT domains/_settings
{
  "index.routing.allocation.total_shards_per_node": 3
}

Be careful with it. The cap still applies when a node fails, so a cap that is too tight leaves shards unassigned exactly when they need to move.

Settings that decide when shards move

These are the ones I actually touch:

Raising the concurrency makes the cluster settle faster and makes searches slower while it does. During the day I keep the defaults. For a planned rebalance at night I raise them, then put them back.

Disk watermarks deserve their own mention, because they move shards without asking anyone:

• low (85%): no new shards are allocated to the node.
• high (90%): Elasticsearch starts moving shards off the node.
• flood_stage (95%): every index with a shard on the node becomes read-only (index.blocks.read_only_allow_delete).

If one node fills up faster than the others, the allocator keeps pushing shards away from it, and the "imbalance" you see is really a disk problem.

Restarting a node without a reshuffle

When a node leaves, Elasticsearch waits for index.unassigned.node_left.delayed_timeout (one minute by default) before it rebuilds the missing replicas elsewhere. If a restart takes longer than that, the data is copied in full, and the shards move back again when the node returns. For a rolling restart I do this:

# Before stopping the node: allocate primaries only
curl -X PUT "$ES/_cluster/settings" -H 'Content-Type: application/json' -d '{
  "persistent": { "cluster.routing.allocation.enable": "primaries" }
}'

# ...restart the node and wait for it to rejoin...

# Afterwards: back to the default
curl -X PUT "$ES/_cluster/settings" -H 'Content-Type: application/json' -d '{
  "persistent": { "cluster.routing.allocation.enable": null }
}'

Then wait for green before moving on to the next node.

Finding out where things are

Three requests answer most questions:

# Shard copies per node, split into primaries (p) and replicas (r)
curl -s "$ES/_cat/shards?h=node,prirep" | sort | uniq -c

# Disk usage and shard count per node
curl -s "$ES/_cat/allocation?v"

# Why a shard is, or is not, where it is
curl -s "$ES/_cluster/allocation/explain?pretty" -H 'Content-Type: application/json' -d '{
  "index": "domains", "shard": 0, "primary": false
}'

allocation/explain is the one I wish I had learned first. Instead of guessing, you get every allocation decider's yes or no, node by node.

What I check, in order

1. Does primaries × (replicas + 1) divide evenly by the number of data nodes?
2. Is any node close to the disk watermarks?
3. Is a busy index concentrated on one node? If so, total_shards_per_node.
4. What does allocation/explain say about the shard that looks misplaced?
5. Only then: the balance weights.