Util: Fix for writeMeta in BulkIndexer helper

This fix adds '_index' to meta when BulkIndexerItem.Index is not empty.

Currently, this value was ignored which leads to index errors when the
default index in BulkIndexerConfig.Index is empty.

Closes #146

(cherry picked from commit fc7f75e)

Author: kiivihal

esutil/bulk_indexer.go

@@ -393,6 +393,15 @@ func (w *worker) writeMeta(item BulkIndexerItem) error {
 		w.buf.Write(w.aux)
 		w.aux = w.aux[:0]
 	}
+	if item.Index != "" {
+		if item.DocumentID != "" {
+			w.buf.WriteRune(',')
+		}
+		w.buf.WriteString(`"_index":`)
+		w.aux = strconv.AppendQuote(w.aux, item.Index)
+		w.buf.Write(w.aux)
+		w.aux = w.aux[:0]
+	}
 	w.buf.WriteRune('}')
 	w.buf.WriteRune('}')
 	w.buf.WriteRune('\n')

esutil/bulk_indexer_integration_test.go

@@ -22,6 +22,8 @@ import (
 )
 
 func TestBulkIndexerIntegration(t *testing.T) {
+	body := `{"body":"Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat."}`
+
 	t.Run("Default", func(t *testing.T) {
 		var countSuccessful uint64
 		indexName := "test-bulk-integration"
@@ -43,7 +45,6 @@ func TestBulkIndexerIntegration(t *testing.T) {
 		})
 
 		numItems := 100000
-		body := `{"body":"Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat."}`
 		start := time.Now().UTC()
 
 		for i := 1; i <= numItems; i++ {
@@ -90,4 +91,69 @@ func TestBulkIndexerIntegration(t *testing.T) {
 			time.Since(start).Truncate(time.Millisecond),
 			1000.0/float64(time.Since(start)/time.Millisecond)*float64(stats.NumFlushed))
 	})
+
+	t.Run("Multiple indices", func(t *testing.T) {
+		es, _ := elasticsearch.NewClient(elasticsearch.Config{
+			Logger: &estransport.ColorLogger{Output: os.Stdout},
+		})
+
+		bi, _ := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
+			Index:  "test-index-a",
+			Client: es,
+		})
+
+		// Default index
+		for i := 1; i <= 10; i++ {
+			err := bi.Add(context.Background(), esutil.BulkIndexerItem{
+				Action:     "index",
+				DocumentID: strconv.Itoa(i),
+				Body:       strings.NewReader(body),
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+
+		// Index 1
+		for i := 1; i <= 10; i++ {
+			err := bi.Add(context.Background(), esutil.BulkIndexerItem{
+				Action: "index",
+				Index:  "test-index-b",
+				Body:   strings.NewReader(body),
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+
+		// Index 2
+		for i := 1; i <= 10; i++ {
+			err := bi.Add(context.Background(), esutil.BulkIndexerItem{
+				Action: "index",
+				Index:  "test-index-c",
+				Body:   strings.NewReader(body),
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+
+		if err := bi.Close(context.Background()); err != nil {
+			t.Errorf("Unexpected error: %s", err)
+		}
+		stats := bi.Stats()
+
+		expectedIndexed := 10 + 10 + 10
+		if stats.NumIndexed != uint64(expectedIndexed) {
+			t.Errorf("Unexpected NumIndexed: want=%d, got=%d", expectedIndexed, stats.NumIndexed)
+		}
+
+		res, err := es.Indices.Exists([]string{"test-index-a", "test-index-b", "test-index-c"})
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		if res.StatusCode != 200 {
+			t.Errorf("Expected indices to exist, but got a [%s] response", res.Status())
+		}
+	})
 }

esutil/bulk_indexer_internal_test.go

@@ -544,6 +544,66 @@ func TestBulkIndexer(t *testing.T) {
 			t.Errorf("Unexpected NumAdded: %d", stats.NumAdded)
 		}
 	})
+
+	t.Run("Worker.writeMeta()", func(t *testing.T) {
+		type args struct {
+			item BulkIndexerItem
+		}
+		tests := []struct {
+			name string
+			args args
+			want string
+		}{
+			{
+				"without _index and _id",
+				args{BulkIndexerItem{Action: "index"}},
+				`{"index":{}}` + "\n",
+			},
+			{
+				"with _id",
+				args{BulkIndexerItem{
+					Action:     "index",
+					DocumentID: "42",
+				}},
+				`{"index":{"_id":"42"}}` + "\n",
+			},
+			{
+				"with _index",
+				args{BulkIndexerItem{
+					Action: "index",
+					Index:  "test",
+				}},
+				`{"index":{"_index":"test"}}` + "\n",
+			},
+			{
+				"with _index and _id",
+				args{BulkIndexerItem{
+					Action:     "index",
+					DocumentID: "42",
+					Index:      "test",
+				}},
+				`{"index":{"_id":"42","_index":"test"}}` + "\n",
+			},
+		}
+		for _, tt := range tests {
+			tt := tt
+
+			t.Run(tt.name, func(t *testing.T) {
+				w := &worker{
+					buf: bytes.NewBuffer(make([]byte, 0, 5e+6)),
+					aux: make([]byte, 0, 512),
+				}
+				if err := w.writeMeta(tt.args.item); err != nil {
+					t.Errorf("Unexpected error: %v", err)
+				}
+
+				if w.buf.String() != tt.want {
+					t.Errorf("worker.writeMeta() %s = got [%s], want [%s]", tt.name, w.buf.String(), tt.want)
+				}
+
+			})
+		}
+	})
 }
 
 type customJSONDecoder struct{}