Add Elasticsearch 7 support

Gopkg.toml

@@ -116,8 +116,8 @@ required = [
   version = "=1.20.1"
 
 [[constraint]]
-  name = "gopkg.in/olivere/elastic.v5"
-  version = "^5.0.53"
+  name = "github.com/olivere/elastic"
+  version = "^6.2.22"
 
 [[constraint]]
   name = "github.com/gocql/gocql"
@@ -174,4 +174,4 @@ required = [
 
 [[override]]
   name = "github.com/Shopify/sarama"
-  revision = "cd910a683f9faa57222e5120d17b60d2e65f7fa9"
+  revision = "cd910a683f9faa57222e5120d17b60d2e65f7fa9"

pkg/es/client.go

@@ -18,7 +18,7 @@ import (
 	"context"
 	"io"
 
-	"gopkg.in/olivere/elastic.v5"
+	"github.com/olivere/elastic"
 )
 
 // Client is an abstraction for elastic.Client
@@ -30,6 +30,7 @@ type Client interface {
 	Search(indices ...string) SearchService
 	MultiSearch() MultiSearchService
 	io.Closer
+	GetVersion() int
 }
 
 // IndicesExistsService is an abstraction for elastic.IndicesExistsService
@@ -60,7 +61,6 @@ type IndexService interface {
 
 // SearchService is an abstraction for elastic.SearchService
 type SearchService interface {
-	Type(typ string) SearchService
 	Size(size int) SearchService
 	Aggregation(name string, aggregation elastic.Aggregation) SearchService
 	IgnoreUnavailable(ignoreUnavailable bool) SearchService

pkg/es/config/config.go

@@ -21,14 +21,15 @@ import (
 	"io/ioutil"
 	"net/http"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/olivere/elastic"
 	"github.com/pkg/errors"
 	"github.com/uber/jaeger-lib/metrics"
 	"go.uber.org/zap"
-	"gopkg.in/olivere/elastic.v5"
 
 	"github.com/jaegertracing/jaeger/pkg/es"
 	"github.com/jaegertracing/jaeger/pkg/es/wrapper"
@@ -154,7 +155,19 @@ func (c *Configuration) NewClient(logger *zap.Logger, metricsFactory metrics.Fac
 	if err != nil {
 		return nil, err
 	}
-	return eswrapper.WrapESClient(rawClient, service), nil
+
+	// Determine ElasticSearch Version
+	pingResult, _, err := rawClient.Ping(c.Servers[0]).Do(context.Background())
+	if err != nil {
+		return nil, err
+	}
+	esVersion, err := strconv.Atoi(string(pingResult.Version.Number[0]))
+	if err != nil {
+		return nil, err
+	}
+	logger.Info("Elasticsearch detected", zap.Int("version", esVersion))
+
+	return eswrapper.WrapESClient(rawClient, service, esVersion), nil
 }
 
 // ApplyDefaults copies settings from source unless its own value is non-zero.

pkg/es/wrapper/wrapper.go

@@ -17,7 +17,7 @@ package eswrapper
 import (
 	"context"
 
-	"gopkg.in/olivere/elastic.v5"
+	"github.com/olivere/elastic"
 
 	"github.com/jaegertracing/jaeger/pkg/es"
 )
@@ -28,11 +28,17 @@ import (
 type ClientWrapper struct {
 	client      *elastic.Client
 	bulkService *elastic.BulkProcessor
+	esVersion   int
+}
+
+// GetVersion returns the ElasticSearch Version
+func (c ClientWrapper) GetVersion() int {
+	return c.esVersion
 }
 
 // WrapESClient creates a ESClient out of *elastic.Client.
-func WrapESClient(client *elastic.Client, s *elastic.BulkProcessor) ClientWrapper {
-	return ClientWrapper{client: client, bulkService: s}
+func WrapESClient(client *elastic.Client, s *elastic.BulkProcessor, esVersion int) ClientWrapper {
+	return ClientWrapper{client: client, bulkService: s, esVersion: esVersion}
 }
 
 // IndexExists calls this function to internal client.
@@ -53,17 +59,25 @@ func (c ClientWrapper) CreateTemplate(ttype string) es.TemplateCreateService {
 // Index calls this function to internal client.
 func (c ClientWrapper) Index() es.IndexService {
 	r := elastic.NewBulkIndexRequest()
-	return WrapESIndexService(r, c.bulkService)
+	return WrapESIndexService(r, c.bulkService, c.esVersion)
 }
 
 // Search calls this function to internal client.
 func (c ClientWrapper) Search(indices ...string) es.SearchService {
-	return WrapESSearchService(c.client.Search(indices...))
+	searchService := c.client.Search(indices...)
+	if c.esVersion == 7 {
+		searchService = searchService.RestTotalHitsAsInt(true)
+	}
+	return WrapESSearchService(searchService)
 }
 
 // MultiSearch calls this function to internal client.
 func (c ClientWrapper) MultiSearch() es.MultiSearchService {
-	return WrapESMultiSearchService(c.client.MultiSearch())
+	multiSearchService := c.client.MultiSearch()
+	if c.esVersion == 7 {
+		multiSearchService = multiSearchService.RestTotalHitsAsInt(true)
+	}
+	return WrapESMultiSearchService(multiSearchService)
 }
 
 // Close closes ESClient and flushes all data to the storage.
@@ -137,21 +151,25 @@ func (c TemplateCreateServiceWrapper) Do(ctx context.Context) (*elastic.IndicesP
 type IndexServiceWrapper struct {
 	bulkIndexReq *elastic.BulkIndexRequest
 	bulkService  *elastic.BulkProcessor
+	esVersion    int
 }
 
 // WrapESIndexService creates an ESIndexService out of *elastic.ESIndexService.
-func WrapESIndexService(indexService *elastic.BulkIndexRequest, bulkService *elastic.BulkProcessor) IndexServiceWrapper {
-	return IndexServiceWrapper{bulkIndexReq: indexService, bulkService: bulkService}
+func WrapESIndexService(indexService *elastic.BulkIndexRequest, bulkService *elastic.BulkProcessor, esVersion int) IndexServiceWrapper {
+	return IndexServiceWrapper{bulkIndexReq: indexService, bulkService: bulkService, esVersion: esVersion}
 }
 
 // Index calls this function to internal service.
 func (i IndexServiceWrapper) Index(index string) es.IndexService {
-	return WrapESIndexService(i.bulkIndexReq.Index(index), i.bulkService)
+	return WrapESIndexService(i.bulkIndexReq.Index(index), i.bulkService, i.esVersion)
 }
 
 // Type calls this function to internal service.
 func (i IndexServiceWrapper) Type(typ string) es.IndexService {
-	return WrapESIndexService(i.bulkIndexReq.Type(typ), i.bulkService)
+	if i.esVersion == 7 {
+		return WrapESIndexService(i.bulkIndexReq, i.bulkService, i.esVersion)
+	}
+	return WrapESIndexService(i.bulkIndexReq.Type(typ), i.bulkService, i.esVersion)
 }
 
 // Add adds the request to bulk service
@@ -171,11 +189,6 @@ func WrapESSearchService(searchService *elastic.SearchService) SearchServiceWrap
 	return SearchServiceWrapper{searchService: searchService}
 }
 
-// Type calls this function to internal service.
-func (s SearchServiceWrapper) Type(typ string) es.SearchService {
-	return WrapESSearchService(s.searchService.Type(typ))
-}
-
 // Size calls this function to internal service.
 func (s SearchServiceWrapper) Size(size int) es.SearchService {
 	return WrapESSearchService(s.searchService.Size(size))

pkg/es/wrapper/wrapper_nolint.go

@@ -21,10 +21,10 @@ import "github.com/jaegertracing/jaeger/pkg/es"
 
 // Id calls this function to internal service.
 func (i IndexServiceWrapper) Id(id string) es.IndexService {
-	return WrapESIndexService(i.bulkIndexReq.Id(id), i.bulkService)
+	return WrapESIndexService(i.bulkIndexReq.Id(id), i.bulkService, i.esVersion)
 }
 
 // BodyJson calls this function to internal service.
 func (i IndexServiceWrapper) BodyJson(body interface{}) es.IndexService {
-	return WrapESIndexService(i.bulkIndexReq.Doc(body), i.bulkService)
+	return WrapESIndexService(i.bulkIndexReq.Doc(body), i.bulkService, i.esVersion)
 }

plugin/storage/es/dependencystore/schema.go

@@ -22,3 +22,17 @@ const dependenciesMapping = `{
       "` + dependencyType + `":{}
    }
 }`
+
+const dependenciesMapping7 = `{
+   "settings":{
+      "index.requests.cache.enable":true
+   },
+   "mappings":{}
+}`
+
+func getMapping(version int) string {
+	if version == 7 {
+		return dependenciesMapping7
+	}
+	return dependenciesMapping
+}

plugin/storage/es/dependencystore/storage.go

@@ -19,9 +19,9 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/olivere/elastic"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
-	"gopkg.in/olivere/elastic.v5"
 
 	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/pkg/es"
@@ -66,7 +66,7 @@ func (s *DependencyStore) WriteDependencies(ts time.Time, dependencies []model.D
 }
 
 func (s *DependencyStore) createIndex(indexName string) error {
-	_, err := s.client.CreateIndex(indexName).Body(dependenciesMapping).Do(s.ctx)
+	_, err := s.client.CreateIndex(indexName).Body(getMapping(s.client.GetVersion())).Do(s.ctx)
 	if err != nil {
 		return errors.Wrap(err, "Failed to create index")
 	}
@@ -84,7 +84,6 @@ func (s *DependencyStore) writeDependencies(indexName string, ts time.Time, depe
 func (s *DependencyStore) GetDependencies(endTs time.Time, lookback time.Duration) ([]model.DependencyLink, error) {
 	indices := getIndices(s.indexPrefix, endTs, lookback)
 	searchResult, err := s.client.Search(indices...).
-		Type(dependencyType).
 		Size(10000). // the default elasticsearch allowed limit
 		Query(buildTSQuery(endTs, lookback)).
 		IgnoreUnavailable(true).