Support for Jaeger tags with binary value

Since we are using the OTEL translator when ingesting Jaeger traces all
the tags with `VType=ValueType_BINARY` are being encoded into base64 and
stored as strings. The context of the type of value is lost after this
translation occurs, meaning that, when we return those tags in a Jaeger
query we return the base64 string as `VType=ValueType_STRING`.

To preserve the type of the value the prefix `__ValueType_BINARY__` will
be added to the base64 string before storing in the DB.

When retrieving a trace we look for tags that are string and have this
prefix, we remove it and decode the resulting base64 string.

Jaeger allows these type of tags for Spans, Process and Logs.

Author: alejandrodnm

CHANGELOG.md

@@ -27,6 +27,8 @@ We use the following categories for changes:
 ### Changed
 - Log throughput in the same line for samples, spans and metric metadata [#1643]
 - The `chunks_created` metrics was removed. [#1634]
+- When querying for Jaeger tags with binary values the binary data will be
+  returned instead of the base64 representation of the string [#1649].
 
 ### Fixed
 - Do not collect telemetry if `timescaledb.telemetry_level=off` [#1612]

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/jackc/pgtype v1.12.0
 	github.com/jackc/pgx/v4 v4.17.0
 	github.com/jaegertracing/jaeger v1.38.0
+	github.com/kr/pretty v0.3.0
 	github.com/oklog/run v1.1.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.57.2
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58
@@ -48,7 +49,7 @@ require (
 	go.opentelemetry.io/otel v1.9.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.7.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.7.0
-	go.opentelemetry.io/otel/sdk v1.7.0
+	go.opentelemetry.io/otel/sdk v1.8.0
 	go.opentelemetry.io/otel/trace v1.9.0
 	go.uber.org/atomic v1.9.0
 	go.uber.org/automaxprocs v1.5.1
@@ -127,6 +128,7 @@ require (
 	github.com/jpillora/backoff v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/julienschmidt/httprouter v1.3.0 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/magiconair/properties v1.8.6 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.12 // indirect
@@ -156,6 +158,7 @@ require (
 	github.com/prometheus/common/sigv4 v0.1.0 // indirect
 	github.com/prometheus/exporter-toolkit v0.7.1 // indirect
 	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/rogpeppe/go-internal v1.6.2 // indirect
 	github.com/rs/cors v1.8.2 // indirect
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
 	github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749 // indirect

go.sum

@@ -0,0 +1,136 @@
+package store
+
+import (
+	"encoding/base64"
+	"fmt"
+	"strings"
+
+	"github.com/jaegertracing/jaeger/model"
+)
+
+const VALUE_TYPE_BINARY = "__ValueType_BINARY__"
+const VALUE_TYPE_BINARY_LEN = len(VALUE_TYPE_BINARY)
+
+// spanBinaryTags is the set of keys from Jaeger model.KeyValue items used in
+// model.Span.Tags, model.Process.Tags and model.Log.Fields that have
+// `VType=ValueType_BINARY` and had to be encoded to be store as strings.
+type spanBinaryTags struct {
+	spanID      int64
+	spanTags    map[string]struct{}
+	processTags map[string]struct{}
+	logsTags    map[int]map[string]struct{}
+}
+
+func (t *spanBinaryTags) isEmpty() bool {
+	return len(t.spanTags) == 0 && len(t.logsTags) == 0 && len(t.processTags) == 0
+}
+
+func isEncodedBinaryValue(v interface{}) bool {
+	if val, isStr := v.(string); isStr {
+		if strings.HasPrefix(val, VALUE_TYPE_BINARY) {
+			return true
+		}
+	}
+	return false
+}
+
+// decodeSpanBinaryTags decodes the tags with binary values that are present in
+// the binaryTags sets for the span, process and logs.
+//
+// The spanBinaryTags struct is necesary in order to not go through all the
+// tags for each Span that the query is returning. We are taking advantage on
+// the fact that when translating from OTEL to Jaeger we had to visit all the
+// tags and were able to construct this set.
+//
+// When writing binary tags we encode the slice of bytes into a base64 string
+// representation and add the prefix `__ValueType_BINARY__`. Decoding implies
+// removing the prefix and decoding the base64 string.
+func decodeSpanBinaryTags(span *model.Span, binaryTags *spanBinaryTags) {
+	if len(binaryTags.spanTags) != 0 {
+		decodeBinaryTags(span.Tags, binaryTags.spanTags)
+	}
+
+	if len(binaryTags.processTags) != 0 {
+		decodeBinaryTags(span.Process.Tags, binaryTags.processTags)
+	}
+
+	if len(binaryTags.logsTags) != 0 {
+		for i, logsTags := range binaryTags.logsTags {
+			decodeBinaryTags(span.Logs[i].Fields, logsTags)
+		}
+	}
+}
+
+func decodeBinaryTags(actualTags []model.KeyValue, binaryTagsToDecode map[string]struct{}) {
+	for i, tag := range actualTags {
+		_, ok := binaryTagsToDecode[tag.Key]
+		if !ok {
+			continue
+		}
+
+		if tag.GetVType() != model.ValueType_STRING {
+			continue
+		}
+
+		encoded := tag.VStr
+		if !strings.HasPrefix(encoded, VALUE_TYPE_BINARY) {
+			continue
+		}
+
+		vBin, err := decodeBinaryTagValue(encoded)
+
+		// If we can't decode it means that we didn't encode it in the
+		// first place, so we should keep it as is.
+		if err != nil {
+			continue
+		}
+		actualTags[i] = model.KeyValue{
+			Key:     tag.Key,
+			VType:   model.ValueType_BINARY,
+			VBinary: vBin,
+		}
+	}
+}
+
+func decodeBinaryTagValue(encoded string) ([]byte, error) {
+	v := encoded[VALUE_TYPE_BINARY_LEN:]
+	return base64.StdEncoding.DecodeString(v)
+}
+
+func encodeBinaryTagToStr(tag model.KeyValue) model.KeyValue {
+	value := fmt.Sprintf("__ValueType_BINARY__%s", base64.StdEncoding.EncodeToString(tag.GetVBinary()))
+	return model.KeyValue{
+		Key:   tag.Key,
+		VType: model.ValueType_STRING,
+		VStr:  value,
+	}
+}
+
+func encodeBinaryTags(span *model.Span) {
+	for i, tag := range span.Tags {
+		if isBinaryTag(tag) {
+			continue
+		}
+		span.Tags[i] = encodeBinaryTagToStr(tag)
+	}
+
+	for _, log := range span.Logs {
+		for i, tag := range log.Fields {
+			if isBinaryTag(tag) {
+				continue
+			}
+			log.Fields[i] = encodeBinaryTagToStr(tag)
+		}
+	}
+
+	for i, tag := range span.Process.Tags {
+		if isBinaryTag(tag) {
+			continue
+		}
+		span.Process.Tags[i] = encodeBinaryTagToStr(tag)
+	}
+}
+
+func isBinaryTag(tag model.KeyValue) bool {
+	return tag.GetVType() != model.ValueType_BINARY
+}

pkg/jaeger/store/binary_tags.go

@@ -36,13 +36,19 @@ func findTraces(ctx context.Context, builder *Builder, conn pgxconn.PgxConn, q *
 
 func scanTraces(rows pgxconn.PgxRows) ([]*model.Trace, error) {
 	traces := ptrace.NewTraces()
+	spansBinaryTags := map[int64]*spanBinaryTags{}
 	for rows.Next() {
 		if rows.Err() != nil {
 			return nil, fmt.Errorf("trace row iterator: %w", rows.Err())
 		}
-		if err := ScanRow(rows, &traces); err != nil {
+		var err error
+		spanBinaryTags, err := ScanRow(rows, &traces)
+		if err != nil {
 			return nil, fmt.Errorf("error scanning trace: %w", err)
 		}
+		if !spanBinaryTags.isEmpty() {
+			spansBinaryTags[spanBinaryTags.spanID] = spanBinaryTags
+		}
 
 	}
 	if rows.Err() != nil {
@@ -54,10 +60,10 @@ func scanTraces(rows pgxconn.PgxRows) ([]*model.Trace, error) {
 		return nil, fmt.Errorf("internal-traces-to-jaeger-proto: %w", err)
 	}
 
-	return batchSliceToTraceSlice(batch), nil
+	return batchSliceToTraceSlice(batch, spansBinaryTags), nil
 }
 
-func batchSliceToTraceSlice(bSlice []*model.Batch) []*model.Trace {
+func batchSliceToTraceSlice(bSlice []*model.Batch, spansWithBinaryTag map[int64]*spanBinaryTags) []*model.Trace {
 	// Mostly Copied from Jaeger's grpc_client.go
 	// https://github.com/jaegertracing/jaeger/blob/067dff713ab635ade66315bbd05518d7b28f40c6/plugin/storage/grpc/shared/grpc_client.go#L179
 	traces := make([]*model.Trace, 0)
@@ -74,6 +80,9 @@ func batchSliceToTraceSlice(bSlice []*model.Batch) []*model.Trace {
 			}
 			//copy over the process from the batch
 			span.Process = batch.Process
+			if binaryTags, ok := spansWithBinaryTag[int64(span.SpanID)]; ok {
+				decodeSpanBinaryTags(span, binaryTags)
+			}
 			trace.Spans = append(trace.Spans, span)
 		}
 	}

pkg/jaeger/store/find_traces.go

@@ -6,6 +6,8 @@ package store
 
 import (
 	"context"
+	"encoding/base64"
+	"fmt"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -48,7 +50,42 @@ func (p *Store) StreamingSpanWriter() spanstore.Writer {
 	return p
 }
 
+func encodeBinaryTagToStr(tag model.KeyValue) model.KeyValue {
+	value := fmt.Sprintf("__ValueType_BINARY__%s", base64.StdEncoding.EncodeToString(tag.GetVBinary()))
+	return model.KeyValue{
+		Key:   tag.Key,
+		VType: model.ValueType_STRING,
+		VStr:  value,
+	}
+}
+
+func encodeBinaryTags(span *model.Span) {
+	for i, tag := range span.Tags {
+		if tag.GetVType() != model.ValueType_BINARY {
+			continue
+		}
+		span.Tags[i] = encodeBinaryTagToStr(tag)
+	}
+
+	for _, log := range span.Logs {
+		for i, tag := range log.Fields {
+			if tag.GetVType() != model.ValueType_BINARY {
+				continue
+			}
+			log.Fields[i] = encodeBinaryTagToStr(tag)
+		}
+	}
+
+	for i, tag := range span.Process.Tags {
+		if tag.GetVType() != model.ValueType_BINARY {
+			continue
+		}
+		span.Process.Tags[i] = encodeBinaryTagToStr(tag)
+	}
+}
+
 func (p *Store) WriteSpan(ctx context.Context, span *model.Span) error {
+	encodeBinaryTags(span)
 	batches := []*model.Batch{
 		{
 			Spans: []*model.Span{span},