Bulk request size parameter elastic indexer

cmd/node/config/external.toml

@@ -4,15 +4,16 @@
     #the node might loose rating (even facing penalties) due to the fact that
     #the indexer is called synchronously and might block due to external causes.
     #Strongly suggested to activate this on a regular observer node.
-    Enabled           = false
-    IndexerCacheSize  = 0
-    URL               = "http://localhost:9200"
-    UseKibana         = false
-    Username          = ""
-    Password          = ""
+    Enabled                   = false
+    IndexerCacheSize          = 0
+    BulkRequestMaxSizeInBytes = 4194304 # 4MB
+    URL                       = "http://localhost:9200"
+    UseKibana                 = false
+    Username                  = ""
+    Password                  = ""
     # EnabledIndexes represents a slice of indexes that will be enabled for indexing. Full list is:
     # ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations"]
-    EnabledIndexes    = ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations"]
+    EnabledIndexes            = ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations"]
 
 # EventNotifierConnector defines settings needed to configure and launch the event notifier component
 [EventNotifierConnector]

config/externalConfig.go

@@ -9,13 +9,14 @@ type ExternalConfig struct {
 
 // ElasticSearchConfig will hold the configuration for the elastic search
 type ElasticSearchConfig struct {
-	Enabled          bool
-	IndexerCacheSize int
-	URL              string
-	UseKibana        bool
-	Username         string
-	Password         string
-	EnabledIndexes   []string
+	Enabled                   bool
+	IndexerCacheSize          int
+	BulkRequestMaxSizeInBytes int
+	URL                       string
+	UseKibana                 bool
+	Username                  string
+	Password                  string
+	EnabledIndexes            []string
 }
 
 // EventNotifierConfig will hold the configuration for the events notifier driver

factory/statusComponents.go

@@ -222,6 +222,7 @@ func (scf *statusComponentsFactory) makeElasticIndexerArgs() *indexerFactory.Arg
 	return &indexerFactory.ArgsIndexerFactory{
 		Enabled:                  elasticSearchConfig.Enabled,
 		IndexerCacheSize:         elasticSearchConfig.IndexerCacheSize,
+		BulkRequestMaxSize:       elasticSearchConfig.BulkRequestMaxSizeInBytes,
 		ShardCoordinator:         scf.shardCoordinator,
 		Url:                      elasticSearchConfig.URL,
 		UserName:                 elasticSearchConfig.Username,

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/ElrondNetwork/arwen-wasm-vm/v1_4 v1.4.34-rc7
 	github.com/ElrondNetwork/concurrent-map v0.1.3
 	github.com/ElrondNetwork/covalent-indexer-go v1.0.6
-	github.com/ElrondNetwork/elastic-indexer-go v1.2.15
+	github.com/ElrondNetwork/elastic-indexer-go v1.2.16
 	github.com/ElrondNetwork/elrond-go-core v1.1.14
 	github.com/ElrondNetwork/elrond-go-crypto v1.0.1
 	github.com/ElrondNetwork/elrond-go-logger v1.0.5

go.sum

@@ -24,8 +24,8 @@ github.com/ElrondNetwork/concurrent-map v0.1.3 h1:j2LtPrNJuerannC1cQDE79STvi/P04
 github.com/ElrondNetwork/concurrent-map v0.1.3/go.mod h1:3XwSwn4JHI0lrKxWLZvtp53Emr8BXYTmNQGwcukHJEE=
 github.com/ElrondNetwork/covalent-indexer-go v1.0.6 h1:+LNKItUc+Pb7WuTbil3VuiLMmdQ1AY7lBJM476PtVNE=
 github.com/ElrondNetwork/covalent-indexer-go v1.0.6/go.mod h1:j3h2g96vqhJAuj3aEX2PWhomae2/o7YfXGEfweNXEeQ=
-github.com/ElrondNetwork/elastic-indexer-go v1.2.15 h1:beLJ0qx2PonDefYDG6pcEQYJTFDeMEiJ06GslKSOmnM=
-github.com/ElrondNetwork/elastic-indexer-go v1.2.15/go.mod h1:jrzgNlt92qgq+GAW5p+ZgW2Rkzama+Ufr9sQ8KDgZfk=
+github.com/ElrondNetwork/elastic-indexer-go v1.2.16 h1:WkzwRe3ev0Q7yExTkBDjGbu6TBc3vQCpubivcRq0/Gs=
+github.com/ElrondNetwork/elastic-indexer-go v1.2.16/go.mod h1:jrzgNlt92qgq+GAW5p+ZgW2Rkzama+Ufr9sQ8KDgZfk=
 github.com/ElrondNetwork/elrond-go-core v1.0.0/go.mod h1:FQMem7fFF4+8pQ6lVsBZq6yO+smD0nV23P4bJpmPjTo=
 github.com/ElrondNetwork/elrond-go-core v1.1.6/go.mod h1:O9FkkTT2H9kxCzfn40TbhoCDXzGmUrRVusMomhK/Y3g=
 github.com/ElrondNetwork/elrond-go-core v1.1.7/go.mod h1:O9FkkTT2H9kxCzfn40TbhoCDXzGmUrRVusMomhK/Y3g=

integrationTests/vm/testIndexer.go

@@ -125,7 +125,7 @@ func (ti *testIndexer) createElasticProcessor(
 	bp, _ := blockProc.NewBlockProcessor(testHasher, testMarshalizer)
 	mp, _ := miniblocks.NewMiniblocksProcessor(shardCoordinator.SelfId(), testHasher, testMarshalizer, false)
 	sp := statistics.NewStatisticsProcessor()
-	vp, _ := validators.NewValidatorsProcessor(pubkeyConv)
+	vp, _ := validators.NewValidatorsProcessor(pubkeyConv, 0)
 	opp, _ := operations.NewOperationsProcessor(false, shardCoordinator)
 	args := &logsevents.ArgsLogsAndEventsProcessor{
 		ShardCoordinator: shardCoordinator,
@@ -257,6 +257,7 @@ func (ti *testIndexer) createDatabaseClient(hasResults bool) elasticProcessor.Da
 		ti.indexerData[index] = buff
 		if !done {
 			done = true
+			ti.saveDoneChan <- struct{}{}
 			return nil
 		}
 		ti.saveDoneChan <- struct{}{}
@@ -273,7 +274,7 @@ func (ti *testIndexer) createDatabaseClient(hasResults bool) elasticProcessor.Da
 // GetIndexerPreparedTransaction -
 func (ti *testIndexer) GetIndexerPreparedTransaction(t *testing.T) *indexerTypes.Transaction {
 	ti.mutex.RLock()
-	txData, ok := ti.indexerData["transactions"]
+	txData, ok := ti.indexerData[""]
 	ti.mutex.RUnlock()
 
 	require.True(t, ok)
@@ -303,7 +304,7 @@ func (ti *testIndexer) GetIndexerPreparedTransaction(t *testing.T) *indexerTypes
 
 func (ti *testIndexer) printReceipt() {
 	ti.mutex.RLock()
-	receipts, ok := ti.indexerData["receipts"]
+	receipts, ok := ti.indexerData[""]
 	ti.mutex.RUnlock()
 
 	if !ok {
@@ -322,7 +323,7 @@ func (ti *testIndexer) printReceipt() {
 
 func (ti *testIndexer) putSCRSInTx(tx *indexerTypes.Transaction) {
 	ti.mutex.RLock()
-	scrData, ok := ti.indexerData["scresults"]
+	scrData, ok := ti.indexerData[""]
 	ti.mutex.RUnlock()
 
 	if !ok {
@@ -333,6 +334,10 @@ func (ti *testIndexer) putSCRSInTx(tx *indexerTypes.Transaction) {
 	require.True(ti.t, len(split) > 2)
 
 	for idx := 1; idx < len(split); idx += 2 {
+		if !bytes.Contains(split[idx], []byte("scresults")) {
+			continue
+		}
+
 		newSCR := &indexerTypes.ScResult{}
 		err := json.Unmarshal(split[idx], newSCR)
 		require.Nil(ti.t, err)