[ES, C*] Check for the presence of a tag

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/cassandra/spanstore/reader.go

@@ -17,6 +17,7 @@ package spanstore
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/opentracing/opentracing-go"
@@ -45,6 +46,12 @@ const (
 		WHERE service_name = ? AND tag_key = ? AND tag_value = ? and start_time > ? and start_time < ?
 		ORDER BY start_time DESC
 		LIMIT ?`
+	queryForTagPresence = `
+		SELECT trace_id
+		FROM tag_index
+		WHERE service_name = ? AND tag_key IN ? and start_time > ? and start_time < ?
+		ORDER BY start_time DESC
+		LIMIT ?`
 	queryByServiceName = `
 		SELECT trace_id
 		FROM service_name_index
@@ -301,9 +308,30 @@ func (s *SpanReader) findTraceIDs(ctx context.Context, traceQuery *spanstore.Tra
 	if len(traceQuery.Tags) > 0 {
 		return s.queryByTagsAndLogs(ctx, traceQuery)
 	}
+	if len(traceQuery.CheckTagsPresent) > 0 {
+		s.queryCheckTagPresence(ctx, traceQuery)
+	}
 	return s.queryByService(ctx, traceQuery)
 }
 
+func (s *SpanReader) queryCheckTagPresence(ctx context.Context, tq *spanstore.TraceQueryParameters) (dbmodel.UniqueTraceIDs, error) {
+	span, _ := startSpanForQuery(ctx, "queryCheckTagPresence", queryByTag)
+	query := s.session.Query(
+		tq.ServiceName,
+		"("+strings.Join(tq.CheckTagsPresent, ",")+")",
+		model.TimeAsEpochMicroseconds(tq.StartTimeMin),
+		model.TimeAsEpochMicroseconds(tq.StartTimeMax),
+		tq.NumTraces*limitMultiple,
+	).PageSize(0)
+
+	t, err := s.executeQuery(span, query, s.metrics.queryTagIndex)
+	if err != nil {
+		return nil, err
+	}
+
+	return t, nil
+}
+
 func (s *SpanReader) queryByTagsAndLogs(ctx context.Context, tq *spanstore.TraceQueryParameters) (dbmodel.UniqueTraceIDs, error) {
 	span, ctx := startSpanForQuery(ctx, "queryByTagsAndLogs", queryByTag)
 	defer span.Finish()

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)