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
`data:application/octet-stream; base64,` 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 return a
`ValueType_BINARY` with the result of decoding the base64 string.

The prefix was choosen from the `The "data" URL scheme` RFC
https://www.rfc-editor.org/rfc/rfc2397#section-2 .

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 MEDIA_TYPE_ENCODED_BINARY = "data:application/octet-stream; base64,"
+const MEDIA_TYPE_ENCODED_BINARY_LEN = len(MEDIA_TYPE_ENCODED_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, MEDIA_TYPE_ENCODED_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, MEDIA_TYPE_ENCODED_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[MEDIA_TYPE_ENCODED_BINARY_LEN:]
+	return base64.StdEncoding.DecodeString(v)
+}
+
+func encodeBinaryTagToStr(tag model.KeyValue) model.KeyValue {
+	value := fmt.Sprintf("%s%s", MEDIA_TYPE_ENCODED_BINARY, 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

@@ -49,6 +49,7 @@ func (p *Store) StreamingSpanWriter() spanstore.Writer {
 }
 
 func (p *Store) WriteSpan(ctx context.Context, span *model.Span) error {
+	encodeBinaryTags(span)
 	batches := []*model.Batch{
 		{
 			Spans: []*model.Span{span},