Elasticsearch (ES) is gaining popularity not only as a search engine but as a document store and analytics platform as well. It is one of the top choices for indexing and querying large structured and unstructured data sets due to the ease of implementation, performance and having a developer-friendly approach.
ES is built to scale dynamically, and it is very easy to add additional resources to an existing cluster. And hence, it’s quite common to see clusters grow to 100s or 1000s of nodes during the life of an application.

The Architecture of an Elasticsearch Cluster

Shards: The smallest unit of scale to design the layout and systematics of storing and distributing data. It is a Lucene index.

ES Index with three shards with three nodes cluster

It is a Lucene Apache Open Source project which is used as a high-performance search engine. The ES Index is the sum of all the shards. When an application queries an ES Index, the request will be routed to each one of those shards by Elasticsearch, making the process of data retrieval faster. However, number of shards need to be carefully considered while allocating an index. Each shard is processed by an individual thread that is taken out of a node local thread pool. Having shards distributed across nodes is fine in terms of parallelism but having too many shards on a single node can have a negative effect on the cluster’s performance. This is because with higher load, the thread pool can get a serious bottleneck effect.
Furthermore, when starting a new index, the associated shards cannot be changed without reindexing all the documents to a new, more organized index.
Replicas: A replica shard is a copy of any given primary shard. Documents are first indexed to one of the primary shards and then forwarded to the corresponding replica. The main reason for having replica shards is resilience. When a node holding a primary shard goes offline, one of its replicas will be promoted to primary.

Distribution of replica shards

When one of the three nodes fail, the data it holds is still available on one of the other nodes. However, resilience by replica has its limits. When data becomes unavailable, ES is unable to recover it from the available replicas and the index will not respond anymore, because at least one of its primary shard is unavailable.
The second reason for replica shards is to serve read requests. When using a search-heavy index, replica shards can boost performance.
From a monitoring perspective, the availability and usage of shards are important elements to observe. Unassigned primary shards in your cluster put your data at risk, as ES routes documents to shards by a hash. When the shard which a document should be routed to is not available, the document won’t be indexed.
Indices: Elasticsearch offers the capability to create an index by just indexing a document. This is very convenient. When creating an index based on a document, ES will create an automatic mapping. Mapping lets you control which fields get indexed and, even more importantly, which fields will not get indexed.
Another very important thing to consider is the layout of the index. Often, one needs to add fresh data and delete the obsolete data. But it should be clear that deleting single documents is a very expensive operation. This is true for updates as well, as an update is nothing more than deleting and reindexing a given document. When working with such time-based data, another option is to create a new index for a time window (i.e. per day) and to delete the obsolete indexes. Deleting a whole index is cheap. It’s deleting the Lucene files associated with the shards of the index.

Monitoring an Elasticsearch Cluster

The first thing in ES cluster is about is its overall performance in terms of response times and throughput of queries. The first step is to look at which threads are running in the cluster. Some threads might be queued up, or even rejected. Rejections point to a problem which deserves further inspection and analysis. One can analyse threads and latency not for the cluster, but also for each individual node. This is because there might be a problem with shard-routing and therefore issues may only be visible on specific nodes.

More metrics to thoroughly look at are on the OS and JVM level. The heap consumption evolves over time for the JVM. Monitor GC running, and time taken. CPU utilization per application as well.

There is a monitoring tool inbuilt in Kibana, where the user can see and interact with every component. These components are being monitored all the way from the host, to a runtime environment, to a specific piece of software or technology. This brings you a lot closer to finding the actual reason for issues in your system, while spending less time on looking at metrics.