apitech
/
grafana
mirror of https://github.com/grafana/grafana
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Loki: Add error source to DataQuery (#77876)

Browse Source 
* Loki: Add error source to DataQuery

* Create middlewares to fix healtcheck test

* Add comment for errors from ReadPrometheusStyleResult
pull/77026/head^2
Ivana Huckova 2 years ago committed by GitHub
parent 95f5bab39b
commit 934456dc1c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 49 additions and 52 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 22
      pkg/tsdb/loki/api.go
  2. 24
      pkg/tsdb/loki/api_test.go
  3. 19
      pkg/tsdb/loki/framing_test.go
  4. 34
      pkg/tsdb/loki/loki.go
  5. 2
      pkg/tsdb/loki/loki_bench_test.go

22
pkg/tsdb/loki/api.go
Unescape View File

@ -13,12 +13,13 @@ import (
"strconv"
"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-plugin-sdk-go/experimental/errorsource"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/tsdb/loki/instrumentation"
@ -155,10 +156,10 @@ 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 {
req, err := makeDataRequest(ctx, api.url, query)
if err != nil {
return nil, err
return errorsource.Response(errorsource.PluginError(err, false))
}
queryAttrs := []any{"start", query.Start, "end", query.End, "step", query.Step, "query", query.Expr, "queryType", query.QueryType, "direction", query.Direction, "maxLines", query.MaxLines, "supportingQueryType", query.SupportingQueryType, "lokiHost", req.URL.Host, "lokiPath", req.URL.Path}
@ -176,7 +177,8 @@ 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
// Here, errors source is provided by errorsource middleware
return errorsource.Response(err)
}
defer func() {
@ -191,7 +193,11 @@ func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery, responseOpts
err := readLokiError(resp.Body)
lp = append(lp, "status", "error", "error", err)
api.log.Error("Error received from Loki", lp...)
return nil, err
// errors should be processed by errorsource middleware
// here we do here something extra - turning non-200 to error
// we will consider this Plugin error, but let's re-evaluate if we need this
// @todo Re-evaluate if we need to turn non-200 to error
return errorsource.Response(errorsource.PluginError(err, false))
} else {
lp = append(lp, "status", "ok")
api.log.Info("Response received from loki", lp...)
@ -211,12 +217,14 @@ func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery, responseOpts
span.SetStatus(codes.Error, err.Error())
instrumentation.UpdatePluginParsingResponseDurationSeconds(ctx, time.Since(start), "error")
api.log.Error("Error parsing response from loki", "error", res.Error, "metricDataplane", responseOpts.metricDataplane, "duration", time.Since(start), "stage", stageParseResponse)
return nil, res.Error
// Here the response.Error is set by converter.ReadPrometheusStyleResult without ErrorSource, which means it will always be PluginError.
// @todo: We should look into when successful response is returned with error field and what type of ErrorSource we should set for that
return res
}
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
}
func makeRawRequest(ctx context.Context, lokiDsUrl string, resourcePath string) (*http.Request, error) {

24
pkg/tsdb/loki/api_test.go
Unescape View File

@ -28,8 +28,8 @@ func TestApiLogVolume(t *testing.T) {
require.Equal(t, "Source=logvolhist", req.Header.Get("X-Query-Tags"))
})
_, err := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryLogsVolume, QueryType: QueryTypeRange}, ResponseOpts{})
require.NoError(t, err)
res := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryLogsVolume, QueryType: QueryTypeRange}, ResponseOpts{})
require.Equal(t, nil, res.Error)
require.True(t, called)
})
@ -40,8 +40,8 @@ func TestApiLogVolume(t *testing.T) {
require.Equal(t, "Source=logsample", req.Header.Get("X-Query-Tags"))
})
_, err := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryLogsSample, QueryType: QueryTypeRange}, ResponseOpts{})
require.NoError(t, err)
res := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryLogsSample, QueryType: QueryTypeRange}, ResponseOpts{})
require.Equal(t, nil, res.Error)
require.True(t, called)
})
@ -52,8 +52,8 @@ func TestApiLogVolume(t *testing.T) {
require.Equal(t, "Source=datasample", req.Header.Get("X-Query-Tags"))
})
_, err := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryDataSample, QueryType: QueryTypeRange}, ResponseOpts{})
require.NoError(t, err)
res := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryDataSample, QueryType: QueryTypeRange}, ResponseOpts{})
require.Equal(t, nil, res.Error)
require.True(t, called)
})
@ -64,8 +64,8 @@ func TestApiLogVolume(t *testing.T) {
require.Equal(t, "", req.Header.Get("X-Query-Tags"))
})
_, err := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryNone, QueryType: QueryTypeRange}, ResponseOpts{})
require.NoError(t, err)
res := api.DataQuery(context.Background(), lokiQuery{Expr: "", SupportingQueryType: SupportingQueryNone, QueryType: QueryTypeRange}, ResponseOpts{})
require.Equal(t, nil, res.Error)
require.True(t, called)
})
}
@ -133,8 +133,8 @@ func TestApiUrlHandling(t *testing.T) {
QueryType: QueryTypeRange,
}
_, err := api.DataQuery(context.Background(), query, ResponseOpts{})
require.NoError(t, err)
res := api.DataQuery(context.Background(), query, ResponseOpts{})
require.Equal(t, nil, res.Error)
require.True(t, called)
})
}
@ -154,8 +154,8 @@ func TestApiUrlHandling(t *testing.T) {
QueryType: QueryTypeInstant,
}
_, err := api.DataQuery(context.Background(), query, ResponseOpts{})
require.NoError(t, err)
res := api.DataQuery(context.Background(), query, ResponseOpts{})
require.Equal(t, nil, res.Error)
require.True(t, called)
})
}

