Tempo: Return new version of dataframe schema directly from the backend (#32116)

* Return dataframe directly from the backend * Streamline some transforms * Fix lint issues * Remove unused lib * Fix datasource test * Fix imports and add some typings * Fix the typings and some tests * Add private doc comment * Remove private tag * Add comments * Fix some API docs issues
2021-03-22 19:09:15 +01:00
parent f8ec947700
commit 3ef9cac640
15 changed files with 616 additions and 109 deletions
@@ -2,7 +2,6 @@ package tempo

 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -12,11 +11,7 @@ import (
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/plugins"

-	jaeger "github.com/jaegertracing/jaeger/model"
-	jaeger_json "github.com/jaegertracing/jaeger/model/converter/json"
-
 	ot_pdata "go.opentelemetry.io/collector/consumer/pdata"
-	ot_jaeger "go.opentelemetry.io/collector/translator/trace/jaeger"
 )

 type tempoExecutor struct {
@@ -77,42 +72,17 @@ func (e *tempoExecutor) DataQuery(ctx context.Context, dsInfo *models.DataSource

 	otTrace := ot_pdata.NewTraces()
 	err = otTrace.FromOtlpProtoBytes(body)
+
 	if err != nil {
 		return plugins.DataResponse{}, fmt.Errorf("failed to convert tempo response to Otlp: %w", err)
 	}

-	jaegerBatches, err := ot_jaeger.InternalTracesToJaegerProto(otTrace)
+	frame, err := TraceToFrame(otTrace)
 	if err != nil {
-		return plugins.DataResponse{}, fmt.Errorf("failed to translate to jaegerBatches %v: %w", traceID, err)
-	}
-
-	jaegerTrace := &jaeger.Trace{
-		Spans:      []*jaeger.Span{},
-		ProcessMap: []jaeger.Trace_ProcessMapping{},
-	}
-
-	// otel proto conversion doesn't set jaeger processes
-	for _, batch := range jaegerBatches {
-		for _, s := range batch.Spans {
-			s.Process = batch.Process
-		}
-
-		jaegerTrace.Spans = append(jaegerTrace.Spans, batch.Spans...)
-		jaegerTrace.ProcessMap = append(jaegerTrace.ProcessMap, jaeger.Trace_ProcessMapping{
-			Process:   *batch.Process,
-			ProcessID: batch.Process.ServiceName,
-		})
-	}
-	jsonTrace := jaeger_json.FromDomain(jaegerTrace)
-
-	traceBytes, err := json.Marshal(jsonTrace)
-	if err != nil {
-		return plugins.DataResponse{}, fmt.Errorf("failed to json.Marshal trace \"%s\" :%w", traceID, err)
-	}
-
-	frames := []*data.Frame{
-		{Name: "Traces", RefID: refID, Fields: []*data.Field{data.NewField("trace", nil, []string{string(traceBytes)})}},
+		return plugins.DataResponse{}, fmt.Errorf("failed to transform trace %v to data frame: %w", traceID, err)
 	}
+	frame.RefID = refID
+	frames := []*data.Frame{frame}
 	queryResult.Dataframes = plugins.NewDecodedDataFrames(frames)

 	return plugins.DataResponse{
@@ -0,0 +1,341 @@
+package tempo
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/translator/conventions"
+	tracetranslator "go.opentelemetry.io/collector/translator/trace"
+)
+
+type KeyValue struct {
+	Value interface{} `json:"value"`
+	Key   string      `json:"key"`
+}
+
+type TraceLog struct {
+	// Millisecond epoch time
+	Timestamp float64     `json:"timestamp"`
+	Fields    []*KeyValue `json:"fields"`
+}
+
+func TraceToFrame(td pdata.Traces) (*data.Frame, error) {
+	// In open telemetry format the spans are grouped first by resource/service they originated in and inside that
+	// resource they are grouped by the instrumentation library which created them.
+
+	resourceSpans := td.ResourceSpans()
+
+	if resourceSpans.Len() == 0 {
+		return nil, nil
+	}
+
+	frame := &data.Frame{
+		Name: "Trace",
+		Fields: []*data.Field{
+			data.NewField("traceID", nil, []string{}),
+			data.NewField("spanID", nil, []string{}),
+			data.NewField("parentSpanID", nil, []string{}),
+			data.NewField("operationName", nil, []string{}),
+			data.NewField("serviceName", nil, []string{}),
+			data.NewField("serviceTags", nil, []string{}),
+			data.NewField("startTime", nil, []float64{}),
+			data.NewField("duration", nil, []float64{}),
+			data.NewField("logs", nil, []string{}),
+			data.NewField("tags", nil, []string{}),
+		},
+		Meta: &data.FrameMeta{
+			// TODO: use constant once available in the SDK
+			PreferredVisualization: "trace",
+		},
+	}
+
+	for i := 0; i < resourceSpans.Len(); i++ {
+		rs := resourceSpans.At(i)
+		rows, err := resourceSpansToRows(rs)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, row := range rows {
+			frame.AppendRow(row...)
+		}
+	}
+
+	return frame, nil
+}
+
+// resourceSpansToRows processes all the spans for a particular resource/service
+func resourceSpansToRows(rs pdata.ResourceSpans) ([][]interface{}, error) {
+	resource := rs.Resource()
+	ilss := rs.InstrumentationLibrarySpans()
+
+	if resource.Attributes().Len() == 0 || ilss.Len() == 0 {
+		return [][]interface{}{}, nil
+	}
+
+	// Approximate the number of the spans as the number of the spans in the first
+	// instrumentation library info.
+	rows := make([][]interface{}, 0, ilss.At(0).Spans().Len())
+
+	for i := 0; i < ilss.Len(); i++ {
+		ils := ilss.At(i)
+
+		// These are finally the actual spans
+		spans := ils.Spans()
+
+		for j := 0; j < spans.Len(); j++ {
+			span := spans.At(j)
+			row, err := spanToSpanRow(span, ils.InstrumentationLibrary(), resource)
+			if err != nil {
+				return nil, err
+			}
+			if row != nil {
+				rows = append(rows, row)
+			}
+		}
+	}
+
+	return rows, nil
+}
+
+func spanToSpanRow(span pdata.Span, libraryTags pdata.InstrumentationLibrary, resource pdata.Resource) ([]interface{}, error) {
+	traceID, err := traceIDToString(span.TraceID())
+	if err != nil {
+		return nil, err
+	}
+
+	spanID, err := spanIDToString(span.SpanID())
+	if err != nil {
+		return nil, err
+	}
+
+	// Should get error only if empty in which case we are ok with empty string
+	parentSpanID, _ := spanIDToString(span.ParentSpanID())
+	startTime := float64(span.StartTime()) / 1_000_000
+	serviceName, serviceTags := resourceToProcess(resource)
+
+	serviceTagsJson, err := json.Marshal(serviceTags)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal service tags: %w", err)
+	}
+
+	spanTags, err := json.Marshal(getSpanTags(span, libraryTags))
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal span tags: %w", err)
+	}
+
+	logs, err := json.Marshal(spanEventsToLogs(span.Events()))
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal span logs: %w", err)
+	}
+
+	return []interface{}{
+		traceID,
+		spanID,
+		parentSpanID,
+		span.Name(),
+		serviceName,
+		toJSONString(serviceTagsJson),
+		startTime,
+		float64(span.EndTime()-span.StartTime()) / 1_000_000,
+		toJSONString(logs),
+		toJSONString(spanTags),
+	}, nil
+}
+
+func toJSONString(json []byte) string {
+	s := string(json)
+	if s == "null" {
+		return ""
+	}
+	return s
+}
+
+// TraceID can be the size of 2 uint64 in OT but we just need a string
+func traceIDToString(traceID pdata.TraceID) (string, error) {
+	traceIDHigh, traceIDLow := tracetranslator.TraceIDToUInt64Pair(traceID)
+	if traceIDLow == 0 && traceIDHigh == 0 {
+		return "", fmt.Errorf("OC span has an all zeros trace ID")
+	}
+	return fmt.Sprintf("%d%d", traceIDHigh, traceIDLow), nil
+}
+
+func spanIDToString(spanID pdata.SpanID) (string, error) {
+	uSpanID := tracetranslator.SpanIDToUInt64(spanID)
+	if uSpanID == 0 {
+		return "", fmt.Errorf("OC span has an all zeros span ID")
+	}
+	return fmt.Sprintf("%d", uSpanID), nil
+}
+
+func resourceToProcess(resource pdata.Resource) (string, []*KeyValue) {
+	attrs := resource.Attributes()
+	serviceName := tracetranslator.ResourceNoServiceName
+	if attrs.Len() == 0 {
+		return serviceName, nil
+	}
+
+	tags := make([]*KeyValue, 0, attrs.Len()-1)
+	attrs.ForEach(func(key string, attr pdata.AttributeValue) {
+		if key == conventions.AttributeServiceName {
+			serviceName = attr.StringVal()
+		}
+		tags = append(tags, &KeyValue{Key: key, Value: getAttributeVal(attr)})
+	})
+
+	return serviceName, tags
+}
+
+func getAttributeVal(attr pdata.AttributeValue) interface{} {
+	switch attr.Type() {
+	case pdata.AttributeValueSTRING:
+		return attr.StringVal()
+	case pdata.AttributeValueINT:
+		return attr.IntVal()
+	case pdata.AttributeValueBOOL:
+		return attr.BoolVal()
+	case pdata.AttributeValueDOUBLE:
+		return attr.DoubleVal()
+	case pdata.AttributeValueMAP, pdata.AttributeValueARRAY:
+		return tracetranslator.AttributeValueToString(attr, false)
+	default:
+		return nil
+	}
+}
+
+func getSpanTags(span pdata.Span, instrumentationLibrary pdata.InstrumentationLibrary) []*KeyValue {
+	var tags []*KeyValue
+
+	libraryTags := getTagsFromInstrumentationLibrary(instrumentationLibrary)
+	if libraryTags != nil {
+		tags = append(tags, libraryTags...)
+	}
+	span.Attributes().ForEach(func(key string, attr pdata.AttributeValue) {
+		tags = append(tags, &KeyValue{Key: key, Value: getAttributeVal(attr)})
+	})
+
+	status := span.Status()
+	possibleNilTags := []*KeyValue{
+		getTagFromSpanKind(span.Kind()),
+		getTagFromStatusCode(status.Code()),
+		getErrorTagFromStatusCode(status.Code()),
+		getTagFromStatusMsg(status.Message()),
+		getTagFromTraceState(span.TraceState()),
+	}
+
+	for _, tag := range possibleNilTags {
+		if tag != nil {
+			tags = append(tags, tag)
+		}
+	}
+	return tags
+}
+
+func getTagsFromInstrumentationLibrary(il pdata.InstrumentationLibrary) []*KeyValue {
+	var keyValues []*KeyValue
+	if ilName := il.Name(); ilName != "" {
+		kv := &KeyValue{
+			Key:   conventions.InstrumentationLibraryName,
+			Value: ilName,
+		}
+		keyValues = append(keyValues, kv)
+	}
+	if ilVersion := il.Version(); ilVersion != "" {
+		kv := &KeyValue{
+			Key:   conventions.InstrumentationLibraryVersion,
+			Value: ilVersion,
+		}
+		keyValues = append(keyValues, kv)
+	}
+
+	return keyValues
+}
+
+func getTagFromSpanKind(spanKind pdata.SpanKind) *KeyValue {
+	var tagStr string
+	switch spanKind {
+	case pdata.SpanKindCLIENT:
+		tagStr = string(tracetranslator.OpenTracingSpanKindClient)
+	case pdata.SpanKindSERVER:
+		tagStr = string(tracetranslator.OpenTracingSpanKindServer)
+	case pdata.SpanKindPRODUCER:
+		tagStr = string(tracetranslator.OpenTracingSpanKindProducer)
+	case pdata.SpanKindCONSUMER:
+		tagStr = string(tracetranslator.OpenTracingSpanKindConsumer)
+	case pdata.SpanKindINTERNAL:
+		tagStr = string(tracetranslator.OpenTracingSpanKindInternal)
+	default:
+		return nil
+	}
+
+	return &KeyValue{
+		Key:   tracetranslator.TagSpanKind,
+		Value: tagStr,
+	}
+}
+
+func getTagFromStatusCode(statusCode pdata.StatusCode) *KeyValue {
+	return &KeyValue{
+		Key:   tracetranslator.TagStatusCode,
+		Value: int64(statusCode),
+	}
+}
+
+func getErrorTagFromStatusCode(statusCode pdata.StatusCode) *KeyValue {
+	if statusCode == pdata.StatusCodeError {
+		return &KeyValue{
+			Key:   tracetranslator.TagError,
+			Value: true,
+		}
+	}
+	return nil
+}
+
+func getTagFromStatusMsg(statusMsg string) *KeyValue {
+	if statusMsg == "" {
+		return nil
+	}
+	return &KeyValue{
+		Key:   tracetranslator.TagStatusMsg,
+		Value: statusMsg,
+	}
+}
+
+func getTagFromTraceState(traceState pdata.TraceState) *KeyValue {
+	if traceState != pdata.TraceStateEmpty {
+		return &KeyValue{
+			Key:   tracetranslator.TagW3CTraceState,
+			Value: string(traceState),
+		}
+	}
+	return nil
+}
+
+func spanEventsToLogs(events pdata.SpanEventSlice) []*TraceLog {
+	if events.Len() == 0 {
+		return nil
+	}
+
+	logs := make([]*TraceLog, 0, events.Len())
+	for i := 0; i < events.Len(); i++ {
+		event := events.At(i)
+		fields := make([]*KeyValue, 0, event.Attributes().Len()+1)
+		if event.Name() != "" {
+			fields = append(fields, &KeyValue{
+				Key:   tracetranslator.TagMessage,
+				Value: event.Name(),
+			})
+		}
+		event.Attributes().ForEach(func(key string, attr pdata.AttributeValue) {
+			fields = append(fields, &KeyValue{Key: key, Value: getAttributeVal(attr)})
+		})
+		logs = append(logs, &TraceLog{
+			Timestamp: float64(event.Timestamp()) / 1_000_000,
+			Fields:    fields,
+		})
+	}
+
+	return logs
+}
@@ -0,0 +1,111 @@
+package tempo
+
+import (
+	"io/ioutil"
+	"testing"
+
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/stretchr/testify/require"
+	ot_pdata "go.opentelemetry.io/collector/consumer/pdata"
+)
+
+func TestTraceToFrame(t *testing.T) {
+	t.Run("should transform tempo protobuf response into dataframe", func(t *testing.T) {
+		// For what ever reason you cannot easily create pdata.Traces for the TraceToFrame from something more readable
+		// like json. You could tediously create the structures manually using all the setters for everything or use
+		// https://github.com/grafana/tempo/tree/master/pkg/tempopb to create the protobuf structs from something like
+		// json. At the moment just saving some real tempo proto response into file and loading was the easiest and
+		// as my patience was diminished trying to figure this out, I say it's good enough.
+		proto, err := ioutil.ReadFile("testData/tempo_proto_response")
+		require.NoError(t, err)
+
+		otTrace := ot_pdata.NewTraces()
+		err = otTrace.FromOtlpProtoBytes(proto)
+		require.NoError(t, err)
+
+		frame, err := TraceToFrame(otTrace)
+		require.NoError(t, err)
+
+		require.Equal(t, 30, frame.Rows())
+		require.ElementsMatch(t, fields, fieldNames(frame))
+
+		bFrame := &BetterFrame{frame}
+		root := rootSpan(bFrame)
+		require.NotNil(t, root)
+
+		require.Equal(t, "HTTP GET - loki_api_v1_query_range", root["operationName"])
+		require.Equal(t, "loki-all", root["serviceName"])
+		require.Equal(t, "[{\"value\":\"loki-all\",\"key\":\"service.name\"},{\"value\":\"Jaeger-Go-2.25.0\",\"key\":\"opencensus.exporterversion\"},{\"value\":\"4d019a031941\",\"key\":\"host.hostname\"},{\"value\":\"172.18.0.6\",\"key\":\"ip\"},{\"value\":\"4b19ace06df8e4de\",\"key\":\"client-uuid\"}]", root["serviceTags"])
+		require.Equal(t, 1616072924070.497, root["startTime"])
+		require.Equal(t, 8.421, root["duration"])
+		require.Equal(t, "", root["logs"])
+		require.Equal(t, "[{\"value\":\"const\",\"key\":\"sampler.type\"},{\"value\":true,\"key\":\"sampler.param\"},{\"value\":200,\"key\":\"http.status_code\"},{\"value\":\"GET\",\"key\":\"http.method\"},{\"value\":\"/loki/api/v1/query_range?direction=BACKWARD\\u0026limit=1000\\u0026query=%7Bcompose_project%3D%22devenv%22%7D%20%7C%3D%22traceID%22\\u0026start=1616070921000000000\\u0026end=1616072722000000000\\u0026step=2\",\"key\":\"http.url\"},{\"value\":\"net/http\",\"key\":\"component\"},{\"value\":\"server\",\"key\":\"span.kind\"},{\"value\":0,\"key\":\"status.code\"}]", root["tags"])
+
+		span := bFrame.FindRowWithValue("spanID", "8185345640360084998")
+
+		require.Equal(t, "GetParallelChunks", span["operationName"])
+		require.Equal(t, "loki-all", span["serviceName"])
+		require.Equal(t, "[{\"value\":\"loki-all\",\"key\":\"service.name\"},{\"value\":\"Jaeger-Go-2.25.0\",\"key\":\"opencensus.exporterversion\"},{\"value\":\"4d019a031941\",\"key\":\"host.hostname\"},{\"value\":\"172.18.0.6\",\"key\":\"ip\"},{\"value\":\"4b19ace06df8e4de\",\"key\":\"client-uuid\"}]", span["serviceTags"])
+		require.Equal(t, 1616072924072.852, span["startTime"])
+		require.Equal(t, 0.094, span["duration"])
+		require.Equal(t, "[{\"timestamp\":1616072924072.856,\"fields\":[{\"value\":1,\"key\":\"chunks requested\"}]},{\"timestamp\":1616072924072.9448,\"fields\":[{\"value\":1,\"key\":\"chunks fetched\"}]}]", span["logs"])
+		require.Equal(t, "[{\"value\":0,\"key\":\"status.code\"}]", span["tags"])
+	})
+}
+
+type Row map[string]interface{}
+type BetterFrame struct {
+	frame *data.Frame
+}
+
+func (f *BetterFrame) GetRow(index int) Row {
+	row := f.frame.RowCopy(index)
+	betterRow := make(map[string]interface{})
+	for i, field := range row {
+		betterRow[f.frame.Fields[i].Name] = field
+	}
+
+	return betterRow
+}
+
+func (f *BetterFrame) FindRow(fn func(row Row) bool) Row {
+	for i := 0; i < f.frame.Rows(); i++ {
+		row := f.GetRow(i)
+		if fn(row) {
+			return row
+		}
+	}
+
+	return nil
+}
+
+func (f *BetterFrame) FindRowWithValue(fieldName string, value interface{}) Row {
+	return f.FindRow(func(row Row) bool {
+		return row[fieldName] == value
+	})
+}
+
+func rootSpan(frame *BetterFrame) Row {
+	return frame.FindRowWithValue("parentSpanID", "")
+}
+
+func fieldNames(frame *data.Frame) []string {
+	var names []string
+	for _, f := range frame.Fields {
+		names = append(names, f.Name)
+	}
+	return names
+}
+
+var fields = []string{
+	"traceID",
+	"spanID",
+	"parentSpanID",
+	"operationName",
+	"serviceName",
+	"serviceTags",
+	"startTime",
+	"duration",
+	"logs",
+	"tags",
+}