Merge pull request #38 from inloco/feature/elasticsearch-disable-sniffing

feat(elasticsearch): adds option to disable sniffing

Author: ggml1

.circleci/config.yml

@@ -1,7 +1,7 @@
 version: 2.1
 defaults: &defaults
   environment:
-    DOCKER_REPO: inlocomedia/kafka-elasticsearch-injector-go
+    DOCKER_REPO: quay.io/inloco/kafka-elasticsearch-injector
 jobs:
   test:
     <<: *defaults
@@ -33,7 +33,7 @@ jobs:
           name: Login to Docker Hub
           command: |
             docker info
-            docker login --username $DOCKER_USER --password $DOCKER_PASSWORD
+            docker login --username $DOCKER_USER --password $DOCKER_PASSWORD quay.io
       - restore_cache:
           keys:
             - go-mod-v4-{{ checksum "go.sum" }}
@@ -106,7 +106,7 @@ jobs:
             docker load -i /tmp/workspace/image.tar
       - run:
           name: "docker login"
-          command: docker login --username $DOCKER_USER --password $DOCKER_PASSWORD
+          command: docker login --username $DOCKER_USER --password $DOCKER_PASSWORD quay.io
       - run:
           name: Push images
           command: |
@@ -128,13 +128,14 @@ workflows:
   test-and-deploy-image-for-commit:
     jobs:
       - test:
-        filters:
-          branches:
-            ignore: /master/
+          context: quay-push-credentials
+          filters:
+            branches:
+              ignore: /master/
       - build-image:
-        filters:
-          branches:
-            ignore: /master/
+          filters:
+            branches:
+              ignore: /master/
       - deploy-image:
           context: media
           requires:
@@ -145,6 +146,7 @@ workflows:
   test-and-deploy-image-for-tag:
     jobs:
       - test:
+          context: quay-push-credentials
           filters:
             tags:
               only: /.+/
@@ -165,4 +167,4 @@ workflows:
             tags:
               only: /.+/
             branches:
-              ignore: /.*/
+              ignore: /.*/

README.md

@@ -25,6 +25,7 @@ To create new injectors for your topics, you should create a new kubernetes depl
 - `ELASTICSEARCH_PASSWORD` Elasticsearch password. **OPTIONAL**
 - `ELASTICSEARCH_SCHEME` scheme to be used when connecting to elasticsearch(http or https). Defaults to http. **OPTIONAL**
 - `ELASTICSEARCH_IGNORE_CERT` if set to "true", ignores certificates when connecting to a secure elasticsearch cluster. Defaults to false. **OPTIONAL**
+- `ELASTICSEARCH_DISABLE_SNIFFING` if set to "true", the client will not sniff Elasticsearch nodes during the node discovery process. Defaults to false. **OPTIONAL**
 - `KAFKA_CONSUMER_CONCURRENCY` Number of parallel goroutines working as a consumer. Default value is 1 **OPTIONAL**
 - `KAFKA_CONSUMER_BATCH_SIZE` Number of records to accumulate before sending them to elasticsearch(for each goroutine). Default value is 100 **OPTIONAL**
 - `ES_INDEX_COLUMN` Record field to append to index name. Ex: to create one ES index per campaign, use "campaign_id" here **OPTIONAL**

go.sum

@@ -53,6 +53,7 @@ github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfb
 github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
@@ -173,6 +174,7 @@ golang.org/x/net v0.0.0-20190311183353-d8887717615a h1:oWX7TPOiFAMXLq8o0ikBYfCJV
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=

src/elasticsearch/config.go

@@ -27,6 +27,7 @@ type Config struct {
 	BulkTimeout        time.Duration
 	Backoff            time.Duration
 	TimeSuffix         TimeIndexSuffix
+	DisableSniffing    bool
 }
 
 func NewConfig() Config {
@@ -68,6 +69,15 @@ func NewConfig() Config {
 			scheme = c
 		}
 	}
+
+	disableSniff := false
+	if c := os.Getenv("ELASTICSEARCH_DISABLE_SNIFFING"); c != "" {
+		res, err := strconv.ParseBool(c)
+		if err == nil {
+			disableSniff = res
+		}
+	}
+
 	return Config{
 		Host:               os.Getenv("ELASTICSEARCH_HOST"),
 		User:               os.Getenv("ELASTICSEARCH_USER"),
@@ -81,5 +91,6 @@ func NewConfig() Config {
 		BulkTimeout:        timeout,
 		Backoff:            backoff,
 		TimeSuffix:         timeSuffix,
+		DisableSniffing:    disableSniff,
 	}
 }

src/elasticsearch/elasticsearch.go

@@ -48,6 +48,9 @@ func (d recordDatabase) GetClient() *elastic.Client {
 		if d.config.Scheme == "https" { // http is default
 			opts = append(opts, elastic.SetScheme(d.config.Scheme))
 		}
+		if d.config.DisableSniffing { // sniffing is enabled by default
+			opts = append(opts, elastic.SetSniff(!d.config.DisableSniffing))
+		}
 		client, err := elastic.NewClient(opts...)
 		if err != nil {
 			level.Error(d.logger).Log("err", err, "message", "could not init elasticsearch client")