19
pkg/tsdb/loki/framing_test.go
Unescape View File

@ -8,7 +8,6 @@ import (
"testing"
"time"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/experimental"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/stretchr/testify/require"
@ -61,14 +60,10 @@ 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), &query, responseOpts, log.New("test"))
require.NoError(t, err)
resp := runQuery(context.Background(), makeMockedAPI(http.StatusOK, "application/json", bytes, nil), &query, responseOpts, log.New("test"))
require.Equal(t, nil, resp.Error)
dr := &backend.DataResponse{
Frames: frames,
Error: err,
}
experimental.CheckGoldenJSONResponse(t, folder, goldenFileName, dr, true)
experimental.CheckGoldenJSONResponse(t, folder, goldenFileName, &resp, true)
}
for _, test := range tt {
@ -126,11 +121,11 @@ 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), &lokiQuery{QueryType: QueryTypeRange, Direction: DirectionBackward}, ResponseOpts{}, log.New("test"))
resp := runQuery(context.Background(), makeMockedAPI(400, test.contentType, test.body, nil), &lokiQuery{QueryType: QueryTypeRange, Direction: DirectionBackward}, ResponseOpts{}, log.New("test"))
require.Len(t, frames, 0)
require.Error(t, err)
require.EqualError(t, err, test.errorMessage)
require.Len(t, resp.Frames, 0)
require.Error(t, resp.Error)
require.EqualError(t, resp.Error, test.errorMessage)
})
}
}

34
pkg/tsdb/loki/loki.go
Unescape View File

@ -13,8 +13,10 @@ import (
"github.com/grafana/dskit/concurrency"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/backend/datasource"
sdkhttpclient "github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana-plugin-sdk-go/experimental/errorsource"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
@ -95,6 +97,7 @@ func newInstanceSettings(httpClientProvider httpclient.Provider) datasource.Inst
return nil, err
}
opts.Middlewares = append(sdkhttpclient.DefaultMiddlewares(), errorsource.Middleware("errorsource"))
client, err := httpClientProvider.New(opts)
if err != nil {
return nil, err
@ -239,37 +242,28 @@ func executeQuery(ctx context.Context, query *lokiQuery, req *backend.QueryDataR
defer span.End()
frames, err := runQuery(ctx, api, query, responseOpts, plog)
queryRes := backend.DataResponse{}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
queryRes.Error = err
} else {
queryRes.Frames = frames
}
return queryRes
res := runQuery(ctx, api, query, responseOpts, plog)
return res
}
// 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)
if err != nil {
plog.Error("Error querying loki", "error", err)
return data.Frames{}, err
func runQuery(ctx context.Context, api *LokiAPI, query *lokiQuery, responseOpts ResponseOpts, plog log.Logger) backend.DataResponse {
dataResponse := api.DataQuery(ctx, *query, responseOpts)
if dataResponse.Error != nil {
plog.Error("Error querying loki", "error", dataResponse.Error)
return dataResponse
}
for _, frame := range frames {
err = adjustFrame(frame, query, !responseOpts.metricDataplane, responseOpts.logsDataplane)
for _, frame := range dataResponse.Frames {
err := adjustFrame(frame, query, !responseOpts.metricDataplane, responseOpts.logsDataplane)
if err != nil {
plog.Error("Error adjusting frame", "error", err)
return data.Frames{}, err
return errorsource.Response(errorsource.PluginError(err, false))
}
}
return frames, nil
return dataResponse
}
func (s *Service) getDSInfo(ctx context.Context, pluginCtx backend.PluginContext) (*datasourceInfo, error) {

2
pkg/tsdb/loki/loki_bench_test.go
Unescape View File

@ -19,7 +19,7 @@ func BenchmarkMatrixJson(b *testing.B) {
b.ResetTimer()
for n := 0; n < b.N; n++ {
_, _ = runQuery(context.Background(), makeMockedAPI(http.StatusOK, "application/json", bytes, nil), &lokiQuery{}, ResponseOpts{}, log.New("test"))
_ = runQuery(context.Background(), makeMockedAPI(http.StatusOK, "application/json", bytes, nil), &lokiQuery{}, ResponseOpts{}, log.New("test"))
}
}

Write Preview
Loading…
Cancel
Save