Merge pull request #1 from pavolloffay/PR1956-es-fix

Read trace ID from ES without leading zeros

plugin/storage/es/spanstore/reader.go

@@ -337,7 +337,7 @@ func (s *SpanReader) multiRead(ctx context.Context, traceIDs []model.TraceID, st
 		}
 		searchRequests := make([]*elastic.SearchRequest, len(traceIDs))
 		for i, traceID := range traceIDs {
-			query := elastic.NewTermQuery("traceID", traceID.String())
+			query := traceIDQuery(traceID)
 			if val, ok := searchAfterTime[traceID]; ok {
 				nextTime = val
 			}
@@ -393,15 +393,36 @@ func (s *SpanReader) multiRead(ctx context.Context, traceIDs []model.TraceID, st
 	return traces, nil
 }
 
+func traceIDQuery(traceID model.TraceID) elastic.Query {
+	traceIDStr := traceID.String()
+	// TODO remove in newer versions, added in Jaeger 1.16
+	// read ID without leading zeros
+	// https://github.com/jaegertracing/jaeger/pull/1956 added leading zeros to IDs
+	if traceIDStr[0] == '0' {
+		var legacyTraceID string
+		if traceID.High == 0 {
+			legacyTraceID = fmt.Sprintf("%x", traceID.Low)
+		} else {
+			legacyTraceID = fmt.Sprintf("%x%016x", traceID.High, traceID.Low)
+		}
+		return elastic.NewBoolQuery().
+			Should(elastic.NewTermQuery(traceIDField, traceIDStr).Boost(2), elastic.NewTermQuery(traceIDField, legacyTraceID))
+	}
+	return elastic.NewTermQuery(traceIDField, traceIDStr)
+}
+
 func convertTraceIDsStringsToModels(traceIDs []string) ([]model.TraceID, error) {
+	traceIDsMap := map[model.TraceID]bool{}
 	traceIDsModels := make([]model.TraceID, len(traceIDs))
 	for i, ID := range traceIDs {
 		traceID, err := model.TraceIDFromString(ID)
 		if err != nil {
 			return nil, errors.Wrap(err, fmt.Sprintf("Making traceID from string '%s' failed", ID))
 		}
-
-		traceIDsModels[i] = traceID
+		if _, ok := traceIDsMap[traceID]; !ok {
+			traceIDsMap[traceID] = true
+			traceIDsModels[i] = traceID
+		}
 	}
 
 	return traceIDsModels, nil

plugin/storage/es/spanstore/reader_test.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
+	"math/bits"
 	"reflect"
 	"testing"
 	"time"
@@ -216,11 +217,16 @@ func TestSpanReader_multiRead_followUp_query(t *testing.T) {
 		spanBytesID2, err := json.Marshal(spanID2)
 		require.NoError(t, err)
 
-		id1Query := elastic.NewTermQuery("traceID", model.TraceID{High: 0, Low: 1}.String())
+		id1Query := elastic.NewBoolQuery().Should(
+			elastic.NewTermQuery(traceIDField, model.TraceID{High: 0, Low: 1}.String()).Boost(2),
+			elastic.NewTermQuery(traceIDField, fmt.Sprintf("%x", 1)))
 		id1Search := elastic.NewSearchRequest().
 			IgnoreUnavailable(true).
 			Source(r.reader.sourceFn(id1Query, model.TimeAsEpochMicroseconds(date.Add(-time.Hour))))
-		id2Query := elastic.NewTermQuery("traceID", model.TraceID{High: 0, Low: 2}.String())
+		//id2Query := elastic.NewTermQuery("traceID", model.TraceID{High: 0, Low: 2}.String())
+		id2Query := elastic.NewBoolQuery().Should(
+			elastic.NewTermQuery(traceIDField, model.TraceID{High: 0, Low: 2}.String()).Boost(2),
+			elastic.NewTermQuery(traceIDField, fmt.Sprintf("%x", 2)))
 		id2Search := elastic.NewSearchRequest().
 			IgnoreUnavailable(true).
 			Source(r.reader.sourceFn(id2Query, model.TimeAsEpochMicroseconds(date.Add(-time.Hour))))
@@ -1080,7 +1086,30 @@ func TestSpanReader_ArchiveTraces_ReadAlias(t *testing.T) {
 	})
 }
 
-func TestNewSpanReader2(t *testing.T) {
-	spanDate := time.Now().UTC().Format("2006")
-	fmt.Println(spanDate)
+func TestTraceIDQuery(t *testing.T) {
+	uintMax := uint64((1 << bits.UintSize) - 1)
+	traceIDNoHigh := model.NewTraceID(0, 1)
+	traceIDHigh := model.NewTraceID(1, 1)
+	traceID := model.NewTraceID(uintMax, uintMax)
+	tests := []struct {
+		traceID model.TraceID
+		query   elastic.Query
+	}{
+		{
+			traceID: traceIDNoHigh,
+			query:   elastic.NewBoolQuery().Should(elastic.NewTermQuery(traceIDField, traceIDNoHigh.String()).Boost(2), elastic.NewTermQuery(traceIDField, fmt.Sprintf("%x", traceIDNoHigh.Low))),
+		},
+		{
+			traceID: traceIDHigh,
+			query:   elastic.NewBoolQuery().Should(elastic.NewTermQuery(traceIDField, traceIDHigh.String()).Boost(2), elastic.NewTermQuery(traceIDField, fmt.Sprintf("%x%016x", traceIDHigh.High, traceIDHigh.Low))),
+		},
+		{
+			traceID: traceID,
+			query:   elastic.NewTermQuery(traceIDField, traceID.String()),
+		},
+	}
+	for _, test := range tests {
+		q := traceIDQuery(test.traceID)
+		assert.Equal(t, test.query, q)
+	}
 }