Merge pull request #7 from inloco/fix/bulk-create-handle-errors

Fix/bulk create handle errors

Author: racevedoo

Gopkg.lock

@@ -29,6 +29,7 @@ To create new injectors for your topics, you should create a new kubernetes depl
 - `LOG_LEVEL` Determines the log level for the app. Should be set to DEBUG, WARN, NONE or INFO. Defaults to INFO. **OPTIONAL**
 - `METRICS_PORT` Port to export app metrics **REQUIRED**
 - `ES_BULK_TIMEOUT` Timeout for elasticsearch bulk writes in the format of golang's `time.ParseDuration`. Default value is 1s **OPTIONAL**
+- `ES_BULK_BACKOFF` Constant backoff when elasticsearch is overloaded. in the format of golang's `time.ParseDuration`. Default value is 1s **OPTIONAL**
 - `KAFKA_CONSUMER_RECORD_TYPE` Kafka record type. Should be set to "avro" or "json". Defaults to avro. **OPTIONAL**
 - `KAFKA_CONSUMER_METRICS_UPDATE_INTERVAL` The interval which the app updates the exported metrics in the format of golang's `time.ParseDuration`. Defaults to 30s. **OPTIONAL**
 

README.md

@@ -1 +1 @@
-0.5.2
+0.5.3

VERSION

@@ -13,6 +13,7 @@ type Config struct {
 	DocIDColumn        string
 	BlacklistedColumns []string
 	BulkTimeout        time.Duration
+	Backoff            time.Duration
 }
 
 func NewConfig() Config {
@@ -24,12 +25,21 @@ func NewConfig() Config {
 			timeout = d
 		}
 	}
+	backoffStr, exists := os.LookupEnv("ES_BULK_BACKOFF")
+	backoff := 1 * time.Second
+	if exists {
+		d, err := time.ParseDuration(backoffStr)
+		if err == nil {
+			backoff = d
+		}
+	}
 	return Config{
 		Host:               os.Getenv("ELASTICSEARCH_HOST"),
 		Index:              os.Getenv("ES_INDEX"),
 		IndexColumn:        os.Getenv("ES_INDEX_COLUMN"),
 		DocIDColumn:        os.Getenv("ES_DOC_ID_COLUMN"),
 		BlacklistedColumns: strings.Split(os.Getenv("ES_BLACKLISTED_COLUMNS"), ","),
 		BulkTimeout:        timeout,
+		Backoff:            backoff,
 	}
 }

src/elasticsearch/config.go

@@ -5,11 +5,12 @@ import (
 
 	"fmt"
 
+	"net/http"
+
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 	"github.com/olivere/elastic"
-	"github.com/pkg/errors"
 )
 
 var esClient *elastic.Client
@@ -21,7 +22,7 @@ type basicDatabase interface {
 
 type RecordDatabase interface {
 	basicDatabase
-	Insert(records []*models.ElasticRecord) error
+	Insert(records []*models.ElasticRecord) (*InsertResponse, error)
 	ReadinessCheck() bool
 }
 
@@ -49,24 +50,62 @@ func (d recordDatabase) CloseClient() {
 	}
 }
 
