Elasticsearch Indexing: Designing Shard Allocations

Elasticsearch is a distributed search and analytics engine built on Apache Lucene. Horizontal scalability in Elasticsearch is achieved by partitioning indexes into smaller units called shards. Each shard is a fully functional, self-contained instance of Apache Lucene.

When configuring a cluster, developers must choose the number of primary shards and replica shards for each index. The choices made directly impact write performance, query latencies, index recovery times, and cluster-wide resource utilization. This guide details how to design shard allocations, implement index mapping controls, write resilient integrations using the Go Elasticsearch client, and remediate critical production failure states.

---

The Mechanics of Sharding and Search Execution

In Elasticsearch, an index is a logical namespace pointing to one or more physical shards.

• Primary Shards: All documents indexed are routed to a primary shard. The routing formula is:

  shard = hash(routing) % number_of_primary_shards
  

  By default, routing is the document’s _id. The number of primary shards must be defined when the index is created and cannot be altered without reindexing.
• Replica Shards: Replicas are copies of primary shards. They serve two main purposes: high availability (preventing data loss if a node fails) and scaling read query throughput. Unlike primary shards, the replica count can be modified dynamically at any time.

                  [Coordinator Node]
                     /          \
            (Route Search)    (Route Search)
                   /              \
       [Node 1: Shard 0 (P)]     [Node 2: Shard 0 (R)]
       [Node 1: Shard 1 (R)]     [Node 2: Shard 1 (P)]

When a search query is executed, the node that receives the request (the coordinator node) broadcasts the query to a copy (either primary or replica) of every shard in the index. Each shard executes the search locally, building a priority queue of matching documents. The coordinator collects, merges, and sorts these local queues (the scatter-gather process) and returns the top results to the client.

Finding the Sweet Spot for Shard Sizes

A common mistake is creating too many small shards. Every shard consumes system resources, including JVM heap memory, open file descriptors, and CPU cycles. Lucene segments store term dictionaries and bloom filters in RAM; if you have thousands of shards containing only a few megabytes of data, your master nodes will suffer from high heap pressure, slowing down cluster metadata updates.

For optimal performance, follow these guidelines:

• Shard Size Limit: Maintain physical shard sizes between 10GB and 50GB.
• Heap-to-Shard Ratio: Keep the total number of active shards below 20 shards per GB of configured JVM heap on each data node.
• Time-Series Roll: For log and event ingestion, use the Rollover API or Index Lifecycle Management (ILM) to roll over indexes when they reach a target size (e.g., 30GB) rather than relying on strict daily calendars.

---

Defining Strict Index Mappings and Settings

By default, Elasticsearch uses dynamic mapping, automatically creating field mappings when a document contains a new field. In production, this behavior can cause type conflicts and bloated cluster states. Instead, always define explicit mappings and set dynamic to strict.

The following payload defines an index mapping for a system log collector with custom analysis filters, three primary shards, and one replica.

{
  "settings": {
    "index": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "30s",
      "codec": "best_compression"
    },
    "analysis": {
      "analyzer": {
        "custom_log_analyzer": {
          "type": "custom",
          "tokenizer": "whitespace",
          "filter": ["lowercase", "stop"]
        }
      }
    }
  },
  "mappings": {
    "dynamic": "strict",
    "properties": {
      "trace_id": { "type": "keyword" },
      "service_name": { "type": "keyword" },
      "timestamp": { "type": "date" },
      "log_level": { "type": "keyword" },
      "message": { 
        "type": "text",
        "analyzer": "custom_log_analyzer"
      },
      "http_status": { "type": "integer" },
      "execution_ms": { "type": "double" }
    }
  }
}

---

Go Client Integration & Custom Search Execution

The Go client must manage index creation, index documents safely, and run search queries using parameterized search JSON buffers. The following complete Go program connects to Elasticsearch, ensures the index exists with correct configurations, indexes a document, and performs a structured search query.

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"strings"
	"time"

	"github.com/elastic/go-elasticsearch/v8"
	"github.com/elastic/go-elasticsearch/v8/esapi"
)

// LogDocument represents the data structure of logs sent to Elasticsearch.
type LogDocument struct {
	TraceID     string    `json:"trace_id"`
	ServiceName string    `json:"service_name"`
	Timestamp   time.Time `json:"timestamp"`
	LogLevel    string    `json:"log_level"`
	Message     string    `json:"message"`
	HTTPStatus  int       `json:"http_status"`
	ExecutionMS float64   `json:"execution_ms"`
}

// SearchResponse represents the top-level elasticsearch search result structure.
type SearchResponse struct {
	Hits struct {
		Total struct {
			Value int `json:"value"`
		} `json:"total"`
		Hits []struct {
			Source json.RawMessage `json:"_source"`
		} `json:"hits"`
	} `json:"hits"`
}

// ESAdapter wraps the official elasticsearch client.
type ESAdapter struct {
	Client *elasticsearch.Client
}

