Rc/2022 july

.github/workflows/pr-build.yml

@@ -3,9 +3,9 @@ name: Go build
 
 on:
   push:
-    branches: [ master, feat/* ]
+    branches: [ master, development, feat/* ]
   pull_request:
-    branches: [ master, feat/* ]
+    branches: [ master, development, feat/* ]
 
 jobs:
   build:

.github/workflows/pr-integration-tests.yml

@@ -2,9 +2,9 @@ name: Integration tests
 
 on:
   push:
-    branches: [ master, feat/* ]
+    branches: [ master, development, feat/* ]
   pull_request:
-    branches: [ master, feat/* ]
+    branches: [ master, development, feat/* ]
 
 jobs:
   test-1:
@@ -26,7 +26,7 @@ jobs:
       - name: Run integration tests with Elasticsearch `v7.16.2`
         run: make integration-tests ES_VERSION=7.16.2
   test-2:
-    name: Elasticsearch v8.1.1
+    name: Elasticsearch v8.3.2
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go 1.x
@@ -41,8 +41,8 @@ jobs:
       - name: Get dependencies
         run: |
           go get -v -t -d ./...
-      - name: Run integration tests with Elasticsearch `v8.1.1`
-        run: make integration-tests ES_VERSION=8.1.1
+      - name: Run integration tests with Elasticsearch `v8.3.2`
+        run: make integration-tests ES_VERSION=8.3.2
   test-3:
     name: OpenSearch v1.2.4
     runs-on: ubuntu-latest
@@ -63,7 +63,7 @@ jobs:
         run: make integration-tests-open-search OPEN_VERSION=1.2.4
 
   test-4:
-    name: OpenSearch v1.3.2
+    name: OpenSearch v2.1.0
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go 1.x
@@ -78,5 +78,5 @@ jobs:
       - name: Get dependencies
         run: |
           go get -v -t -d ./...
-      - name: Run integration tests with OpenSearch `v1.3.2`
-        run: make integration-tests-open-search OPEN_VERSION=1.3.2
+      - name: Run integration tests with OpenSearch `v2.1.0`
+        run: make integration-tests-open-search OPEN_VERSION=2.1.0

.github/workflows/pr-tests.yml

@@ -2,9 +2,9 @@ name: Tests
 
 on:
   push:
-    branches: [ master, feat/* ]
+    branches: [ master, development, feat/* ]
   pull_request:
-    branches: [ master, feat/* ]
+    branches: [ master, development, feat/* ]
 
 jobs:
   test:

client/elasticClient.go

@@ -18,6 +18,7 @@ import (
 // TODO add more unit tests
 
 const (
+	esConflictsPolicy      = "proceed"
 	errPolicyAlreadyExists = "document already exists"
 )
 
@@ -30,7 +31,7 @@ type (
 
 type elasticClient struct {
 	elasticBaseUrl string
-	es             *elasticsearch.Client
+	client         *elasticsearch.Client
 
 	// countScroll is used to be incremented after each scroll so the scroll duration is different each time,
 	// bypassing any possible caching based on the same request
@@ -49,7 +50,7 @@ func NewElasticClient(cfg elasticsearch.Config) (*elasticClient, error) {
 	}
 
 	ec := &elasticClient{
-		es:             es,
+		client:         es,
 		elasticBaseUrl: cfg.Addresses[0],
 	}
 
@@ -94,7 +95,7 @@ func (ec *elasticClient) CheckAndCreateAlias(alias string, indexName string) err
 
 // DoRequest will do a request to elastic server
 func (ec *elasticClient) DoRequest(req *esapi.IndexRequest) error {
-	res, err := req.Do(context.Background(), ec.es)
+	res, err := req.Do(context.Background(), ec.client)
 	if err != nil {
 		return err
 	}
@@ -108,10 +109,10 @@ func (ec *elasticClient) DoBulkRequest(buff *bytes.Buffer, index string) error {
 
 	options := make([]func(*esapi.BulkRequest), 0)
 	if index != "" {
-		options = append(options, ec.es.Bulk.WithIndex(index))
+		options = append(options, ec.client.Bulk.WithIndex(index))
 	}
 
-	res, err := ec.es.Bulk(
+	res, err := ec.client.Bulk(
 		reader,
 		options...,
 	)
@@ -132,9 +133,9 @@ func (ec *elasticClient) DoMultiGet(ids []string, index string, withSource bool,
 		return err
 	}
 
-	res, err := ec.es.Mget(
+	res, err := ec.client.Mget(
 		&body,
-		ec.es.Mget.WithIndex(index),
+		ec.client.Mget.WithIndex(index),
 	)
 	if err != nil {
 		log.Warn("elasticClient.DoMultiGet",
@@ -152,46 +153,54 @@ func (ec *elasticClient) DoMultiGet(ids []string, index string, withSource bool,
 	return nil
 }
 
-// DoBulkRemove will do a bulk remove to elasticsearch server
-func (ec *elasticClient) DoBulkRemove(index string, hashes []string) error {
-	obj := prepareHashesForBulkRemove(hashes)
-	body, err := encode(obj)
-	if err != nil {
-		return err
+// DoQueryRemove will do a query remove to elasticsearch server
+func (ec *elasticClient) DoQueryRemove(index string, body *bytes.Buffer) error {
+	if err := ec.doRefresh(index); err != nil {
+		log.Warn("elasticClient.doRefresh", "cannot to refresh", err.Error())
 	}
 
-	res, err := ec.es.DeleteByQuery(
+	res, err := ec.client.DeleteByQuery(
 		[]string{index},
-		&body,
-		ec.es.DeleteByQuery.WithIgnoreUnavailable(true),
+		body,
+		ec.client.DeleteByQuery.WithIgnoreUnavailable(true),
+		ec.client.DeleteByQuery.WithConflicts(esConflictsPolicy),
 	)
 
 	if err != nil {
-		log.Warn("elasticClient.DoBulkRemove",
-			"cannot do bulk remove", err.Error())
+		log.Warn("elasticClient.DoQueryRemove", "cannot do query remove", err.Error())
 		return err
 	}
 
-	var decodedBody objectsMap
-	err = parseResponse(res, &decodedBody, elasticDefaultErrorResponseHandler)
+	err = parseResponse(res, nil, elasticDefaultErrorResponseHandler)
 	if err != nil {
-		log.Warn("elasticClient.DoBulkRemove",
-			"error parsing response", err.Error())
+		log.Warn("elasticClient.DoQueryRemove", "error parsing response", err.Error())
 		return err
 	}
 
 	return nil
 }
 
+func (ec *elasticClient) doRefresh(index string) error {
+	res, err := ec.client.Indices.Refresh(
+		ec.client.Indices.Refresh.WithIndex(index),
+		ec.client.Indices.Refresh.WithIgnoreUnavailable(true),
+	)
+	if err != nil {
+		return err
+	}
+
+	return parseResponse(res, nil, elasticDefaultErrorResponseHandler)
+}
+
 // TemplateExists checks weather a template is already created
 func (ec *elasticClient) templateExists(index string) bool {
-	res, err := ec.es.Indices.ExistsTemplate([]string{index})
+	res, err := ec.client.Indices.ExistsTemplate([]string{index})
 	return exists(res, err)
 }
 
 // IndexExists checks if a given index already exists
 func (ec *elasticClient) indexExists(index string) bool {
-	res, err := ec.es.Indices.Exists([]string{index})
+	res, err := ec.client.Indices.Exists([]string{index})
 	return exists(res, err)
 }
 
@@ -205,7 +214,7 @@ func (ec *elasticClient) PolicyExists(policy string) bool {
 	)
 
 	req := newRequest(http.MethodGet, policyRoute, nil)
-	res, err := ec.es.Transport.Perform(req)
+	res, err := ec.client.Transport.Perform(req)
 	if err != nil {
 		log.Warn("elasticClient.PolicyExists",
 			"error performing request", err.Error())
@@ -237,7 +246,7 @@ func (ec *elasticClient) aliasExists(alias string) bool {
 	)
 
 	req := newRequest(http.MethodHead, aliasRoute, nil)
-	res, err := ec.es.Transport.Perform(req)
+	res, err := ec.client.Transport.Perform(req)
 	if err != nil {
 		log.Warn("elasticClient.AliasExists",
 			"error performing request", err.Error())
@@ -255,7 +264,7 @@ func (ec *elasticClient) aliasExists(alias string) bool {
 
 // CreateIndex creates an elasticsearch index
 func (ec *elasticClient) createIndex(index string) error {
-	res, err := ec.es.Indices.Create(index)
+	res, err := ec.client.Indices.Create(index)
 	if err != nil {
 		return err
 	}
@@ -274,7 +283,7 @@ func (ec *elasticClient) createPolicy(policyName string, policy *bytes.Buffer) e
 	req := newRequest(http.MethodPut, policyRoute, policy)
 	req.Header[headerContentType] = headerContentTypeJSON
 	req.Header[headerXSRF] = []string{"false"}
-	res, err := ec.es.Transport.Perform(req)
+	res, err := ec.client.Transport.Perform(req)
 	if err != nil {
 		return err
 	}
@@ -301,7 +310,7 @@ func (ec *elasticClient) createPolicy(policyName string, policy *bytes.Buffer) e
 
 // CreateIndexTemplate creates an elasticsearch index template
 func (ec *elasticClient) createIndexTemplate(templateName string, template io.Reader) error {
-	res, err := ec.es.Indices.PutTemplate(templateName, template)
+	res, err := ec.client.Indices.PutTemplate(templateName, template)
 	if err != nil {
 		return err
 	}
@@ -311,7 +320,7 @@ func (ec *elasticClient) createIndexTemplate(templateName string, template io.Re
 
 // CreateAlias creates an index alias
 func (ec *elasticClient) createAlias(alias string, index string) error {
-	res, err := ec.es.Indices.PutAlias([]string{index}, alias)
+	res, err := ec.client.Indices.PutAlias([]string{index}, alias)
 	if err != nil {
 		return err
 	}

client/elasticClientScroll.go

@@ -15,9 +15,9 @@ import (
 
 // DoCountRequest will get the number of elements that correspond with the provided query
 func (ec *elasticClient) DoCountRequest(index string, body []byte) (uint64, error) {
-	res, err := ec.es.Count(
-		ec.es.Count.WithIndex(index),
-		ec.es.Count.WithBody(bytes.NewBuffer(body)),
+	res, err := ec.client.Count(
+		ec.client.Count.WithIndex(index),
+		ec.client.Count.WithBody(bytes.NewBuffer(body)),
 	)
 	if err != nil {
 		return 0, err
@@ -44,13 +44,13 @@ func (ec *elasticClient) DoScrollRequest(
 	handlerFunc func(responseBytes []byte) error,
 ) error {
 	ec.countScroll++
-	res, err := ec.es.Search(
-		ec.es.Search.WithSize(9000),
-		ec.es.Search.WithScroll(10*time.Minute+time.Duration(ec.countScroll)*time.Millisecond),
-		ec.es.Search.WithContext(context.Background()),
-		ec.es.Search.WithIndex(index),
-		ec.es.Search.WithBody(bytes.NewBuffer(body)),
-		ec.es.Search.WithSource(strconv.FormatBool(withSource)),
+	res, err := ec.client.Search(
+		ec.client.Search.WithSize(9000),
+		ec.client.Search.WithScroll(10*time.Minute+time.Duration(ec.countScroll)*time.Millisecond),
+		ec.client.Search.WithContext(context.Background()),
+		ec.client.Search.WithIndex(index),
+		ec.client.Search.WithBody(bytes.NewBuffer(body)),
+		ec.client.Search.WithSource(strconv.FormatBool(withSource)),
 	)
 	if err != nil {
 		return err
@@ -103,9 +103,9 @@ func (ec *elasticClient) iterateScroll(
 
 func (ec *elasticClient) getScrollResponse(scrollID string) ([]byte, error) {
 	ec.countScroll++
-	res, err := ec.es.Scroll(
-		ec.es.Scroll.WithScrollID(scrollID),
-		ec.es.Scroll.WithScroll(2*time.Minute+time.Duration(ec.countScroll)*time.Millisecond),
+	res, err := ec.client.Scroll(
+		ec.client.Scroll.WithScrollID(scrollID),
+		ec.client.Scroll.WithScroll(2*time.Minute+time.Duration(ec.countScroll)*time.Millisecond),
 	)
 	if err != nil {
 		return nil, err
@@ -115,8 +115,8 @@ func (ec *elasticClient) getScrollResponse(scrollID string) ([]byte, error) {
 }
 
 func (ec *elasticClient) clearScroll(scrollID string) error {
-	resp, err := ec.es.ClearScroll(
-		ec.es.ClearScroll.WithScrollID(scrollID),
+	resp, err := ec.client.ClearScroll(
+		ec.client.ClearScroll.WithScrollID(scrollID),
 	)
 	if err != nil {
 		return err

client/queries.go

@@ -28,13 +28,3 @@ func getDocumentsByIDsQuery(hashes []string, withSource bool) objectsMap {
 		"docs": interfaceSlice,
 	}
 }
-
-func prepareHashesForBulkRemove(hashes []string) objectsMap {
-	return objectsMap{
-		"query": objectsMap{
-			"ids": objectsMap{
-				"values": hashes,
-			},
-		},
-	}
-}

constants.go

@@ -43,6 +43,8 @@ const (
 	DelegatorsIndex = "delegators"
 	// OperationsIndex is the Elasticsearch index for transactions and smart contract results
 	OperationsIndex = "operations"
+	// CollectionsIndex is the Elasticsearch index for collections
+	CollectionsIndex = "collections"
 
 	// TransactionsPolicy is the Elasticsearch policy for the transactions
 	TransactionsPolicy = "transactions_policy"

converters/json.go

@@ -1,14 +1,41 @@
 package converters
 
-import "encoding/json"
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+
+	logger "github.com/ElrondNetwork/elrond-go-logger"
+)
+
+const defaultStr = "default"
+
+var log = logger.GetOrCreate("indexer/converters")
 
 // JsonEscape will format the provided string in a json compatible string
 func JsonEscape(i string) string {
 	b, err := json.Marshal(i)
 	if err != nil {
-		return ""
+		log.Warn("converters.JsonEscape something went wrong",
+			"input", i,
+			"error", err,
+		)
+		return defaultStr
 	}
 
 	// Trim the beginning and trailing " character
 	return string(b[1 : len(b)-1])
 }
+
+// PrepareHashesForQueryRemove will prepare the provided hashes for query remove
+func PrepareHashesForQueryRemove(hashes []string) *bytes.Buffer {
+	if len(hashes) == 0 {
+		hashes = []string{}
+	}
+
+	serializedHashes, _ := json.Marshal(hashes)
+	query := `{"query": {"ids": {"values": %s}}}`
+	deleteQuery := fmt.Sprintf(query, serializedHashes)
+
+	return bytes.NewBuffer([]byte(deleteQuery))
+}