[ES] Add support to check for presence of a tag

Signed-off-by: Annanay <annanayagarwal@gmail.com>

cmd/query/app/grpc_handler.go

@@ -79,14 +79,15 @@ func (g *GRPCHandler) ArchiveTrace(ctx context.Context, r *api_v2.ArchiveTraceRe
 func (g *GRPCHandler) FindTraces(r *api_v2.FindTracesRequest, stream api_v2.QueryService_FindTracesServer) error {
 	query := r.GetQuery()
 	queryParams := spanstore.TraceQueryParameters{
-		ServiceName:   query.ServiceName,
-		OperationName: query.OperationName,
-		Tags:          query.Tags,
-		StartTimeMin:  query.StartTimeMin,
-		StartTimeMax:  query.StartTimeMax,
-		DurationMin:   query.DurationMin,
-		DurationMax:   query.DurationMax,
-		NumTraces:     int(query.SearchDepth),
+		ServiceName:      query.ServiceName,
+		OperationName:    query.OperationName,
+		Tags:             query.Tags,
+		StartTimeMin:     query.StartTimeMin,
+		StartTimeMax:     query.StartTimeMax,
+		DurationMin:      query.DurationMin,
+		DurationMax:      query.DurationMax,
+		NumTraces:        int(query.SearchDepth),
+		CheckTagsPresent: query.CheckTagsPresent,
 	}
 	traces, err := g.queryService.FindTraces(stream.Context(), &queryParams)
 	if err != nil {

model/proto/api_v2/query.proto

@@ -95,6 +95,7 @@ message TraceQueryParameters {
     (gogoproto.nullable) = false
   ];
   int32 search_depth = 8;
+  repeated string check_tags_present = 9;
 }
 
 message FindTracesRequest {

plugin/storage/es/spanstore/reader.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/olivere/elastic"
@@ -577,6 +578,12 @@ func (s *SpanReader) buildFindTraceIDsQuery(traceQuery *spanstore.TraceQueryPara
 		tagQuery := s.buildTagQuery(k, v)
 		boolQuery.Must(tagQuery)
 	}
+
+	for _, v := range traceQuery.CheckTagsPresent {
+		checkTagPresentQuery := s.buildCheckTagPresentQuery(v)
+		boolQuery.Must(checkTagPresentQuery)
+	}
+
 	return boolQuery
 }
 
@@ -633,6 +640,13 @@ func (s *SpanReader) buildObjectQuery(field string, k string, v string) elastic.
 	return elastic.NewBoolQuery().Must(keyQuery)
 }
 
+func (s *SpanReader) buildCheckTagPresentQuery(field string) elastic.Query {
+	// Check in tags as well as process tags
+	checkTagQuery := elastic.NewExistsQuery(strings.Join([]string{"tag", field}, "."))
+	checkProcesTagQuery := elastic.NewExistsQuery(strings.Join([]string{"process", "tag", field}, "."))
+	return elastic.NewBoolQuery().Should(checkTagQuery, checkProcesTagQuery)
+}
+
 func logErrorToSpan(span opentracing.Span, err error) {
 	ottag.Error.Set(span, true)
 	span.LogFields(otlog.Error(err))

plugin/storage/es/spanstore/reader_test.go

@@ -915,6 +915,9 @@ func TestSpanReader_buildFindTraceIDsQuery(t *testing.T) {
 			Tags: map[string]string{
 				"hello": "world",
 			},
+			CheckTagsPresent: []string{
+				"hello",
+			},
 		}
 
 		actualQuery := r.reader.buildFindTraceIDsQuery(traceQuery)
@@ -927,6 +930,7 @@ func TestSpanReader_buildFindTraceIDsQuery(t *testing.T) {
 				r.reader.buildServiceNameQuery("s"),
 				r.reader.buildOperationNameQuery("o"),
 				r.reader.buildTagQuery("hello", "world"),
+				r.reader.buildCheckTagPresentQuery("hello"),
 			)
 		expected, err := expectedQuery.Source()
 		require.NoError(t, err)

storage/spanstore/interface.go

@@ -44,14 +44,15 @@ type Reader interface {
 
 // TraceQueryParameters contains parameters of a trace query.
 type TraceQueryParameters struct {
-	ServiceName   string
-	OperationName string
-	Tags          map[string]string
-	StartTimeMin  time.Time
-	StartTimeMax  time.Time
-	DurationMin   time.Duration
-	DurationMax   time.Duration
-	NumTraces     int
+	ServiceName      string
+	OperationName    string
+	Tags             map[string]string
+	StartTimeMin     time.Time
+	StartTimeMax     time.Time
+	DurationMin      time.Duration
+	DurationMax      time.Duration
+	NumTraces        int
+	CheckTagsPresent []string
 }
 
 // OperationQueryParameters contains parameters of query operations, empty spanKind means get operations for all kinds of span.