Merge pull request #30 from inloco/hotfix/workers-locking

Hotfix/workers locking

Author: João Filipe Moura

README.md

@@ -113,6 +113,9 @@ The exported metrics are:
 - `kafka_consumer_records_consumed_successfully`: number of records consumed successfully by this instance.
 - `kafka_consumer_endpoint_latency_histogram_seconds`: endpoint latency in seconds (insertion to elasticsearch).
 - `kafka_consumer_buffer_full`: indicates whether the app buffer is full(meaning that elasticsearch is not being able to keep up with the topic volume).
+- `elasticsearch_events_retried`: number of events that needed to be retryed to sent to Elasticsearch
+- `elasticsearch_document_already_exists`: number of events that tryed to be inserted on elasticsearch but already existed
+- `elasticsearch_bad_request`: the number of requests that failed due to malformed events
 
 ## Development
 

src/elasticsearch/elasticsearch.go

@@ -5,6 +5,8 @@ import (
 	"crypto/tls"
 	"fmt"
 
+	"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
+
 	"github.com/pkg/errors"
 
 	"net/http"
@@ -29,8 +31,9 @@ type RecordDatabase interface {
 }
 
 type recordDatabase struct {
-	logger log.Logger
-	config Config
+	metricsPublisher metrics.MetricsPublisher
+	logger           log.Logger
+	config           Config
 }
 
 func (d recordDatabase) GetClient() *elastic.Client {
@@ -76,6 +79,7 @@ func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse
 	timeout := d.config.BulkTimeout
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
+
 	res, err := bulkRequest.Do(ctx)
 	if err == elastic.ErrNoClient || errors.Cause(err) == elastic.ErrNoClient {
 		_ = level.Warn(d.logger).Log("message", "no elasticsearch node available", "err", err)
@@ -104,7 +108,12 @@ func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse
 				recordMap[rec.ID] = rec
 			}
 			for _, f := range failed {
+				if f.Status == http.StatusBadRequest {
+					d.metricsPublisher.ElasticsearchBadRequests(1)
+					continue
+				}
 				if f.Status == http.StatusConflict {
+					d.metricsPublisher.ElasticsearchConflicts(1)
 					continue
 				}
 				retry = append(retry, recordMap[f.Id])
@@ -117,6 +126,7 @@ func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse
 				level.Warn(d.logger).Log("message", "insert failed: elasticsearch is overloaded", "retry_count", len(retry))
 			}
 		}
+		d.metricsPublisher.ElasticsearchRetries(len(retry))
 		return &InsertResponse{alreadyExistsIds, retry, overloaded}, nil
 	}
 
@@ -145,6 +155,10 @@ func (d recordDatabase) buildBulkRequest(records []*models.ElasticRecord) (*elas
 	return bulkRequest, nil
 }
 
