ElasticSearch 5 support

common/elasticsearch/elasticsearch.go

@@ -20,9 +20,10 @@ import (
 	"time"
 
 	"github.com/golang/glog"
-	"github.com/pborman/uuid"
 
-	"gopkg.in/olivere/elastic.v3"
+	"errors"
+	elastic2 "gopkg.in/olivere/elastic.v3"
+	elastic5 "gopkg.in/olivere/elastic.v5"
 	"os"
 )
 
@@ -32,10 +33,9 @@ const (
 )
 
 type ElasticSearchService struct {
-	EsClient      *elastic.Client
-	bulkProcessor *elastic.BulkProcessor
-	baseIndex     string
-	ClusterName   string
+	EsClient    *esClient
+	baseIndex   string
+	ClusterName string
 }
 
 func (esSvc *ElasticSearchService) Index(date time.Time) string {
@@ -46,7 +46,7 @@ func (esSvc *ElasticSearchService) IndexAlias(date time.Time, typeName string) s
 }
 
 func (esSvc *ElasticSearchService) FlushData() error {
-	return esSvc.bulkProcessor.Flush()
+	return esSvc.EsClient.FlushBulk()
 }
 
 // SaveDataIntoES save metrics and events to ES by using ES client
@@ -58,44 +58,63 @@ func (esSvc *ElasticSearchService) SaveData(date time.Time, typeName string, sin
 	indexName := esSvc.Index(date)
 
 	// Use the IndexExists service to check if a specified index exists.
-	exists, err := esSvc.EsClient.IndexExists(indexName).Do()
+	exists, err := esSvc.EsClient.IndexExists(indexName)
 	if err != nil {
 		return err
 	}
+
 	if !exists {
 		// Create a new index.
-		createIndex, err := esSvc.EsClient.CreateIndex(indexName).BodyString(mapping).Do()
+		createIndex, err := esSvc.EsClient.CreateIndex(indexName, mapping)
 		if err != nil {
 			return err
 		}
-		if !createIndex.Acknowledged {
-			return fmt.Errorf("Failed to create Index in ES cluster: %s", err)
+
+		ack := false
+		switch i := createIndex.(type) {
+		case *elastic2.IndicesCreateResult:
+			ack = i.Acknowledged
+		case *elastic5.IndicesCreateResult:
+			ack = i.Acknowledged
+		}
+		if !ack {
+			return errors.New("Failed to acknoledge index creation")
 		}
 	}
 
-	aliases, err := esSvc.EsClient.Aliases().Index(indexName).Do()
+	aliases, err := esSvc.EsClient.GetAliases(indexName)
 	if err != nil {
 		return err
 	}
 	aliasName := esSvc.IndexAlias(date, typeName)
-	if !aliases.Indices[indexName].HasAlias(aliasName) {
-		createAlias, err := esSvc.EsClient.Alias().Add(indexName, esSvc.IndexAlias(date, typeName)).Do()
+
+	hasAlias := false
+	switch a := aliases.(type) {
+	case *elastic2.AliasesResult:
+		hasAlias = a.Indices[indexName].HasAlias(aliasName)
+	case *elastic5.AliasesResult:
+		hasAlias = a.Indices[indexName].HasAlias(aliasName)
+	}
+	if !hasAlias {
+		createAlias, err := esSvc.EsClient.AddAlias(indexName, esSvc.IndexAlias(date, typeName))
 		if err != nil {
 			return err
 		}
-		if !createAlias.Acknowledged {
-			return fmt.Errorf("Failed to create Index Alias in ES cluster: %s", err)
+
+		ack := false
+		switch i := createAlias.(type) {
+		case *elastic2.AliasResult:
+			ack = i.Acknowledged
+		case *elastic5.AliasResult:
+			ack = i.Acknowledged
+		}
+		if !ack {
+			return errors.New("Failed to acknoledge index alias creation")
 		}
 	}
 
 	for _, data := range sinkData {
-		indexID := uuid.NewUUID()
-		req := elastic.NewBulkIndexRequest().
-			Index(indexName).
-			Type(typeName).
-			Id(indexID.String()).
-			Doc(data)
-		esSvc.bulkProcessor.Add(req)
+		esSvc.EsClient.AddBulkReq(indexName, typeName, data)
 	}
 
 	return nil
@@ -111,6 +130,14 @@ func CreateElasticSearchService(uri *url.URL) (*ElasticSearchService, error) {
 		return nil, fmt.Errorf("Failed to parser url's query string: %s", err)
 	}
 
+	version := 5
+	if len(opts["ver"]) > 0 {
+		version, err = strconv.Atoi(opts["ver"][0])
+		if err != nil {
+			return nil, fmt.Errorf("Failed to parse URL's version value into an int: %v", err)
+		}
+	}
+
 	esSvc.ClusterName = ESClusterName
 	if len(opts["cluster_name"]) > 0 {
 		esSvc.ClusterName = opts["cluster_name"][0]
@@ -122,45 +149,53 @@ func CreateElasticSearchService(uri *url.URL) (*ElasticSearchService, error) {
 		esSvc.baseIndex = opts["index"][0]
 	}
 
+	var startupFnsV5 []elastic5.ClientOptionFunc
+	var startupFnsV2 []elastic2.ClientOptionFunc
+
 	// Set the URL endpoints of the ES's nodes. Notice that when sniffing is
 	// enabled, these URLs are used to initially sniff the cluster on startup.
-	var startupFns []elastic.ClientOptionFunc
 	if len(opts["nodes"]) > 0 {
-		startupFns = append(startupFns, elastic.SetURL(opts["nodes"]...))
+		startupFnsV2 = append(startupFnsV2, elastic2.SetURL(opts["nodes"]...))
+		startupFnsV5 = append(startupFnsV5, elastic5.SetURL(opts["nodes"]...))
 	} else if uri.Scheme != "" && uri.Host != "" {
-		startupFns = append(startupFns, elastic.SetURL(uri.Scheme+"://"+uri.Host))
+		startupFnsV2 = append(startupFnsV2, elastic2.SetURL(uri.Scheme+"://"+uri.Host))
+		startupFnsV5 = append(startupFnsV5, elastic5.SetURL(uri.Scheme+"://"+uri.Host))
 	} else {
-		return nil, fmt.Errorf("There is no node assigned for connecting ES cluster")
+		return nil, errors.New("There is no node assigned for connecting ES cluster")
 	}
 
 	// If the ES cluster needs authentication, the username and secret
 	// should be set in sink config.Else, set the Authenticate flag to false
 	if len(opts["esUserName"]) > 0 && len(opts["esUserSecret"]) > 0 {
-		startupFns = append(startupFns, elastic.SetBasicAuth(opts["esUserName"][0], opts["esUserSecret"][0]))
+		startupFnsV2 = append(startupFnsV2, elastic2.SetBasicAuth(opts["esUserName"][0], opts["esUserSecret"][0]))
+		startupFnsV5 = append(startupFnsV5, elastic5.SetBasicAuth(opts["esUserName"][0], opts["esUserSecret"][0]))
 	}
 
 	if len(opts["maxRetries"]) > 0 {
 		maxRetries, err := strconv.Atoi(opts["maxRetries"][0])
 		if err != nil {
-			return nil, fmt.Errorf("Failed to parse URL's maxRetries value into an int")
+			return nil, errors.New("Failed to parse URL's maxRetries value into an int")
 		}
-		startupFns = append(startupFns, elastic.SetMaxRetries(maxRetries))
+		startupFnsV2 = append(startupFnsV2, elastic2.SetMaxRetries(maxRetries))
+		startupFnsV5 = append(startupFnsV5, elastic5.SetMaxRetries(maxRetries))
 	}
 
 	if len(opts["healthCheck"]) > 0 {
 		healthCheck, err := strconv.ParseBool(opts["healthCheck"][0])
 		if err != nil {
-			return nil, fmt.Errorf("Failed to parse URL's healthCheck value into a bool")
+			return nil, errors.New("Failed to parse URL's healthCheck value into a bool")
 		}
-		startupFns = append(startupFns, elastic.SetHealthcheck(healthCheck))
+		startupFnsV2 = append(startupFnsV2, elastic2.SetHealthcheck(healthCheck))
+		startupFnsV5 = append(startupFnsV5, elastic5.SetHealthcheck(healthCheck))
 	}
 
 	if len(opts["startupHealthcheckTimeout"]) > 0 {
 		timeout, err := time.ParseDuration(opts["startupHealthcheckTimeout"][0] + "s")
 		if err != nil {
 			return nil, fmt.Errorf("Failed to parse URL's startupHealthcheckTimeout: %s", err.Error())
 		}
-		startupFns = append(startupFns, elastic.SetHealthcheckTimeoutStartup(timeout))
+		startupFnsV2 = append(startupFnsV2, elastic2.SetHealthcheckTimeoutStartup(timeout))
+		startupFnsV5 = append(startupFnsV5, elastic5.SetHealthcheckTimeoutStartup(timeout))
 	}
 
 	if os.Getenv("AWS_ACCESS_KEY_ID") != "" || os.Getenv("AWS_ACCESS_KEY") != "" ||
@@ -172,58 +207,45 @@ func CreateElasticSearchService(uri *url.URL) (*ElasticSearchService, error) {
 			return nil, err
 		}
 
-		startupFns = append(startupFns, elastic.SetHttpClient(awsClient), elastic.SetSniff(false))
+		startupFnsV2 = append(startupFnsV2, elastic2.SetHttpClient(awsClient), elastic2.SetSniff(false))
+		startupFnsV5 = append(startupFnsV5, elastic5.SetHttpClient(awsClient), elastic5.SetSniff(false))
 	} else {
 		if len(opts["sniff"]) > 0 {
 			sniff, err := strconv.ParseBool(opts["sniff"][0])
 			if err != nil {
-				return nil, fmt.Errorf("Failed to parse URL's sniff value into a bool")
+				return nil, errors.New("Failed to parse URL's sniff value into a bool")
 			}
-			startupFns = append(startupFns, elastic.SetSniff(sniff))
+			startupFnsV2 = append(startupFnsV2, elastic2.SetSniff(sniff))
+			startupFnsV5 = append(startupFnsV5, elastic5.SetSniff(sniff))
 		}
 	}
 
-	esSvc.EsClient, err = elastic.NewClient(startupFns...)
-	if err != nil {
-		return nil, fmt.Errorf("Failed to create ElasticSearch client: %v", err)
-	}
-
 	bulkWorkers := 5
 	if len(opts["bulkWorkers"]) > 0 {
 		bulkWorkers, err = strconv.Atoi(opts["bulkWorkers"][0])
 		if err != nil {
-			return nil, fmt.Errorf("Failed to parse URL's bulkWorkers value into an int")
+			return nil, errors.New("Failed to parse URL's bulkWorkers value into an int")
 		}
 	}
-	esSvc.bulkProcessor, err = esSvc.EsClient.BulkProcessor().
-		Name("ElasticSearchWorker").
-		Workers(bulkWorkers).
-		After(bulkAfterCB).
-		BulkActions(1000).               // commit if # requests >= 1000
-		BulkSize(2 << 20).               // commit if size of requests >= 2 MB
-		FlushInterval(10 * time.Second). // commit every 10s
-		Do()
+
+	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, pipeline)
+	default:
+		return nil, UnsupportedVersion{}
+	}
 	if err != nil {
-		return nil, fmt.Errorf("Failed to an ElasticSearch Bulk Processor: %v", err)
+		return nil, fmt.Errorf("Failed to create ElasticSearch client: %v", err)
 	}
 
 	glog.V(2).Infof("ElasticSearch sink configure successfully")
 
 	return &esSvc, nil
 }
-
-func bulkAfterCB(executionId int64, requests []elastic.BulkableRequest, response *elastic.BulkResponse, err error) {
-	if err != nil {
-		glog.Warningf("Failed to execute bulk operation to ElasticSearch: %v", err)
-	}
-
-	if response.Errors {
-		for _, list := range response.Items {
-			for name, itm := range list {
-				if itm.Error != nil {
-					glog.V(3).Infof("Failed to execute bulk operation to ElasticSearch on %s: %v", name, itm.Error)
-				}
-			}
-		}
-	}
-}