Fix concurrent conflicts in elasticsearch (#53) (#58)

This patch makes use of concurrent package to fix conflicting writes/reads
to/from ES buffer when bulk index is configured

Cherry picked from commit e61141a

Co-authored-by: Martin Mágr <[email protected]>

plugins/application/elasticsearch/main.go

@@ -13,6 +13,7 @@ import (
 	jsoniter "github.com/json-iterator/go"
 	"github.com/pkg/errors"
 
+	"github.com/infrawatch/sg-core/pkg/concurrent"
 	"github.com/infrawatch/sg-core/plugins/application/elasticsearch/pkg/lib"
 )
 
@@ -51,15 +52,15 @@ type Elasticsearch struct {
 	configuration *lib.AppConfig
 	logger        *logging.Logger
 	client        *lib.Client
-	buffer        map[string][]string
+	buffer        *concurrent.Map
 	dump          chan esIndex
 }
 
 // New constructor
 func New(logger *logging.Logger) application.Application {
 	return &Elasticsearch{
 		logger: logger,
-		buffer: make(map[string][]string),
+		buffer: concurrent.NewMap(),
 		dump:   make(chan esIndex, 100),
 	}
 }
@@ -88,19 +89,20 @@ func (es *Elasticsearch) ReceiveEvent(event data.Event) {
 	// buffer or index record
 	var recordList []string
 	if es.configuration.BufferSize > 1 {
-		if _, ok := es.buffer[event.Index]; !ok {
-			es.buffer[event.Index] = make([]string, 0, es.configuration.BufferSize)
+		if !es.buffer.Contains(event.Index) {
+			es.buffer.Set(event.Index, make([]string, 0, es.configuration.BufferSize))
 		}
 
-		es.buffer[event.Index] = append(es.buffer[event.Index], record)
-		if len(es.buffer[event.Index]) < es.configuration.BufferSize {
+		recordList = (es.buffer.Get(event.Index)).([]string)
+		recordList = append(recordList, record)
+		if len(recordList) < es.configuration.BufferSize {
 			// buffer is not full, don't send
 			es.logger.Metadata(logging.Metadata{"plugin": appname, "record": record})
 			es.logger.Debug("buffering record")
+			es.buffer.Set(event.Index, recordList)
 			return
 		}
-		recordList = es.buffer[event.Index]
-		delete(es.buffer, event.Index)
+		es.buffer.Delete(event.Index)
 	} else {
 		recordList = []string{record}
 	}

plugins/application/elasticsearch/main_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 
 	"github.com/infrawatch/apputils/logging"
+	"github.com/infrawatch/sg-core/pkg/concurrent"
 	"github.com/infrawatch/sg-core/pkg/data"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -232,7 +233,7 @@ func TestElasticsearchApp(t *testing.T) {
 		results := make(chan esIndex, len(eventCases))
 		app := &Elasticsearch{
 			logger: logger,
-			buffer: make(map[string][]string),
+			buffer: concurrent.NewMap(),
 			dump:   results,
 		}
 		err := app.Config([]byte(testConf))
@@ -254,7 +255,7 @@ func TestElasticsearchApp(t *testing.T) {
 		results := make(chan esIndex, len(logCases))
 		app := &Elasticsearch{
 			logger: logger,
-			buffer: make(map[string][]string),
+			buffer: concurrent.NewMap(),
 			dump:   results,
 		}
 		err := app.Config([]byte(testConf))