-func NewDatabase(logger log.Logger, config Config) RecordDatabase {
-	return recordDatabase{logger: logger, config: config}
+func NewDatabase(logger log.Logger, config Config, metrics metrics.MetricsPublisher) RecordDatabase {
+	return recordDatabase{
+		metricsPublisher: metrics,
+		logger:           logger,
+		config:           config,
+	}
 }

src/elasticsearch/elasticsearch_test.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/inloco/kafka-elasticsearch-injector/src/kafka/fixtures"
 	"github.com/inloco/kafka-elasticsearch-injector/src/logger_builder"
+	"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 	"github.com/olivere/elastic/v7"
 	"github.com/stretchr/testify/assert"
@@ -28,7 +29,8 @@ var config = Config{
 	BlacklistedColumns: []string{},
 	BulkTimeout:        10 * time.Second,
 }
-var db = NewDatabase(logger, config)
+
+var db = NewDatabase(logger, config, metrics.NewMetricsPublisher())
 var template = `
 {
 	"template": "my-topic-*",

src/injector/service.go

@@ -28,7 +28,7 @@ func NewService(logger log.Logger, metrics metrics.MetricsPublisher) Service {
 	return instrumentingMiddleware{
 		metricsPublisher: metrics,
 		next: basicService{
-			store.NewStore(logger),
+			store.NewStore(logger, metrics),
 		},
 	}
 }

src/injector/store/store.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/go-kit/kit/log"
 	"github.com/inloco/kafka-elasticsearch-injector/src/elasticsearch"
+	"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 )
 
@@ -24,6 +25,7 @@ func (s basicStore) Insert(records []*models.Record) error {
 	if err != nil {
 		return err
 	}
+
 	for {
 		res, err := s.db.Insert(elasticRecords)
 		if err != nil {
@@ -45,10 +47,10 @@ func (s basicStore) ReadinessCheck() bool {
 	return s.db.ReadinessCheck()
 }
 
-func NewStore(logger log.Logger) Store {
+func NewStore(logger log.Logger, metricsPublisher metrics.MetricsPublisher) Store {
 	config := elasticsearch.NewConfig()
 	return basicStore{
-		db:      elasticsearch.NewDatabase(logger, config),
+		db:      elasticsearch.NewDatabase(logger, config, metricsPublisher),
 		codec:   elasticsearch.NewCodec(logger, config),
 		backoff: config.Backoff,
 	}

src/kafka/consumer_test.go

@@ -51,13 +51,14 @@ func (be *fixtureEndpoints) Insert() endpoint.Endpoint {
 }
 
 var (
-	logger = logger_builder.NewLogger("consumer-test")
-	config = elasticsearch.Config{
+	metricsPublisher = metrics.NewMetricsPublisher()
+	logger           = logger_builder.NewLogger("consumer-test")
+	config           = elasticsearch.Config{
 		Host:        "http://localhost:9200",
 		Index:       fixtures.DefaultTopic,
 		BulkTimeout: 10 * time.Second,
 	}
-	db        = elasticsearch.NewDatabase(logger, config)
+	db        = elasticsearch.NewDatabase(logger, config, metricsPublisher)
 	codec     = elasticsearch.NewCodec(logger, config)
 	service   = fixtureService{db, codec}
 	endpoints = &fixtureEndpoints{
@@ -90,7 +91,7 @@ func TestMain(m *testing.M) {
 		BatchSize:             1,
 		MetricsUpdateInterval: 30 * time.Second,
 	}
-	k = NewKafka("localhost:9092", consumer, metrics.NewMetricsPublisher())
+	k = NewKafka("localhost:9092", consumer, metricsPublisher)
 	retCode := m.Run()
 	os.Exit(retCode)
 }

src/metrics/metrics.go

@@ -18,6 +18,9 @@ type metrics struct {
 	recordsConsumed          *kitprometheus.Counter
 	endpointLatencyHistogram *kitprometheus.Summary
 	bufferFullGauge          *kitprometheus.Gauge
+	elasticsearchRetries     *kitprometheus.Counter
+	elasticsearchConflicts   *kitprometheus.Counter
+	elasticsearchBadRequest  *kitprometheus.Counter
 	lock                     sync.RWMutex
 	topicPartitionToOffset   map[string]map[int32]int64
 }
@@ -69,12 +72,27 @@ func (m *metrics) BufferFull(full bool) {
 	m.bufferFullGauge.Set(val)
 }
 
+func (m *metrics) ElasticsearchRetries(count int) {
+	m.elasticsearchRetries.Add(float64(count))
+}
+
+func (m *metrics) ElasticsearchConflicts(count int) {
+	m.elasticsearchConflicts.Add(float64(count))
+}
+
+func (m *metrics) ElasticsearchBadRequests(count int) {
+	m.elasticsearchBadRequest.Add(float64(count))
+}
+
 type MetricsPublisher interface {
 	PublishOffsetMetrics(highWaterMarks map[string]map[int32]int64)
 	UpdateOffset(topic string, partition int32, delay int64)
 	IncrementRecordsConsumed(count int)
 	RecordEndpointLatency(latency float64)
 	BufferFull(full bool)
+	ElasticsearchRetries(count int)
+	ElasticsearchConflicts(count int)
+	ElasticsearchBadRequests(cont int)
 }
 
 func NewMetricsPublisher() MetricsPublisher {
@@ -95,13 +113,28 @@ func NewMetricsPublisher() MetricsPublisher {
 		Name: "kafka_consumer_buffer_full",
 		Help: "Kafka consumer boolean indicating if app buffer is full",
 	}, []string{})
+	elasticsearchRetriesCounter := kitprometheus.NewCounterFrom(stdprometheus.CounterOpts{
+		Name: "elasticsearch_events_retryed",
+		Help: "number of events that needed to be retryed sending to Elasticsearch",
+	}, []string{})
+	elasticsearchConflictsCounter := kitprometheus.NewCounterFrom(stdprometheus.CounterOpts{
+		Name: "elasticsearch_document_already_exists",
+		Help: "number of events that tried to be inserted on elasticsearch but alredy existed",
+	}, []string{})
+	elasticsearchBadRequestCounter := kitprometheus.NewCounterFrom(stdprometheus.CounterOpts{
+		Name: "elasticsearch_bad_request",
+		Help: "the number of malformed events",
+	}, []string{})
 	return &metrics{
 		logger:                   logger,
 		partitionDelay:           partitionDelay,
 		recordsConsumed:          recordsConsumed,
 		endpointLatencyHistogram: endpointLatencySummary,
 		bufferFullGauge:          bufferFullGauge,
 		lock:                     sync.RWMutex{},
-		topicPartitionToOffset: make(map[string]map[int32]int64),
+		elasticsearchRetries:     elasticsearchRetriesCounter,
+		elasticsearchConflicts:   elasticsearchConflictsCounter,
+		elasticsearchBadRequest:  elasticsearchBadRequestCounter,
+		topicPartitionToOffset:   make(map[string]map[int32]int64),
 	}
 }