public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CloudMonitoring: Add support for preprocessing (#33011)

Browse Source 
* add support for handling preprocessors in the backend

* add preprocessor tests

* use uppercase for constants

* add super label component

* remove error message from query editor since its not working (probably cause onDataError doesnt work anymore)

* use cheat sheet instead of help

* add return type annotation for projects

* add support for preprocessing. replace segment comp with select. change components names and refactoring

* cleanup

* more pr feedback

* fix annotations editor

* rename aggregation component

* fix broken test

* remove unnecessary cast

* fix strict errors

* fix more strict errors

* remove not used prop

* update docs

* use same inline label for annotation editor

* fix react prop warning

* disable preprocessing for distribution types

* using new default values for reducer

* auto select 'rate' if metric kind is not gauge

* fix create label format

* pr feedback

* more pr feedback

* update images
This commit is contained in:
Erik Sundell
2021-05-19 08:16:05 +02:00
committed by GitHub
parent e3188458d5
commit 5042dc3b52
45 changed files with 1385 additions and 923 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tsdb/cloudmonitoring/cloudmonitoring.go
+41 -10
View File
@@ -57,11 +57,13 @@ var (
)
const (
gceAuthentication string = "gce"
jwtAuthentication string = "jwt"
metricQueryType string = "metrics"
sloQueryType string = "slo"
mqlEditorMode string = "mql"
gceAuthentication string = "gce"
jwtAuthentication string = "jwt"
metricQueryType string = "metrics"
sloQueryType string = "slo"
mqlEditorMode string = "mql"
crossSeriesReducerDefault string = "REDUCE_NONE"
perSeriesAlignerDefault string = "ALIGN_MEAN"
)
func init() {
@@ -210,6 +212,7 @@ func (e *Executor) buildQueryExecutors(tsdbQuery plugins.DataQuery) ([]cloudMoni
if err := json.Unmarshal(model, &q); err != nil {
return nil, fmt.Errorf("could not unmarshal CloudMonitoringQuery json: %w", err)
}
q.MetricQuery.PreprocessorType = toPreprocessorType(q.MetricQuery.Preprocessor)
var target string
params := url.Values{}
params.Add("interval.startTime", startTime.UTC().Format(time.RFC3339))
@@ -345,16 +348,44 @@ func buildSLOFilterExpression(q sloQuery) string {
func setMetricAggParams(params *url.Values, query *metricQuery, durationSeconds int, intervalMs int64) {
if query.CrossSeriesReducer == "" {
query.CrossSeriesReducer = "REDUCE_NONE"
query.CrossSeriesReducer = crossSeriesReducerDefault
}
if query.PerSeriesAligner == "" {
query.PerSeriesAligner = "ALIGN_MEAN"
query.PerSeriesAligner = perSeriesAlignerDefault
}
params.Add("aggregation.crossSeriesReducer", query.CrossSeriesReducer)
params.Add("aggregation.perSeriesAligner", query.PerSeriesAligner)
params.Add("aggregation.alignmentPeriod", calculateAlignmentPeriod(query.AlignmentPeriod, intervalMs, durationSeconds))
alignmentPeriod := calculateAlignmentPeriod(query.AlignmentPeriod, intervalMs, durationSeconds)
// In case a preprocessor is defined, the preprocessor becomes the primary aggregation
// and the aggregation that is specified in the UI becomes the secondary aggregation
// Rules are specified in this issue: https://github.com/grafana/grafana/issues/30866
if query.PreprocessorType != PreprocessorTypeNone {
params.Add("secondaryAggregation.alignmentPeriod", alignmentPeriod)
params.Add("secondaryAggregation.crossSeriesReducer", query.CrossSeriesReducer)
params.Add("secondaryAggregation.perSeriesAligner", query.PerSeriesAligner)
primaryCrossSeriesReducer := crossSeriesReducerDefault
if len(query.GroupBys) > 0 {
primaryCrossSeriesReducer = query.CrossSeriesReducer
}
params.Add("aggregation.crossSeriesReducer", primaryCrossSeriesReducer)
aligner := "ALIGN_RATE"
if query.PreprocessorType == PreprocessorTypeDelta {
aligner = "ALIGN_DELTA"
}
params.Add("aggregation.perSeriesAligner", aligner)
for _, groupBy := range query.GroupBys {
params.Add("secondaryAggregation.groupByFields", groupBy)
}
} else {
params.Add("aggregation.crossSeriesReducer", query.CrossSeriesReducer)
params.Add("aggregation.perSeriesAligner", query.PerSeriesAligner)
}
params.Add("aggregation.alignmentPeriod", alignmentPeriod)
for _, groupBy := range query.GroupBys {
params.Add("aggregation.groupByFields", groupBy)
pkg/tsdb/cloudmonitoring/cloudmonitoring_test.go
+163
View File
@@ -1043,6 +1043,169 @@ func TestCloudMonitoring(t *testing.T) {
assert.Contains(t, value, `zone=monitoring.regex.full_match("us-central1-a~")`)
})
})
t.Run("and query preprocessor is not defined", func(t *testing.T) {
tsdbQuery := getBaseQuery()
tsdbQuery.Queries[0].Model = simplejson.NewFromAny(map[string]interface{}{
"metricType": "a/metric/type",
"crossSeriesReducer": "REDUCE_MIN",
"perSeriesAligner": "REDUCE_SUM",
"alignmentPeriod": "+60s",
"groupBys": []string{"labelname"},
"view": "FULL",
})
qes, err := executor.buildQueryExecutors(tsdbQuery)
require.NoError(t, err)
queries := getCloudMonitoringQueriesFromInterface(t, qes)
assert.Equal(t, 1, len(queries))
assert.Equal(t, "REDUCE_MIN", queries[0].Params["aggregation.crossSeriesReducer"][0])
assert.Equal(t, "REDUCE_SUM", queries[0].Params["aggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["aggregation.alignmentPeriod"][0])
assert.Equal(t, "labelname", queries[0].Params["aggregation.groupByFields"][0])
assert.NotContains(t, queries[0].Params, "secondaryAggregation.crossSeriesReducer")
assert.NotContains(t, "REDUCE_SUM", queries[0].Params, "secondaryAggregation.perSeriesAligner")
assert.NotContains(t, "+60s", queries[0].Params, "secondaryAggregation.alignmentPeriod")
assert.NotContains(t, "labelname", queries[0].Params, "secondaryAggregation.groupByFields")
})
t.Run("and query preprocessor is set to none", func(t *testing.T) {
tsdbQuery := getBaseQuery()
tsdbQuery.Queries[0].Model = simplejson.NewFromAny(map[string]interface{}{
"metricType": "a/metric/type",
"crossSeriesReducer": "REDUCE_MIN",
"perSeriesAligner": "REDUCE_SUM",
"alignmentPeriod": "+60s",
"groupBys": []string{"labelname"},
"view": "FULL",
"preprocessor": "none",
})
qes, err := executor.buildQueryExecutors(tsdbQuery)
require.NoError(t, err)
queries := getCloudMonitoringQueriesFromInterface(t, qes)
assert.Equal(t, 1, len(queries))
assert.Equal(t, "REDUCE_MIN", queries[0].Params["aggregation.crossSeriesReducer"][0])
assert.Equal(t, "REDUCE_SUM", queries[0].Params["aggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["aggregation.alignmentPeriod"][0])
assert.Equal(t, "labelname", queries[0].Params["aggregation.groupByFields"][0])
assert.NotContains(t, queries[0].Params, "secondaryAggregation.crossSeriesReducer")
assert.NotContains(t, "REDUCE_SUM", queries[0].Params, "secondaryAggregation.perSeriesAligner")
assert.NotContains(t, "+60s", queries[0].Params, "secondaryAggregation.alignmentPeriod")
assert.NotContains(t, "labelname", queries[0].Params, "secondaryAggregation.groupByFields")
})
t.Run("and query preprocessor is set to rate and there's no group bys", func(t *testing.T) {
tsdbQuery := getBaseQuery()
tsdbQuery.Queries[0].Model = simplejson.NewFromAny(map[string]interface{}{
"metricType": "a/metric/type",
"crossSeriesReducer": "REDUCE_SUM",
"perSeriesAligner": "REDUCE_MIN",
"alignmentPeriod": "+60s",
"groupBys": []string{},
"view": "FULL",
"preprocessor": "rate",
})
qes, err := executor.buildQueryExecutors(tsdbQuery)
require.NoError(t, err)
queries := getCloudMonitoringQueriesFromInterface(t, qes)
assert.Equal(t, 1, len(queries))
assert.Equal(t, "REDUCE_NONE", queries[0].Params["aggregation.crossSeriesReducer"][0])
assert.Equal(t, "ALIGN_RATE", queries[0].Params["aggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["aggregation.alignmentPeriod"][0])
assert.Equal(t, "REDUCE_SUM", queries[0].Params["secondaryAggregation.crossSeriesReducer"][0])
assert.Equal(t, "REDUCE_MIN", queries[0].Params["secondaryAggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["secondaryAggregation.alignmentPeriod"][0])
})
t.Run("and query preprocessor is set to rate and group bys exist", func(t *testing.T) {
tsdbQuery := getBaseQuery()
tsdbQuery.Queries[0].Model = simplejson.NewFromAny(map[string]interface{}{
"metricType": "a/metric/type",
"crossSeriesReducer": "REDUCE_SUM",
"perSeriesAligner": "REDUCE_MIN",
"alignmentPeriod": "+60s",
"groupBys": []string{"labelname"},
"view": "FULL",
"preprocessor": "rate",
})
qes, err := executor.buildQueryExecutors(tsdbQuery)
require.NoError(t, err)
queries := getCloudMonitoringQueriesFromInterface(t, qes)
assert.Equal(t, 1, len(queries))
assert.Equal(t, "REDUCE_SUM", queries[0].Params["aggregation.crossSeriesReducer"][0])
assert.Equal(t, "ALIGN_RATE", queries[0].Params["aggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["aggregation.alignmentPeriod"][0])
assert.Equal(t, "labelname", queries[0].Params["aggregation.groupByFields"][0])
assert.Equal(t, "REDUCE_SUM", queries[0].Params["secondaryAggregation.crossSeriesReducer"][0])
assert.Equal(t, "REDUCE_MIN", queries[0].Params["secondaryAggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["secondaryAggregation.alignmentPeriod"][0])
assert.Equal(t, "labelname", queries[0].Params["secondaryAggregation.groupByFields"][0])
})
t.Run("and query preprocessor is set to delta and there's no group bys", func(t *testing.T) {
tsdbQuery := getBaseQuery()
tsdbQuery.Queries[0].Model = simplejson.NewFromAny(map[string]interface{}{
"metricType": "a/metric/type",
"crossSeriesReducer": "REDUCE_MIN",
"perSeriesAligner": "REDUCE_SUM",
"alignmentPeriod": "+60s",
"groupBys": []string{},
"view": "FULL",
"preprocessor": "delta",
})
qes, err := executor.buildQueryExecutors(tsdbQuery)
require.NoError(t, err)
queries := getCloudMonitoringQueriesFromInterface(t, qes)
assert.Equal(t, 1, len(queries))
assert.Equal(t, "REDUCE_NONE", queries[0].Params["aggregation.crossSeriesReducer"][0])
assert.Equal(t, "ALIGN_DELTA", queries[0].Params["aggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["aggregation.alignmentPeriod"][0])
assert.Equal(t, "REDUCE_MIN", queries[0].Params["secondaryAggregation.crossSeriesReducer"][0])
assert.Equal(t, "REDUCE_SUM", queries[0].Params["secondaryAggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["secondaryAggregation.alignmentPeriod"][0])
})
t.Run("and query preprocessor is set to delta and group bys exist", func(t *testing.T) {
tsdbQuery := getBaseQuery()
tsdbQuery.Queries[0].Model = simplejson.NewFromAny(map[string]interface{}{
"metricType": "a/metric/type",
"crossSeriesReducer": "REDUCE_MIN",
"perSeriesAligner": "REDUCE_SUM",
"alignmentPeriod": "+60s",
"groupBys": []string{"labelname"},
"view": "FULL",
"preprocessor": "delta",
})
qes, err := executor.buildQueryExecutors(tsdbQuery)
require.NoError(t, err)
queries := getCloudMonitoringQueriesFromInterface(t, qes)
assert.Equal(t, 1, len(queries))
assert.Equal(t, "REDUCE_MIN", queries[0].Params["aggregation.crossSeriesReducer"][0])
assert.Equal(t, "ALIGN_DELTA", queries[0].Params["aggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["aggregation.alignmentPeriod"][0])
assert.Equal(t, "labelname", queries[0].Params["aggregation.groupByFields"][0])
assert.Equal(t, "REDUCE_MIN", queries[0].Params["secondaryAggregation.crossSeriesReducer"][0])
assert.Equal(t, "REDUCE_SUM", queries[0].Params["secondaryAggregation.perSeriesAligner"][0])
assert.Equal(t, "+60s", queries[0].Params["secondaryAggregation.alignmentPeriod"][0])
assert.Equal(t, "labelname", queries[0].Params["secondaryAggregation.groupByFields"][0])
})
}
func loadTestFile(path string) (cloudMonitoringResponse, error) {
pkg/tsdb/cloudmonitoring/preprocessor.go
+22
View File
@@ -0,0 +1,22 @@
package cloudmonitoring
type preprocessorType int
const (
PreprocessorTypeNone preprocessorType = iota
PreprocessorTypeRate
PreprocessorTypeDelta
)
func toPreprocessorType(preprocessorTypeString string) preprocessorType {
switch preprocessorTypeString {
case "none":
return PreprocessorTypeNone
case "rate":
return PreprocessorTypeRate
case "delta":
return PreprocessorTypeDelta
default:
return PreprocessorTypeNone
}
}
pkg/tsdb/cloudmonitoring/time_series_filter.go
+4
View File
@@ -95,6 +95,10 @@ func (timeSeriesFilter *cloudMonitoringTimeSeriesFilter) parseResponse(queryRes
frame.RefID = timeSeriesFilter.RefID
frame.Meta = &data.FrameMeta{
ExecutedQueryString: executedQueryString,
Custom: map[string]interface{}{
"alignmentPeriod": timeSeriesFilter.Params.Get("aggregation.alignmentPeriod"),
"perSeriesAligner": timeSeriesFilter.Params.Get("aggregation.perSeriesAligner"),
},
}
for key, value := range series.Metric.Labels {
pkg/tsdb/cloudmonitoring/types.go
+2
View File
@@ -56,6 +56,8 @@ type (
View string
EditorMode string
Query string
Preprocessor string
PreprocessorType preprocessorType
}
sloQuery struct {