feat(elastic_codecs): adding codecs that transforms a kafka record into a record to be inserted in elasticsearch

Author: Mateus Moury

src/elasticsearch/codec.go

@@ -0,0 +1,81 @@
+package elasticsearch
+
+import (
+	"fmt"
+
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
+	"github.com/inloco/kafka-elasticsearch-injector/src/models"
+)
+
+type Codec interface {
+	EncodeElasticRecords(records []*models.Record) ([]*models.ElasticRecord, error)
+}
+
+type basicCodec struct {
+	config Config
+	logger log.Logger
+}
+
+func NewCodec(logger log.Logger, config Config) Codec {
+	return basicCodec{logger: logger, config: config}
+}
+
+func (c basicCodec) EncodeElasticRecords(records []*models.Record) ([]*models.ElasticRecord, error) {
+	elasticRecords := make([]*models.ElasticRecord, len(records))
+	for idx, record := range records {
+		index, err := c.getDatabaseIndex(record)
+		if err != nil {
+			return nil, err
+		}
+
+		docID, err := c.getDatabaseDocID(record)
+		if err != nil {
+			return nil, err
+		}
+
+		elasticRecords[idx] = &models.ElasticRecord{
+			Index: index,
+			Type:  record.Topic,
+			ID:    docID,
+			Json:  record.FilteredFieldsJSON(c.config.BlacklistedColumns),
+		}
+	}
+
+	return elasticRecords, nil
+}
+
+func (c basicCodec) getDatabaseIndex(record *models.Record) (string, error) {
+	indexPrefix := c.config.Index
+	if indexPrefix == "" {
+		indexPrefix = record.Topic
+	}
+
+	indexColumn := c.config.IndexColumn
+	indexSuffix := record.FormatTimestamp()
+	if indexColumn != "" {
+		newIndexSuffix, err := record.GetValueForField(indexColumn)
+		if err != nil {
+			level.Error(c.logger).Log("err", err, "message", "Could not get column value from record.")
+			return "", err
+		}
+		indexSuffix = newIndexSuffix
+	}
+
+	return fmt.Sprintf("%s-%s", indexPrefix, indexSuffix), nil
+}
+
+func (c basicCodec) getDatabaseDocID(record *models.Record) (string, error) {
+	docID := record.GetId()
+
+	docIDColumn := c.config.DocIDColumn
+	if docIDColumn != "" {
+		newDocID, err := record.GetValueForField(docIDColumn)
+		if err != nil {
+			level.Error(c.logger).Log("err", err, "message", "Could not get doc id value from record.")
+			return "", err
+		}
+		docID = newDocID
+	}
+	return docID, nil
+}

src/elasticsearch/codec_test.go

@@ -0,0 +1,79 @@
+package elasticsearch
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"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"
+)
+
+var codecLogger = logger_builder.NewLogger("elasticsearch-test")
+
+func TestCodec_EncodeElasticRecords(t *testing.T) {
+	codec := &basicCodec{
+		config: Config{},
+		logger: codecLogger,
+	}
+	record, id, value := fixtures.NewRecord(time.Now())
+
+	elasticRecords, err := codec.EncodeElasticRecords([]*models.Record{record})
+	if assert.NoError(t, err) && assert.Len(t, elasticRecords, 1) {
+		elasticRecord := elasticRecords[0]
+		assert.Equal(t, fmt.Sprintf("%s-%s", record.Topic, record.FormatTimestamp()), elasticRecord.Index)
+		assert.Equal(t, record.Topic, elasticRecord.Type)
+		assert.Equal(t, fmt.Sprintf("%d:%d", record.Partition, record.Offset), elasticRecord.ID)
+		assert.Equal(t, id, elasticRecord.Json["id"])
+		assert.Equal(t, value, elasticRecord.Json["value"])
+	}
+}
+
+func TestCodec_EncodeElasticRecords_ColumnsBlacklist(t *testing.T) {
+	codec := &basicCodec{
+		config: Config{BlacklistedColumns: []string{"value"}},
+		logger: codecLogger,
+	}
+	record, _, _ := fixtures.NewRecord(time.Now())
+
+	elasticRecords, err := codec.EncodeElasticRecords([]*models.Record{record})
+	if assert.NoError(t, err) && assert.Len(t, elasticRecords, 1) {
+		elasticRecord := elasticRecords[0]
+		assert.Contains(t, elasticRecord.Json, "id")
+		assert.NotContains(t, elasticRecord.Json, "value")
+	}
+}
+
+func TestCodec_EncodeElasticRecords_IndexColumn(t *testing.T) {
+	indexPrefix := "prefix"
+
+	codec := &basicCodec{
+		config: Config{Index: indexPrefix, IndexColumn: "id"},
+		logger: codecLogger,
+	}
+	record, id, _ := fixtures.NewRecord(time.Now())
+
+	elasticRecords, err := codec.EncodeElasticRecords([]*models.Record{record})
+	if assert.NoError(t, err) && assert.Len(t, elasticRecords, 1) {
+		elasticRecord := elasticRecords[0]
+		assert.Equal(t, fmt.Sprintf("%v-%v", indexPrefix, id), elasticRecord.Index)
+	}
+}
+
+func TestCodec_EncodeElasticRecords_DocIDColumn(t *testing.T) {
+	codec := &basicCodec{
+		config: Config{DocIDColumn: "id"},
+		logger: codecLogger,
+	}
+	record, id, _ := fixtures.NewRecord(time.Now())
+
+	elasticRecords, err := codec.EncodeElasticRecords([]*models.Record{record})
+	if assert.NoError(t, err) && assert.Len(t, elasticRecords, 1) {
+		elasticRecord := elasticRecords[0]
+		assert.Equal(t, strconv.Itoa(int(id)), elasticRecord.ID)
+	}
+}