Merge pull request #271 from multiversx/get-write-index-for-delete-by…

…-query

Bug fix delete by query

client/elasticClient.go

@@ -153,8 +153,14 @@ func (ec *elasticClient) DoQueryRemove(ctx context.Context, index string, body *
 		log.Warn("elasticClient.doRefresh", "cannot do refresh", err)
 	}
 
+	writeIndex, err := ec.getWriteIndex(index)
+	if err != nil {
+		log.Warn("elasticClient.getWriteIndex", "cannot do get write index", err)
+		return err
+	}
+
 	res, err := ec.client.DeleteByQuery(
-		[]string{index},
+		[]string{writeIndex},
 		body,
 		ec.client.DeleteByQuery.WithIgnoreUnavailable(true),
 		ec.client.DeleteByQuery.WithConflicts(esConflictsPolicy),
@@ -323,6 +329,39 @@ func (ec *elasticClient) createAlias(alias string, index string) error {
 	return parseResponse(res, nil, elasticDefaultErrorResponseHandler)
 }
 
+func (ec *elasticClient) getWriteIndex(alias string) (string, error) {
+	res, err := ec.client.Indices.GetAlias(
+		ec.client.Indices.GetAlias.WithIndex(alias),
+	)
+	if err != nil {
+		return "", err
+	}
+
+	var indexData map[string]struct {
+		Aliases map[string]struct {
+			IsWriteIndex bool `json:"is_write_index"`
+		} `json:"aliases"`
+	}
+	err = parseResponse(res, &indexData, elasticDefaultErrorResponseHandler)
+	if err != nil {
+		return "", err
+	}
+
+	for index, details := range indexData {
+		if len(indexData) == 1 {
+			return index, nil
+		}
+
+		for _, indexAlias := range details.Aliases {
+			if indexAlias.IsWriteIndex {
+				return index, nil
+			}
+		}
+	}
+
+	return alias, nil
+}
+
 // UpdateByQuery will update all the documents that match the provided query from the provided index
 func (ec *elasticClient) UpdateByQuery(ctx context.Context, index string, buff *bytes.Buffer) error {
 	reader := bytes.NewReader(buff.Bytes())

client/elasticClient_test.go

@@ -2,7 +2,7 @@ package client
 
 import (
 	"context"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -53,7 +53,7 @@ func TestElasticClient_DoMultiGet(t *testing.T) {
 		jsonFile, err := os.Open("./testsData/response-multi-get.json")
 		require.Nil(t, err)
 
-		byteValue, _ := ioutil.ReadAll(jsonFile)
+		byteValue, _ := io.ReadAll(jsonFile)
 		_, _ = w.Write(byteValue)
 	}
 
@@ -75,3 +75,51 @@ func TestElasticClient_DoMultiGet(t *testing.T) {
 	_, ok := resMap["docs"]
 	require.True(t, ok)
 }
+
+func TestElasticClient_GetWriteIndexMultipleIndicesBehind(t *testing.T) {
+	handler := http.NotFound
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		handler(w, r)
+	}))
+	defer ts.Close()
+
+	handler = func(w http.ResponseWriter, r *http.Request) {
+		jsonFile, err := os.Open("./testsData/response-get-alias.json")
+		require.Nil(t, err)
+
+		byteValue, _ := io.ReadAll(jsonFile)
+		_, _ = w.Write(byteValue)
+	}
+
+	esClient, _ := NewElasticClient(elasticsearch.Config{
+		Addresses: []string{ts.URL},
+		Logger:    &logging.CustomLogger{},
+	})
+	res, err := esClient.getWriteIndex("blocks")
+	require.Nil(t, err)
+	require.Equal(t, "blocks-000004", res)
+}
+
+func TestElasticClient_GetWriteIndexOneIndex(t *testing.T) {
+	handler := http.NotFound
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		handler(w, r)
+	}))
+	defer ts.Close()
+
+	handler = func(w http.ResponseWriter, r *http.Request) {
+		jsonFile, err := os.Open("./testsData/response-get-alias-only-one-index.json")
+		require.Nil(t, err)
+
+		byteValue, _ := io.ReadAll(jsonFile)
+		_, _ = w.Write(byteValue)
+	}
+
+	esClient, _ := NewElasticClient(elasticsearch.Config{
+		Addresses: []string{ts.URL},
+		Logger:    &logging.CustomLogger{},
+	})
+	res, err := esClient.getWriteIndex("delegators")
+	require.Nil(t, err)
+	require.Equal(t, "delegators-000001", res)
+}

client/testsData/response-get-alias-only-one-index.json

@@ -0,0 +1,7 @@
+{
+  "delegators-000001" : {
+    "aliases" : {
+      "delegators" : { }
+    }
+  }
+}

client/testsData/response-get-alias.json

@@ -0,0 +1,30 @@
+{
+  "blocks-000003": {
+    "aliases": {
+      "blocks": {
+        "is_write_index": false
+      }
+    }
+  },
+  "blocks-000004": {
+    "aliases": {
+      "blocks": {
+        "is_write_index": true
+      }
+    }
+  },
+  "blocks-000002": {
+    "aliases": {
+      "blocks": {
+        "is_write_index": false
+      }
+    }
+  },
+  "blocks-000001": {
+    "aliases": {
+      "blocks": {
+        "is_write_index": false
+      }
+    }
+  }
+}