fix: concurrent access issue when metrics collection happen

Author: erayarslan

config/config.go

@@ -9,19 +9,19 @@ import (
 )
 
 type Elasticsearch struct {
-	Username                    string            `yaml:"username"`
-	Password                    string            `yaml:"password"`
 	BatchByteSizeLimit          any               `yaml:"batchByteSizeLimit"`
+	BatchCommitTickerDuration   *time.Duration    `yaml:"batchCommitTickerDuration"`
 	CollectionIndexMapping      map[string]string `yaml:"collectionIndexMapping"`
 	MaxConnsPerHost             *int              `yaml:"maxConnsPerHost"`
 	MaxIdleConnDuration         *time.Duration    `yaml:"maxIdleConnDuration"`
 	DiscoverNodesInterval       *time.Duration    `yaml:"discoverNodesInterval"`
 	TypeName                    string            `yaml:"typeName"`
-	Urls                        []string          `yaml:"urls"`
+	Password                    string            `yaml:"password"`
+	Username                    string            `yaml:"username"`
 	RejectionLog                RejectionLog      `yaml:"rejectionLog"`
+	Urls                        []string          `yaml:"urls"`
 	BatchSizeLimit              int               `yaml:"batchSizeLimit"`
 	BatchTickerDuration         time.Duration     `yaml:"batchTickerDuration"`
-	BatchCommitTickerDuration   *time.Duration    `yaml:"batchCommitTickerDuration"`
 	ConcurrentRequest           int               `yaml:"concurrentRequest"`
 	CompressionEnabled          bool              `yaml:"compressionEnabled"`
 	DisableDiscoverNodesOnStart bool              `yaml:"disableDiscoverNodesOnStart"`

elasticsearch/bulk/bulk.go

@@ -439,19 +439,27 @@ func fillErrorDataWithBulkRequestError(batchActions []*document.ESActionDocument
 	return errorData
 }
 
+func (b *Bulk) LockMetrics() {
+	b.metricCounterMutex.Lock()
+}
+
+func (b *Bulk) UnlockMetrics() {
+	b.metricCounterMutex.Unlock()
+}
+
 func (b *Bulk) finalizeProcess(batchActions []*document.ESActionDocument, errorData map[string]string) {
 	for _, action := range batchActions {
 		key := getActionKey(*action)
 		if _, ok := errorData[key]; ok {
-			b.countError(action)
+			go b.countError(action)
 			if b.sinkResponseHandler != nil {
 				b.sinkResponseHandler.OnError(&dcpElasticsearch.SinkResponseHandlerContext{
 					Action: action,
 					Err:    fmt.Errorf(errorData[key]),
 				})
 			}
 		} else {
-			b.countSuccess(action)
+			go b.countSuccess(action)
 			if b.sinkResponseHandler != nil {
 				b.sinkResponseHandler.OnSuccess(&dcpElasticsearch.SinkResponseHandlerContext{
 					Action: action,
@@ -462,8 +470,8 @@ func (b *Bulk) finalizeProcess(batchActions []*document.ESActionDocument, errorD
 }
 
 func (b *Bulk) countError(action *document.ESActionDocument) {
-	b.metricCounterMutex.Lock()
-	defer b.metricCounterMutex.Unlock()
+	b.LockMetrics()
+	defer b.UnlockMetrics()
 
 	if action.Type == document.Index || action.Type == document.DocUpdate || action.Type == document.ScriptUpdate {
 		b.metric.IndexingErrorActionCounter[action.IndexName]++
@@ -473,8 +481,8 @@ func (b *Bulk) countError(action *document.ESActionDocument) {
 }
 
 func (b *Bulk) countSuccess(action *document.ESActionDocument) {
-	b.metricCounterMutex.Lock()
-	defer b.metricCounterMutex.Unlock()
+	b.LockMetrics()
+	defer b.UnlockMetrics()
 
 	if action.Type == document.Index || action.Type == document.DocUpdate || action.Type == document.ScriptUpdate {
 		b.metric.IndexingSuccessActionCounter[action.IndexName]++

elasticsearch/bulk/bulk_test.go

@@ -7,6 +7,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/Trendyol/go-dcp-elasticsearch/elasticsearch"
 	"github.com/Trendyol/go-dcp-elasticsearch/elasticsearch/document"
@@ -165,6 +166,7 @@ func TestBulk_executeSinkResponseHandler(t *testing.T) {
 
 		// When
 		sut.finalizeProcess(batchActions, errorData)
+		time.Sleep(1 * time.Second)
 
 		// Then
 		if handler.successResult != "1:someIndex" {

metric/collector.go

@@ -19,6 +19,9 @@ func (s *Collector) Describe(ch chan<- *prometheus.Desc) {
 }
 
 func (s *Collector) Collect(ch chan<- prometheus.Metric) {
+	s.bulk.LockMetrics()
+	defer s.bulk.UnlockMetrics()
+
 	bulkMetric := s.bulk.GetMetric()
 
 	ch <- prometheus.MustNewConstMetric(