Fix Jaeger trace querying by converted Jaeger tags

When translating Jaeger traces to OTEL format, some Jaeger tags get
converted to the to OTEL specific fields and gets removed from the tag map. This
change takes this conversion into account when querying traces using these
tags.

Author: antekresic

CHANGELOG.md

@@ -19,6 +19,9 @@ We use the following categories for changes:
 - `-enable-feature=promql-per-step-stats` feature for statistics in PromQL evaluation
 - Add `readinessProbe` in helm chart [#1266]
 
+### Fixed
+- Fix trace querying by status code tag [#1384]
+
 ## [0.11.0] - 2022-05-11
 
 ## Added

go.mod

@@ -40,6 +40,7 @@ require (
 	github.com/thanos-io/thanos v0.25.2
 	go.opentelemetry.io/collector/model v0.49.0
 	go.opentelemetry.io/collector/pdata v0.49.0
+	go.opentelemetry.io/collector/semconv v0.52.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.32.0
 	go.opentelemetry.io/otel v1.7.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.6.3

go.sum

@@ -2074,6 +2074,12 @@ go.opentelemetry.io/collector/model v0.49.0 h1:mbUSNgpaBE3GWmzGsRb5t0xILpXIVYv7s
 go.opentelemetry.io/collector/model v0.49.0/go.mod h1:nOYQv9KoFPs6ihJwOi24qB209EOhS9HkwhGj54YiEAw=
 go.opentelemetry.io/collector/pdata v0.49.0 h1:aYj5rOlRC0x7lGXbc185LMsMMoY/pjOTXr5s1O2SzXs=
 go.opentelemetry.io/collector/pdata v0.49.0/go.mod h1:YwmKuiFhNgtmhRdpi8Q8FAWPa0AwJTCSlssSsAtuRcY=
+go.opentelemetry.io/collector/semconv v0.52.0 h1:ogYkQ6WL01xQ4GGSwWQejNTQwy/Pwcd6jCKFLSd7svA=
+go.opentelemetry.io/collector/semconv v0.52.0/go.mod h1:SxK0rUnUP7YeDakexzbE/vhimTOHwE6m/4aKKd9e27Q=
+go.opentelemetry.io/contrib v0.20.0 h1:ubFQUn0VCZ0gPwIoJfBJVpeBlyRMxu8Mm/huKWYd9p0=
+go.opentelemetry.io/contrib v0.20.0/go.mod h1:G/EtFaa6qaN7+LxqfIAT3GiZa7Wv5DTBUzl5H4LY0Kc=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.20.0/go.mod h1:oVGt1LRbBOBq1A5BQLlUg9UaU/54aiHw8cgjV3aWZ/E=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.28.0/go.mod h1:vEhqr0m4eTc+DWxfsXoXue2GBgV2uUwVznkGIHW/e5w=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.31.0/go.mod h1:SY9qHHUES6W3oZnO1H2W8NvsSovIoXRg/A1AH9px8+I=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0/go.mod h1:PFmBsWbldL1kiWZk9+0LBZz2brhByaGsvp6pRICMlPE=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.32.0 h1:mac9BKRqwaX6zxHPDe3pvmWpwuuIM0vuXv2juCnQevE=

pkg/jaeger/query/get_operations.go

@@ -41,7 +41,7 @@ func getOperations(ctx context.Context, conn pgxconn.PgxConn, query spanstore.Op
 	args := []interface{}{query.ServiceName}
 	kindQual := "TRUE"
 	if len(query.SpanKind) > 0 {
-		pgEnum, err := getPGKindEnum(query.SpanKind)
+		pgEnum, err := spanKindStringToInternal(query.SpanKind)
 		if err != nil {
 			return operationsResp, fmt.Errorf("converting query enum: %w", err)
 		}

pkg/jaeger/query/trace_query.go

@@ -1,7 +1,6 @@
 // This file and its contents are licensed under the Apache License 2.0.
 // Please see the included NOTICE for copyright information and
 // LICENSE for a copy of the license.
-
 package query
 
 import (
@@ -12,6 +11,9 @@ import (
 	"github.com/jackc/pgtype"
 	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	semconv "go.opentelemetry.io/collector/semconv/v1.6.1"
 )
 
 const (
@@ -106,6 +108,13 @@ const (
 		%s
 	) AS complete_trace ON (TRUE)
 	`
+
+	// Keys used to represent OTLP constructs from Jaeger tags which are then dropped from the tag map.
+	TagError         = "error"
+	TagHostname      = "hostname"
+	TagJaegerVersion = "jaeger.version"
+	TagSpanKind      = "span.kind"
+	TagW3CTraceState = "w3c.tracestate"
 )
 
 type Builder struct {
@@ -148,8 +157,7 @@ func (b *Builder) findTracesQuery(q *spanstore.TraceQueryParameters) (string, []
 }
 
 func (b *Builder) findTraceIDsQuery(q *spanstore.TraceQueryParameters) (string, []interface{}) {
-	subquery, params := b.buildTraceIDSubquery(q)
-	return subquery, params
+	return b.buildTraceIDSubquery(q)
 }
 
 func (b *Builder) getTraceQuery(traceID model.TraceID) (string, []interface{}, error) {
@@ -192,6 +200,50 @@ func (b *Builder) buildTraceIDSubquery(q *spanstore.TraceQueryParameters) (strin
 		operation_clauses = append(operation_clauses, qual)
 	}
 
+	if len(q.Tags) > 0 {
+		for k, v := range q.Tags {
+			switch k {
+			case TagError:
+				var sc pdata.StatusCode
+				switch v {
+				case "true":
+					sc = ptrace.StatusCodeError
+				case "false":
+					sc = ptrace.StatusCodeUnset
+				default:
+					// Ignore tag if we don't match boolean.
+					continue
+				}
+				params = append(params, statusCodeToInternal(sc))
+				qual := fmt.Sprintf(`s.status_code = $%d`, len(params))
+				clauses = append(clauses, qual)
+			case TagHostname:
+				// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/a13030ca6208fa7d4477f0805f3b95e271f32b24/pkg/translator/jaeger/jaegerproto_to_traces.go#L109
+				params = append(params, semconv.AttributeHostName, "\""+v+"\"")
+				qual := fmt.Sprintf(`_ps_trace.tag_map_denormalize(s.span_tags)->$%d = $%d`, len(params)-1, len(params))
+				clauses = append(clauses, qual)
+			case TagJaegerVersion:
+				// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/a13030ca6208fa7d4477f0805f3b95e271f32b24/pkg/translator/jaeger/jaegerproto_to_traces.go#L119
+				params = append(params, "opencensus.exporterversion", "\"Jaeger-"+v+"\"")
+				qual := fmt.Sprintf(`_ps_trace.tag_map_denormalize(s.span_tags)->$%d = $%d`, len(params)-1, len(params))
+				clauses = append(clauses, qual)
+			case TagSpanKind:
+				params = append(params, jSpanKindToInternal(v))
+				qual := fmt.Sprintf("span_kind = $%d", len(params))
+				operation_clauses = append(operation_clauses, qual)
+			case TagW3CTraceState:
+				params = append(params, v)
+				qual := fmt.Sprintf(`s.trace_state = $%d`, len(params))
+				clauses = append(clauses, qual)
+			default:
+				//TODO make sure this is optimized correctly
+				params = append(params, k, "\""+v+"\"")
+				qual := fmt.Sprintf(`_ps_trace.tag_map_denormalize(s.span_tags)->$%d = $%d`, len(params)-1, len(params))
+				clauses = append(clauses, qual)
+			}
+		}
+	}
+
 	if len(operation_clauses) > 0 {
 		qual := fmt.Sprintf(`
 		   s.operation_id IN (
@@ -205,15 +257,6 @@ func (b *Builder) buildTraceIDSubquery(q *spanstore.TraceQueryParameters) (strin
 		clauses = append(clauses, qual)
 	}
 
-	if len(q.Tags) > 0 {
-		for k, v := range q.Tags {
-			//TODO make sure this is optimized correctly
-			params = append(params, k, "\""+v+"\"")
-			qual := fmt.Sprintf(`_ps_trace.tag_map_denormalize(s.span_tags)->$%d = $%d`, len(params)-1, len(params))
-			clauses = append(clauses, qual)
-		}
-	}
-
 	//todo check the inclusive semantics here
 	var defaultTime time.Time
 	if q.StartTimeMin != defaultTime {

pkg/jaeger/query/trace_scan.go

@@ -163,11 +163,7 @@ func populateSpan(
 	ref.SetName(dbResult.spanName)
 
 	if dbResult.kind.Status == pgtype.Present {
-		kind, err := makeKind(dbResult.kind.String)
-		if err != nil {
-			return err
-		}
-		ref.SetKind(kind)
+		ref.SetKind(internalToSpanKind(dbResult.kind.String))
 	}
 
 	ref.SetStartTimestamp(pdata.NewTimestampFromTime(dbResult.startTime))
@@ -202,11 +198,7 @@ func populateSpan(
 
 func setStatus(ref ptrace.Span, dbRes *spanDBResult) error {
 	if dbRes.statusCode != "" {
-		code, err := makeStatusCode(dbRes.statusCode)
-		if err != nil {
-			return err
-		}
-		ref.Status().SetCode(code)
+		ref.Status().SetCode(internalToStatusCode(dbRes.statusCode))
 	}
 	if dbRes.statusMessage.Status != pgtype.Null {
 		message := dbRes.statusMessage.String
@@ -300,54 +292,3 @@ func makeSpanId(s *int64) pdata.SpanID {
 	b8 := trace.Int64ToByteArray(*s)
 	return pdata.NewSpanID(b8)
 }
-
-func getPGKindEnum(jaegerKind string) (string, error) {
-	switch jaegerKind {
-	case ptrace.SpanKindClient.String():
-		return "client", nil
-	case ptrace.SpanKindServer.String():
-		return "server", nil
-	case ptrace.SpanKindInternal.String():
-		return "internal", nil
-	case ptrace.SpanKindConsumer.String():
-		return "consumer", nil
-	case ptrace.SpanKindProducer.String():
-		return "producer", nil
-	case ptrace.SpanKindUnspecified.String():
-		return "unspecified", nil
-	default:
-		return "", fmt.Errorf("unknown span kind: %v", jaegerKind)
-	}
-}
-
-func makeKind(s string) (pdata.SpanKind, error) {
-	switch s {
-	case "client":
-		return ptrace.SpanKindClient, nil
-	case "server":
-		return ptrace.SpanKindServer, nil
-	case "internal":
-		return ptrace.SpanKindInternal, nil
-	case "consumer":
-		return ptrace.SpanKindConsumer, nil
-	case "producer":
-		return ptrace.SpanKindProducer, nil
-	case "unspecified":
-		return ptrace.SpanKindUnspecified, nil
-	default:
-		return ptrace.SpanKindUnspecified, fmt.Errorf("unknown span kind: %s", s)
-	}
-}
-
-func makeStatusCode(s string) (pdata.StatusCode, error) {
-	switch s {
-	case "ok":
-		return ptrace.StatusCodeOk, nil
-	case "error":
-		return ptrace.StatusCodeError, nil
-	case "unset":
-		return ptrace.StatusCodeUnset, nil
-	default:
-		return ptrace.StatusCodeUnset, fmt.Errorf("unknown status-code kind: %s", s)
-	}
-}

pkg/jaeger/query/translation.go

@@ -0,0 +1,82 @@
+// This file and its contents are licensed under the Apache License 2.0.
+// Please see the included NOTICE for copyright information and
+// LICENSE for a copy of the license.
+package query
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+var (
+	// Map of status codes to internal DB values.
+	statusCodeInternalValue = map[string]pdata.StatusCode{
+		"ok":    ptrace.StatusCodeOk,
+		"error": ptrace.StatusCodeError,
+		"unset": ptrace.StatusCodeUnset,
+	}
+	// Map of span kinds to internal DB values.
+	spanKindInternalValue = map[string]pdata.SpanKind{
+		"client":      ptrace.SpanKindClient,
+		"server":      ptrace.SpanKindServer,
+		"internal":    ptrace.SpanKindInternal,
+		"consumer":    ptrace.SpanKindConsumer,
+		"producer":    ptrace.SpanKindProducer,
+		"unspecified": ptrace.SpanKindUnspecified,
+	}
+	// Map of Jaeger span kind tag strings to internal DB values.
+	// This is kept explicit even though the values are identical.
+	jSpanKindToInternalValue = map[string]string{
+		"client":   "client",
+		"server":   "server",
+		"internal": "internal",
+		"consumer": "consumer",
+		"producer": "producer",
+	}
+)
+
+func internalToStatusCode(s string) pdata.StatusCode {
+	v, ok := statusCodeInternalValue[s]
+	if !ok {
+		panic(fmt.Sprintf("status code %s does not have internal representation", s))
+	}
+	return v
+}
+
+func statusCodeToInternal(sc pdata.StatusCode) string {
+	for k, v := range statusCodeInternalValue {
+		if sc == v {
+			return k
+		}
+	}
+	panic(fmt.Sprintf("status code %s does not have internal representation", sc.String()))
+}
+
+func internalToSpanKind(s string) pdata.SpanKind {
+	v, ok := spanKindInternalValue[s]
+	if !ok {
+		panic(fmt.Sprintf("span kind %s does not have internal representation", s))
+	}
+	return v
+}
+
+func spanKindStringToInternal(kind string) (string, error) {
+	for k, v := range spanKindInternalValue {
+		if v.String() == kind {
+			return k, nil
+		}
+	}
+	return "", fmt.Errorf("unknown span kind: %s", kind)
+}
+
+// JSpanKindToInternal translates Jaeger span kind tag value to internal enum value used for storage.
+// Corresponds to:
+// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/a13030ca6208fa7d4477f0805f3b95e271f32b24/pkg/translator/jaeger/jaegerproto_to_traces.go#L303
+func jSpanKindToInternal(spanKind string) string {
+	if v, ok := jSpanKindToInternalValue[spanKind]; ok {
+		return v
+	}
+	return "unspecified"
+}

pkg/tests/end_to_end_tests/dataset_traces_test.go

@@ -35,7 +35,13 @@ var (
 
 	service0 = "service-name-0"
 
-	spanAttributes      = pdata.NewAttributeMapFromMap(map[string]pcommon.Value{"span-attr": pcommon.NewValueString("span-attr-val")})
+	spanAttributes = pdata.NewAttributeMapFromMap(
+		map[string]pcommon.Value{
+			"span-attr":                  pcommon.NewValueString("span-attr-val"),
+			"host.name":                  pcommon.NewValueString("hostname1"),
+			"opencensus.exporterversion": pcommon.NewValueString("Jaeger-1.0.0"),
+		},
+	)
 	spanEventAttributes = pdata.NewAttributeMapFromMap(map[string]pcommon.Value{"span-event-attr": pcommon.NewValueString("span-event-attr-val")})
 	spanLinkAttributes  = pdata.NewAttributeMapFromMap(map[string]pcommon.Value{"span-link-attr": pcommon.NewValueString("span-link-attr-val")})
 )
@@ -44,6 +50,18 @@ func getTraceId(bSlice [16]byte) string {
 	return hex.EncodeToString(bSlice[:])
 }
 
+// generateBrokenTestTraces switches start and end times for every span to check if
+// we handle this broken situation correctly and reverse the times while ingesting.
+func generateBrokenTestTraces() pdata.Traces {
+	data := generateTestTrace()
+	startTime := data.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).StartTimestamp()
+	endTime := data.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).EndTimestamp()
+
+	data.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).SetStartTimestamp(endTime)
+	data.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).SetEndTimestamp(startTime)
+	return data
+}
+
 func generateTestTrace() pdata.Traces {
 	rand.Seed(1)
 	spanCount := 4
@@ -139,11 +157,11 @@ func fillSpanOne(span pdata.Span) {
 	span.SetTraceID(pdata.NewTraceID(traceID1))
 	span.SetSpanID(pdata.NewSpanID(generateRandSpanID()))
 	span.SetName("operationA")
-	// test the logic that detects and fixes end < start
-	span.SetStartTimestamp(testSpanEndTimestamp)
-	span.SetEndTimestamp(testSpanStartTimestamp)
+	span.SetStartTimestamp(testSpanStartTimestamp)
+	span.SetEndTimestamp(testSpanEndTimestamp)
 	span.SetDroppedAttributesCount(1)
 	span.SetTraceState("span-trace-state1")
+	span.SetKind(ptrace.SpanKindClient)
 	initSpanAttributes(span.Attributes())
 	evs := span.Events()
 	ev0 := evs.AppendEmpty()

pkg/tests/end_to_end_tests/generate_jaeger_response_test.go

@@ -70,7 +70,7 @@ func insertDataIntoJaeger(endpoint string, data pdata.Traces) error {
 
 	client := jaegerproto.NewCollectorServiceClient(conn)
 
-	batches, err := jaegertranslator.InternalTracesToJaegerProto(data)
+	batches, err := jaegertranslator.ProtoFromTraces(data)
 	if err != nil {
 		panic(err)
 	}