Merge pull request #534 from spacefreak86/fix-summarize-values

fix SummarizeValues

Author: deniszh

pkg/expr/functions/legendValue/function.go

@@ -57,7 +57,7 @@ func (f *legendValue) Do(ctx context.Context, e parser.Expr, from, until int32,
 	for _, a := range arg {
 		var values []string
 		for _, method := range methods {
-			summaryVal, _, err := helper.SummarizeValues(method, a.Values)
+			summaryVal, _, err := helper.SummarizeValues(method, a.Values, a.IsAbsent)
 			if err != nil {
 				return []*types.MetricData{}, err
 			}

pkg/expr/functions/sortBy/function.go

@@ -42,11 +42,11 @@ func (f *sortBy) Do(ctx context.Context, e parser.Expr, from, until int32, value
 	for i, a := range arg {
 		switch e.Target() {
 		case "sortByTotal":
-			vals[i], _, _ = helper.SummarizeValues("sum", a.Values)
+			vals[i], _, _ = helper.SummarizeValues("sum", a.Values, a.IsAbsent)
 		case "sortByMaxima":
-			vals[i], _, _ = helper.SummarizeValues("max", a.Values)
+			vals[i], _, _ = helper.SummarizeValues("max", a.Values, a.IsAbsent)
 		case "sortByMinima":
-			min, _, _ := helper.SummarizeValues("min", a.Values)
+			min, _, _ := helper.SummarizeValues("min", a.Values, a.IsAbsent)
 			vals[i] = 1 / min
 		}
 	}

pkg/expr/functions/summarize/function.go

@@ -117,12 +117,14 @@ func (f *summarize) Do(ctx context.Context, e parser.Expr, from, until int32, va
 		t := arg.StartTime // unadjusted
 		bucketEnd := start + bucketSize
 		values := make([]float64, 0, bucketSize/arg.StepTime)
+		absent := make([]bool, 0, bucketSize/arg.StepTime)
 		ridx := 0
 		bucketItems := 0
 		for i, v := range arg.Values {
 			bucketItems++
 			if !arg.IsAbsent[i] {
 				values = append(values, v)
+				absent = append(absent, false)
 			}
 
 			t += arg.StepTime
@@ -132,20 +134,21 @@ func (f *summarize) Do(ctx context.Context, e parser.Expr, from, until int32, va
 			}
 
 			if t >= bucketEnd {
-				r.Values[ridx], r.IsAbsent[ridx], err = helper.SummarizeValues(summarizeFunction, values)
+				r.Values[ridx], r.IsAbsent[ridx], err = helper.SummarizeValues(summarizeFunction, values, absent)
 				if err != nil {
 					return []*types.MetricData{}, err
 				}
 				ridx++
 				bucketEnd += bucketSize
 				bucketItems = 0
 				values = values[:0]
+				absent = absent[:0]
 			}
 		}
 
 		// last partial bucket
 		if bucketItems > 0 {
-			r.Values[ridx], r.IsAbsent[ridx], err = helper.SummarizeValues(summarizeFunction, values)
+			r.Values[ridx], r.IsAbsent[ridx], err = helper.SummarizeValues(summarizeFunction, values, absent)
 			if err != nil {
 				return []*types.MetricData{}, err
 			}

pkg/expr/helper/helper.go

@@ -188,7 +188,7 @@ func AggregateSeriesWithWildcards(name string, args []*types.MetricData, fields
 }
 
 // SummarizeValues summarizes values
-func SummarizeValues(f string, values []float64) (float64, bool, error) {
+func SummarizeValues(f string, values []float64, absent []bool) (float64, bool, error) {
 	rv := 0.0
 
 	if len(values) == 0 {
@@ -201,10 +201,16 @@ func SummarizeValues(f string, values []float64) (float64, bool, error) {
 			rv += av
 		}
 	case "avg", "average":
-		for _, av := range values {
-			rv += av
+		total := 0
+		for i, av := range values {
+			if !absent[i] {
+				rv += av
+				total++
+			}
+		}
+		if total > 0 {
+			rv /= float64(total)
 		}
-		rv /= float64(len(values))
 	case "max":
 		rv = math.Inf(-1)
 		for _, av := range values {
@@ -214,20 +220,31 @@ func SummarizeValues(f string, values []float64) (float64, bool, error) {
 		}
 	case "min":
 		rv = math.Inf(1)
-		for _, av := range values {
-			if av < rv {
-				rv = av
+		for i, av := range values {
+			if !absent[i] {
+				if av < rv {
+					rv = av
+				}
 			}
 		}
 	case "last":
-		if len(values) > 0 {
-			rv = values[len(values)-1]
+		for i := len(values) - 1; i >= 0; i-- {
+			if !absent[i] {
+				rv = values[i]
+				break
+			}
 		}
 	case "count":
-		rv = float64(len(values))
+		total := 0
+		for i := range values {
+			if !absent[i] {
+				total++
+			}
+		}
+		rv = float64(total)
 	case "median":
-		val, absent := Percentile(values, 50, true)
-		return val, absent, nil
+		val, abs := Percentile(values, 50, true)
+		return val, abs, nil
 	default:
 		looks_like_percentile, err := regexp.MatchString(`^p\d\d?$`, f)
 		if err != nil {
@@ -239,8 +256,8 @@ func SummarizeValues(f string, values []float64) (float64, bool, error) {
 			if err != nil {
 				return 0, true, parser.ParseError(err.Error())
 			}
-			val, absent := Percentile(values, percent, true)
-			return val, absent, nil
+			val, abs := Percentile(values, percent, true)
+			return val, abs, nil
 		} else {
 			return 0, true, parser.ParseError(fmt.Sprintf("unsupported aggregation function: %s", f))
 		}