[ES] Separate the CoreSpanWriter from SpanWriter (jaegertracing#6883)

## Which problem is this PR solving?
Fixes a part of: jaegertracing#6458

## Description of the changes
- We have to prepeare the writer to move to `v2` and for that we need to
separate `SpanWriter`

## How was this change tested?
- Unit Tests

## Checklist
- [x] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [x] I have signed all commits
- [x] I have added unit tests for the new functionality
- [x] I have run lint and test steps successfully
  - for `jaeger`: `make lint test`
  - for `jaeger-ui`: `npm run lint` and `npm run test`

---------

Signed-off-by: Manik2708 <[email protected]>
Co-authored-by: Yuri Shkuro <[email protected]>
Signed-off-by: amol-verma-allen <[email protected]>

Author: 2 people

.mockery.yaml

@@ -82,4 +82,5 @@ packages:
   github.com/jaegertracing/jaeger/internal/storage/v1/elasticsearch/spanstore:
     interfaces:
       CoreSpanReader:
+      CoreSpanWriter:
 

internal/storage/v1/elasticsearch/spanstore/internal/dbmodel/from_domain_test.go

@@ -146,3 +146,57 @@ func TestConvertKeyValueValue(t *testing.T) {
 		})
 	}
 }
+
+func TestNewSpanTags(t *testing.T) {
+	testCases := []struct {
+		spanConverter FromDomain
+		expected      Span
+		name          string
+	}{
+		{
+			spanConverter: NewFromDomain(true, []string{}, ""),
+			expected: Span{
+				Tag: map[string]any{"foo": "bar"}, Tags: []KeyValue{},
+				Process: Process{Tag: map[string]any{"bar": "baz"}, Tags: []KeyValue{}},
+			},
+			name: "allTagsAsFields",
+		},
+		{
+			spanConverter: NewFromDomain(false, []string{"foo", "bar", "rere"}, ""),
+			expected: Span{
+				Tag: map[string]any{"foo": "bar"}, Tags: []KeyValue{},
+				Process: Process{Tag: map[string]any{"bar": "baz"}, Tags: []KeyValue{}},
+			},
+			name: "definedTagNames",
+		},
+		{
+			spanConverter: NewFromDomain(false, []string{}, ""),
+			expected: Span{
+				Tags: []KeyValue{{
+					Key:   "foo",
+					Type:  StringType,
+					Value: "bar",
+				}},
+				Process: Process{Tags: []KeyValue{{
+					Key:   "bar",
+					Type:  StringType,
+					Value: "baz",
+				}}},
+			},
+			name: "noAllTagsAsFields",
+		},
+	}
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			s := &model.Span{
+				Tags:    []model.KeyValue{{Key: "foo", VStr: "bar"}},
+				Process: &model.Process{Tags: []model.KeyValue{{Key: "bar", VStr: "baz"}}},
+			}
+			mSpan := test.spanConverter.FromDomainEmbedProcess(s)
+			assert.Equal(t, test.expected.Tag, mSpan.Tag)
+			assert.Equal(t, test.expected.Tags, mSpan.Tags)
+			assert.Equal(t, test.expected.Process.Tag, mSpan.Process.Tag)
+			assert.Equal(t, test.expected.Process.Tags, mSpan.Process.Tags)
+		})
+	}
+}

internal/storage/v1/elasticsearch/spanstore/mocks/CoreSpanWriter.go

@@ -41,13 +41,13 @@ func TestWriteService(t *testing.T) {
 			},
 		}
 
-		w.writer.spanWriter.writeService(indexName, jsonSpan)
+		w.writer.writeService(indexName, jsonSpan)
 
 		indexService.AssertNumberOfCalls(t, "Add", 1)
 		assert.Equal(t, "", w.logBuffer.String())
 
 		// test that cache works, will call the index service only once.
-		w.writer.spanWriter.writeService(indexName, jsonSpan)
+		w.writer.writeService(indexName, jsonSpan)
 		indexService.AssertNumberOfCalls(t, "Add", 1)
 	})
 }
@@ -76,7 +76,7 @@ func TestWriteServiceError(*testing.T) {
 			},
 		}
 
-		w.writer.spanWriter.writeService(indexName, jsonSpan)
+		w.writer.writeService(indexName, jsonSpan)
 	})
 }
 

internal/storage/v1/elasticsearch/spanstore/service_operation_test.go

