SSE: DSNode to update result with names to make each value identifiable by labels (only Graphite and TestData) (#71246)

* introduce a function checkIfSeriesNeedToBeFixed to scan all value fields in the response and provide a function that updates Series so they can be uniquely identifiable. Only Graphite and TestData are checked.

* update `convertDataFramesToResults` to run this function and provide it to WideToMany
* update WideToMany to run the fix function if it is not nil

pkg/expr/nodes.go

@@ -20,6 +20,9 @@ import (
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 )
 
+// label that is used when all mathexp.Series have 0 labels to make them identifiable by labels. The value of this label is extracted from value field names
+const nameLabelName = "__name__"
+
 var (
 	logger = log.New("expr")
 )
@@ -295,7 +298,6 @@ func convertDataFramesToResults(ctx context.Context, frames data.Frames, datasou
 		return "no-data", mathexp.Results{Values: mathexp.Values{mathexp.NewNoData()}}, nil
 	}
 
-	vals := make([]mathexp.Value, 0)
 	var dt data.FrameType
 	dt, useDataplane, _ := shouldUseDataplane(frames, logger, s.features.IsEnabled(featuremgmt.FlagDisableSSEDataplane))
 	if useDataplane {
@@ -325,6 +327,7 @@ func convertDataFramesToResults(ctx context.Context, frames data.Frames, datasou
 			if err != nil {
 				return "", mathexp.Results{}, err
 			}
+			vals := make([]mathexp.Value, 0, len(numberSet))
 			for _, n := range numberSet {
 				vals = append(vals, n)
 			}
@@ -334,16 +337,33 @@ func convertDataFramesToResults(ctx context.Context, frames data.Frames, datasou
 		}
 	}
 
+	filtered := make([]*data.Frame, 0, len(frames))
+	totalLen := 0
 	for _, frame := range frames {
+		schema := frame.TimeSeriesSchema()
 		// Check for TimeSeriesTypeNot in InfluxDB queries. A data frame of this type will cause
 		// the WideToMany() function to error out, which results in unhealthy alerts.
 		// This check should be removed once inconsistencies in data source responses are solved.
-		if frame.TimeSeriesSchema().Type == data.TimeSeriesTypeNot && datasourceType == datasources.DS_INFLUXDB {
+		if schema.Type == data.TimeSeriesTypeNot && datasourceType == datasources.DS_INFLUXDB {
 			logger.Warn("Ignoring InfluxDB data frame due to missing numeric fields")
 			continue
 		}
-		var series []mathexp.Series
+		if schema.Type != data.TimeSeriesTypeWide {
-		series, err := WideToMany(frame)
+			return "", mathexp.Results{}, fmt.Errorf("input data must be a wide series but got type %s (input refid)", schema.Type)
+		}
+		filtered = append(filtered, frame)
+		totalLen += len(schema.ValueIndices)
+	}
+
+	if len(filtered) == 0 {
+		return "no data", mathexp.Results{Values: mathexp.Values{mathexp.NoData{Frame: frames[0]}}}, nil
+	}
+
+	maybeFixerFn := checkIfSeriesNeedToBeFixed(filtered, datasourceType)
+
+	vals := make([]mathexp.Value, 0, totalLen)
+	for _, frame := range filtered {
+		series, err := WideToMany(frame, maybeFixerFn)
 		if err != nil {
 			return "", mathexp.Results{}, err
 		}
@@ -351,14 +371,17 @@ func convertDataFramesToResults(ctx context.Context, frames data.Frames, datasou
 			vals = append(vals, ser)
 		}
 	}
-
+	dataType := "single frame series"
-	return "series set", mathexp.Results{
+	if len(filtered) > 1 {
-		Values: vals, // TODO vals can be empty. Should we replace with no-data?
+		dataType = "multi frame series"
+	}
+	return dataType, mathexp.Results{
+		Values: vals,
 	}, nil
 }
 
 func isAllFrameVectors(datasourceType string, frames data.Frames) bool {
-	if datasourceType != "prometheus" {
+	if datasourceType != datasources.DS_PROMETHEUS {
 		return false
 	}
 	allVector := false
@@ -466,7 +489,7 @@ func extractNumberSet(frame *data.Frame) ([]mathexp.Number, error) {
 // is created for each value type column of wide frame.
 //
 // This might not be a good idea long term, but works now as an adapter/shim.
-func WideToMany(frame *data.Frame) ([]mathexp.Series, error) {
+func WideToMany(frame *data.Frame, fixSeries func(series mathexp.Series, valueField *data.Field)) ([]mathexp.Series, error) {
 	tsSchema := frame.TimeSeriesSchema()
 	if tsSchema.Type != data.TimeSeriesTypeWide {
 		return nil, fmt.Errorf("input data must be a wide series but got type %s (input refid)", tsSchema.Type)
@@ -477,10 +500,13 @@ func WideToMany(frame *data.Frame) ([]mathexp.Series, error) {
 		if err != nil {
 			return nil, err
 		}
+		if fixSeries != nil {
+			fixSeries(s, frame.Fields[tsSchema.ValueIndices[0]])
+		}
 		return []mathexp.Series{s}, nil
 	}
 
-	series := []mathexp.Series{}
+	series := make([]mathexp.Series, 0, len(tsSchema.ValueIndices))
 	for _, valIdx := range tsSchema.ValueIndices {
 		l := frame.Rows()
 		f := data.NewFrameOfFieldTypes(frame.Name, l, frame.Fields[tsSchema.TimeIndex].Type(), frame.Fields[valIdx].Type())
@@ -500,8 +526,79 @@ func WideToMany(frame *data.Frame) ([]mathexp.Series, error) {
 		if err != nil {
 			return nil, err
 		}
+		if fixSeries != nil {
+			fixSeries(s, frame.Fields[valIdx])
+		}
 		series = append(series, s)
 	}
 
 	return series, nil
 }
+
+// checkIfSeriesNeedToBeFixed scans all value fields of all provided frames and determines whether the resulting mathexp.Series
+// needs to be updated so each series could be identifiable by labels.
+// NOTE: applicable only to only datasources.DS_GRAPHITE and datasources.DS_TESTDATA data sources
+// returns a function that patches the mathexp.Series with information from data.Field from which it was created if the all series need to be fixed. Otherwise, returns nil
+func checkIfSeriesNeedToBeFixed(frames []*data.Frame, datasourceType string) func(series mathexp.Series, valueField *data.Field) {
+	if !(datasourceType == datasources.DS_GRAPHITE || datasourceType == datasources.DS_TESTDATA) {
+		return nil
+	}
+
+	// get all value fields
+	var valueFields []*data.Field
+	for _, frame := range frames {
+		tsSchema := frame.TimeSeriesSchema()
+		for _, index := range tsSchema.ValueIndices {
+			field := frame.Fields[index]
+			// if at least one value field contains labels, the result does not need to be fixed.
+			if len(field.Labels) > 0 {
+				return nil
+			}
+			if valueFields == nil {
+				valueFields = make([]*data.Field, 0, len(frames)*len(tsSchema.ValueIndices))
+			}
+			valueFields = append(valueFields, field)
+		}
+	}
+
+	// selectors are in precedence order.
+	nameSelectors := []func(f *data.Field) string{
+		func(f *data.Field) string {
+			if f == nil || f.Config == nil {
+				return ""
+			}
+			return f.Config.DisplayNameFromDS
+		},
+		func(f *data.Field) string {
+			if f == nil || f.Config == nil {
+				return ""
+			}
+			return f.Config.DisplayName
+		},
+		func(f *data.Field) string {
+			return f.Name
+		},
+	}
+
+	// now look for the first selector that would make all value fields be unique
+	for _, selector := range nameSelectors {
+		names := make(map[string]struct{}, len(valueFields))
+		good := true
+		for _, field := range valueFields {
+			name := selector(field)
+			if _, ok := names[name]; ok || name == "" {
+				good = false
+				break
+			}
+			names[name] = struct{}{}
+		}
+		if good {
+			return func(series mathexp.Series, valueField *data.Field) {
+				series.SetLabels(data.Labels{
+					nameLabelName: selector(valueField),
+				})
+			}
+		}
+	}
+	return nil
+}

pkg/expr/nodes_test.go

@@ -1,10 +1,22 @@
 package expr
 
 import (
+	"context"
 	"errors"
+	"fmt"
 	"testing"
+	"time"
 
+	"github.com/grafana/grafana-plugin-sdk-go/data"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/grafana/pkg/expr/mathexp"
+	"github.com/grafana/grafana/pkg/infra/log/logtest"
+	"github.com/grafana/grafana/pkg/infra/tracing"
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
+	"github.com/grafana/grafana/pkg/setting"
 )
 
 type expectedError struct{}
@@ -40,3 +52,232 @@ func TestQueryError_Unwrap(t *testing.T) {
 		assert.True(t, errors.As(e, &expectedAsError))
 	})
 }
+
+func TestCheckIfSeriesNeedToBeFixed(t *testing.T) {
+	createFrame := func(m ...func(field *data.Field)) []*data.Frame {
+		f := data.NewFrame("",
+			data.NewField("Time", nil, []time.Time{}))
+		for i := 0; i < 100; i++ {
+			fld := data.NewField(fmt.Sprintf("fld-%d", i), nil, []*float64{})
+			fld.Config = &data.FieldConfig{}
+			for _, change := range m {
+				change(fld)
+			}
+			f.Fields = append(f.Fields, fld)
+		}
+		return []*data.Frame{f}
+	}
+	withLabels := func(field *data.Field) {
+		field.Labels = map[string]string{
+			"field": field.Name,
+		}
+	}
+	withDisplayNameFromDS := func(field *data.Field) {
+		field.Config.DisplayNameFromDS = fmt.Sprintf("dnds-%s", field.Name)
+	}
+	withDisplayName := func(field *data.Field) {
+		field.Config.DisplayName = fmt.Sprintf("dn-%s", field.Name)
+	}
+	withoutName := func(field *data.Field) {
+		field.Name = ""
+	}
+
+	getLabelName := func(f func(series mathexp.Series, valueField *data.Field)) string {
+		s := mathexp.NewSeries("A", nil, 0)
+		field := &data.Field{
+			Name: "Name",
+			Config: &data.FieldConfig{
+				DisplayNameFromDS: "DisplayNameFromDS",
+				DisplayName:       "DisplayName",
+			},
+		}
+		f(s, field)
+		return s.GetLabels()[nameLabelName]
+	}
+
+	testCases := []struct {
+		name         string
+		frames       []*data.Frame
+		expectedName string
+	}{
+		{
+			name:         "should return nil if at least one value field has labels",
+			frames:       createFrame(withLabels, withDisplayNameFromDS, withDisplayName),
+			expectedName: "",
+		},
+		{
+			name:         "should return nil if names are empty",
+			frames:       createFrame(withoutName),
+			expectedName: "",
+		},
+		{
+			name:         "should return patcher with DisplayNameFromDS first",
+			frames:       createFrame(withDisplayNameFromDS, withDisplayName),
+			expectedName: "DisplayNameFromDS",
+		},
+		{
+			name: "should return patcher with DisplayName if DisplayNameFromDS is not unique",
+			frames: func() []*data.Frame {
+				frames := createFrame(withDisplayNameFromDS, withDisplayName)
+				f := frames[0]
+				f.Fields[2].Config.DisplayNameFromDS = "test"
+				f.Fields[3].Config.DisplayNameFromDS = "test"
+				return frames
+			}(),
+			expectedName: "DisplayName",
+		},
+		{
+			name:         "should return patcher with DisplayName if is empty",
+			frames:       createFrame(withDisplayName),
+			expectedName: "DisplayName",
+		},
+		{
+			name: "should return patcher with Name if DisplayName and DisplayNameFromDS are not unique",
+			frames: func() []*data.Frame {
+				frames := createFrame(withDisplayNameFromDS, withDisplayName)
+				f := frames[0]
+				f.Fields[1].Config.DisplayNameFromDS = f.Fields[2].Config.DisplayNameFromDS
+				f.Fields[1].Config.DisplayName = f.Fields[2].Config.DisplayName
+				return frames
+			}(),
+			expectedName: "Name",
+		},
+		{
+			name:         "should return patcher with Name if DisplayName and DisplayNameFromDS are empty",
+			frames:       createFrame(),
+			expectedName: "Name",
+		},
+		{
+			name: "should return nil if all fields are not unique",
+			frames: func() []*data.Frame {
+				frames := createFrame(withDisplayNameFromDS, withDisplayName)
+				f := frames[0]
+				f.Fields[1].Config.DisplayNameFromDS = f.Fields[2].Config.DisplayNameFromDS
+				f.Fields[1].Config.DisplayName = f.Fields[2].Config.DisplayName
+				f.Fields[1].Name = f.Fields[2].Name
+				return frames
+			}(),
+			expectedName: "",
+		},
+	}
+
+	supportedDatasources := []string{
+		datasources.DS_GRAPHITE,
+		datasources.DS_TESTDATA,
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			for _, datasource := range supportedDatasources {
+				fixer := checkIfSeriesNeedToBeFixed(tc.frames, datasource)
+				if tc.expectedName == "" {
+					require.Nil(t, fixer)
+				} else {
+					require.Equal(t, tc.expectedName, getLabelName(fixer))
+				}
+			}
+		})
+	}
+}
+
+func TestConvertDataFramesToResults(t *testing.T) {
+	s := &Service{
+		cfg:      setting.NewCfg(),
+		features: &featuremgmt.FeatureManager{},
+		tracer:   tracing.InitializeTracerForTest(),
+		metrics:  newMetrics(nil),
+	}
+
+	t.Run("should add name label if no labels and specific data source", func(t *testing.T) {
+		supported := []string{datasources.DS_GRAPHITE, datasources.DS_TESTDATA}
+		t.Run("when only field name is specified", func(t *testing.T) {
+			t.Run("use value field names if one frame - many series", func(t *testing.T) {
+				supported := []string{datasources.DS_GRAPHITE, datasources.DS_TESTDATA}
+
+				frames := []*data.Frame{
+					data.NewFrame("test",
+						data.NewField("time", nil, []time.Time{time.Unix(1, 0)}),
+						data.NewField("test-value1", nil, []*float64{fp(2)}),
+						data.NewField("test-value2", nil, []*float64{fp(2)})),
+				}
+
+				for _, dtype := range supported {
+					t.Run(dtype, func(t *testing.T) {
+						resultType, res, err := convertDataFramesToResults(context.Background(), frames, dtype, s, &logtest.Fake{})
+						require.NoError(t, err)
+						assert.Equal(t, "single frame series", resultType)
+						require.Len(t, res.Values, 2)
+
+						var names []string
+						for _, value := range res.Values {
+							require.IsType(t, mathexp.Series{}, value)
+							lbls := value.GetLabels()
+							require.Contains(t, lbls, nameLabelName)
+							names = append(names, lbls[nameLabelName])
+						}
+						require.EqualValues(t, []string{"test-value1", "test-value2"}, names)
+					})
+				}
+			})
+			t.Run("should use frame name if one frame - one series", func(t *testing.T) {
+				frames := []*data.Frame{
+					data.NewFrame("test-frame1",
+						data.NewField("time", nil, []time.Time{time.Unix(1, 0)}),
+						data.NewField("test-value1", nil, []*float64{fp(2)})),
+					data.NewFrame("test-frame2",
+						data.NewField("time", nil, []time.Time{time.Unix(1, 0)}),
+						data.NewField("test-value2", nil, []*float64{fp(2)})),
+				}
+
+				for _, dtype := range supported {
+					t.Run(dtype, func(t *testing.T) {
+						resultType, res, err := convertDataFramesToResults(context.Background(), frames, dtype, s, &logtest.Fake{})
+						require.NoError(t, err)
+						assert.Equal(t, "multi frame series", resultType)
+						require.Len(t, res.Values, 2)
+
+						var names []string
+						for _, value := range res.Values {
+							require.IsType(t, mathexp.Series{}, value)
+							lbls := value.GetLabels()
+							require.Contains(t, lbls, nameLabelName)
+							names = append(names, lbls[nameLabelName])
+						}
+						require.EqualValues(t, []string{"test-frame1", "test-frame2"}, names)
+					})
+				}
+			})
+		})
+		t.Run("should use fields DisplayNameFromDS when it is unique", func(t *testing.T) {
+			f1 := data.NewField("test-value1", nil, []*float64{fp(2)})
+			f1.Config = &data.FieldConfig{DisplayNameFromDS: "test-value1"}
+			f2 := data.NewField("test-value2", nil, []*float64{fp(2)})
+			f2.Config = &data.FieldConfig{DisplayNameFromDS: "test-value2"}
+			frames := []*data.Frame{
+				data.NewFrame("test-frame1",
+					data.NewField("time", nil, []time.Time{time.Unix(1, 0)}),
+					f1),
+				data.NewFrame("test-frame2",
+					data.NewField("time", nil, []time.Time{time.Unix(1, 0)}),
+					f2),
+			}
+
+			for _, dtype := range supported {
+				t.Run(dtype, func(t *testing.T) {
+					resultType, res, err := convertDataFramesToResults(context.Background(), frames, dtype, s, &logtest.Fake{})
+					require.NoError(t, err)
+					assert.Equal(t, "multi frame series", resultType)
+					require.Len(t, res.Values, 2)
+
+					var names []string
+					for _, value := range res.Values {
+						require.IsType(t, mathexp.Series{}, value)
+						lbls := value.GetLabels()
+						require.Contains(t, lbls, nameLabelName)
+						names = append(names, lbls[nameLabelName])
+					}
+					require.EqualValues(t, []string{"test-value1", "test-value2"}, names)
+				})
+			}
+		})
+	})
+}

pkg/expr/service_test.go

@@ -26,7 +26,7 @@ import (
 func TestService(t *testing.T) {
 	dsDF := data.NewFrame("test",
 		data.NewField("time", nil, []time.Time{time.Unix(1, 0)}),
-		data.NewField("value", nil, []*float64{fp(2)}))
+		data.NewField("value", data.Labels{"test": "label"}, []*float64{fp(2)}))
 
 	me := &mockEndpoint{
 		Frames: []*data.Frame{dsDF},
@@ -78,7 +78,7 @@ func TestService(t *testing.T) {
 
 	bDF := data.NewFrame("",
 		data.NewField("Time", nil, []time.Time{time.Unix(1, 0)}),
-		data.NewField("B", nil, []*float64{fp(4)}))
+		data.NewField("B", data.Labels{"test": "label"}, []*float64{fp(4)}))
 	bDF.RefID = "B"
 	bDF.SetMeta(&data.FrameMeta{
 		Type:        data.FrameTypeTimeSeriesMulti,

pkg/services/datasources/models.go

@@ -28,6 +28,7 @@ const (
 	DS_ES_OPEN_DISTRO = "grafana-es-open-distro-datasource"
 	DS_ES_OPENSEARCH  = "grafana-opensearch-datasource"
 	DS_AZURE_MONITOR  = "grafana-azure-monitor-datasource"
+	DS_TESTDATA       = "testdata"
 	// CustomHeaderName is the prefix that is used to store the name of a custom header.
 	CustomHeaderName = "httpHeaderName"
 	// CustomHeaderValue is the prefix that is used to store the value of a custom header.

pkg/services/ngalert/backtesting/eval_data.go

@@ -21,7 +21,7 @@ type dataEvaluator struct {
 }
 
 func newDataEvaluator(refID string, frame *data.Frame) (*dataEvaluator, error) {
-	series, err := expr.WideToMany(frame)
+	series, err := expr.WideToMany(frame, nil)
 	if err != nil {
 		return nil, err
 	}