[jaeger-v2] add elasticsearch & opensearch e2e integration test

cmd/jaeger/config-elasticsearch.yaml

@@ -18,12 +18,15 @@ extensions:
         server_urls: http://localhost:9200
         log_level: "error"
         index_prefix: "jaeger-main"
-        use_aliases: true
+        use_aliases: false
+        create_mappings: true
       es_archive:
         server_urls: http://localhost:9200
         log_level: "error"
         index_prefix: "jaeger-archive"
-        use_aliases: true
+        use_aliases: false
+        create_mappings: true
+
 receivers:
   otlp:
     protocols:

cmd/jaeger/internal/integration/es_test.go

@@ -0,0 +1,48 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"testing"
+
+	"github.com/jaegertracing/jaeger/plugin/storage/integration"
+)
+
+const (
+	host          = "0.0.0.0"
+	queryPort     = "9200"
+	queryHostPort = host + ":" + queryPort
+	queryURL      = "http://" + queryHostPort
+)
+
+type ESStorageIntegration struct {
+	E2EStorageIntegration
+	esClient *integration.EsClient
+}
+
+func (s *ESStorageIntegration) initializeES(t *testing.T) {
+	s.esClient = integration.StartEsClient(t, queryURL)
+	s.CleanUp = func(t *testing.T) {
+		s.esClient.DeleteAllIndixes(t)
+	}
+	s.esClient.DeleteAllIndixes(t)
+	// TODO: remove this flag after ES supports returning spanKind
+	//  Issue https://github.com/jaegertracing/jaeger/issues/1923
+	s.GetOperationsMissingSpanKind = true
+}
+
+func TestESStorage(t *testing.T) {
+	integration.SkipUnlessEnv(t, "elasticsearch", "opensearch")
+
+	s := &ESStorageIntegration{}
+	s.initializeES(t)
+	s.Fixtures = integration.LoadAndParseQueryTestCases(t, "fixtures/queries_es.json")
+	s.ConfigFile = "cmd/jaeger/config-elasticsearch.yaml"
+	s.SkipBinaryAttrs = true
+	s.e2eInitialize(t)
+	t.Cleanup(func() {
+		s.e2eCleanUp(t)
+	})
+	s.RunSpanStoreTests(t)
+}

plugin/storage/es/factory.go

@@ -95,8 +95,8 @@ func NewFactoryWithConfig(
 		return nil, err
 	}
 
