From 3dc7cfdc18114f653587b7a24b4eeb490a4eb83f Mon Sep 17 00:00:00 2001 From: Ivana Huckova <30407135+ivanahuckova@users.noreply.github.com> Date: Mon, 8 Jan 2024 18:32:16 +0100 Subject: [PATCH] Loki: Implement error source (#80143) --- pkg/tsdb/loki/api.go | 21 +++++++++--- pkg/tsdb/loki/framing_test.go | 60 +++++++++++++++++++++++++++++------ pkg/tsdb/loki/loki.go | 19 +++++------ 3 files changed, 75 insertions(+), 25 deletions(-) diff --git a/pkg/tsdb/loki/api.go b/pkg/tsdb/loki/api.go index 51b43943ce7..0f57d65c109 100644 --- a/pkg/tsdb/loki/api.go +++ b/pkg/tsdb/loki/api.go @@ -11,14 +11,15 @@ import ( "net/url" "path" "strconv" + "syscall" "time" - "github.com/grafana/grafana-plugin-sdk-go/data" jsoniter "github.com/json-iterator/go" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace" + "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/tsdb/loki/instrumentation" @@ -160,7 +161,7 @@ func readLokiError(body io.ReadCloser) error { return makeLokiError(bytes) } -func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery, responseOpts ResponseOpts) (data.Frames, error) { +func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery, responseOpts ResponseOpts) (*backend.DataResponse, error) { req, err := makeDataRequest(ctx, api.url, query, api.requestStructuredMetadata) if err != nil { return nil, err @@ -181,7 +182,13 @@ func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery, responseOpts lp = append(lp, "statusCode", resp.StatusCode) } api.log.Error("Error received from Loki", lp...) - return nil, err + res := backend.DataResponse{ + Error: err, + } + if errors.Is(err, syscall.ECONNREFUSED) { + res.ErrorSource = backend.ErrorSourceDownstream + } + return &res, nil } defer func() { @@ -194,9 +201,13 @@ func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery, responseOpts lp = append(lp, queryAttrs...) if resp.StatusCode/100 != 2 { err := readLokiError(resp.Body) + res := backend.DataResponse{ + Error: err, + ErrorSource: backend.ErrorSourceFromHTTPStatus(resp.StatusCode), + } lp = append(lp, "status", "error", "error", err) api.log.Error("Error received from Loki", lp...) - return nil, err + return &res, nil } else { lp = append(lp, "status", "ok") api.log.Info("Response received from loki", lp...) @@ -221,7 +232,7 @@ func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery, responseOpts instrumentation.UpdatePluginParsingResponseDurationSeconds(ctx, time.Since(start), "ok") api.log.Info("Response parsed from loki", "duration", time.Since(start), "metricDataplane", responseOpts.metricDataplane, "framesLength", len(res.Frames), "stage", stageParseResponse) - return res.Frames, nil + return &res, nil } func makeRawRequest(ctx context.Context, lokiDsUrl string, resourcePath string) (*http.Request, error) { diff --git a/pkg/tsdb/loki/framing_test.go b/pkg/tsdb/loki/framing_test.go index 5480fe3abcc..077444c0f82 100644 --- a/pkg/tsdb/loki/framing_test.go +++ b/pkg/tsdb/loki/framing_test.go @@ -63,13 +63,9 @@ func TestSuccessResponse(t *testing.T) { bytes, err := os.ReadFile(responseFileName) require.NoError(t, err) - frames, err := runQuery(context.Background(), makeMockedAPI(http.StatusOK, "application/json", bytes, nil, false), &query, responseOpts, log.New("test")) + dr, err := runQuery(context.Background(), makeMockedAPI(http.StatusOK, "application/json", bytes, nil, false), &query, responseOpts, log.New("test")) require.NoError(t, err) - dr := &backend.DataResponse{ - Frames: frames, - Error: err, - } experimental.CheckGoldenJSONResponse(t, folder, goldenFileName, dr, false) } @@ -128,11 +124,57 @@ func TestErrorResponse(t *testing.T) { for _, test := range tt { t.Run(test.name, func(t *testing.T) { - frames, err := runQuery(context.Background(), makeMockedAPI(400, test.contentType, test.body, nil, false), &lokiQuery{QueryType: QueryTypeRange, Direction: DirectionBackward}, ResponseOpts{}, log.New("test")) + dr, err := runQuery(context.Background(), makeMockedAPI(400, test.contentType, test.body, nil, false), &lokiQuery{QueryType: QueryTypeRange, Direction: DirectionBackward}, ResponseOpts{}, log.New("test")) + require.NoError(t, err) + require.Len(t, dr.Frames, 0) + require.Equal(t, dr.Error.Error(), test.errorMessage) + require.Equal(t, dr.ErrorSource, backend.ErrorSourceDownstream) + }) + } +} - require.Len(t, frames, 0) - require.Error(t, err) - require.EqualError(t, err, test.errorMessage) +func TestErrorsFromResponseCodes(t *testing.T) { + tt := []struct { + name string + statusCode int + errorSource backend.ErrorSource + }{ + { + name: "parse response with status code 400 into correct error", + statusCode: 400, + errorSource: backend.ErrorSourceDownstream, + }, + { + name: "parse response with status code 406 into correct error", + statusCode: 406, + errorSource: backend.ErrorSourcePlugin, + }, + { + name: "parse response with status code 413 into correct error", + statusCode: 413, + errorSource: backend.ErrorSourcePlugin, + }, + { + name: "parse response with status code 500 into correct error", + statusCode: 500, + errorSource: backend.ErrorSourceDownstream, + }, + { + name: "parse response with status code 501 into correct error", + statusCode: 501, + errorSource: backend.ErrorSourcePlugin, + }, + } + + errorString := "parse error at line 1, col 8: something is wrong" + contentType := "application/json; charset=UTF-8" + + for _, test := range tt { + t.Run(test.name, func(t *testing.T) { + dr, _ := runQuery(context.Background(), makeMockedAPI(test.statusCode, contentType, []byte(errorString), nil, false), &lokiQuery{QueryType: QueryTypeRange, Direction: DirectionBackward}, ResponseOpts{}, log.New("test")) + require.Len(t, dr.Frames, 0) + require.Equal(t, dr.Error.Error(), errorString) + require.Equal(t, dr.ErrorSource, test.errorSource) }) } } diff --git a/pkg/tsdb/loki/loki.go b/pkg/tsdb/loki/loki.go index 6a6e4f59128..68e1466623b 100644 --- a/pkg/tsdb/loki/loki.go +++ b/pkg/tsdb/loki/loki.go @@ -239,37 +239,34 @@ func executeQuery(ctx context.Context, query *lokiQuery, req *backend.QueryDataR defer span.End() - frames, err := runQuery(ctx, api, query, responseOpts, plog) - queryRes := backend.DataResponse{} + queryRes, err := runQuery(ctx, api, query, responseOpts, plog) if err != nil { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) queryRes.Error = err - } else { - queryRes.Frames = frames } - return queryRes + return *queryRes } // we extracted this part of the functionality to make it easy to unit-test it -func runQuery(ctx context.Context, api *LokiAPI, query *lokiQuery, responseOpts ResponseOpts, plog log.Logger) (data.Frames, error) { - frames, err := api.DataQuery(ctx, *query, responseOpts) +func runQuery(ctx context.Context, api *LokiAPI, query *lokiQuery, responseOpts ResponseOpts, plog log.Logger) (*backend.DataResponse, error) { + res, err := api.DataQuery(ctx, *query, responseOpts) if err != nil { plog.Error("Error querying loki", "error", err) - return data.Frames{}, err + return res, err } - for _, frame := range frames { + for _, frame := range res.Frames { err = adjustFrame(frame, query, !responseOpts.metricDataplane, responseOpts.logsDataplane) if err != nil { plog.Error("Error adjusting frame", "error", err) - return data.Frames{}, err + return res, err } } - return frames, nil + return res, nil } func (s *Service) getDSInfo(ctx context.Context, pluginCtx backend.PluginContext) (*datasourceInfo, error) {