Graphite: Support backend queries with different time ranges (#103984)

* Refactor to handle queries with various time ranges

* Lint

* Remove comment

pkg/tsdb/graphite/graphite.go

@@ -97,26 +97,29 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
 		return nil, err
 	}
 
-	// take the first query in the request list, since all query should share the same timerange
-	q := req.Queries[0]
+	emptyQueries := []string{}
+	graphiteQueries := map[string]struct {
+		req      *http.Request
+		formData url.Values
+	}{}
+	for _, query := range req.Queries {
+		graphiteReq, formData, emptyQuery, err := s.createGraphiteRequest(ctx, query, logger, dsInfo)
+		if err != nil {
+			return nil, err
+		}
 
-	/*
-		graphite doc about from and until, with sdk we are getting absolute instead of relative time
-		https://graphite-api.readthedocs.io/en/latest/api.html#from-until
-	*/
-	from, until := epochMStoGraphiteTime(q.TimeRange)
-	formData := url.Values{
-		"from":          []string{from},
-		"until":         []string{until},
-		"format":        []string{"json"},
-		"maxDataPoints": []string{fmt.Sprintf("%d", q.MaxDataPoints)},
-		"target":        []string{},
-	}
+		if emptyQuery != nil {
+			emptyQueries = append(emptyQueries, fmt.Sprintf("Query: %v has no target", emptyQuery))
+			continue
+		}
 
-	// Convert datasource query to graphite target request
-	targetList, emptyQueries, origRefIds, err := s.processQueries(logger, req.Queries)
-	if err != nil {
-		return nil, err
+		graphiteQueries[query.RefID] = struct {
+			req      *http.Request
+			formData url.Values
+		}{
+			req:      graphiteReq,
+			formData: formData,
+		}
 	}
 
 	var result = backend.QueryDataResponse{}
@@ -134,52 +137,47 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
 			return &result, nil
 		}
 	}
-	formData["target"] = targetList
 
-	if setting.Env == setting.Dev {
-		logger.Debug("Graphite request", "params", formData)
-	}
+	frames := data.Frames{}
 
-	graphiteReq, err := s.createRequest(ctx, logger, dsInfo, formData)
-	if err != nil {
-		return &result, err
-	}
-
-	ctx, span := s.tracer.Start(ctx, "graphite query")
-	defer span.End()
-
-	targetStr := strings.Join(formData["target"], ",")
-	span.SetAttributes(
-		attribute.String("target", targetStr),
-		attribute.String("from", from),
-		attribute.String("until", until),
-		attribute.Int64("datasource_id", dsInfo.Id),
-		attribute.Int64("org_id", req.PluginContext.OrgID),
-	)
-	s.tracer.Inject(ctx, graphiteReq.Header, span)
-
-	res, err := dsInfo.HTTPClient.Do(graphiteReq)
-	if res != nil {
-		span.SetAttributes(attribute.Int("graphite.response.code", res.StatusCode))
-	}
-	if err != nil {
-		span.RecordError(err)
-		span.SetStatus(codes.Error, err.Error())
-		return &result, err
-	}
-
-	defer func() {
-		err := res.Body.Close()
+	for refId, graphiteReq := range graphiteQueries {
+		ctx, span := s.tracer.Start(ctx, "graphite query")
+		defer span.End()
+		targetStr := strings.Join(graphiteReq.formData["target"], ",")
+		span.SetAttributes(
+			attribute.String("refId", refId),
+			attribute.String("target", targetStr),
+			attribute.String("from", graphiteReq.formData["from"][0]),
+			attribute.String("until", graphiteReq.formData["until"][0]),
+			attribute.Int64("datasource_id", dsInfo.Id),
+			attribute.Int64("org_id", req.PluginContext.OrgID),
+		)
+		s.tracer.Inject(ctx, graphiteReq.req.Header, span)
+		res, err := dsInfo.HTTPClient.Do(graphiteReq.req)
+		if res != nil {
+			span.SetAttributes(attribute.Int("graphite.response.code", res.StatusCode))
+		}
 		if err != nil {
-			logger.Warn("Failed to close response body", "error", err)
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+			return &result, err
 		}
-	}()
 
-	frames, err := s.toDataFrames(logger, res, origRefIds)
-	if err != nil {
-		span.RecordError(err)
-		span.SetStatus(codes.Error, err.Error())
-		return &result, err
+		defer func() {
+			err := res.Body.Close()
+			if err != nil {
+				logger.Warn("Failed to close response body", "error", err)
+			}
+		}()
+
+		queryFrames, err := s.toDataFrames(logger, res, refId)
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+			return &result, err
+		}
+
+		frames = append(frames, queryFrames...)
 	}
 
 	result = backend.QueryDataResponse{
@@ -200,46 +198,82 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
 	return &result, nil
 }
 
