[es cit] Use storage factory instead of manual creation

plugin/storage/integration/elasticsearch_test.go

@@ -28,19 +28,16 @@ import (
 	"github.com/olivere/elastic"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	sdktrace "go.opentelemetry.io/otel/sdk/trace"
-	"go.opentelemetry.io/otel/sdk/trace/tracetest"
-	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 
 	estemplate "github.com/jaegertracing/jaeger/pkg/es"
 	eswrapper "github.com/jaegertracing/jaeger/pkg/es/wrapper"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
 	"github.com/jaegertracing/jaeger/pkg/testutils"
+	"github.com/jaegertracing/jaeger/plugin/storage/es"
 	"github.com/jaegertracing/jaeger/plugin/storage/es/dependencystore"
 	"github.com/jaegertracing/jaeger/plugin/storage/es/mappings"
 	"github.com/jaegertracing/jaeger/plugin/storage/es/samplingstore"
-	"github.com/jaegertracing/jaeger/plugin/storage/es/spanstore"
 )
 
 const (
@@ -56,6 +53,8 @@ const (
 	spanTemplateName         = "jaeger-span"
 	serviceTemplateName      = "jaeger-service"
 	dependenciesTemplateName = "jaeger-dependencies"
+	primaryNamespace         = "es"
+	archiveNamespace         = "es-archive"
 )
 
 type ESStorageIntegration struct {
@@ -67,19 +66,19 @@ type ESStorageIntegration struct {
 	logger        *zap.Logger
 }
 
-func (s *ESStorageIntegration) tracerProvider() (trace.TracerProvider, *tracetest.InMemoryExporter, func()) {
-	exporter := tracetest.NewInMemoryExporter()
-	tp := sdktrace.NewTracerProvider(
-		sdktrace.WithSampler(sdktrace.AlwaysSample()),
-		sdktrace.WithSyncer(exporter),
-	)
-	closer := func() {
-		if err := tp.Shutdown(context.Background()); err != nil {
-			s.logger.Error("failed to close tracer", zap.Error(err))
-		}
-	}
-	return tp, exporter, closer
-}
+// func (s *ESStorageIntegration) tracerProvider() (trace.TracerProvider, *tracetest.InMemoryExporter, func()) {
+// 	exporter := tracetest.NewInMemoryExporter()
+// 	tp := sdktrace.NewTracerProvider(
+// 		sdktrace.WithSampler(sdktrace.AlwaysSample()),
+// 		sdktrace.WithSyncer(exporter),
+// 	)
+// 	closer := func() {
+// 		if err := tp.Shutdown(context.Background()); err != nil {
+// 			s.logger.Error("failed to close tracer", zap.Error(err))
+// 		}
+// 	}
+// 	return tp, exporter, closer
+// }
 
 func (s *ESStorageIntegration) getVersion() (uint, error) {
 	pingResult, _, err := s.client.Ping(queryURL).Do(context.Background())
@@ -105,14 +104,38 @@ func (s *ESStorageIntegration) initializeES(t *testing.T, allTagsAsFields bool)
 		elastic.SetSniff(false))
 	require.NoError(t, err)
 	s.logger, _ = testutils.NewLogger()
-
 	s.client = rawClient
 	s.v8Client, err = elasticsearch8.NewClient(elasticsearch8.Config{
 		Addresses:            []string{queryURL},
 		DiscoverNodesOnStart: false,
 	})
 	require.NoError(t, err)
 
+	// Initialize ES Factory
+	opts := es.NewOptions(primaryNamespace, archiveNamespace)
+	cfg := opts.Primary.Configuration
+
+	f, err := es.NewFactoryWithConfig(cfg, metrics.NullFactory, s.logger)
+	if err != nil {
+		return err
+	}
+	// Create Span Writer and Reader
+	s.SpanWriter, err = f.CreateSpanWriter()
+	if err != nil {
+		return err
+	}
+	s.SpanReader, err = f.CreateSpanReader()
+	if err != nil {
+		return err
+	}
+	s.ArchiveSpanWriter, err = f.CreateArchiveSpanWriter()
+	if err != nil {
+		return err
+	}
+	s.ArchiveSpanReader, err = f.CreateArchiveSpanReader()
+	if err != nil {
+		return err
+	}
 	s.initSpanstore(t, allTagsAsFields)
 	s.initSamplingStore(t)
 
@@ -124,6 +147,7 @@ func (s *ESStorageIntegration) initializeES(t *testing.T, allTagsAsFields bool)
 	// TODO: remove this flag after ES support returning spanKind when get operations
 	s.GetOperationsMissingSpanKind = true
 	s.SkipArchiveTest = false
+
 	return nil
 }
 
@@ -178,61 +202,61 @@ func (s *ESStorageIntegration) initSpanstore(t *testing.T, allTagsAsFields bool)
 		IndexPrefix:     indexPrefix,
 		UseILM:          false,
 	}