-func (d recordDatabase) Insert(records []*models.ElasticRecord) error {
+type InsertResponse struct {
+	AlreadyExists []string
+	Retry         []*models.ElasticRecord
+	Overloaded    bool
+}
+
+func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse, error) {
 	bulkRequest, err := d.buildBulkRequest(records)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	timeout := d.config.BulkTimeout
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 	res, err := bulkRequest.Do(ctx)
-	if err == nil {
-		if res.Errors {
-			for _, f := range res.Failed() {
-				return errors.New(fmt.Sprintf("%v", f.Error))
+
+	if err != nil {
+		return nil, err
+	}
+	if res.Errors {
+		created := res.Created()
+		var alreadyExistsIds []string
+		for _, c := range created {
+			if c.Status == http.StatusConflict {
+				alreadyExistsIds = append(alreadyExistsIds, c.Id)
+			}
+		}
+		if len(alreadyExistsIds) > 0 {
+			level.Warn(d.logger).Log("message", "document already exists", "doc_count", len(alreadyExistsIds))
+		}
+		failed := res.Failed()
+		var retry []*models.ElasticRecord
+		overloaded := false
+		if len(failed) > 0 {
+			recordMap := make(map[string]*models.ElasticRecord)
+			for _, rec := range records {
+				recordMap[rec.ID] = rec
+			}
+			for _, f := range failed {
+				if f.Status == http.StatusConflict {
+					continue
+				}
+				retry = append(retry, recordMap[f.Id])
+				if f.Status == http.StatusTooManyRequests {
+					//es is overloaded, backoff
+					overloaded = true
+				}
+			}
+			if overloaded {
+				level.Warn(d.logger).Log("message", "insert failed: elasticsearch is overloaded", "retry_count", len(retry))
 			}
 		}
+		return &InsertResponse{alreadyExistsIds, retry, overloaded}, nil
 	}
 
-	return err
+	return &InsertResponse{[]string{}, []*models.ElasticRecord{}, false}, nil
 }
 
 func (d recordDatabase) ReadinessCheck() bool {
@@ -82,7 +121,8 @@ func (d recordDatabase) ReadinessCheck() bool {
 func (d recordDatabase) buildBulkRequest(records []*models.ElasticRecord) (*elastic.BulkService, error) {
 	bulkRequest := d.GetClient().Bulk()
 	for _, record := range records {
-		bulkRequest.Add(elastic.NewBulkIndexRequest().Index(record.Index).
+		bulkRequest.Add(elastic.NewBulkIndexRequest().OpType("create").
+			Index(record.Index).
 			Type(record.Type).
 			Id(record.ID).
 			Doc(record.Json))

src/elasticsearch/elasticsearch.go

@@ -11,6 +11,8 @@ import (
 
 	"encoding/json"
 
+	"strconv"
+
 	"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/models"
@@ -72,8 +74,30 @@ func TestRecordDatabase_ReadinessCheck(t *testing.T) {
 
 func TestRecordDatabase_Insert(t *testing.T) {
 	record, id := fixtures.NewElasticRecord()
-	err := db.Insert([]*models.ElasticRecord{record})
+	_, err := db.Insert([]*models.ElasticRecord{record})
+	db.GetClient().Refresh("_all").Do(context.Background())
+	var recordFromES fixtures.FixtureRecord
+	if assert.NoError(t, err) {
+		count, err := db.GetClient().Count(record.Index).Do(context.Background())
+		if assert.NoError(t, err) {
+			assert.Equal(t, int64(1), count)
+		}
+		res, err := db.GetClient().Get().Index(record.Index).Type(record.Type).Id(record.ID).Do(context.Background())
+		if assert.NoError(t, err) {
+			json.Unmarshal(*res.Source, &recordFromES)
+		}
+		assert.Equal(t, recordFromES.Id, id)
+	}
+	db.GetClient().DeleteByQuery(record.Index).Query(elastic.MatchAllQuery{}).Do(context.Background())
+}
+
+func TestRecordDatabase_Insert_RepeatedId(t *testing.T) {
+	record, id := fixtures.NewElasticRecord()
+	_, err := db.Insert([]*models.ElasticRecord{record})
 	db.GetClient().Refresh("_all").Do(context.Background())
+	res, err := db.Insert([]*models.ElasticRecord{record})
+	assert.Len(t, res.AlreadyExists, 1)
+	assert.Contains(t, res.AlreadyExists, strconv.Itoa(int(id)))
 	var recordFromES fixtures.FixtureRecord
 	if assert.NoError(t, err) {
 		count, err := db.GetClient().Count(record.Index).Do(context.Background())
@@ -91,7 +115,7 @@ func TestRecordDatabase_Insert(t *testing.T) {
 
 func TestRecordDatabase_Insert_Multiple(t *testing.T) {
 	record, id := fixtures.NewElasticRecord()
-	err := db.Insert([]*models.ElasticRecord{record, record})
+	_, err := db.Insert([]*models.ElasticRecord{record, record})
 	db.GetClient().Refresh("_all").Do(context.Background())
 	var recordFromES fixtures.FixtureRecord
 	if assert.NoError(t, err) {

src/elasticsearch/elasticsearch_test.go

@@ -1,6 +1,8 @@
 package store
 
 import (
+	"time"
+
 	"github.com/go-kit/kit/log"
 	"github.com/inloco/kafka-elasticsearch-injector/src/elasticsearch"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
@@ -12,16 +14,31 @@ type Store interface {
 }
 
 type basicStore struct {
-	db    elasticsearch.RecordDatabase
-	codec elasticsearch.Codec
+	db      elasticsearch.RecordDatabase
+	codec   elasticsearch.Codec
+	backoff time.Duration
 }
 
 func (s basicStore) Insert(records []*models.Record) error {
 	elasticRecords, err := s.codec.EncodeElasticRecords(records)
 	if err != nil {
 		return err
 	}
-	return s.db.Insert(elasticRecords)
+	for {
+		res, err := s.db.Insert(elasticRecords)
+		if err != nil {
+			return err
+		}
+		if len(res.Retry) == 0 {
+			break
+		}
+		//some records failed to index, backoff(if overloaded) then retry
+		if res.Overloaded {
+			time.Sleep(s.backoff)
+		}
+		s.db.Insert(res.Retry)
+	}
+	return nil
 }
 
 func (s basicStore) ReadinessCheck() bool {
@@ -31,7 +48,8 @@ func (s basicStore) ReadinessCheck() bool {
 func NewStore(logger log.Logger) Store {
 	config := elasticsearch.NewConfig()
 	return basicStore{
-		db:    elasticsearch.NewDatabase(logger, config),
-		codec: elasticsearch.NewCodec(logger, config),
+		db:      elasticsearch.NewDatabase(logger, config),
+		codec:   elasticsearch.NewCodec(logger, config),
+		backoff: config.Backoff,
 	}
 }

src/injector/store/store.go

@@ -34,7 +34,8 @@ func (s fixtureService) Insert(records []*models.Record) error {
 	if err != nil {
 		return err
 	}
-	return s.db.Insert(elasticRecords)
+	_, err = s.db.Insert(elasticRecords)
+	return err
 }
 
 func (s fixtureService) ReadinessCheck() bool {