Make Jaeger Event queryable using name and tags

So far we were matching tags present only in span, however Jaeger supports query on tags present in Event too.

Fixes #1552

Signed-off-by: Arunprasad Rajkumar <ar.arunprasad@gmail.com>

CHANGELOG.md

@@ -17,12 +17,12 @@ We use the following categories for changes:
 
 ### Added
 - Implement Jaeger gRPC remote storage writer interface [#1543]
-
 - Helm chart now ships a JSON Schema for imposing a structure of the values.yaml file [#1551]
 
 ### Changed
 
 ### Fixed
+- Make Jaeger Event queryable using name and tags [#1553]
 
 ## [0.13.0] - 2022-07-20
 

pkg/jaeger/store/trace_query.go

@@ -85,12 +85,13 @@ const (
 		trace_sub.trace_id, start_time_max - $%[1]d::interval as time_low, start_time_max + $%[1]d::interval as time_high
 	FROM (
 		SELECT
-			trace_id,
+			s.trace_id,
 			max(start_time) as start_time_max
 		FROM _ps_trace.span s
+		LEFT JOIN _ps_trace.event e ON(s.trace_id = e.trace_id AND s.span_id = e.span_id)
 		WHERE
 			%[2]s
-		GROUP BY trace_id
+		GROUP BY s.trace_id
 	) as trace_sub
 	ORDER BY trace_sub.start_time_max DESC
 	`
@@ -117,6 +118,7 @@ const (
 	TagJaegerVersion = "jaeger.version"
 	TagSpanKind      = "span.kind"
 	TagW3CTraceState = "w3c.tracestate"
+	TagEventName     = "event"
 )
 
 var digitCheck = regexp.MustCompile(`^\d*\.?\d+$`) // Ints or Floats.
@@ -239,6 +241,10 @@ func (b *Builder) buildTraceIDSubquery(q *spanstore.TraceQueryParameters) (strin
 				params = append(params, v)
 				qual := fmt.Sprintf(`s.trace_state = $%d`, len(params))
 				clauses = append(clauses, qual)
+			case TagEventName:
+				params = append(params, v)
+				qual := fmt.Sprintf(`e.name = $%d`, len(params))
+				clauses = append(clauses, qual)
 			default:
 				//TODO make sure this is optimized correctly
 				val := "\"" + v + "\""
@@ -249,8 +255,9 @@ func (b *Builder) buildTraceIDSubquery(q *spanstore.TraceQueryParameters) (strin
 				params = append(params, k, val)
 				// Note: We do not need to check resource tags in above cases, since they
 				// come from Jaeger conversion that are specific to span_tags.
+				// Tags are spread into _ps_trace.span and _ps_trace.event tables.
 				qual := fmt.Sprintf(
-					`(_ps_trace.tag_map_denormalize(s.span_tags)->$%[1]d = $%[2]d OR _ps_trace.tag_map_denormalize(s.resource_tags)->$%[1]d = $%[2]d)`,
+					`(_ps_trace.tag_map_denormalize(s.span_tags)->$%[1]d = $%[2]d OR _ps_trace.tag_map_denormalize(s.resource_tags)->$%[1]d = $%[2]d OR _ps_trace.tag_map_denormalize(e.tags)->$%[1]d = $%[2]d)`,
 					len(params)-1, len(params))
 				clauses = append(clauses, qual)
 			}

pkg/tests/end_to_end_tests/generate_jaeger_response_test.go

@@ -5,7 +5,7 @@
 // +build gen_response
 
 // Below line generates the Jaeger API responses.
-//go:generate go test . -run TestGenerateJaegerAPIResponses
+//go:generate go test . -run TestGenerateJaegerAPIResponses -tags gen_response
 
 package end_to_end_tests
 

pkg/tests/end_to_end_tests/ingest_trace_test.go

@@ -144,6 +144,65 @@ func findTraceTest(t testing.TB, q *store.Store) {
 	require.NoError(t, err)
 	require.Equal(t, 1, len(traces))
 
+	// Query Event by name.
+	request = &spanstore.TraceQueryParameters{
+		Tags: map[string]string{
+			"event": "event",
+		},
+	}
+	traces, err = q.FindTraces(context.Background(), request)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(traces))
+
+	// Query Event by tag, service and operation.
+	request = &spanstore.TraceQueryParameters{
+		ServiceName:   "service-name-0",
+		OperationName: "operationA",
+		Tags: map[string]string{
+			"span-event-attr": "span-event-attr-val",
+		},
+	}
+	traces, err = q.FindTraces(context.Background(), request)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(traces))
+
+	// Query Event by name, service and operation.
+	request = &spanstore.TraceQueryParameters{
+		ServiceName:   "service-name-0",
+		OperationName: "operationA",
+		Tags: map[string]string{
+			"event": "event-with-attr",
+		},
+	}
+	traces, err = q.FindTraces(context.Background(), request)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(traces))
+
+	// Query Event by name, non existent tag, service and operation.
+	request = &spanstore.TraceQueryParameters{
+		ServiceName:   "service-name-0",
+		OperationName: "operationA",
+		Tags: map[string]string{
+			"event":           "event-with-attr",
+			"span-event-attr": "not-exist",
+		},
+	}
+	traces, err = q.FindTraces(context.Background(), request)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(traces))
+
+	// Query non existent Event by name, service and operation.
+	request = &spanstore.TraceQueryParameters{
+		ServiceName:   "service-name-0",
+		OperationName: "operationA",
+		Tags: map[string]string{
+			"event": "not-exists",
+		},
+	}
+	traces, err = q.FindTraces(context.Background(), request)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(traces))
+
 	request = &spanstore.TraceQueryParameters{
 		ServiceName:   "service-name-0",
 		OperationName: "operationA",

pkg/tests/end_to_end_tests/trace_query_integration_test.go

@@ -81,6 +81,22 @@ var traceQueryCases = []traceQuery{
 		traceID: traceID1,
 		tag:     &tag{"isExpired", "true"},
 	},
+	{
+		name:    "get trace by event name",
+		start:   timestamp.FromTime(testSpanStartTime) * 1000,
+		end:     timestamp.FromTime(testSpanEndTime) * 1000,
+		service: service0,
+		traceID: traceID1,
+		tag:     &tag{"event", "event-with-attr"},
+	},
+	{
+		name:    "get trace by event attribute",
+		start:   timestamp.FromTime(testSpanStartTime) * 1000,
+		end:     timestamp.FromTime(testSpanEndTime) * 1000,
+		service: service0,
+		traceID: traceID1,
+		tag:     &tag{"span-event-attr", "span-event-attr-val"},
+	},
 }
 
 func TestCompareTraceQueryResponse(t *testing.T) {