From 8f42e59e052e2787d43bab03e4141af9e990b7a3 Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@gmail.com>
Date: Fri, 13 Nov 2020 15:21:51 +0200
Subject: [PATCH] app/vmselect/promql: remove spikes from `increase()` and
 `delta()` results on time series with spare irregular data points

Do not take into account spare data point value if the next point will is located too far from the current point.

Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/894
---
 app/vmselect/promql/rollup.go      | 12 ++++++++-
 app/vmselect/promql/rollup_test.go | 43 +++++++++++++++++++-----------
 2 files changed, 38 insertions(+), 17 deletions(-)

diff --git a/app/vmselect/promql/rollup.go b/app/vmselect/promql/rollup.go
index 247743ddc5..c711f04ad8 100644
--- a/app/vmselect/promql/rollup.go
+++ b/app/vmselect/promql/rollup.go
@@ -338,9 +338,12 @@ type rollupFuncArg struct {
 	// Timestamps for values.
 	timestamps []int64
 
-	// Actual value preceeding values without restrictions on staleness interval.
+	// Real value preceeding values without restrictions on staleness interval.
 	realPrevValue float64
 
+	// Real value which goes after values.
+	realNextValue float64
+
 	// Current timestamp for rollup evaluation.
 	currTimestamp int64
 
@@ -558,6 +561,11 @@ func (rc *rollupConfig) doInternal(dstValues []float64, tsm *timeseriesMap, valu
 		} else {
 			rfa.realPrevValue = nan
 		}
+		if j < len(values) {
+			rfa.realNextValue = values[j]
+		} else {
+			rfa.realNextValue = nan
+		}
 		rfa.currTimestamp = tEnd
 		value := rc.Func(rfa)
 		rfa.idx++
@@ -1282,6 +1290,8 @@ func rollupDelta(rfa *rollupFuncArg) float64 {
 		d := float64(10)
 		if len(values) > 1 {
 			d = values[1] - values[0]
+		} else if !math.IsNaN(rfa.realNextValue) {
+			d = rfa.realNextValue - values[0]
 		}
 		if math.Abs(values[0]) < 10*(math.Abs(d)+1) {
 			prevValue = 0
diff --git a/app/vmselect/promql/rollup_test.go b/app/vmselect/promql/rollup_test.go
index e0eaacb6cd..dff89e2c8e 100644
--- a/app/vmselect/promql/rollup_test.go
+++ b/app/vmselect/promql/rollup_test.go
@@ -1103,12 +1103,13 @@ func testRowsEqual(t *testing.T, values []float64, timestamps []int64, valuesExp
 }
 
 func TestRollupDelta(t *testing.T) {
-	f := func(prevValue, realPrevValue float64, values []float64, resultExpected float64) {
+	f := func(prevValue, realPrevValue, realNextValue float64, values []float64, resultExpected float64) {
 		t.Helper()
 		rfa := &rollupFuncArg{
 			prevValue:     prevValue,
 			values:        values,
 			realPrevValue: realPrevValue,
+			realNextValue: realNextValue,
 		}
 		result := rollupDelta(rfa)
 		if math.IsNaN(result) {
@@ -1121,26 +1122,36 @@ func TestRollupDelta(t *testing.T) {
 			t.Fatalf("unexpected result; got %v; want %v", result, resultExpected)
 		}
 	}
-	f(nan, nan, nil, nan)
+	f(nan, nan, nan, nil, nan)
 
 	// Small initial value
-	f(nan, nan, []float64{1}, 1)
-	f(nan, nan, []float64{10}, 10)
-	f(nan, nan, []float64{100}, 100)
-	f(nan, nan, []float64{1, 2, 3}, 3)
-	f(1, nan, []float64{1, 2, 3}, 2)
-	f(nan, nan, []float64{5, 6, 8}, 8)
-	f(2, nan, []float64{5, 6, 8}, 6)
+	f(nan, nan, nan, []float64{1}, 1)
+	f(nan, nan, nan, []float64{10}, 10)
+	f(nan, nan, nan, []float64{100}, 100)
+	f(nan, nan, nan, []float64{1, 2, 3}, 3)
+	f(1, nan, nan, []float64{1, 2, 3}, 2)
+	f(nan, nan, nan, []float64{5, 6, 8}, 8)
+	f(2, nan, nan, []float64{5, 6, 8}, 6)
 
 	// Too big initial value must be skipped.
-	f(nan, nan, []float64{1000}, 0)
-	f(nan, nan, []float64{1000, 1001, 1002}, 2)
+	f(nan, nan, nan, []float64{1000}, 0)
+	f(nan, nan, nan, []float64{1000, 1001, 1002}, 2)
 
-	// Delta calculations against non-nan realPrevValue
-	f(nan, 900, []float64{1000}, 100)
-	f(nan, 900, []float64{1000, 1001, 1002}, 102)
+	// Non-nan realPrevValue
+	f(nan, 900, nan, []float64{1000}, 100)
+	f(nan, 1000, nan, []float64{1000}, 0)
+	f(nan, 1100, nan, []float64{1000}, -100)
+	f(nan, 900, nan, []float64{1000, 1001, 1002}, 102)
+
+	// Small delta between realNextValue and values
+	f(nan, nan, 990, []float64{1000}, 0)
+	f(nan, nan, 1005, []float64{1000}, 0)
+
+	// Big delta between relaNextValue and values
+	f(nan, nan, 800, []float64{1000}, 1000)
+	f(nan, nan, 1300, []float64{1000}, 1000)
 
 	// Empty values
-	f(1, nan, nil, 0)
-	f(100, nan, nil, 0)
+	f(1, nan, nan, nil, 0)
+	f(100, nan, nan, nil, 0)
 }