From 30d5482a484eb08ead32e56ebf955742427d6f4e Mon Sep 17 00:00:00 2001 From: Andrej Ocenas Date: Mon, 31 Jan 2022 19:51:30 +0100 Subject: [PATCH] Prometheus: Fix some of the alerting queries that use reduce/math operation (#44380) (#44672) * Prometheus: Dont fill response with nulls for alerting queries * Refactor based on reviews (cherry picked from commit 780591cc1259684982de937766ab196820194f93) --- .../prometheus/json_to_dataframes_test.go | 2 +- pkg/tsdb/prometheus/time_series_query.go | 50 ++++++++++++++++--- pkg/tsdb/prometheus/time_series_query_test.go | 45 ++++++++++++++--- 3 files changed, 84 insertions(+), 13 deletions(-) diff --git a/pkg/tsdb/prometheus/json_to_dataframes_test.go b/pkg/tsdb/prometheus/json_to_dataframes_test.go index aaf782ec0ff..3b46f055d15 100644 --- a/pkg/tsdb/prometheus/json_to_dataframes_test.go +++ b/pkg/tsdb/prometheus/json_to_dataframes_test.go @@ -109,7 +109,7 @@ func testScenario(t *testing.T, name string) { api, err := makeMockedApi(responseBytes) require.NoError(t, err) - result, err := runQueries(context.Background(), api, []*PrometheusQuery{&query}) + result, err := runQueries(context.Background(), api, []*PrometheusQuery{&query}, true) require.NoError(t, err) require.Len(t, result.Responses, 1) diff --git a/pkg/tsdb/prometheus/time_series_query.go b/pkg/tsdb/prometheus/time_series_query.go index 3375865c32a..0dfaef2915e 100644 --- a/pkg/tsdb/prometheus/time_series_query.go +++ b/pkg/tsdb/prometheus/time_series_query.go @@ -47,7 +47,7 @@ const ( ExemplarQueryType TimeSeriesQueryType = "exemplar" ) -func runQueries(ctx context.Context, client apiv1.API, queries []*PrometheusQuery) (*backend.QueryDataResponse, error) { +func runQueries(ctx context.Context, client apiv1.API, queries []*PrometheusQuery, fillNulls bool) (*backend.QueryDataResponse, error) { result := backend.QueryDataResponse{ Responses: backend.Responses{}, } @@ -101,7 +101,7 @@ func runQueries(ctx context.Context, client apiv1.API, queries []*PrometheusQuer } } - frames, err := parseTimeSeriesResponse(response, query) + frames, err := parseTimeSeriesResponse(response, query, fillNulls) if err != nil { return &result, err } @@ -128,7 +128,12 @@ func (s *Service) executeTimeSeriesQuery(ctx context.Context, req *backend.Query return &result, err } - return runQueries(ctx, client, queries) + fillNulls := true + if req.Headers["FromAlert"] == "true" { + fillNulls = false + } + + return runQueries(ctx, client, queries, fillNulls) } func formatLegend(metric model.Metric, query *PrometheusQuery) string { @@ -202,7 +207,7 @@ func (s *Service) parseTimeSeriesQuery(queryContext *backend.QueryDataRequest, d return qs, nil } -func parseTimeSeriesResponse(value map[TimeSeriesQueryType]interface{}, query *PrometheusQuery) (data.Frames, error) { +func parseTimeSeriesResponse(value map[TimeSeriesQueryType]interface{}, query *PrometheusQuery, fillNulls bool) (data.Frames, error) { var ( frames = data.Frames{} nextFrames = data.Frames{} @@ -214,7 +219,11 @@ func parseTimeSeriesResponse(value map[TimeSeriesQueryType]interface{}, query *P switch v := value.(type) { case model.Matrix: - nextFrames = matrixToDataFrames(v, query, nextFrames) + if fillNulls { + nextFrames = matrixToDataFramesWithNullFill(v, query, nextFrames) + } else { + nextFrames = matrixToDataFrames(v, query, nextFrames) + } case model.Vector: nextFrames = vectorToDataFrames(v, query, nextFrames) case *model.Scalar: @@ -308,7 +317,7 @@ func interpolateVariables(model *QueryModel, interval time.Duration, timeRange t return expr } -func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery, frames data.Frames) data.Frames { +func matrixToDataFramesWithNullFill(matrix model.Matrix, query *PrometheusQuery, frames data.Frames) data.Frames { for _, v := range matrix { tags := make(map[string]string, len(v.Metric)) for k, v := range v.Metric { @@ -358,6 +367,35 @@ func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery, frames data return frames } +func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery, frames data.Frames) data.Frames { + for _, v := range matrix { + tags := make(map[string]string, len(v.Metric)) + for k, v := range v.Metric { + tags[string(k)] = string(v) + } + timeField := data.NewFieldFromFieldType(data.FieldTypeTime, len(v.Values)) + valueField := data.NewFieldFromFieldType(data.FieldTypeNullableFloat64, len(v.Values)) + + for i, k := range v.Values { + timeField.Set(i, time.Unix(k.Timestamp.Unix(), 0).UTC()) + value := float64(k.Value) + + if !math.IsNaN(value) { + valueField.Set(i, &value) + } + } + + name := formatLegend(v.Metric, query) + timeField.Name = data.TimeSeriesTimeFieldName + valueField.Name = data.TimeSeriesValueFieldName + valueField.Config = &data.FieldConfig{DisplayNameFromDS: name} + valueField.Labels = tags + frames = append(frames, newDataFrame(name, "matrix", timeField, valueField)) + } + + return frames +} + func scalarToDataFrames(scalar *model.Scalar, query *PrometheusQuery, frames data.Frames) data.Frames { timeVector := []time.Time{time.Unix(scalar.Timestamp.Unix(), 0).UTC()} values := []float64{float64(scalar.Value)} diff --git a/pkg/tsdb/prometheus/time_series_query_test.go b/pkg/tsdb/prometheus/time_series_query_test.go index a98f995759f..2f5ec14f160 100644 --- a/pkg/tsdb/prometheus/time_series_query_test.go +++ b/pkg/tsdb/prometheus/time_series_query_test.go @@ -556,7 +556,7 @@ func TestPrometheus_parseTimeSeriesResponse(t *testing.T) { query := &PrometheusQuery{ LegendFormat: "legend {{app}}", } - res, err := parseTimeSeriesResponse(value, query) + res, err := parseTimeSeriesResponse(value, query, true) require.NoError(t, err) // Test fields @@ -594,7 +594,7 @@ func TestPrometheus_parseTimeSeriesResponse(t *testing.T) { End: time.Unix(5, 0).UTC(), UtcOffsetSec: 0, } - res, err := parseTimeSeriesResponse(value, query) + res, err := parseTimeSeriesResponse(value, query, true) require.NoError(t, err) require.Len(t, res, 1) @@ -631,7 +631,7 @@ func TestPrometheus_parseTimeSeriesResponse(t *testing.T) { End: time.Unix(4, 0).UTC(), UtcOffsetSec: 0, } - res, err := parseTimeSeriesResponse(value, query) + res, err := parseTimeSeriesResponse(value, query, true) require.NoError(t, err) require.Len(t, res, 1) @@ -643,6 +643,39 @@ func TestPrometheus_parseTimeSeriesResponse(t *testing.T) { require.Nil(t, res[0].Fields[1].At(2)) }) + t.Run("matrix response with from alerting missed data points should be parsed correctly", func(t *testing.T) { + values := []p.SamplePair{ + {Value: 1, Timestamp: 1000}, + {Value: 4, Timestamp: 4000}, + } + value := make(map[TimeSeriesQueryType]interface{}) + value[RangeQueryType] = p.Matrix{ + &p.SampleStream{ + Metric: p.Metric{"app": "Application", "tag2": "tag2"}, + Values: values, + }, + } + query := &PrometheusQuery{ + LegendFormat: "", + Step: 1 * time.Second, + Start: time.Unix(1, 0).UTC(), + End: time.Unix(4, 0).UTC(), + UtcOffsetSec: 0, + } + res, err := parseTimeSeriesResponse(value, query, false) + + require.NoError(t, err) + require.Len(t, res, 1) + require.Equal(t, res[0].Name, "{app=\"Application\", tag2=\"tag2\"}") + require.Len(t, res[0].Fields, 2) + require.Len(t, res[0].Fields[0].Labels, 0) + require.Equal(t, res[0].Fields[0].Name, "Time") + require.Len(t, res[0].Fields[1].Labels, 2) + require.Equal(t, res[0].Fields[1].Labels.String(), "app=Application, tag2=tag2") + require.Equal(t, res[0].Fields[1].Name, "Value") + require.Equal(t, res[0].Fields[1].Config.DisplayNameFromDS, "{app=\"Application\", tag2=\"tag2\"}") + }) + t.Run("matrix response with NaN value should be changed to null", func(t *testing.T) { value := make(map[TimeSeriesQueryType]interface{}) value[RangeQueryType] = p.Matrix{ @@ -660,7 +693,7 @@ func TestPrometheus_parseTimeSeriesResponse(t *testing.T) { End: time.Unix(4, 0).UTC(), UtcOffsetSec: 0, } - res, err := parseTimeSeriesResponse(value, query) + res, err := parseTimeSeriesResponse(value, query, true) require.NoError(t, err) var nilPointer *float64 @@ -680,7 +713,7 @@ func TestPrometheus_parseTimeSeriesResponse(t *testing.T) { query := &PrometheusQuery{ LegendFormat: "legend {{app}}", } - res, err := parseTimeSeriesResponse(value, query) + res, err := parseTimeSeriesResponse(value, query, true) require.NoError(t, err) require.Len(t, res, 1) @@ -707,7 +740,7 @@ func TestPrometheus_parseTimeSeriesResponse(t *testing.T) { } query := &PrometheusQuery{} - res, err := parseTimeSeriesResponse(value, query) + res, err := parseTimeSeriesResponse(value, query, true) require.NoError(t, err) require.Len(t, res, 1)