// NewESAdapter initializes the Elasticsearch client configuration.
func NewESAdapter(addresses []string) (*ESAdapter, error) {
	cfg := elasticsearch.Config{
		Addresses: addresses,
		RetryOnStatus: []int{502, 503, 504},
		MaxRetries:    3,
		RetryBackoff: func(attempt int) time.Duration {
			return time.Duration(attempt) * 100 * time.Millisecond
		},
	}
	client, err := elasticsearch.NewClient(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to instantiate ES client: %w", err)
	}
	return &ESAdapter{Client: client}, nil
}

// CreateIndexWithMappings creates the index if it does not already exist.
func (es *ESAdapter) CreateIndexWithMappings(ctx context.Context, indexName string, mappingsJSON string) error {
	req := esapi.IndicesExistsRequest{
		Index: []string{indexName},
	}
	res, err := req.Do(ctx, es.Client)
	if err != nil {
		return fmt.Errorf("error verifying index existence: %w", err)
	}
	defer res.Body.Close()

	if res.StatusCode == 200 {
		return nil // Index already exists
	}

	createReq := esapi.IndicesCreateRequest{
		Index: indexName,
		Body:  strings.NewReader(mappingsJSON),
	}
	createRes, err := createReq.Do(ctx, es.Client)
	if err != nil {
		return fmt.Errorf("error creating index: %w", err)
	}
	defer createRes.Body.Close()

	if createRes.IsError() {
		body, _ := io.ReadAll(createRes.Body)
		return fmt.Errorf("failed index creation API call: %s", string(body))
	}

	return nil
}

// IndexLog serializes and writes a log document to Elasticsearch.
func (es *ESAdapter) IndexLog(ctx context.Context, indexName string, docID string, doc LogDocument) error {
	data, err := json.Marshal(doc)
	if err != nil {
		return fmt.Errorf("failed to marshal document: %w", err)
	}

	req := esapi.IndexRequest{
		Index:      indexName,
		DocumentID: docID,
		Body:       bytes.NewReader(data),
		Refresh:    "wait_for", // block until the document is searchable
	}

	res, err := req.Do(ctx, es.Client)
	if err != nil {
		return fmt.Errorf("failed to index document: %w", err)
	}
	defer res.Body.Close()

	if res.IsError() {
		body, _ := io.ReadAll(res.Body)
		return fmt.Errorf("failed indexing API call: %s", string(body))
	}

	return nil
}

// SearchLogs executes a query to locate matching log messages.
func (es *ESAdapter) SearchLogs(ctx context.Context, indexName string, serviceName string, minExecutionMS float64) (*SearchResponse, error) {
	queryMap := map[string]interface{}{
		"query": map[string]interface{}{
			"bool": map[string]interface{}{
				"must": []map[string]interface{}{
					{
						"term": map[string]interface{}{
							"service_name": serviceName,
						},
					},
					{
						"range": map[string]interface{}{
							"execution_ms": map[string]interface{}{
								"gte": minExecutionMS,
							},
						},
					},
				},
			},
		},
	}

	queryJSON, err := json.Marshal(queryMap)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal query map: %w", err)
	}

	req := esapi.SearchRequest{
		Index: []string{indexName},
		Body:  bytes.NewReader(queryJSON),
	}

	res, err := req.Do(ctx, es.Client)
	if err != nil {
		return nil, fmt.Errorf("search API invocation failed: %w", err)
	}
	defer res.Body.Close()

	if res.IsError() {
		body, _ := io.ReadAll(res.Body)
		return nil, fmt.Errorf("failed search API: %s", string(body))
	}

	var searchRes SearchResponse
	if err := json.NewDecoder(res.Body).Decode(&searchRes); err != nil {
		return nil, fmt.Errorf("failed to parse search response: %w", err)
	}

	return &searchRes, nil
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	addresses := []string{"http://10.0.2.15:9200"}
	adapter, err := NewESAdapter(addresses)
	if err != nil {
		log.Fatalf("ES init failed: %v", err)
	}

	mappings := `{
		"settings": {
			"index": {
				"number_of_shards": 2,
				"number_of_replicas": 1
			}
		},
		"mappings": {
			"dynamic": "strict",
			"properties": {
				"trace_id": { "type": "keyword" },
				"service_name": { "type": "keyword" },
				"timestamp": { "type": "date" },
				"log_level": { "type": "keyword" },
				"message": { "type": "text" },
				"http_status": { "type": "integer" },
				"execution_ms": { "type": "double" }
			}
		}
	}`

	indexName := "application-logs"
	err = adapter.CreateIndexWithMappings(ctx, indexName, mappings)
	if err != nil {
		log.Fatalf("Could not setup index: %v", err)
	}

	doc := LogDocument{
		TraceID:     "tx_78103-bc92",
		ServiceName: "user-authentication",
		Timestamp:   time.Now(),
		LogLevel:    "INFO",
		Message:     "User authenticated successfully via oauth2 provider",
		HTTPStatus:  200,
		ExecutionMS: 45.12,
	}

	err = adapter.IndexLog(ctx, indexName, "log_1001923", doc)
	if err != nil {
		log.Printf("Failed to index log: %v", err)
	} else {
		fmt.Println("Log document successfully indexed and committed.")
	}

	results, err := adapter.SearchLogs(ctx, indexName, "user-authentication", 10.0)
	if err != nil {
		log.Fatalf("Search failed: %v", err)
	}

	fmt.Printf("Search query matched %d total logs.\n", results.Hits.Total.Value)
}

