Operating a cluster

Elasticsearch is a distributed system you operate, and its two hardest facts are decided before the first document: how many shards an index has, which cannot be changed, and how it will be reindexed when the mapping changes, which it will. Everything else — replicas, sizing, the pressure that indexing and search put on each other, what to watch — follows from those two.

#Shards and replicas

An index is split into primary shards, each a complete Lucene index holding a slice of the documents, spread across the nodes; each primary has zero or more replica shards, copies on other nodes that serve reads and take over when a primary's node is lost. A search fans out to one copy of every shard, each returns its top hits, and the coordinating node merges them — so search latency is the slowest shard's, and the shard count is the parallelism.

PUT /articles-v3
{ "settings": { "number_of_shards": 1, "number_of_replicas": 1 }, "mappings": { ... } }

The number of primaries is fixed at creation (the _split and _shrink APIs exist, with constraints, and are themselves a reindex in disguise). The number of replicas can change any time. A green cluster has every shard and replica allocated; yellow has every primary but a missing replica (one node down, or a single-node cluster with replicas: 1, which can never be green); red has a missing primary and a slice of the data unavailable.

#Sizing: the decisions you cannot change later

Two rules from the operators who run large clusters, and they pull against each other:

• Shards between about 10 and 50 GB, because a shard is the unit of recovery and rebalancing, and moving a 300 GB shard after a node failure takes hours during which the cluster is yellow.
• As few shards as possible, because every shard costs memory (heap for its segment metadata, file handles, a share of the coordination on every search) whether or not it is busy. A cluster with ten thousand tiny shards is slow for no reason the data explains, and the default of five primaries per index in old versions created exactly that; the default is one now.

So: estimate the index's size at the horizon you can see (documents × average size, plus the inverted index, usually 1× to 1.5× the raw data), divide by the target shard size, round up, and for anything under 30 GB use one primary. Time-based data — logs, events — does not size one index at all: it uses an index per day or week (or ILM, index lifecycle management, which rolls over on size or age) so that each index is small, old ones are deleted or moved to cheaper nodes, and the shard count grows with time rather than with a guess. Node-wise, the heap is at most 32 GB (compressed pointers) and at most half the machine's RAM, because Lucene uses the filesystem cache for the other half, and that cache is what makes search fast.

#Aliases: the indirection that makes everything else possible

An alias is a name that points at one or more indices, and clients use the alias, never the index:

POST /_aliases
{ "actions": [ { "remove": { "index": "articles-v2", "alias": "articles" } },
               { "add":    { "index": "articles-v3", "alias": "articles" } } ] }

That swap is atomic: every query on articles before it hit v2, every query after hits v3, and no query saw neither. Set an alias for reads and, for time-based data, a write alias with is_write_index that rollover moves. A client that names a concrete index has coupled itself to the one thing you will need to replace.

#Reindexing without downtime

A mapping cannot change for an existing field, an analyser cannot change for an existing index, and the shard count is fixed — so a real change is a new index and a copy:

1. Create articles-v3 with the new mapping and settings.
2. POST /_reindex { "source": { "index": "articles-v2" }, "dest": { "index": "articles-v3" } }, which reads every document from the old index and writes it to the new one, in batches, resumable, with slices: auto to parallelise. On a large index this runs for hours, and the old index keeps serving through the alias the whole time.
3. Handle the writes that arrive during the copy. Either dual-write from the application to both indices for the duration, or stop writes briefly, reindex the tail (_reindex with a query on updatedAt since the copy began), and switch. The outbox lesson's relay is what a serious system uses here: the search index is a projection of the source of truth, rebuilt from the events, and dual-writing from the application is the dual-write problem in a new costume.
4. Swap the alias, in the one atomic call above.
5. Verify, then delete v2 — not before, because the swap back is the rollback.

Reindexing is routine, and a team that has not done it is a team that will do it for the first time under pressure. Do it once on purpose, with the runbook written, before the mapping change that forces it. The _update_by_query API is the small sibling for a change that needs no new mapping — a computed field, a bulk correction — and it is a rewrite of every matched document, with the segment cost that implies.

#Search versus indexing pressure

Indexing and searching compete for the same nodes. A bulk load of a million documents fills the indexing queues and the CPU with segment writes and merges; searches on the same nodes get slower, and if the bulk is large enough the thread pool queues reject with 429 es_rejected_execution_exception, which the client must treat as backpressure — the backpressure lesson's rule, in a 429. The levers: bulk in batches of a few thousand documents or 5 to 15 MB, not one at a time and not a hundred thousand at once; raise refresh_interval (or set it to -1) during a bulk load so the engine is not making segments searchable every second for data nobody is searching yet, and put it back after; hot-warm architectures with dedicated nodes for the index being written and cheaper nodes for the ones only read; and, at the extreme, separate clusters for ingest and query fed by the same events. Merges are the background cost that surprises people — the force_merge API compacts a read-only index after its load and should never run on one still being written.

#Monitoring: what to watch

• Cluster health (_cluster/health): green/yellow/red, unassigned shards, and the pending tasks count.
• Node stats (_nodes/stats): JVM heap in use — old-generation pressure and long GC pauses are the first sign of too many shards or a too-large aggregation; the filesystem cache hit rate; disk watermarks, because a node past the flood stage (95%) makes every index read-only and that is an outage that looks like a bug.
• Thread pools: search and write queue depths and rejections, which is the pressure section made visible.
• Per-index: indexing rate, search rate, query latency percentiles (the observability course's rule: never the average), refresh and merge times.
• Slow logs: the index and search slow logs with thresholds, which is where the wildcard query and the 10,000-bucket aggregation confess.

Snapshots to object storage on a schedule are the backup, and the backup lesson's rule holds: a restore that has never been run is a hope. Version upgrades are rolling, one node at a time, and the mapping for a major version's breaking changes is the release notes — read before, not after.