Upgrade storage integration test: Use V2 Archive ReaderWriter (#6489)

## Which problem is this PR solving?
- Part of #6366

## Description of the changes
 For es v1 storage integration test:
  - [x] replaced ArchiveSpanReader with ArchiveTraceReader
  - [x] replaced ArchiveSpanWriter with ArchiveTraceWriter

## How was this change tested?
- CI tests

## Checklist
- [x] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [x] I have signed all commits
- [ ] 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: Emmanuel Emonueje Ebenezer <[email protected]>

plugin/storage/integration/elasticsearch_test.go

@@ -27,7 +27,7 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/plugin/storage/es"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
-	"github.com/jaegertracing/jaeger/storage/spanstore"
+	"github.com/jaegertracing/jaeger/storage_v2/tracestore"
 	"github.com/jaegertracing/jaeger/storage_v2/v1adapter"
 )
 
@@ -54,8 +54,8 @@ type ESStorageIntegration struct {
 	client   *elastic.Client
 	v8Client *elasticsearch8.Client
 
-	ArchiveSpanReader spanstore.Reader
-	ArchiveSpanWriter spanstore.Writer
+	ArchiveTraceReader tracestore.Reader
+	ArchiveTraceWriter tracestore.Writer
 
 	factory        *es.Factory
 	archiveFactory *es.Factory
@@ -148,10 +148,12 @@ func (s *ESStorageIntegration) initSpanstore(t *testing.T, allTagsAsFields bool)
 	spanReader, err := f.CreateSpanReader()
 	require.NoError(t, err)
 	s.TraceReader = v1adapter.NewTraceReader(spanReader)
-	s.ArchiveSpanReader, err = af.CreateSpanReader()
+	archiveSpanReader, err := af.CreateSpanReader()
 	require.NoError(t, err)
-	s.ArchiveSpanWriter, err = af.CreateSpanWriter()
+	s.ArchiveTraceReader = v1adapter.NewTraceReader(archiveSpanReader)
+	archiveSpanWriter, err := af.CreateSpanWriter()
 	require.NoError(t, err)
+	s.ArchiveTraceWriter = v1adapter.NewTraceWriter(archiveSpanWriter)
 
 	dependencyReader, err := f.CreateDependencyReader()
 	require.NoError(t, err)
@@ -263,23 +265,32 @@ func (s *ESStorageIntegration) testArchiveTrace(t *testing.T) {
 	s.skipIfNeeded(t)
 	defer s.cleanUp(t)
 	tID := model.NewTraceID(uint64(11), uint64(22))
-	expected := &model.Span{
-		OperationName: "archive_span",
-		StartTime:     time.Now().Add(-maxSpanAge * 5).Truncate(time.Microsecond),
-		TraceID:       tID,
-		SpanID:        model.NewSpanID(55),
-		References:    []model.SpanRef{},
-		Process:       model.NewProcess("archived_service", model.KeyValues{}),
+	expected := &model.Trace{
+		Spans: []*model.Span{
+			{
+				OperationName: "archive_span",
+				StartTime:     time.Now().Add(-maxSpanAge * 5).Truncate(time.Microsecond),
+				TraceID:       tID,
+				SpanID:        model.NewSpanID(55),
+				References:    []model.SpanRef{},
+				Process:       model.NewProcess("archived_service", model.KeyValues{}),
+			},
+		},
 	}
 
-	require.NoError(t, s.ArchiveSpanWriter.WriteSpan(context.Background(), expected))
+	require.NoError(t, s.ArchiveTraceWriter.WriteTraces(context.Background(), v1adapter.V1TraceToOtelTrace(expected)))
 
 	var actual *model.Trace
 	found := s.waitForCondition(t, func(_ *testing.T) bool {
 		var err error
-		actual, err = s.ArchiveSpanReader.GetTrace(context.Background(), spanstore.GetTraceParameters{TraceID: tID})
+		iterTraces := s.ArchiveTraceReader.GetTraces(context.Background(), tracestore.GetTraceParams{TraceID: v1adapter.FromV1TraceID(tID)})
+		traces, err := v1adapter.V1TracesFromSeq2(iterTraces)
+		if len(traces) == 0 {
+			return false
+		}
+		actual = traces[0]
 		return err == nil && len(actual.Spans) == 1
 	})
 	require.True(t, found)
-	CompareTraces(t, &model.Trace{Spans: []*model.Span{expected}}, actual)
+	CompareTraces(t, expected, actual)
 }

plugin/storage/integration/integration.go

@@ -189,9 +189,10 @@ func (s *StorageIntegration) testGetLargeSpan(t *testing.T) {
 	found := s.waitForCondition(t, func(_ *testing.T) bool {
 		iterTraces := s.TraceReader.GetTraces(context.Background(), tracestore.GetTraceParams{TraceID: expectedTraceID})
 		traces, err := v1adapter.V1TracesFromSeq2(iterTraces)
-		if len(traces) > 0 {
-			actual = traces[0]
+		if len(traces) == 0 {
+			return false
 		}
+		actual = traces[0]
 		return err == nil && len(actual.Spans) >= len(expected.Spans)
 	})
 
@@ -269,9 +270,10 @@ func (s *StorageIntegration) testGetTrace(t *testing.T) {
 			t.Log(err)
 			return false
 		}
-		if len(traces) > 0 {
-			actual = traces[0]
+		if len(traces) == 0 {
+			return false
 		}
+		actual = traces[0]
 		return len(actual.Spans) == len(expected.Spans)
 	})
 	if !assert.True(t, found) {