feat(elasticsearch): remove types to support elasticsearch 7.0

Author: racevedoo

Gopkg.lock

@@ -15,5 +15,5 @@ docker/run:
 	count=0; \
           until curl localhost:9200 || ((count ++ >= 10)); \
           do echo "Retrying: Verify if Elasticsearch is ready"; sleep 5; done
-	curl -XPOST "localhost:9200/_template/my-topic" --data '{"template":"my-topic-*","settings":{"refresh_interval":"30s","number_of_replicas":0},"mappings":{"_default_":{"_all":{"enabled":"false"},"_source":{"enabled":"true"},"properties":{"@timestamp":{"format":"epoch_millis","ignore_malformed":true,"type":"date"}},"dynamic_templates":[{"strings":{"match_mapping_type":"string","mapping":{"type":"text","index":false}}}]}}}'
+	curl -XPOST -H "Content-Type: application/json" "localhost:9200/_template/my-topic" --data '{"template":"my-topic-*","settings":{"refresh_interval":"30s","number_of_replicas":0},"mappings":{"_source":{"enabled":"true"},"properties":{"@timestamp":{"format":"epoch_millis","ignore_malformed":true,"type":"date"}},"dynamic_templates":[{"strings":{"match_mapping_type":"string","mapping":{"type":"keyword","index":true}}}]}}'
 	docker-compose up -d producer app

Makefile

@@ -25,12 +25,23 @@ services:
     ports:
       - "8081:8081"
   elasticsearch:
-    image: 'elasticsearch:5-alpine'
+    image: 'docker.elastic.co/elasticsearch/elasticsearch-oss:7.0.0'
     container_name: elasticsearch
+    environment:
+      - cluster.name=docker-cluster
+      - bootstrap.memory_lock=true
+      - discovery.type=single-node
+      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
     ports:
-      - '9200:9200'
+      - "9200:9200"
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    volumes:
+      - esdata1:/usr/share/elasticsearch/data
   kibana:
-    image: 'kibana:5'
+    image: 'docker.elastic.co/kibana/kibana-oss:7.0.0'
     container_name: kibana
     depends_on:
       - elasticsearch
@@ -70,4 +81,8 @@ services:
     environment:
       - KAFKA_ADDRESS=kafka:9092
       - SCHEMA_REGISTRY_URL=http://schema-registry:8081
-      - LOG_LEVEL=DEBUG
+      - LOG_LEVEL=DEBUG
+
+volumes:
+  esdata1:
+    driver: local

docker-compose.yml

@@ -36,7 +36,7 @@ func (c basicCodec) EncodeElasticRecords(records []*models.Record) ([]*models.El
 
 		elasticRecords[idx] = &models.ElasticRecord{
 			Index: index,
-			Type:  record.Topic,
+			Type:  "_doc",
 			ID:    docID,
 			Json:  record.FilteredFieldsJSON(c.config.BlacklistedColumns),
 		}

src/elasticsearch/codec.go

@@ -2,9 +2,10 @@ package elasticsearch
 
 import (
 	"context"
-
 	"fmt"
 
+	"github.com/pkg/errors"
+
 	"net/http"
 
 	"github.com/go-kit/kit/log"
@@ -53,7 +54,7 @@ func (d recordDatabase) CloseClient() {
 type InsertResponse struct {
 	AlreadyExists []string
 	Retry         []*models.ElasticRecord
-	Overloaded    bool
+	Backoff       bool
 }
 
 func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse, error) {
@@ -65,7 +66,10 @@ func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse
 	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)
+		return &InsertResponse{AlreadyExists: nil, Retry: records, Backoff: true}, nil
+	}
 	if err != nil {
 		return nil, err
 	}

src/elasticsearch/elasticsearch.go

@@ -33,10 +33,10 @@ func (s basicStore) Insert(records []*models.Record) error {
 			break
 		}
 		//some records failed to index, backoff(if overloaded) then retry
-		if res.Overloaded {
+		if res.Backoff {
 			time.Sleep(s.backoff)
 		}
-		s.db.Insert(res.Retry)
+		elasticRecords = res.Retry
 	}
 	return nil
 }