diff --git a/pkg/tsdb/graphite/graphite.go b/pkg/tsdb/graphite/graphite.go index 6483689b8cb..f2b2b4d4097 100644 --- a/pkg/tsdb/graphite/graphite.go +++ b/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 ` ` - // 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 - 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 { diff --git a/pkg/tsdb/graphite/graphite_test.go b/pkg/tsdb/graphite/graphite_test.go index 5f9a97d815f..3a76af65acf 100644 --- a/pkg/tsdb/graphite/graphite_test.go +++ b/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) - }) }