feat(monitor): result reducer for query result (#21076)

pkg/apis/monitor/alert.go

@@ -96,6 +96,8 @@ type AlertQuery struct {
 	Model MetricQuery `json:"model"`
 	From  string      `json:"from"`
 	To    string      `json:"to"`
+	// 查询结果 reducer，执行 p95 这些操作
+	ResultReducer *Condition `json:"result_reducer"`
 }
 
 type AlertCreateInput struct {

pkg/apis/monitor/reducer.go

@@ -0,0 +1,48 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package monitor
+
+import "yunion.io/x/pkg/util/sets"
+
+type ReducerType string
+
+const (
+	REDUCER_AVG            ReducerType = "avg"
+	REDUCER_SUM            ReducerType = "sum"
+	REDUCER_MIN            ReducerType = "min"
+	REDUCER_MAX            ReducerType = "max"
+	REDUCER_COUNT          ReducerType = "count"
+	REDUCER_LAST           ReducerType = "last"
+	REDUCER_MEDIAN         ReducerType = "median"
+	REDUCER_DIFF           ReducerType = "diff"
+	REDUCER_PERCENT_DIFF   ReducerType = "percent_diff"
+	REDUCER_COUNT_NON_NULL ReducerType = "count_non_null"
+	REDUCER_PERCENTILE     ReducerType = "percentile"
+)
+
+var ValidateReducerTypes = sets.NewString()
+
+func init() {
+	for _, rt := range []ReducerType{REDUCER_AVG, REDUCER_SUM, REDUCER_MIN,
+		REDUCER_MAX, REDUCER_COUNT, REDUCER_LAST, REDUCER_MEDIAN, REDUCER_DIFF,
+		REDUCER_PERCENT_DIFF, REDUCER_COUNT_NON_NULL, REDUCER_PERCENTILE} {
+		ValidateReducerTypes.Insert(string(rt))
+	}
+}
+
+type ReducedResult struct {
+	Reducer Condition `json:"reducer"`
+	Result  []float64 `json:"result"`
+}

pkg/apis/monitor/unifiedmonitor_query.go

@@ -96,9 +96,10 @@ type SimpleQueryOutput struct {
 }
 
 type MetricsQueryResult struct {
-	SeriesTotal int64
-	Series      TimeSeriesSlice
-	Metas       []QueryResultMeta
+	SeriesTotal   int64
+	Series        TimeSeriesSlice
+	Metas         []QueryResultMeta
+	ReducedResult *ReducedResult
 }
 
 type TimeSeriesPoints []TimePoint
@@ -190,7 +191,8 @@ func NewTimeSeriesPointsFromArgs(values ...float64) TimeSeriesPoints {
 }
 
 type QueryResultMeta struct {
-	RawQuery string `json:"raw_query"`
+	RawQuery           string  `json:"raw_query"`
+	ResultReducerValue float64 `json:"result_reducer_value"`
 }
 
 const ConditionTypeMetricQuery = "metricquery"

pkg/mcclient/modules/monitor/helper.go

@@ -282,9 +282,10 @@ type AlertQuery struct {
 	policy       string
 	resultFormat string
 
-	selects *AlertQuerySelects
-	where   *AlertQueryWhere
-	groupBy *AlertQueryGroupBy
+	selects       *AlertQuerySelects
+	where         *AlertQueryWhere
+	groupBy       *AlertQueryGroupBy
+	resultReducer *monitor.Condition
 }
 
 func NewAlertQuery(database string, measurement string) *AlertQuery {
@@ -395,6 +396,14 @@ func (q *AlertQuery) GroupBy() *AlertQueryGroupBy {
 	return g
 }
 
+func (q *AlertQuery) Reducer(rType string, params []float64) *AlertQuery {
+	q.resultReducer = &monitor.Condition{
+		Type:   rType,
+		Params: params,
+	}
+	return q
+}
+
 type AlertQuerySelects struct {
 	parts []*AlertQuerySelect
 }
@@ -646,6 +655,10 @@ func (input *MetricQueryInput) GroupBy() *AlertQueryGroupBy {
 	return input.query.GroupBy()
 }
 
+func (input *MetricQueryInput) Reducer(rType string, params []float64) *AlertQuery {
+	return input.query.Reducer(rType, params)
+}
+
 func (input *MetricQueryInput) ToQueryData() *monitor.MetricQueryInput {
 	data := &monitor.MetricQueryInput{
 		From:            input.from,
@@ -661,7 +674,10 @@ func (input *MetricQueryInput) ToQueryData() *monitor.MetricQueryInput {
 		SkipCheckSeries: input.skipCheckSeries,
 	}
 	data.MetricQuery = []*monitor.AlertQuery{
-		{Model: input.query.ToMetricQuery()},
+		{
+			Model:         input.query.ToMetricQuery(),
+			ResultReducer: input.query.resultReducer,
+		},
 	}
 
 	jsonData := jsonutils.Marshal(data).(*jsonutils.JSONDict)

pkg/mcclient/options/monitor/unifiedmonitor.go

@@ -15,6 +15,7 @@
 package monitor
 
 import (
+	"strconv"
 	"strings"
 	"time"
 
@@ -62,6 +63,7 @@ type MetricQueryOptions struct {
 	GroupBy         []string `help:"group by tag"`
 	UseMean         bool     `help:"calcuate mean result for field"`
 	SkipCheckSeries bool     `help:"skip checking series: not fetch extra tags from region service"`
+	Reducer         string   `help:"series result reducer. e.g.: sum, percentile(95)"`
 }
 
 func (o MetricQueryOptions) GetQueryInput() (*api.MetricQueryInput, error) {
@@ -112,5 +114,41 @@ func (o MetricQueryOptions) GetQueryInput() (*api.MetricQueryInput, error) {
 		groupBy.TAG(tag)
 	}
 
+	if o.Reducer != "" {
+		r, err := o.parseReducer(o.Reducer)
+		if err != nil {
+			return nil, errors.Wrapf(err, "invalid reducer: %q", o.Reducer)
+		}
+		input.Reducer(r.Type, r.Params)
+	}
+
 	return input.ToQueryData(), nil
 }
+
+func (o MetricQueryOptions) parseReducer(reducer string) (*api.Condition, error) {
+	if reducer == "" {
+		return nil, errors.Errorf("invalid reducer %q", reducer)
+	}
+	parts := strings.Split(reducer, "(")
+	if len(parts) < 1 {
+		return nil, errors.Errorf("invalid reducer %q", reducer)
+	}
+	rType := parts[0]
+	cond := &api.Condition{
+		Type: rType,
+	}
+	if len(parts) > 1 {
+		params := []float64{}
+		paramStr := parts[1]
+		paramsStr := strings.Split(strings.TrimSuffix(paramStr, ")"), ",")
+		for _, param := range paramsStr {
+			f, err := strconv.ParseFloat(strings.ReplaceAll(param, " ", ""), 64)
+			if err != nil {
+				return nil, errors.Wrapf(err, "invalid reducer param %q", param)
+			}
+			params = append(params, f)
+		}
+		cond.Params = params
+	}
+	return cond, nil
+}

pkg/mcclient/options/monitor/unifiedmonitor_test.go

@@ -0,0 +1,57 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package monitor
+
+import (
+	"reflect"
+	"testing"
+
+	api "yunion.io/x/onecloud/pkg/apis/monitor"
+)
+
+func TestMetricQueryOptions_parseReducer(t *testing.T) {
+	tests := []struct {
+		reducer string
+		want    *api.Condition
+		wantErr bool
+	}{
+		{
+			reducer: "avg",
+			want: &api.Condition{
+				Type: "avg",
+			},
+		},
+		{
+			reducer: "percentile(95)",
+			want: &api.Condition{
+				Type:   "percentile",
+				Params: []float64{95},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.reducer, func(t *testing.T) {
+			o := MetricQueryOptions{}
+			got, err := o.parseReducer(tt.reducer)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("parseReducer() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("parseReducer() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/monitor/alerting/conditions/commonalertreducer.go

@@ -14,6 +14,8 @@
 
 package conditions
 
+import "yunion.io/x/onecloud/pkg/apis/monitor"
+
 func NewCommonAlertReducer(t string) *queryReducer {
-	return &queryReducer{Type: t}
+	return &queryReducer{Type: monitor.ReducerType(t)}
 }

pkg/monitor/alerting/conditions/mathreducer.go

@@ -35,8 +35,8 @@ func (s *mathReducer) GetParams() []float64 {
 	return s.Params
 }
 
-func (s *mathReducer) GetType() string {
-	return s.Type
+func (s *mathReducer) GetType() monitor.ReducerType {
+	return monitor.ReducerType(s.Type)
 }
 
 func (s *mathReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {

pkg/monitor/alerting/conditions/metricquery.go

@@ -66,6 +66,11 @@ func NewMetricQueryCondition(models []*monitor.AlertCondition) (*MetricQueryCond
 		if err := validators.ValidateToValue(qc.Query.To); err != nil {
 			return nil, errors.Wrapf(err, "validate to value %q", qc.Query.To)
 		}
+		reducer, err := NewAlertReducer(&model.Reducer)
+		if err != nil {
+			return nil, errors.Wrapf(err, "NewAlertReducer")
+		}
+		qc.Reducer = reducer
 		qc.setResType()
 		cond.QueryCons = append(cond.QueryCons, *qc)
 	}
@@ -100,6 +105,23 @@ func (query *MetricQueryCondition) ExecuteQuery(userCred mcclient.TokenCredentia
 			return nil, errors.Wrapf(err, "query.executeQuery from %s", ds.Type)
 		}
 		log.Debugf("query metrics from TSDB %q elapsed: %s", ds.Type, time.Since(startTime))
+		if firstCond.Reducer.GetType() != "" {
+			if queryResult.reducedResult == nil {
+				queryResult.reducedResult = &monitor.ReducedResult{
+					Reducer: monitor.Condition{
+						Type:   string(firstCond.Reducer.GetType()),
+						Params: firstCond.Reducer.GetParams(),
+					},
+					Result: make([]float64, len(queryResult.series)),
+				}
+			}
+			for i, ss := range queryResult.series {
+				resultReducerValue, _ := firstCond.Reducer.Reduce(ss)
+				if resultReducerValue != nil {
+					queryResult.reducedResult.Result[i] = *resultReducerValue
+				}
+			}
+		}
 		return queryResult, nil
 	}
 
@@ -109,10 +131,10 @@ func (query *MetricQueryCondition) ExecuteQuery(userCred mcclient.TokenCredentia
 			return nil, errors.Wrap(err, "queryTSDB")
 		}
 		metrics := monitor.MetricsQueryResult{
-			Series: make(monitor.TimeSeriesSlice, 0),
-			Metas:  qr.metas,
+			Series:        qr.series,
+			Metas:         qr.metas,
+			ReducedResult: qr.reducedResult,
 		}
-		metrics.Series = qr.series
 		return &metrics, nil
 	}
 
@@ -170,8 +192,9 @@ func (query *MetricQueryCondition) ExecuteQuery(userCred mcclient.TokenCredentia
 	//	metrics.Series = append(metrics.Series, serie)
 	//}
 	metrics := monitor.MetricsQueryResult{
-		Series: make(monitor.TimeSeriesSlice, 0),
-		Metas:  qr.metas,
+		Series:        make(monitor.TimeSeriesSlice, 0),
+		Metas:         qr.metas,
+		ReducedResult: qr.reducedResult,
 	}
 	mtx := sync.Mutex{}
 	workqueue.Parallelize(4, len(qr.series), func(piece int) {
@@ -187,6 +210,7 @@ func (query *MetricQueryCondition) ExecuteQuery(userCred mcclient.TokenCredentia
 	})
 	log.Debugf("fill metrics tag elapsed: %s", time.Since(startTime))
 	log.Debugf("all steps elapsed: %s", time.Since(allStartTime))
+
 	return &metrics, nil
 }
 

pkg/monitor/alerting/conditions/query.go

@@ -98,7 +98,7 @@ func (c *QueryCondition) GenerateFormatCond(meta *monitor.QueryResultMeta, metri
 	return &FormatCond{
 		QueryMeta:    meta,
 		QueryKeyInfo: metric,
-		Reducer:      c.Reducer.GetType(),
+		Reducer:      string(c.Reducer.GetType()),
 		Evaluator:    c.Evaluator,
 	}
 }
@@ -376,8 +376,9 @@ func (c *QueryCondition) jointPointStr(series monitor.TimeSeries, value string,
 }
 
 type queryResult struct {
-	series monitor.TimeSeriesSlice
-	metas  []monitor.QueryResultMeta
+	series        monitor.TimeSeriesSlice
+	metas         []monitor.QueryResultMeta
+	reducedResult *monitor.ReducedResult
 }
 
 func (c *QueryCondition) executeQuery(evalCtx *alerting.EvalContext, timeRange *tsdb.TimeRange) (*queryResult, error) {

pkg/monitor/alerting/conditions/reducer.go

@@ -27,7 +27,7 @@ import (
 
 type Reducer interface {
 	Reduce(series *monitor.TimeSeries) (*float64, []string)
-	GetType() string
+	GetType() monitor.ReducerType
 	GetParams() []float64
 }
 
@@ -35,7 +35,7 @@ type Reducer interface {
 type queryReducer struct {
 	// Type is how the timeseries should be reduced.
 	// Ex: avg, sum, max, min, count
-	Type   string
+	Type   monitor.ReducerType
 	Params []float64
 }
 
@@ -43,7 +43,7 @@ func (s *queryReducer) GetParams() []float64 {
 	return s.Params
 }
 
-func (s *queryReducer) GetType() string {
+func (s *queryReducer) GetType() monitor.ReducerType {
 	return s.Type
 }
 
@@ -64,7 +64,7 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 	allNull := true
 	valArr := make([]string, 0)
 	switch s.Type {
-	case "avg":
+	case monitor.REDUCER_AVG:
 		validPointsCount := 0
 		for _, point := range series.Points {
 			if point.IsValid() {
@@ -76,14 +76,14 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 		if validPointsCount > 0 {
 			value = value / float64(validPointsCount)
 		}
-	case "sum":
+	case monitor.REDUCER_SUM:
 		for _, point := range series.Points {
 			if point.IsValid() {
 				value += point.Value()
 				allNull = false
 			}
 		}
-	case "min":
+	case monitor.REDUCER_MIN:
 		value = math.MaxFloat64
 		for _, point := range series.Points {
 			if point.IsValid() {
@@ -94,7 +94,7 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 				}
 			}
 		}
-	case "max":
+	case monitor.REDUCER_MAX:
 		value = -math.MaxFloat64
 		for _, point := range series.Points {
 			if point.IsValid() {
@@ -105,10 +105,10 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 				}
 			}
 		}
-	case "count":
+	case monitor.REDUCER_COUNT:
 		value = float64(len(series.Points))
 		allNull = false
-	case "last":
+	case monitor.REDUCER_LAST:
 		points := series.Points
 		for i := len(points) - 1; i >= 0; i-- {
 			if points[i].IsValid() {
@@ -118,7 +118,7 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 				break
 			}
 		}
-	case "median":
+	case monitor.REDUCER_MEDIAN:
 		var values []float64
 		for _, v := range series.Points {
 			if v.IsValid() {
@@ -135,11 +135,11 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 				value = (values[(length/2)-1] + values[length/2]) / 2
 			}
 		}
-	case "diff":
+	case monitor.REDUCER_DIFF:
 		allNull, value = calculateDiff(series, allNull, value, diff)
-	case "percent_diff":
+	case monitor.REDUCER_PERCENT_DIFF:
 		allNull, value = calculateDiff(series, allNull, value, percentDiff)
-	case "count_non_null":
+	case monitor.REDUCER_COUNT_NON_NULL:
 		for _, v := range series.Points {
 			if v.IsValid() {
 				value++
@@ -149,7 +149,7 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 		if value > 0 {
 			allNull = false
 		}
-	case "percentile":
+	case monitor.REDUCER_PERCENTILE:
 		var values []float64
 		for _, v := range series.Points {
 			if v.IsValid() {
@@ -178,14 +178,14 @@ func (s *queryReducer) Reduce(series *monitor.TimeSeries) (*float64, []string) {
 
 func newSimpleReducer(cond *monitor.Condition) *queryReducer {
 	return &queryReducer{
-		Type:   cond.Type,
+		Type:   monitor.ReducerType(cond.Type),
 		Params: cond.Params,
 	}
 }
 
 func newSimpleReducerByType(typ string) *queryReducer {
 	return &queryReducer{
-		Type:   typ,
+		Type:   monitor.ReducerType(typ),
 		Params: []float64{},
 	}
 }
@@ -235,5 +235,5 @@ func NewAlertReducer(cond *monitor.Condition) (Reducer, error) {
 		return newMathReducer(cond)
 	}
 
-	return nil, errors.Wrapf(errors.Error("reducer operator is ilegal"), "operator: %s", cond.Operators[0])
+	return nil, errors.Wrapf(errors.Error("reducer operator is illegal"), "operator: %s", cond.Operators[0])
 }

pkg/monitor/alerting/conditions/suggestrulereducer.go

@@ -27,7 +27,7 @@ type suggestRuleReducer struct {
 
 func NewSuggestRuleReducer(t string, duration time.Duration) Reducer {
 	return &suggestRuleReducer{
-		queryReducer: &queryReducer{Type: t},
+		queryReducer: &queryReducer{Type: monitor.ReducerType(t)},
 		duration:     duration,
 	}
 }

pkg/monitor/models/unifiedmonitor.go

@@ -334,6 +334,9 @@ func doQuery(userCred mcclient.TokenCredential, query monitor.MetricQueryInput)
 			Type:  monitor.ConditionTypeMetricQuery,
 			Query: *q,
 		}
+		if q.ResultReducer != nil {
+			condition.Reducer = *q.ResultReducer
+		}
 		conds = append(conds, &condition)
 	}
 	factory := mq.GetQueryFactories()[monitor.ConditionTypeMetricQuery]

pkg/monitor/validators/validators.go

@@ -125,6 +125,11 @@ func ValidateSelectOfMetricQuery(input monitor.AlertQuery) error {
 			return httperrors.NewInputParameterError("select for nothing in query")
 		}
 	}
+	if input.ResultReducer != nil {
+		if !monitor.ValidateReducerTypes.Has(input.ResultReducer.Type) {
+			return httperrors.NewInputParameterError("invalid result reducer type %s", input.ResultReducer.Type)
+		}
+	}
 	return nil
 }