---

Metrics Comparison Table

The metrics below compare different sharding strategies for an identical 100GB dataset consisting of 200,000,000 document records, run on a 3-node Elasticsearch cluster.

Performance Vector	Under-Sharded (1 Shard)	Optimized Sharding (3 Shards)	Over-Sharded (100 Shards)	Operational Takeaway
Physical Shard Size	100 GB	~ 33.3 GB	1 GB	Keep shards < 50GB to maintain recovery speed.
Max Indexing Rate (docs/sec)	12,500 docs/sec	34,800 docs/sec	8,900 docs/sec	Small shards generate high segment merge contention.
Search Query Latency (p95)	620 ms	48 ms	280 ms	High shard counts force costly scatter-gather loops.
Master Node Heap Utilization	12%	15%	78%	Excess shards bloat cluster state metadata in RAM.
Shard Recovery Time (Node Failure)	3.5 Hours	14 Minutes	48 Minutes	Large shards cause disk bottle-necks during recovery.
Term Dictionary Memory Footprint	Low (Optimized Lucene segments)	Low (Optimized Lucene segments)	Extreme (Dozens of open segment structures)	Avoid high overhead by rolling indices dynamically.

---

What Breaks in Production

Operating an Elasticsearch cluster at scale requires monitoring physical thresholds. Below are key failure states, symptoms, and remediation methods.

1. Over-Sharding and Cluster Status “RED”

When a cluster accumulates thousands of shards, the master node must manage metadata for each segment. This leads to heap exhaustion, slow cluster operations, and node crashes.

• Symptoms: Cluster status is RED or YELLOW, master nodes fail with JVM OutOfMemoryError, and the cluster is slow to react to changes (e.g. index creations or node additions are delayed).
• Remediation:
  1. Determine shard counts per node using:

     curl -X GET "localhost:9200/_cat/shards?v=true"
     

  2. Implement the Shrink API to reduce the primary shard count of historical, read-only indexes.
  3. Clean up older indexes using Index Lifecycle Management (ILM) policies to merge daily indexes into larger weekly/monthly indexes, or delete expired data.
  4. Force-merge segments of read-only indexes using:

     curl -X POST "localhost:9200/my-old-index/_forcemerge?max_num_segments=1"
     

2. Index Mapping Conflicts and Ingestion Failures

When dynamic mapping is enabled, Elasticsearch may map a field to a type based on the first document it indexes. If subsequent documents contain a different type, those documents are rejected.

• Symptoms: Elasticsearch returns a 400 Bad Request containing: mapper_parsing_exception: object mapping for [user_info] tried to parse field [user_info] as object, but found a concrete value Ingestion queues (e.g. Logstash or Vector) back up, causing data loss or delivery delays.
• Remediation:
  1. Define explicit index templates that enforce mappings for all indexes matching a specific pattern.
  2. Set "dynamic": "strict" in mappings to fail fast and prevent corrupt fields from altering the schema.
  3. Use the Reindex API to copy data from the old index with conflicts to a newly created index with corrected mappings:

     POST /_reindex
     {
       "source": { "index": "logs-old" },
       "dest": { "index": "logs-new" }
     }
     

3. Disk Watermark Violations and Write Blocks

Elasticsearch monitors disk space on data nodes. If disk space runs low, it applies safety locks to prevent filesystem corruption.

• Symptoms: All writes to the index fail, and Elasticsearch returns: cluster_block_exception: blocked by: [FORBIDDEN/12/index read-only / allow delete] Logs report that the low, high, or flood-stage disk watermarks have been exceeded.
• Remediation:
  1. By default, the high watermark is 90% and flood-stage is 95%. Add disk space or scale cluster nodes.
  2. Clear the write lock manually once space is cleared:

     PUT /_all/_settings
     {
       "index.blocks.read_only_allow_delete": null
     }
     

  3. Adjust transient watermarks in emergencies (use with caution):

     PUT /_cluster/settings
     {
       "transient": {
         "cluster.routing.allocation.disk.watermark.low": "85%",
         "cluster.routing.allocation.disk.watermark.high": "90%",
         "cluster.routing.allocation.disk.watermark.flood_stage": "92%"
       }
     }
     

---

FAQs

What is the primary benefit of this design pattern?

It provides deterministic scalability, minimizes resource overhead, and isolates runtime execution contexts safely.

How do we verify the performance improvements?

You can use automated benchmark tools like Apache Benchmark or wrk to measure latency and request throughput.