-	cfg.MaxDocCount = defaultMaxDocCount
-	cfg.Enabled = true
+	defaultConfig := getDefaultConfig()
+	cfg.ApplyDefaults(&defaultConfig)
 
 	archive := make(map[string]*namespaceConfig)
 	archive[archiveNamespace] = &namespaceConfig{

plugin/storage/es/options.go

@@ -98,33 +98,7 @@ type namespaceConfig struct {
 // NewOptions creates a new Options struct.
 func NewOptions(primaryNamespace string, otherNamespaces ...string) *Options {
 	// TODO all default values should be defined via cobra flags
-	defaultConfig := config.Configuration{
-		Username:                     "",
-		Password:                     "",
-		Sniffer:                      false,
-		MaxSpanAge:                   72 * time.Hour,
-		AdaptiveSamplingLookback:     72 * time.Hour,
-		NumShards:                    5,
-		NumReplicas:                  1,
-		PrioritySpanTemplate:         0,
-		PriorityServiceTemplate:      0,
-		PriorityDependenciesTemplate: 0,
-		BulkSize:                     5 * 1000 * 1000,
-		BulkWorkers:                  1,
-		BulkActions:                  1000,
-		BulkFlushInterval:            time.Millisecond * 200,
-		Tags: config.TagsAsFields{
-			DotReplacement: "@",
-		},
-		Enabled:              true,
-		CreateIndexTemplates: true,
-		Version:              0,
-		Servers:              []string{defaultServerURL},
-		RemoteReadClusters:   []string{},
-		MaxDocCount:          defaultMaxDocCount,
-		LogLevel:             "error",
-		SendGetBodyAs:        defaultSendGetBodyAs,
-	}
+	defaultConfig := getDefaultConfig()
 	options := &Options{
 		Primary: namespaceConfig{
 			Configuration: defaultConfig,
@@ -426,3 +400,35 @@ func initDateLayout(rolloverFreq, sep string) string {
 	}
 	return indexLayout
 }
+
+func getDefaultConfig() config.Configuration {
+	return config.Configuration{
+		Username:                     "",
+		Password:                     "",
+		Sniffer:                      false,
+		MaxSpanAge:                   72 * time.Hour,
+		AdaptiveSamplingLookback:     72 * time.Hour,
+		NumShards:                    5,
+		NumReplicas:                  1,
+		PrioritySpanTemplate:         0,
+		PriorityServiceTemplate:      0,
+		PriorityDependenciesTemplate: 0,
+		BulkSize:                     5 * 1000 * 1000,
+		BulkWorkers:                  1,
+		BulkActions:                  1000,
+		BulkFlushInterval:            time.Millisecond * 200,
+		Tags: config.TagsAsFields{
+			DotReplacement: "@",
+		},
+		Enabled:              true,
+		CreateIndexTemplates: true,
+		Version:              0,
+		UseReadWriteAliases:  false,
+		UseILM:               false,
+		Servers:              []string{defaultServerURL},
+		RemoteReadClusters:   []string{},
+		MaxDocCount:          defaultMaxDocCount,
+		LogLevel:             "error",
+		SendGetBodyAs:        defaultSendGetBodyAs,
+	}
+}

plugin/storage/integration/elasticsearch_client.go

@@ -0,0 +1,40 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"context"
+	"testing"
+
+	elasticsearch8 "github.com/elastic/go-elasticsearch/v8"
+	"github.com/olivere/elastic"
+	"github.com/stretchr/testify/require"
+)
+
+type EsClient struct {
+	client   *elastic.Client
+	v8Client *elasticsearch8.Client
+}
+
+func StartEsClient(t *testing.T, queryURL string) *EsClient {
+	rawClient, err := elastic.NewClient(
+		elastic.SetURL(queryURL),
+		elastic.SetSniff(false))
+	require.NoError(t, err)
+
+	rawV8Client, err := elasticsearch8.NewClient(elasticsearch8.Config{
+		Addresses:            []string{queryURL},
+		DiscoverNodesOnStart: false,
+	})
+	require.NoError(t, err)
+	return &EsClient{
+		client:   rawClient,
+		v8Client: rawV8Client,
+	}
+}
+
+func (s *EsClient) DeleteAllIndixes(t *testing.T) {
+	_, err := s.client.DeleteIndex("*").Do(context.Background())
+	require.NoError(t, err)
+}

plugin/storage/integration/elasticsearch_test.go

@@ -25,8 +25,7 @@
 	"testing"
 	"time"
 
-	elasticsearch8 "github.com/elastic/go-elasticsearch/v8"
 	"github.com/olivere/elastic"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -58,14 +57,12 @@
 
 type ESStorageIntegration struct {
 	StorageIntegration
-
-	client   *elastic.Client
-	v8Client *elasticsearch8.Client
-	logger   *zap.Logger
+	logger        *zap.Logger
+	client        *EsClient
 }
 
 func (s *ESStorageIntegration) getVersion() (uint, error) {
-	pingResult, _, err := s.client.Ping(queryURL).Do(context.Background())
+	pingResult, _, err := s.client.client.Ping(queryURL).Do(context.Background())
 	if err != nil {
 		return 0, err
 	}
@@ -83,19 +80,9 @@
 }
 
 func (s *ESStorageIntegration) initializeES(t *testing.T, allTagsAsFields bool) {
-	rawClient, err := elastic.NewClient(
-		elastic.SetURL(queryURL),
-		elastic.SetSniff(false))
-	require.NoError(t, err)
+	s.client = StartEsClient(t, queryURL)
 	s.logger, _ = testutils.NewLogger()
 
-	s.client = rawClient
-	s.v8Client, err = elasticsearch8.NewClient(elasticsearch8.Config{
-		Addresses:            []string{queryURL},
-		DiscoverNodesOnStart: false,
-	})
-	require.NoError(t, err)
-
 	s.initSpanstore(t, allTagsAsFields)
 
 	s.CleanUp = func(t *testing.T) {
@@ -109,8 +96,7 @@
 }
 
 func (s *ESStorageIntegration) esCleanUp(t *testing.T, allTagsAsFields bool) {
-	_, err := s.client.DeleteIndex("*").Do(context.Background())
-	require.NoError(t, err)
+	s.client.DeleteAllIndixes(t)
 	s.initSpanstore(t, allTagsAsFields)
 }
 
@@ -204,17 +190,17 @@
 	require.NoError(t, err)
 	// TODO abstract this into pkg/es/client.IndexManagementLifecycleAPI
 	if esVersion <= 7 {
-		serviceTemplateExists, err := s.client.IndexTemplateExists(indexPrefix + "-jaeger-service").Do(context.Background())
+		serviceTemplateExists, err := s.client.client.IndexTemplateExists(indexPrefix + "-jaeger-service").Do(context.Background())
 		require.NoError(t, err)
 		assert.True(t, serviceTemplateExists)
-		spanTemplateExists, err := s.client.IndexTemplateExists(indexPrefix + "-jaeger-span").Do(context.Background())
+		spanTemplateExists, err := s.client.client.IndexTemplateExists(indexPrefix + "-jaeger-span").Do(context.Background())
 		require.NoError(t, err)
 		assert.True(t, spanTemplateExists)
 	} else {
-		serviceTemplateExistsResponse, err := s.v8Client.API.Indices.ExistsIndexTemplate(indexPrefix + "-jaeger-service")
+		serviceTemplateExistsResponse, err := s.client.v8Client.API.Indices.ExistsIndexTemplate(indexPrefix + "-jaeger-service")
 		require.NoError(t, err)
 		assert.Equal(t, 200, serviceTemplateExistsResponse.StatusCode)
-		spanTemplateExistsResponse, err := s.v8Client.API.Indices.ExistsIndexTemplate(indexPrefix + "-jaeger-span")
+		spanTemplateExistsResponse, err := s.client.v8Client.API.Indices.ExistsIndexTemplate(indexPrefix + "-jaeger-span")
 		require.NoError(t, err)
 		assert.Equal(t, 200, spanTemplateExistsResponse.StatusCode)
 	}
@@ -229,14 +215,14 @@
 		if prefix != "" {
 			prefixWithSeparator += "-"
 		}
-		_, err := s.v8Client.Indices.DeleteIndexTemplate(prefixWithSeparator + spanTemplateName)
+		_, err := s.client.v8Client.Indices.DeleteIndexTemplate(prefixWithSeparator + spanTemplateName)
 		require.NoError(t, err)
-		_, err = s.v8Client.Indices.DeleteIndexTemplate(prefixWithSeparator + serviceTemplateName)
+		_, err = s.client.v8Client.Indices.DeleteIndexTemplate(prefixWithSeparator + serviceTemplateName)
 		require.NoError(t, err)
-		_, err = s.v8Client.Indices.DeleteIndexTemplate(prefixWithSeparator + dependenciesTemplateName)
+		_, err = s.client.v8Client.Indices.DeleteIndexTemplate(prefixWithSeparator + dependenciesTemplateName)
 		require.NoError(t, err)
 	} else {
-		_, err := s.client.IndexDeleteTemplate("*").Do(context.Background())
+		_, err := s.client.client.IndexDeleteTemplate("*").Do(context.Background())
 		require.NoError(t, err)
 	}
 	return nil

plugin/storage/integration/es_index_cleaner_test.go

@@ -251,8 +251,10 @@ func createESV8Client() (*elasticsearch8.Client, error) {
 
 func cleanESIndexTemplates(t *testing.T, client *elastic.Client, v8Client *elasticsearch8.Client, prefix string) {
 	s := &ESStorageIntegration{
-		client:   client,
-		v8Client: v8Client,
+		client: &EsClient{
+			client:   client,
+			v8Client: v8Client,
+		},
 	}
 	s.logger, _ = testutils.NewLogger()
 	s.cleanESIndexTemplates(t, prefix)

scripts/es-integration-test.sh

@@ -134,6 +134,7 @@ main() {
 
   bring_up_storage "${distro}" "${version}"
   STORAGE=${distro} make storage-integration-test
+  STORAGE=${distro} SPAN_STORAGE_TYPE=${distro} make jaeger-v2-storage-integration-test
   make index-cleaner-integration-test
   make index-rollover-integration-test
 }