@@ -11,9 +11,7 @@ import (
 
 	"go.uber.org/zap"
 
-	"github.com/jaegertracing/jaeger-idl/model/v1"
 	"github.com/jaegertracing/jaeger/internal/cache"
-	"github.com/jaegertracing/jaeger/internal/storage/v1/api/spanstore/spanstoremetrics"
 	"github.com/jaegertracing/jaeger/internal/storage/v1/elasticsearch/spanstore/internal/dbmodel"
 	"github.com/jaegertracing/jaeger/pkg/es"
 	cfg "github.com/jaegertracing/jaeger/pkg/es/config"
@@ -27,25 +25,25 @@ const (
 	indexCacheTTLDefault   = 48 * time.Hour
 )
 
-type spanWriterMetrics struct {
-	indexCreate *spanstoremetrics.WriteMetrics
-}
-
 type serviceWriter func(string, *dbmodel.Span)
 
 // SpanWriter is a wrapper around elastic.Client
 type SpanWriter struct {
-	client        func() es.Client
-	logger        *zap.Logger
-	writerMetrics spanWriterMetrics // TODO: build functions to wrap around each Do fn
+	client func() es.Client
+	logger *zap.Logger
 	// indexCache       cache.Cache
 	serviceWriter    serviceWriter
-	spanConverter    dbmodel.FromDomain
 	spanServiceIndex spanAndServiceIndexFn
 }
 
-type SpanWriterV1 struct {
-	spanWriter *SpanWriter
+// CoreSpanWriter is a DB-Level abstraction which directly deals with database level operations
+type CoreSpanWriter interface {
+	// CreateTemplates creates index templates.
+	CreateTemplates(spanTemplate, serviceTemplate string, indexPrefix cfg.IndexPrefix) error
+	// WriteSpan writes a span and its corresponding service:operation in ElasticSearch
+	WriteSpan(spanStartTime time.Time, span *dbmodel.Span)
+	// Close closes CoreSpanWriter
+	Close() error
 }
 
 // SpanWriterParams holds constructor parameters for NewSpanWriter
@@ -64,13 +62,6 @@ type SpanWriterParams struct {
 	ServiceCacheTTL     time.Duration
 }
 
-// NewSpanWriterV1 returns the SpanWriterV1 for use
-func NewSpanWriterV1(p SpanWriterParams) *SpanWriterV1 {
-	return &SpanWriterV1{
-		spanWriter: NewSpanWriter(p),
-	}
-}
-
 // NewSpanWriter creates a new SpanWriter for use
 func NewSpanWriter(p SpanWriterParams) *SpanWriter {
 	serviceCacheTTL := p.ServiceCacheTTL
@@ -89,22 +80,13 @@ func NewSpanWriter(p SpanWriterParams) *SpanWriter {
 
 	serviceOperationStorage := NewServiceOperationStorage(p.Client, p.Logger, serviceCacheTTL)
 	return &SpanWriter{
-		client: p.Client,
-		logger: p.Logger,
-		writerMetrics: spanWriterMetrics{
-			indexCreate: spanstoremetrics.NewWriter(p.MetricsFactory, "index_create"),
-		},
+		client:           p.Client,
+		logger:           p.Logger,
 		serviceWriter:    serviceOperationStorage.Write,
-		spanConverter:    dbmodel.NewFromDomain(p.AllTagsAsFields, p.TagKeysAsFields, p.TagDotReplacement),
 		spanServiceIndex: getSpanAndServiceIndexFn(p, writeAliasSuffix),
 	}
 }
 
-// CreateTemplates creates index templates.
-func (s *SpanWriterV1) CreateTemplates(spanTemplate, serviceTemplate string, indexPrefix cfg.IndexPrefix) error {
-	return s.spanWriter.CreateTemplates(spanTemplate, serviceTemplate, indexPrefix)
-}
-
 // CreateTemplates creates index templates.
 func (s *SpanWriter) CreateTemplates(spanTemplate, serviceTemplate string, indexPrefix cfg.IndexPrefix) error {
 	jaegerSpanIdx := indexPrefix.Apply("jaeger-span")
@@ -146,23 +128,11 @@ func (s *SpanWriter) WriteSpan(spanStartTime time.Time, span *dbmodel.Span) {
 	s.logger.Debug("Wrote span to ES index", zap.String("index", spanIndexName))
 }
 
-// WriteSpan writes a span and its corresponding service:operation in ElasticSearch
-func (s *SpanWriterV1) WriteSpan(_ context.Context, span *model.Span) error {
-	jsonSpan := s.spanWriter.spanConverter.FromDomainEmbedProcess(span)
-	s.spanWriter.WriteSpan(span.StartTime, jsonSpan)
-	return nil
-}
-
 // Close closes SpanWriter
 func (s *SpanWriter) Close() error {
 	return s.client().Close()
 }
 
-// Close closes SpanWriter
-func (s *SpanWriterV1) Close() error {
-	return s.spanWriter.Close()
-}
-
 func keyInCache(key string, c cache.Cache) bool {
 	return c.Get(key) != nil
 }