Add Elasticsearch version flag

pkg/es/client.go

@@ -31,7 +31,7 @@ type Client interface {
 	Search(indices ...string) SearchService
 	MultiSearch() MultiSearchService
 	io.Closer
-	GetVersion() int
+	GetVersion() uint
 }
 
 // IndicesExistsService is an abstraction for elastic.IndicesExistsService

pkg/es/config/config.go

@@ -63,6 +63,7 @@ type Configuration struct {
 	TLS                   TLSConfig
 	UseReadWriteAliases   bool
 	CreateIndexTemplates  bool
+	Version               uint
 }
 
 // TLSConfig describes the configuration properties to connect tls enabled ElasticSearch cluster
@@ -89,6 +90,7 @@ type ClientBuilder interface {
 	GetTokenFilePath() string
 	IsEnabled() bool
 	IsCreateIndexTemplates() bool
+	GetVersion() uint
 }
 
 // NewClient creates a new ElasticSearch client
@@ -157,18 +159,21 @@ func (c *Configuration) NewClient(logger *zap.Logger, metricsFactory metrics.Fac
 		return nil, err
 	}
 
-	// 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
+	if c.Version == 0 {
+		// 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))
+		c.Version = uint(esVersion)
 	}
-	logger.Info("Elasticsearch detected", zap.Int("version", esVersion))
 
-	return eswrapper.WrapESClient(rawClient, service, esVersion), nil
+	return eswrapper.WrapESClient(rawClient, service, c.Version), nil
 }
 
 // ApplyDefaults copies settings from source unless its own value is non-zero.
@@ -243,6 +248,11 @@ func (c *Configuration) GetAllTagsAsFields() bool {
 	return c.AllTagsAsFields
 }
 