-	spanMapping, serviceMapping, err := mappingBuilder.GetSpanServiceMappings()
-	require.NoError(t, err)
+	// spanMapping, serviceMapping, err := mappingBuilder.GetSpanServiceMappings()
+	// require.NoError(t, err)
 	clientFn := func() estemplate.Client { return client }
 
 	// Initializing Span Reader and Writer
-	w := spanstore.NewSpanWriter(
-		spanstore.SpanWriterParams{
-			Client:            clientFn,
-			Logger:            s.logger,
-			MetricsFactory:    metrics.NullFactory,
-			IndexPrefix:       indexPrefix,
-			AllTagsAsFields:   allTagsAsFields,
-			TagDotReplacement: tagKeyDeDotChar,
-			Archive:           false,
-		})
-	err = w.CreateTemplates(spanMapping, serviceMapping, indexPrefix)
-	require.NoError(t, err)
-	tracer, _, closer := s.tracerProvider()
-	defer closer()
-	s.SpanWriter = w
-	s.SpanReader = spanstore.NewSpanReader(spanstore.SpanReaderParams{
-		Client:            clientFn,
-		Logger:            s.logger,
-		MetricsFactory:    metrics.NullFactory,
-		IndexPrefix:       indexPrefix,
-		MaxSpanAge:        maxSpanAge,
-		TagDotReplacement: tagKeyDeDotChar,
-		MaxDocCount:       defaultMaxDocCount,
-		Tracer:            tracer.Tracer("test"),
-		Archive:           false,
-	})
 
-	// Initializing Archive Span Reader and Writer
-	s.ArchiveSpanWriter = spanstore.NewSpanWriter(
-		spanstore.SpanWriterParams{
-			Client:            clientFn,
-			Logger:            s.logger,
-			MetricsFactory:    metrics.NullFactory,
-			IndexPrefix:       indexPrefix,
-			AllTagsAsFields:   allTagsAsFields,
-			TagDotReplacement: tagKeyDeDotChar,
-			Archive:           true,
-		})
-	s.ArchiveSpanReader = spanstore.NewSpanReader(spanstore.SpanReaderParams{
-		Client:            clientFn,
-		Logger:            s.logger,
-		MetricsFactory:    metrics.NullFactory,
-		IndexPrefix:       indexPrefix,
-		MaxSpanAge:        maxSpanAge,
-		TagDotReplacement: tagKeyDeDotChar,
-		MaxDocCount:       defaultMaxDocCount,
-		Tracer:            tracer.Tracer("test"),
-		Archive:           true,
-	})
+	// w := spanstore.NewSpanWriter(
+	// 	spanstore.SpanWriterParams{
+	// 		Client:            clientFn,
+	// 		Logger:            s.logger,
+	// 		MetricsFactory:    metrics.NullFactory,
+	// 		IndexPrefix:       indexPrefix,
+	// 		AllTagsAsFields:   allTagsAsFields,
+	// 		TagDotReplacement: tagKeyDeDotChar,
+	// 		Archive:           false,
+	// 	})
+	// err = w.CreateTemplates(spanMapping, serviceMapping, indexPrefix)
+	// require.NoError(t, err)
+	// tracer, _, closer := s.tracerProvider()
+	// defer closer()
+	// s.SpanWriter = w
+	// s.SpanReader = spanstore.NewSpanReader(spanstore.SpanReaderParams{
+	// 	Client:            clientFn,
+	// 	Logger:            s.logger,
+	// 	MetricsFactory:    metrics.NullFactory,
+	// 	IndexPrefix:       indexPrefix,
+	// 	MaxSpanAge:        maxSpanAge,
+	// 	TagDotReplacement: tagKeyDeDotChar,
+	// 	MaxDocCount:       defaultMaxDocCount,
+	// 	Tracer:            tracer.Tracer("test"),
+	// 	Archive:           false,
+	// })
 
+	// Initializing Archive Span Reader and Writer
+	// s.ArchiveSpanWriter = spanstore.NewSpanWriter(
+	// 	spanstore.SpanWriterParams{
+	// 		Client:            clientFn,
+	// 		Logger:            s.logger,
+	// 		MetricsFactory:    metrics.NullFactory,
+	// 		IndexPrefix:       indexPrefix,
+	// 		AllTagsAsFields:   allTagsAsFields,
+	// 		TagDotReplacement: tagKeyDeDotChar,
+	// 		Archive:           true,
+	// 	})
+	// s.ArchiveSpanReader = spanstore.NewSpanReader(spanstore.SpanReaderParams{
+	// 	Client:            clientFn,
+	// 	Logger:            s.logger,
+	// 	MetricsFactory:    metrics.NullFactory,
+	// 	IndexPrefix:       indexPrefix,
+	// 	MaxSpanAge:        maxSpanAge,
+	// 	TagDotReplacement: tagKeyDeDotChar,
+	// 	MaxDocCount:       defaultMaxDocCount,
+	// 	Tracer:            tracer.Tracer("test"),
+	// 	Archive:           true,
+	// })
 	dependencyStore := dependencystore.NewDependencyStore(dependencystore.DependencyStoreParams{
 		Client:          clientFn,
 		Logger:          s.logger,