app/vmselect: add sum_eq_over_time, sum_gt_over_time and sum_le_over_time functions to MetricsQL

See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4641

app/vmselect/promql/exec_test.go

@@ -6108,6 +6108,39 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
+	t.Run(`sum_gt_over_time`, func(t *testing.T) {
+		t.Parallel()
+		q := `round(sum_gt_over_time(rand(0)[200s:10s], 0.7), 0.1)`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{5.9, 5.2, 8.5, 5.1, 4.9, 4.5},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
+	t.Run(`sum_le_over_time`, func(t *testing.T) {
+		t.Parallel()
+		q := `round(sum_le_over_time(rand(0)[200s:10s], 0.7), 0.1)`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{4.2, 4.9, 3.2, 5.8, 4.1, 5.3},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
+	t.Run(`sum_eq_over_time`, func(t *testing.T) {
+		t.Parallel()
+		q := `round(sum_eq_over_time(rand(0)[200s:10s], 0.7), 0.1)`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0, 0, 0, 0, 0, 0},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`increases_over_time`, func(t *testing.T) {
 		t.Parallel()
 		q := `increases_over_time(rand(0)[200s:10s])`

app/vmselect/promql/rollup.go

@@ -79,12 +79,15 @@ var rollupFuncs = map[string]newRollupFunc{
 	"rollup_rate":             newRollupFuncOneOrTwoArgs(rollupFake), // + rollupFuncsRemoveCounterResets
 	"rollup_scrape_interval":  newRollupFuncOneOrTwoArgs(rollupFake),
 	"scrape_interval":         newRollupFuncOneArg(rollupScrapeInterval),
+	"share_eq_over_time":      newRollupShareEQ,
 	"share_gt_over_time":      newRollupShareGT,
 	"share_le_over_time":      newRollupShareLE,
-	"share_eq_over_time":      newRollupShareEQ,
 	"stale_samples_over_time": newRollupFuncOneArg(rollupStaleSamples),
 	"stddev_over_time":        newRollupFuncOneArg(rollupStddev),
 	"stdvar_over_time":        newRollupFuncOneArg(rollupStdvar),
+	"sum_eq_over_time":        newRollupSumEQ,
+	"sum_gt_over_time":        newRollupSumGT,
+	"sum_le_over_time":        newRollupSumLE,
 	"sum_over_time":           newRollupFuncOneArg(rollupSum),
 	"sum2_over_time":          newRollupFuncOneArg(rollupSum2),
 	"tfirst_over_time":        newRollupFuncOneArg(rollupTfirst),
@@ -1089,59 +1092,89 @@ func newRollupDurationOverTime(args []interface{}) (rollupFunc, error) {
 }
 
 func newRollupShareLE(args []interface{}) (rollupFunc, error) {
-	return newRollupShareFilter(args, countFilterLE)
+	return newRollupAvgFilter(args, countFilterLE)
 }
 
-func countFilterLE(values []float64, le float64) int {
+func countFilterLE(values []float64, le float64) float64 {
 	n := 0
 	for _, v := range values {
 		if v <= le {
 			n++
 		}
 	}
-	return n
+	return float64(n)
 }
 
 func newRollupShareGT(args []interface{}) (rollupFunc, error) {
-	return newRollupShareFilter(args, countFilterGT)
+	return newRollupAvgFilter(args, countFilterGT)
 }
 
-func countFilterGT(values []float64, gt float64) int {
+func countFilterGT(values []float64, gt float64) float64 {
 	n := 0
 	for _, v := range values {
 		if v > gt {
 			n++
 		}
 	}
-	return n
+	return float64(n)
 }
 
 func newRollupShareEQ(args []interface{}) (rollupFunc, error) {
-	return newRollupShareFilter(args, countFilterEQ)
+	return newRollupAvgFilter(args, countFilterEQ)
 }
 
-func countFilterEQ(values []float64, eq float64) int {
+func sumFilterEQ(values []float64, eq float64) float64 {
+	var sum float64
+	for _, v := range values {
+		if v == eq {
+			sum += v
+		}
+	}
+	return sum
+}
+
+func sumFilterLE(values []float64, le float64) float64 {
+	var sum float64
+	for _, v := range values {
+		if v <= le {
+			sum += v
+		}
+	}
+	return sum
+}
+
+func sumFilterGT(values []float64, gt float64) float64 {
+	var sum float64
+	for _, v := range values {
+		if v > gt {
+			sum += v
+		}
+	}
+	return sum
+}
+
+func countFilterEQ(values []float64, eq float64) float64 {
 	n := 0
 	for _, v := range values {
 		if v == eq {
 			n++
 		}
 	}
-	return n
+	return float64(n)
 }
 
-func countFilterNE(values []float64, ne float64) int {
+func countFilterNE(values []float64, ne float64) float64 {
 	n := 0
 	for _, v := range values {
 		if v != ne {
 			n++
 		}
 	}
-	return n
+	return float64(n)
 }
 
-func newRollupShareFilter(args []interface{}, countFilter func(values []float64, limit float64) int) (rollupFunc, error) {
+func newRollupAvgFilter(args []interface{}, f func(values []float64, limit float64) float64) (rollupFunc, error) {
-	rf, err := newRollupCountFilter(args, countFilter)
+	rf, err := newRollupFilter(args, f)
 	if err != nil {
 		return nil, err
 	}
@@ -1151,23 +1184,35 @@ func newRollupShareFilter(args []interface{}, countFilter func(values []float64,
 	}, nil
 }
 
+func newRollupCountEQ(args []interface{}) (rollupFunc, error) {
+	return newRollupFilter(args, countFilterEQ)
+}
+
 func newRollupCountLE(args []interface{}) (rollupFunc, error) {
-	return newRollupCountFilter(args, countFilterLE)
+	return newRollupFilter(args, countFilterLE)
 }
 
 func newRollupCountGT(args []interface{}) (rollupFunc, error) {
-	return newRollupCountFilter(args, countFilterGT)
+	return newRollupFilter(args, countFilterGT)
-}
-
-func newRollupCountEQ(args []interface{}) (rollupFunc, error) {
-	return newRollupCountFilter(args, countFilterEQ)
 }
 
 func newRollupCountNE(args []interface{}) (rollupFunc, error) {
-	return newRollupCountFilter(args, countFilterNE)
+	return newRollupFilter(args, countFilterNE)
 }
 
-func newRollupCountFilter(args []interface{}, countFilter func(values []float64, limit float64) int) (rollupFunc, error) {
+func newRollupSumEQ(args []interface{}) (rollupFunc, error) {
+	return newRollupFilter(args, sumFilterEQ)
+}
+
+func newRollupSumLE(args []interface{}) (rollupFunc, error) {
+	return newRollupFilter(args, sumFilterLE)
+}
+
+func newRollupSumGT(args []interface{}) (rollupFunc, error) {
+	return newRollupFilter(args, sumFilterGT)
+}
+
+func newRollupFilter(args []interface{}, f func(values []float64, limit float64) float64) (rollupFunc, error) {
 	if err := expectRollupArgsNum(args, 2); err != nil {
 		return nil, err
 	}
@@ -1183,7 +1228,7 @@ func newRollupCountFilter(args []interface{}, countFilter func(values []float64,
 			return nan
 		}
 		limit := limits[rfa.idx]
-		return float64(countFilter(values, limit))
+		return f(values, limit)
 	}
 	return rf, nil
 }

app/vmselect/promql/rollup_test.go

@@ -407,6 +407,75 @@ func TestRollupCountNEOverTime(t *testing.T) {
 	f(12, 11)
 }
 
+func TestRollupSumLEOverTime(t *testing.T) {
+	f := func(le, vExpected float64) {
+		t.Helper()
+		les := []*timeseries{{
+			Values:     []float64{le},
+			Timestamps: []int64{123},
+		}}
+		var me metricsql.MetricExpr
+		args := []interface{}{&metricsql.RollupExpr{Expr: &me}, les}
+		testRollupFunc(t, "sum_le_over_time", args, vExpected)
+	}
+
+	f(-123, 0)
+	f(0, 0)
+	f(10, 0)
+	f(12, 12)
+	f(30, 33)
+	f(50, 289)
+	f(100, 442)
+	f(123, 565)
+	f(1000, 565)
+}
+
+func TestRollupSumGTOverTime(t *testing.T) {
+	f := func(le, vExpected float64) {
+		t.Helper()
+		les := []*timeseries{{
+			Values:     []float64{le},
+			Timestamps: []int64{123},
+		}}
+		var me metricsql.MetricExpr
+		args := []interface{}{&metricsql.RollupExpr{Expr: &me}, les}
+		testRollupFunc(t, "sum_gt_over_time", args, vExpected)
+	}
+
+	f(-123, 565)
+	f(0, 565)
+	f(10, 565)
+	f(12, 553)
+	f(30, 532)
+	f(50, 276)
+	f(100, 123)
+	f(123, 0)
+	f(1000, 0)
+}
+
+func TestRollupSumEQOverTime(t *testing.T) {
+	f := func(le, vExpected float64) {
+		t.Helper()
+		les := []*timeseries{{
+			Values:     []float64{le},
+			Timestamps: []int64{123},
+		}}
+		var me metricsql.MetricExpr
+		args := []interface{}{&metricsql.RollupExpr{Expr: &me}, les}
+		testRollupFunc(t, "sum_eq_over_time", args, vExpected)
+	}
+
+	f(-123, 0)
+	f(0, 0)
+	f(10, 0)
+	f(12, 12)
+	f(30, 0)
+	f(50, 0)
+	f(100, 0)
+	f(123, 123)
+	f(1000, 0)
+}
+
 func TestRollupQuantileOverTime(t *testing.T) {
 	f := func(phi, vExpected float64) {
 		t.Helper()

docs/CHANGELOG.md

@@ -41,6 +41,7 @@ The sandbox cluster installation is running under the constant load generated by
 * FEATURE: [VictoriaMetrics cluster](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html): add `-disableReroutingOnUnavailable` command-line flag to `vminsert`, which can be used for reducing resource usage spikes at `vmstorage` nodes during rolling restart. See [these docs](https://docs.victoriametrics.com/cluster-victoriametrics/#improving-re-routing-performance-during-restart). Thanks to @Muxa1L for [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5713).
 * FEATURE: add `-search.resetRollupResultCacheOnStartup` command-line flag for resetting [query cache](https://docs.victoriametrics.com/#rollup-result-cache) on startup. See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/834).
 * FEATURE: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): propagate label filters across [label_set](https://docs.victoriametrics.com/MetricsQL.html#label_set) and [alias](https://docs.victoriametrics.com/MetricsQL.html#alias) functions. For example, `label_set(q1, "a", "b") + q2{c="d"}` is automatically transformed to `label_set(q1{c="d"}, "a", "b") + q2{a="b",c="d"}` now. This should improve performance for such queries. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1827#issuecomment-1654095358).
+* FEATURE: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): add [sum_eq_over_time](https://docs.victoriametrics.com/MetricsQL.html#sum_eq_over_time), [sum_gt_over_time](https://docs.victoriametrics.com/MetricsQL.html#sum_gt_over_time) and [sum_le_over_time](https://docs.victoriametrics.com/MetricsQL.html#sum_le_over_time) functions. See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4641).
 * FEATURE: [dashboards/vmagent](https://grafana.com/grafana/dashboards/12683): add `Targets scraped/s` stat panel showing the number of targets scraped by the vmagent per-second.
 * FEATURE: [dashboards/all](https://grafana.com/orgs/victoriametrics): add new panel `CPU spent on GC`. It should help identifying cases when too much CPU is spent on garbage collection, and advice users on how this can be addressed.
 * FEATURE: [vmalert](https://docs.victoriametrics.com/#vmalert): support [filtering](https://prometheus.io/docs/prometheus/2.49/querying/api/#rules) for `/api/v1/rules` API. See [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5749) by @victoramsantos.

docs/MetricsQL.md

@@ -243,7 +243,7 @@ from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.ht
 
 Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
 
-See also [count_over_time](#count_over_time).
+See also [count_over_time](#count_over_time) and [share_eq_over_time](#share_eq_over_time).
 
 #### count_gt_over_time
 
@@ -253,7 +253,7 @@ from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.ht
 
 Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
 
-See also [count_over_time](#count_over_time).
+See also [count_over_time](#count_over_time) and [share_gt_over_time](#share_gt_over_time).
 
 #### count_le_over_time
 
@@ -263,7 +263,7 @@ from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.ht
 
 Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
 
-See also [count_over_time](#count_over_time).
+See also [count_over_time](#count_over_time) and [share_le_over_time](#share_le_over_time).
 
 #### count_ne_over_time
 
@@ -743,7 +743,7 @@ This function is useful for calculating SLI and SLO. Example: `share_gt_over_tim
 
 Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
 
-See also [share_le_over_time](#share_le_over_time).
+See also [share_le_over_time](#share_le_over_time) and [count_gt_over_time](#count_gt_over_time).
 
 #### share_le_over_time
 
@@ -756,7 +756,7 @@ the share of time series values for the last 24 hours when memory usage was belo
 
 Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
 
-See also [share_gt_over_time](#share_gt_over_time).
+See also [share_gt_over_time](#share_gt_over_time) and [count_le_over_time](#count_le_over_time).
 
 #### share_eq_over_time
 
@@ -766,6 +766,8 @@ from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.ht
 
 Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
 
+See also [count_eq_over_time](#count_eq_over_time).
+
 #### stale_samples_over_time
 
 `stale_samples_over_time(series_selector[d])` is a [rollup function](#rollup-functions), which calculates the number
@@ -792,6 +794,33 @@ Metric names are stripped from the resulting rollups. Add [keep_metric_names](#k
 
 This function is supported by PromQL. See also [stddev_over_time](#stddev_over_time).
 
+#### sum_eq_over_time
+
+`sum_eq_over_time(series_selector[d], eq)` is a [rollup function](#rollup-function), which calculates the sum of raw sample values equal to `eq`
+on the given lookbehind window `d` per each time series returned from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.html#filtering).
+
+Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
+
+See also [sum_over_time](#sum_over_time) and [count_eq_over_time](#count_eq_over_time).
+
+#### sum_gt_over_time
+
+`sum_gt_over_time(series_selector[d], gt)` is a [rollup function](#rollup-function), which calculates the sum of raw sample values bigger than `gt`
+on the given lookbehind window `d` per each time series returned from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.html#filtering).
+
+Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
+
+See also [sum_over_time](#sum_over_time) and [count_gt_over_time](#count_gt_over_time).
+
+#### sum_le_over_time
+
+`sum_le_over_time(series_selector[d], le)` is a [rollup function](#rollup-function), which calculates the sum of raw sample values smaller or equal to `le`
+on the given lookbehind window `d` per each time series returned from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.html#filtering).
+
+Metric names are stripped from the resulting rollups. Add [keep_metric_names](#keep_metric_names) modifier in order to keep metric names.
+
+See also [sum_over_time](#sum_over_time) and [count_le_over_time](#count_le_over_time).
+
 #### sum_over_time
 
 `sum_over_time(series_selector[d])` is a [rollup function](#rollup-functions), which calculates the sum of raw sample values

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/VictoriaMetrics/easyproto v0.1.4
 	github.com/VictoriaMetrics/fastcache v1.12.2
 	github.com/VictoriaMetrics/metrics v1.31.0
-	github.com/VictoriaMetrics/metricsql v0.71.0
+	github.com/VictoriaMetrics/metricsql v0.72.1
 	github.com/aws/aws-sdk-go-v2 v1.24.1
 	github.com/aws/aws-sdk-go-v2/config v1.26.6
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.15.15

go.sum

@@ -71,8 +71,8 @@ github.com/VictoriaMetrics/fastcache v1.12.2/go.mod h1:AmC+Nzz1+3G2eCPapF6UcsnkT
 github.com/VictoriaMetrics/metrics v1.24.0/go.mod h1:eFT25kvsTidQFHb6U0oa0rTrDRdz4xTYjpL8+UPohys=
 github.com/VictoriaMetrics/metrics v1.31.0 h1:X6+nBvAP0UB+GjR0Ht9hhQ3pjL1AN4b8dt9zFfzTsUo=
 github.com/VictoriaMetrics/metrics v1.31.0/go.mod h1:r7hveu6xMdUACXvB8TYdAj8WEsKzWB0EkpJN+RDtOf8=
-github.com/VictoriaMetrics/metricsql v0.71.0 h1:i6uJPTPY2CTGvuWlxD0+l9HrhPHFbRPqvEopp6YzCSI=
+github.com/VictoriaMetrics/metricsql v0.72.1 h1:fLIHgzezXgD4NjY5ksF4lRkHILW88uI5Lz0Q+N2ucnY=
-github.com/VictoriaMetrics/metricsql v0.71.0/go.mod h1:k4UaP/+CjuZslIjd+kCigNG9TQmUqh5v0TP/nMEy90I=
+github.com/VictoriaMetrics/metricsql v0.72.1/go.mod h1:k4UaP/+CjuZslIjd+kCigNG9TQmUqh5v0TP/nMEy90I=
 github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1ow=
 github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=

vendor/github.com/VictoriaMetrics/metricsql/rollup.go

@@ -70,6 +70,9 @@ var rollupFuncs = map[string]bool{
 	"stale_samples_over_time": true,
 	"stddev_over_time":        true,
 	"stdvar_over_time":        true,
+	"sum_eq_over_time":        true,
+	"sum_gt_over_time":        true,
+	"sum_le_over_time":        true,
 	"sum_over_time":           true,
 	"sum2_over_time":          true,
 	"tfirst_over_time":        true,

vendor/modules.txt

@@ -102,7 +102,7 @@ github.com/VictoriaMetrics/fastcache
 # github.com/VictoriaMetrics/metrics v1.31.0
 ## explicit; go 1.17
 github.com/VictoriaMetrics/metrics
-# github.com/VictoriaMetrics/metricsql v0.71.0
+# github.com/VictoriaMetrics/metricsql v0.72.1
 ## explicit; go 1.13
 github.com/VictoriaMetrics/metricsql
 github.com/VictoriaMetrics/metricsql/binaryop