From d9d01f976bc51a44bda46e16c53a22b3e08dec9a 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 247743ddc..c711f04ad 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 e0eaacb6c..dff89e2c8 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)
 }