+// GetVersion returns Elasticsearch version
+func (c *Configuration) GetVersion() uint {
+	return c.Version
+}
+
 // GetTagDotReplacement returns character is used to replace dots in tag keys, when
 // the tag is stored as object field.
 func (c *Configuration) GetTagDotReplacement() string {

pkg/es/wrapper/wrapper.go

@@ -29,16 +29,16 @@ import (
 type ClientWrapper struct {
 	client      *elastic.Client
 	bulkService *elastic.BulkProcessor
-	esVersion   int
+	esVersion   uint
 }
 
 // GetVersion returns the ElasticSearch Version
-func (c ClientWrapper) GetVersion() int {
+func (c ClientWrapper) GetVersion() uint {
 	return c.esVersion
 }
 
 // WrapESClient creates a ESClient out of *elastic.Client.
-func WrapESClient(client *elastic.Client, s *elastic.BulkProcessor, esVersion int) ClientWrapper {
+func WrapESClient(client *elastic.Client, s *elastic.BulkProcessor, esVersion uint) ClientWrapper {
 	return ClientWrapper{client: client, bulkService: s, esVersion: esVersion}
 }
 
@@ -152,11 +152,11 @@ func (c TemplateCreateServiceWrapper) Do(ctx context.Context) (*elastic.IndicesP
 type IndexServiceWrapper struct {
 	bulkIndexReq *elastic.BulkIndexRequest
 	bulkService  *elastic.BulkProcessor
-	esVersion    int
+	esVersion    uint
 }
 
 // WrapESIndexService creates an ESIndexService out of *elastic.ESIndexService.
-func WrapESIndexService(indexService *elastic.BulkIndexRequest, bulkService *elastic.BulkProcessor, esVersion int) IndexServiceWrapper {
+func WrapESIndexService(indexService *elastic.BulkIndexRequest, bulkService *elastic.BulkProcessor, esVersion uint) IndexServiceWrapper {
 	return IndexServiceWrapper{bulkIndexReq: indexService, bulkService: bulkService, esVersion: esVersion}
 }
 

plugin/storage/es/dependencystore/schema.go

@@ -31,7 +31,7 @@ const dependenciesMapping7 = `{
    "mappings":{}
 }`
 
-func getMapping(version int) string {
+func getMapping(version uint) string {
 	if version == 7 {
 		return dependenciesMapping7
 	}

plugin/storage/es/dependencystore/storage_test.go

@@ -76,7 +76,7 @@ func TestWriteDependencies(t *testing.T) {
 		createIndexError error
 		writeError       error
 		expectedError    string
-		esVersion        int
+		esVersion        uint
 	}{
 		{
 			createIndexError: errors.New("index not created"),

plugin/storage/es/factory.go

@@ -199,7 +199,7 @@ func createSpanWriter(
 }
 
 // GetMappings returns span and service mappings
-func GetMappings(shards, replicas int64, esVersion int) (string, string) {
+func GetMappings(shards, replicas int64, esVersion uint) (string, string) {
 	if esVersion == 7 {
 		return fixMapping(loadMapping("/jaeger-span-7.json"), shards, replicas),
 			fixMapping(loadMapping("/jaeger-service-7.json"), shards, replicas)

plugin/storage/es/factory_test.go

@@ -52,7 +52,7 @@ func (m *mockClientBuilder) NewClient(logger *zap.Logger, metricsFactory metrics
 		tService.On("Body", mock.Anything).Return(tService)
 		tService.On("Do", context.Background()).Return(nil, m.createTemplateError)
 		c.On("CreateTemplate", mock.Anything).Return(tService)
-		c.On("GetVersion").Return(6)
+		c.On("GetVersion").Return(uint(6))
 		return c, nil
 	}
 	return nil, m.err

plugin/storage/es/options.go

@@ -54,6 +54,7 @@ const (
 	suffixReadAlias           = ".use-aliases"
 	suffixCreateIndexTemplate = ".create-index-templates"
 	suffixEnabled             = ".enabled"
+	suffixVersion             = ".version"
 )
 
 // TODO this should be moved next to config.Configuration struct (maybe ./flags package)
@@ -96,6 +97,7 @@ func NewOptions(primaryNamespace string, otherNamespaces ...string) *Options {
 				TagDotReplacement:    "@",
 				Enabled:              true,
 				CreateIndexTemplates: true,
+				Version:              0,
 			},
 			servers:   "http://127.0.0.1:9200",
 			namespace: primaryNamespace,
@@ -221,6 +223,10 @@ func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
 		nsConfig.namespace+suffixCreateIndexTemplate,
 		nsConfig.CreateIndexTemplates,
 		"Create index templates at application startup. Set to false when templates are installed manually.")
+	flagSet.Uint(
+		nsConfig.namespace+suffixVersion,
+		0,
+		"The major Elasticsearch version. If not specified, the value will be auto-detected from Elasticsearch.")
 	if nsConfig.namespace == archiveNamespace {
 		flagSet.Bool(
 			nsConfig.namespace+suffixEnabled,
@@ -264,6 +270,7 @@ func initFromViper(cfg *namespaceConfig, v *viper.Viper) {
 	cfg.UseReadWriteAliases = v.GetBool(cfg.namespace + suffixReadAlias)
 	cfg.Enabled = v.GetBool(cfg.namespace + suffixEnabled)
 	cfg.CreateIndexTemplates = v.GetBool(cfg.namespace + suffixCreateIndexTemplate)
+	cfg.Version = uint(v.GetInt(cfg.namespace + suffixVersion))
 	// TODO: Need to figure out a better way for do this.
 	cfg.AllowTokenFromContext = v.GetBool(spanstore.StoragePropagationKey)
 }

plugin/storage/integration/elasticsearch_test.go

@@ -56,7 +56,7 @@ type ESStorageIntegration struct {
 	logger        *zap.Logger
 }
 
-func (s *ESStorageIntegration) getVersion() (int, error) {
+func (s *ESStorageIntegration) getVersion() (uint, error) {
 	pingResult, _, err := s.client.Ping(queryURL).Do(context.Background())
 	if err != nil {
 		return 0, err
@@ -65,7 +65,7 @@ func (s *ESStorageIntegration) getVersion() (int, error) {
 	if err != nil {
 		return 0, err
 	}
-	return esVersion, nil
+	return uint(esVersion), nil
 }
 
 func (s *ESStorageIntegration) initializeES(allTagsAsFields, archive bool) error {

plugin/storage/integration/token_propagation_test.go

@@ -20,11 +20,8 @@ import (
 	"context"
 	"encoding/json"
 	"net/http"
-	"os"
-	"os/exec"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/olivere/elastic"
 	"github.com/stretchr/testify/assert"
@@ -40,16 +37,6 @@ type esTokenPropagationTestHandler struct {
 }
 
 func (h *esTokenPropagationTestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	// Return the elasticsearch version
-	if r.URL.Path == "/" {
-		ret := new(elastic.PingResult)
-		ret.Version.Number = "7.3.0"
-		json_ret, _ := json.Marshal(ret)
-		w.Header().Add("Content-Type", "application/json; charset=UTF-8")
-		w.Write(json_ret)
-		return
-	}
-
 	authValue := r.Header.Get("Authorization")
 	if authValue != "" {
 		headerValue := strings.Split(authValue, " ")
@@ -102,19 +89,6 @@ func TestBearTokenPropagation(t *testing.T) {
 	defer srv.Shutdown(context.Background())
 
 	go createElasticSearchMock(srv, t)
-	// Wait for http server to start
-	time.Sleep(100 * time.Millisecond)
-
-	// Path relative to plugin/storage/integration/token_propagation_test.go
-	cmd := exec.Command("../../../cmd/query/query-linux", "--es.server-urls=http://127.0.0.1:9200", "--es.tls=false", "--query.bearer-token-propagation=true")
-	cmd.Env = []string{"SPAN_STORAGE_TYPE=elasticsearch"}
-	cmd.Stdout = os.Stdout
-	cmd.Stderr = os.Stderr
-	err := cmd.Start()
-	assert.NoError(t, err)
-
-	// Wait for query service to start
-	time.Sleep(100 * time.Millisecond)
 
 	// Test cases.
 	for _, testCase := range testCases {
@@ -130,6 +104,4 @@ func TestBearTokenPropagation(t *testing.T) {
 			assert.Equal(t, resp.StatusCode, http.StatusOK)
 		}
 	}
-
-	cmd.Process.Kill()
 }

scripts/travis/es-integration-test.sh

@@ -20,4 +20,8 @@ echo "Executing token propatagion test"
 # Mock UI, needed only for build query service.
 make build-crossdock-ui-placeholder
 GOOS=linux make build-query
+
+SPAN_STORAGE_TYPE=elasticsearch ./cmd/query/query-linux --es.server-urls=http://127.0.0.1:9200 --es.tls=false --es.version=7 --query.bearer-token-propagation=true &
+PID=$(echo $!)
 make token-propagation-integration-test
+kill -9 ${PID}