-// processQueries converts each datasource query to a graphite query target. It returns the list of
-// targets, a list of invalid queries, and a mapping of formatted refIds (used in the target query)
-// to original query refIds, later used to associate ressponses with the original queries
-func (s *Service) processQueries(logger log.Logger, queries []backend.DataQuery) ([]string, []string, map[string]string, error) {
-	emptyQueries := make([]string, 0)
-	origRefIds := make(map[string]string, 0)
-	targets := make([]string, 0)
+// processQuery converts a Graphite data source query to a Graphite query target. It returns the target,
+// and the model if the target is invalid
+func (s *Service) processQuery(logger log.Logger, query backend.DataQuery) (string, *simplejson.Json, error) {
+	model, err := simplejson.NewJson(query.JSON)
+	if err != nil {
+		return "", nil, err
+	}
+	logger.Debug("Graphite", "query", model)
+	currTarget := ""
+	if fullTarget, err := model.Get(TargetFullModelField).String(); err == nil {
+		currTarget = fullTarget
+	} else {
+		currTarget = model.Get(TargetModelField).MustString()
+	}
+	if currTarget == "" {
+		logger.Debug("Graphite", "empty query target", model)
+		return "", model, nil
+	}
+	target := fixIntervalFormat(currTarget)
 
-	for _, query := range queries {
-		model, err := simplejson.NewJson(query.JSON)
-		if err != nil {
-			return nil, nil, nil, err
-		}
-		logger.Debug("Graphite", "query", model)
-		currTarget := ""
-		if fullTarget, err := model.Get(TargetFullModelField).String(); err == nil {
-			currTarget = fullTarget
-		} else {
-			currTarget = model.Get(TargetModelField).MustString()
-		}
-		if currTarget == "" {
-			logger.Debug("Graphite", "empty query target", model)
-			emptyQueries = append(emptyQueries, fmt.Sprintf("Query: %v has no target", model))
-			continue
-		}
-		target := fixIntervalFormat(currTarget)
+	return target, nil, nil
+}
 
-		// This is a somewhat inglorious way to ensure we can associate results with the right query
-		// By using aliasSub, we can get back a resolved series Target name (accounting for other aliases)
-		// And the original refId. Since there are no restrictions on refId, we need to format it to make it
-		// easy to find in the response
-		formattedRefId := strings.ReplaceAll(query.RefID, " ", "_")
-		origRefIds[formattedRefId] = query.RefID
-		// This will set the alias to `<resolvedSeriesName> <formattedRefId>`
-		// e.g. aliasSub(alias(myquery, "foo"), "(^.*$)", "\1 A") will return "foo A"
-		target = fmt.Sprintf("aliasSub(%s,\"(^.*$)\",\"\\1 %s\")", target, formattedRefId)
-		targets = append(targets, target)
+func (s *Service) createRequest(ctx context.Context, l log.Logger, dsInfo *datasourceInfo, data url.Values) (*http.Request, error) {
+	u, err := url.Parse(dsInfo.URL)
+	if err != nil {
+		return nil, err
+	}
+	u.Path = path.Join(u.Path, "render")
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), strings.NewReader(data.Encode()))
+	if err != nil {
+		logger.Info("Failed to create request", "error", err)
+		return nil, fmt.Errorf("failed to create request: %w", err)
 	}
 
-	return targets, emptyQueries, origRefIds, nil
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	return req, err
+}
+
+func (s *Service) createGraphiteRequest(ctx context.Context, query backend.DataQuery, logger log.Logger, dsInfo *datasourceInfo) (*http.Request, url.Values, *simplejson.Json, error) {
+	/*
+		graphite doc about from and until, with sdk we are getting absolute instead of relative time
+		https://graphite-api.readthedocs.io/en/latest/api.html#from-until
+	*/
+	from, until := epochMStoGraphiteTime(query.TimeRange)
+	formData := url.Values{
+		"from":          []string{from},
+		"until":         []string{until},
+		"format":        []string{"json"},
+		"maxDataPoints": []string{fmt.Sprintf("%d", query.MaxDataPoints)},
+		"target":        []string{},
+	}
+
+	target, emptyQuery, err := s.processQuery(logger, query)
+	if err != nil {
+		return nil, formData, nil, err
+	}
+
+	if emptyQuery != nil {
+		logger.Debug("Graphite", "empty query target", emptyQuery)
+		return nil, formData, emptyQuery, nil
+	}
+
+	formData["target"] = []string{target}
+
+	if setting.Env == setting.Dev {
+		logger.Debug("Graphite request", "params", formData)
+	}
+
+	graphiteReq, err := s.createRequest(ctx, logger, dsInfo, formData)
+	if err != nil {
+		return nil, formData, nil, err
+	}
+
+	return graphiteReq, formData, emptyQuery, nil
 }
 
 func (s *Service) parseResponse(logger log.Logger, res *http.Response) ([]TargetResponseDTO, error) {
@@ -268,7 +302,7 @@ func (s *Service) parseResponse(logger log.Logger, res *http.Response) ([]Target
 	return data, nil
 }
 
-func (s *Service) toDataFrames(logger log.Logger, response *http.Response, origRefIds map[string]string) (frames data.Frames, error error) {
+func (s *Service) toDataFrames(logger log.Logger, response *http.Response, refId string) (frames data.Frames, error error) {
 	responseData, err := s.parseResponse(logger, response)
 	if err != nil {
 		return nil, err
@@ -278,18 +312,6 @@ func (s *Service) toDataFrames(logger log.Logger, response *http.Response, origR
 	for _, series := range responseData {
 		timeVector := make([]time.Time, 0, len(series.DataPoints))
 		values := make([]*float64, 0, len(series.DataPoints))
-		// series.Target will be in the format <resolvedSeriesName> <formattedRefId>
-		ls := strings.LastIndex(series.Target, " ")
-		if ls == -1 {
-			return nil, fmt.Errorf("received graphite response with invalid target format: %s", series.Target)
-		}
-		target := series.Target[:ls]
-		formattedRefId := series.Target[ls+1:]
-		refId, ok := origRefIds[formattedRefId]
-		if !ok {
-			logger.Warn("Unable to find refId associated with provided formattedRefId", "formattedRefId", formattedRefId)
-			refId = formattedRefId // fallback - shouldn't happen except for in tests
-		}
 
 		for _, dataPoint := range series.DataPoints {
 			var timestamp, value, err = parseDataTimePoint(dataPoint)
@@ -303,7 +325,7 @@ func (s *Service) toDataFrames(logger log.Logger, response *http.Response, origR
 		tags := make(map[string]string)
 		for name, value := range series.Tags {
 			if name == "name" {
-				value = target
+				value = series.Target
 			}
 			switch value := value.(type) {
 			case string:
@@ -315,7 +337,7 @@ func (s *Service) toDataFrames(logger log.Logger, response *http.Response, origR
 
 		frames = append(frames, data.NewFrame(refId,
 			data.NewField("time", nil, timeVector),
-			data.NewField("value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: target})).SetMeta(
+			data.NewField("value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: series.Target})).SetMeta(
 			&data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}))
 
 		if setting.Env == setting.Dev {
@@ -325,23 +347,6 @@ func (s *Service) toDataFrames(logger log.Logger, response *http.Response, origR
 	return frames, nil
 }
 
-func (s *Service) createRequest(ctx context.Context, l log.Logger, dsInfo *datasourceInfo, data url.Values) (*http.Request, error) {
-	u, err := url.Parse(dsInfo.URL)
-	if err != nil {
-		return nil, err
-	}
-	u.Path = path.Join(u.Path, "render")
-
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), strings.NewReader(data.Encode()))
-	if err != nil {
-		logger.Info("Failed to create request", "error", err)
-		return nil, fmt.Errorf("failed to create request: %w", err)
-	}
-
-	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-	return req, err
-}
-
 func fixIntervalFormat(target string) string {
 	rMinute := regexp.MustCompile(`'(\d+)m'`)
 	target = rMinute.ReplaceAllStringFunc(target, func(m string) string {

pkg/tsdb/graphite/graphite_test.go

@@ -3,7 +3,6 @@ package graphite
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -58,7 +57,7 @@ func TestFixIntervalFormat(t *testing.T) {
 	}
 }
 
-func TestProcessQueries(t *testing.T) {
+func TestProcessQuery(t *testing.T) {
 	service := &Service{}
 	log := logger.FromContext(context.Background())
 	t.Run("Parses single valid query", func(t *testing.T) {
@@ -70,61 +69,26 @@ func TestProcessQueries(t *testing.T) {
 				}`),
 			},
 		}
-		targets, invalids, mapping, err := service.processQueries(log, queries)
+		target, jsonModel, err := service.processQuery(log, queries[0])
 		assert.NoError(t, err)
-		assert.Empty(t, invalids)
-		assert.Len(t, mapping, 1)
-		assert.Len(t, targets, 1)
-		assert.Equal(t, "aliasSub(app.grafana.*.dashboards.views.1M.count,\"(^.*$)\",\"\\1 A\")", targets[0])
+		assert.Nil(t, jsonModel)
+		assert.Equal(t, "app.grafana.*.dashboards.views.1M.count", target)
 	})
 
-	t.Run("Parses multiple valid queries with refId mappings", func(t *testing.T) {
+	t.Run("Returns if target is empty", func(t *testing.T) {
 		queries := []backend.DataQuery{
 			{
 				RefID: "A",
 				JSON: []byte(`{
-					"target": "app.grafana.*.dashboards.views.1M.count"
-				}`),
-			},
-			{
-				RefID: "query B",
-				JSON: []byte(`{
-					"target": "aliasByNode(hitcount(averageSeries(app.grafana.*.dashboards.views.count), '1mon'), 4)"
+					"target": ""
 				}`),
 			},
 		}
-		targets, invalids, mapping, err := service.processQueries(log, queries)
+		jsonEqual, _ := simplejson.NewJson([]byte(`{"target": ""}`))
+		target, jsonModel, err := service.processQuery(log, queries[0])
 		assert.NoError(t, err)
-		assert.Empty(t, invalids)
-		assert.Len(t, mapping, 2)
-		assert.Len(t, targets, 2)
-		assert.Equal(t, "aliasSub(app.grafana.*.dashboards.views.1M.count,\"(^.*$)\",\"\\1 A\")", targets[0])
-		assert.Equal(t, "aliasSub(aliasByNode(hitcount(averageSeries(app.grafana.*.dashboards.views.count), '1mon'), 4),\"(^.*$)\",\"\\1 query_B\")", targets[1])
-	})
-
-	t.Run("Parses multiple queries with one invalid", func(t *testing.T) {
-		queries := []backend.DataQuery{
-			{
-				RefID: "A",
-				JSON: []byte(`{
-					"target": "app.grafana.*.dashboards.views.1M.count"
-				}`),
-			},
-			{
-				RefID: "B",
-				JSON: []byte(`{
-					"query": "app.grafana.*.dashboards.views.1M.count"
-				}`),
-			},
-		}
-		targets, invalids, mapping, err := service.processQueries(log, queries)
-		assert.NoError(t, err)
-		assert.Len(t, invalids, 1)
-		assert.Len(t, mapping, 1)
-		assert.Len(t, targets, 1)
-		json, _ := simplejson.NewJson(queries[1].JSON)
-		expectedInvalid := fmt.Sprintf("Query: %v has no target", json)
-		assert.Equal(t, expectedInvalid, invalids[0])
+		assert.Equal(t, jsonEqual, jsonModel)
+		assert.Equal(t, "", target)
 	})
 
 	t.Run("QueryData with no valid queries returns bad request response", func(t *testing.T) {
@@ -217,12 +181,13 @@ func TestConvertResponses(t *testing.T) {
 		body := `
 		[
 			{
-				"target": "target A",
+				"target": "target",
 				"datapoints": [[50, 1], [null, 2], [100, 3]]
 			}
 		]`
 		a := 50.0
 		b := 100.0
+		refId := "A"
 		expectedFrame := data.NewFrame("A",
 			data.NewField("time", nil, []time.Time{time.Unix(1, 0).UTC(), time.Unix(2, 0).UTC(), time.Unix(3, 0).UTC()}),
 			data.NewField("value", data.Labels{}, []*float64{&a, nil, &b}).SetConfig(&data.FieldConfig{DisplayNameFromDS: "target"}),
@@ -230,7 +195,7 @@ func TestConvertResponses(t *testing.T) {
 		expectedFrames := data.Frames{expectedFrame}
 
 		httpResponse := &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body))}
-		dataFrames, err := service.toDataFrames(logger, httpResponse, map[string]string{})
+		dataFrames, err := service.toDataFrames(logger, httpResponse, refId)
 
 		require.NoError(t, err)
 		if !reflect.DeepEqual(expectedFrames, dataFrames) {
@@ -244,13 +209,14 @@ func TestConvertResponses(t *testing.T) {
 		body := `
 		[
 			{
-				"target": "target A",
+				"target": "target",
 				"tags": { "fooTag": "fooValue", "barTag": "barValue", "int": 100, "float": 3.14 },
 				"datapoints": [[50, 1], [null, 2], [100, 3]]
 			}
 		]`
 		a := 50.0
 		b := 100.0
+		refId := "A"
 		expectedFrame := data.NewFrame("A",
 			data.NewField("time", nil, []time.Time{time.Unix(1, 0).UTC(), time.Unix(2, 0).UTC(), time.Unix(3, 0).UTC()}),
 			data.NewField("value", data.Labels{
@@ -263,7 +229,7 @@ func TestConvertResponses(t *testing.T) {
 		expectedFrames := data.Frames{expectedFrame}
 
 		httpResponse := &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body))}
-		dataFrames, err := service.toDataFrames(logger, httpResponse, map[string]string{})
+		dataFrames, err := service.toDataFrames(logger, httpResponse, refId)
 
 		require.NoError(t, err)
 		if !reflect.DeepEqual(expectedFrames, dataFrames) {
@@ -272,79 +238,4 @@ func TestConvertResponses(t *testing.T) {
 			t.Errorf("Data frames should have been equal but was, expected:\n%s\nactual:\n%s", expectedFramesJSON, dataFramesJSON)
 		}
 	})
-
-	t.Run("Converts response with multiple targets", func(*testing.T) {
-		body := `
-		[
-			{
-				"target": "target 1 A",
-				"datapoints": [[50, 1], [null, 2], [100, 3]]
-			},
-			{
-				"target": "target 2 B",
-				"datapoints": [[50, 1], [null, 2], [100, 3]]
-			}
-		]`
-		a := 50.0
-		b := 100.0
-		expectedFrameA := data.NewFrame("A",
-			data.NewField("time", nil, []time.Time{time.Unix(1, 0).UTC(), time.Unix(2, 0).UTC(), time.Unix(3, 0).UTC()}),
-			data.NewField("value", data.Labels{}, []*float64{&a, nil, &b}).SetConfig(&data.FieldConfig{DisplayNameFromDS: "target 1"}),
-		).SetMeta(&data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti})
-		expectedFrameB := data.NewFrame("B",
-			data.NewField("time", nil, []time.Time{time.Unix(1, 0).UTC(), time.Unix(2, 0).UTC(), time.Unix(3, 0).UTC()}),
-			data.NewField("value", data.Labels{}, []*float64{&a, nil, &b}).SetConfig(&data.FieldConfig{DisplayNameFromDS: "target 2"}),
-		).SetMeta(&data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti})
-		expectedFrames := data.Frames{expectedFrameA, expectedFrameB}
-
-		httpResponse := &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body))}
-		dataFrames, err := service.toDataFrames(logger, httpResponse, map[string]string{})
-
-		require.NoError(t, err)
-		if !reflect.DeepEqual(expectedFrames, dataFrames) {
-			expectedFramesJSON, _ := json.Marshal(expectedFrames)
-			dataFramesJSON, _ := json.Marshal(dataFrames)
-			t.Errorf("Data frames should have been equal but was, expected:\n%s\nactual:\n%s", expectedFramesJSON, dataFramesJSON)
-		}
-	})
-
-	t.Run("Converts response with refId mapping", func(*testing.T) {
-		body := `
-		[
-			{
-				"target": "target A_A",
-				"datapoints": [[50, 1], [null, 2], [100, 3]]
-			}
-		]`
-		a := 50.0
-		b := 100.0
-		expectedFrame := data.NewFrame("A A",
-			data.NewField("time", nil, []time.Time{time.Unix(1, 0).UTC(), time.Unix(2, 0).UTC(), time.Unix(3, 0).UTC()}),
-			data.NewField("value", data.Labels{}, []*float64{&a, nil, &b}).SetConfig(&data.FieldConfig{DisplayNameFromDS: "target"}),
-		).SetMeta(&data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti})
-		expectedFrames := data.Frames{expectedFrame}
-
-		httpResponse := &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body))}
-		dataFrames, err := service.toDataFrames(logger, httpResponse, map[string]string{"A_A": "A A"})
-
-		require.NoError(t, err)
-		if !reflect.DeepEqual(expectedFrames, dataFrames) {
-			expectedFramesJSON, _ := json.Marshal(expectedFrames)
-			dataFramesJSON, _ := json.Marshal(dataFrames)
-			t.Errorf("Data frames should have been equal but was, expected:\n%s\nactual:\n%s", expectedFramesJSON, dataFramesJSON)
-		}
-	})
-
-	t.Run("Chokes on response with invalid target name", func(*testing.T) {
-		body := `
-		[
-			{
-				"target": "target",
-				"datapoints": [[50, 1], [null, 2], [100, 3]]
-			}
-		]`
-		httpResponse := &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body))}
-		_, err := service.toDataFrames(logger, httpResponse, map[string]string{})
-		require.Error(t, err)
-	})
 }