Merge pull request #10 from inloco/feature/hour-index

feat(codec): add hour index option

Author: racevedoo

Gopkg.lock

@@ -30,6 +30,7 @@ To create new injectors for your topics, you should create a new kubernetes depl
 - `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**
+- `ES_TIME_SUFFIX` Indicates what time unit to append to index names on elasticsearch. Supported values are `day` and `hour`. Default value is `day` **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.3
+0.6.0

VERSION

@@ -52,7 +52,10 @@ func (c basicCodec) getDatabaseIndex(record *models.Record) (string, error) {
 	}
 
 	indexColumn := c.config.IndexColumn
-	indexSuffix := record.FormatTimestamp()
+	indexSuffix := record.FormatTimestampDay()
+	if c.config.TimeSuffix == TimeSuffixHour {
+		indexSuffix = record.FormatTimestampHour()
+	}
 	if indexColumn != "" {
 		newIndexSuffix, err := record.GetValueForField(indexColumn)
 		if err != nil {

src/elasticsearch/codec.go

@@ -25,7 +25,27 @@ func TestCodec_EncodeElasticRecords(t *testing.T) {
 	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, fmt.Sprintf("%s-%s", record.Topic, record.FormatTimestampDay()), 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_EncodeElasticRecordsHourSuffix(t *testing.T) {
+	codec := &basicCodec{
+		config: Config{
+			TimeSuffix: TimeSuffixHour,
+		},
+		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.FormatTimestampHour()), 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"])

src/elasticsearch/codec_test.go

@@ -6,6 +6,13 @@ import (
 	"time"
 )
 
+type TimeIndexSuffix int
+
+const (
+	TimeSuffixDay  TimeIndexSuffix = 0
+	TimeSuffixHour TimeIndexSuffix = 1
+)
+
 type Config struct {
 	Host               string
 	Index              string
@@ -14,6 +21,7 @@ type Config struct {
 	BlacklistedColumns []string
 	BulkTimeout        time.Duration
 	Backoff            time.Duration
+	TimeSuffix         TimeIndexSuffix
 }
 
 func NewConfig() Config {
@@ -33,6 +41,13 @@ func NewConfig() Config {
 			backoff = d
 		}
 	}
+	timeSuffix := TimeSuffixDay
+	if suffix := os.Getenv("ES_TIME_SUFFIX"); suffix != "" {
+		switch suffix {
+		case "hour":
+			timeSuffix = TimeSuffixHour
+		}
+	}
 	return Config{
 		Host:               os.Getenv("ELASTICSEARCH_HOST"),
 		Index:              os.Getenv("ES_INDEX"),
@@ -41,5 +56,6 @@ func NewConfig() Config {
 		BlacklistedColumns: strings.Split(os.Getenv("ES_BLACKLISTED_COLUMNS"), ","),
 		BulkTimeout:        timeout,
 		Backoff:            backoff,
+		TimeSuffix:         timeSuffix,
 	}
 }

src/elasticsearch/config.go

@@ -15,10 +15,14 @@ type Record struct {
 	Json      map[string]interface{}
 }
 
-func (r *Record) FormatTimestamp() string {
+func (r *Record) FormatTimestampDay() string {
 	return r.Timestamp.Format("2006-01-02")
 }
 
+func (r *Record) FormatTimestampHour() string {
+	return r.Timestamp.Format("2006-01-02T15")
+}
+
 func (r *Record) GetId() string {
 	return fmt.Sprintf("%d:%d", r.Partition, r.Offset)
 }