ElasticSearch: Add supports for Ignest pipelines

This commit adds support for ElasticSearch ingest pipelines.  This
only works on ES5+.

Author: AlmogBaku

common/elasticsearch/elasticsearch.go

@@ -228,11 +228,16 @@ func CreateElasticSearchService(uri *url.URL) (*ElasticSearchService, error) {
 		}
 	}
 
+	pipeline := ""
+	if len(opts["pipeline"]) > 0 {
+		pipeline = opts["pipeline"][0]
+	}
+
 	switch version {
 	case 2:
 		esSvc.EsClient, err = newEsClientV2(startupFnsV2, bulkWorkers)
 	case 5:
-		esSvc.EsClient, err = newEsClientV5(startupFnsV5, bulkWorkers)
+		esSvc.EsClient, err = newEsClientV5(startupFnsV5, bulkWorkers, pipeline)
 	default:
 		return nil, UnsupportedVersion{}
 	}

common/elasticsearch/elasticsearch_test.go

@@ -146,6 +146,32 @@ func TestCreateElasticSearchServiceForDefaultClusterName(t *testing.T) {
 	}
 }
 
+func TestCreateElasticSearchServiceWithIngestPipeline(t *testing.T) {
+	const pipeline = "test"
+
+	esURI := "?nodes=https://foo.com:20468&nodes=https://bar.com:20468&" +
+		"pipeline=" + pipeline + "&" +
+		"sniff=false&healthCheck=false"
+
+	url, err := url.Parse(esURI)
+	if err != nil {
+		t.Fatalf("Error when parsing URL: %s", err.Error())
+	}
+
+	esSvc, err := CreateElasticSearchService(url)
+	if err != nil {
+		t.Fatalf("Error when creating config: %s", err.Error())
+	}
+
+	if esSvc.EsClient.pipeline != pipeline {
+		t.Fatalf("Ingest pipline is not equal. Expected: %s, Got: %s", pipeline, esSvc.EsClient.pipeline)
+	}
+
+	if esSvc.EsClient.version < 5 {
+		t.Fatalf("ElasticSearch client not using version 5+ (required for pipeline). Got: %d", esSvc.EsClient.version)
+	}
+}
+
 func TestCreateElasticSearchServiceSingleDnsEntrypointV5(t *testing.T) {
 	clusterName := "sandbox"
 	esURI := fmt.Sprintf("https://foo.com:9200?"+

common/elasticsearch/esVersionManager.go

@@ -35,12 +35,13 @@ type esClient struct {
 	clientV5        *elastic5.Client
 	bulkProcessorV2 *elastic2.BulkProcessor
 	bulkProcessorV5 *elastic5.BulkProcessor
+	pipeline        string
 }
 
 func NewMockClient() *esClient {
 	return &esClient{}
 }
-func newEsClientV5(startupFns []elastic5.ClientOptionFunc, bulkWorkers int) (*esClient, error) {
+func newEsClientV5(startupFns []elastic5.ClientOptionFunc, bulkWorkers int, pipeline string) (*esClient, error) {
 	client, err := elastic5.NewClient(startupFns...)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to an ElasticSearch Client: %v", err)
@@ -56,7 +57,7 @@ func newEsClientV5(startupFns []elastic5.ClientOptionFunc, bulkWorkers int) (*es
 	if err != nil {
 		return nil, fmt.Errorf("Failed to an ElasticSearch Bulk Processor: %v", err)
 	}
-	return &esClient{version: 5, clientV5: client, bulkProcessorV5: bps}, nil
+	return &esClient{version: 5, clientV5: client, bulkProcessorV5: bps, pipeline: pipeline}, nil
 }
 func newEsClientV2(startupFns []elastic2.ClientOptionFunc, bulkWorkers int) (*esClient, error) {
 	client, err := elastic2.NewClient(startupFns...)
@@ -131,11 +132,16 @@ func (es *esClient) AddBulkReq(index, typeName string, data interface{}) error {
 			Doc(data))
 		return nil
 	case 5:
-		es.bulkProcessorV5.Add(elastic5.NewBulkIndexRequest().
+		req := elastic5.NewBulkIndexRequest().
 			Index(index).
 			Type(typeName).
 			Id(uuid.NewUUID().String()).
-			Doc(data))
+			Doc(data)
+		if es.pipeline != "" {
+			req.Pipeline(es.pipeline)
+		}
+
+		es.bulkProcessorV5.Add(req)
 		return nil
 	default:
 		return UnsupportedVersion{}

docs/sink-configuration.md

@@ -197,7 +197,7 @@ Besides this, the following options can be set in query string:
 * `ver` - ElasticSearch cluster version, can be either `2` or `5`. The default is `5`
 * `bulkWorkers` - number of workers for bulk processing. Default value is `5`.
 * `cluster_name` - cluster name for different Kubernetes clusters. Default value is `default`.
-
+* `pipeline` - (optional; >ES5) Ingest Pipeline to process the documents. The default is disabled(empty value)
 
 #### AWS Integration
 In order to use AWS Managed Elastic we need to use one of the following methods: