From 0755cb3b506e81c10eaeb63a8d1921c21c79cf0b Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@gmail.com>
Date: Tue, 21 Jul 2020 17:22:36 +0300
Subject: [PATCH] app/vmselect/promql: skip the first value in time series
 passed to `increase()` if it exceeds by more than 10x the delta between the
 next value and the first value

This should prvent from inflated `increase()` results for time series that start from big initial values.
Such cases may occur when a label value changes in a metric without counter reset.
---
 app/vmselect/promql/rollup.go      | 12 +++++++--
 app/vmselect/promql/rollup_test.go | 39 ++++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/app/vmselect/promql/rollup.go b/app/vmselect/promql/rollup.go
index dda5d02b8..1cf6820f4 100644
--- a/app/vmselect/promql/rollup.go
+++ b/app/vmselect/promql/rollup.go
@@ -1187,11 +1187,19 @@ func rollupDeltaInternal(rfa *rollupFuncArg, canUseRealPrevValue bool) float64 {
 			return nan
 		}
 		// Assume that the previous non-existing value was 0
-		// only if the first value is quite small.
+		// only if the first value doesn't exceed too much the delta with the next value.
+		//
 		// This should prevent from improper increase() results for os-level counters
 		// such as cpu time or bytes sent over the network interface.
 		// These counters may start long ago before the first value appears in the db.
-		if values[0] < 1e6 {
+		//
+		// This also should prevent from improper increase() results when a part of label values are changed
+		// without counter reset.
+		d := float64(10)
+		if len(values) > 1 {
+			d = values[1] - values[0]
+		}
+		if math.Abs(values[0]) < 10*(math.Abs(d)+1) {
 			prevValue = 0
 			if canUseRealPrevValue && !math.IsNaN(rfa.realPrevValue) {
 				prevValue = rfa.realPrevValue
diff --git a/app/vmselect/promql/rollup_test.go b/app/vmselect/promql/rollup_test.go
index 1e7c42d71..65aeee9d1 100644
--- a/app/vmselect/promql/rollup_test.go
+++ b/app/vmselect/promql/rollup_test.go
@@ -1024,3 +1024,42 @@ func testRowsEqual(t *testing.T, values []float64, timestamps []int64, valuesExp
 		}
 	}
 }
+
+func TestRollupIncrease(t *testing.T) {
+	f := func(prevValue float64, values []float64, resultExpected float64) {
+		t.Helper()
+		rfa := &rollupFuncArg{
+			prevValue:     prevValue,
+			realPrevValue: prevValue,
+			values:        values,
+		}
+		result := rollupIncrease(rfa)
+		if math.IsNaN(result) {
+			if !math.IsNaN(resultExpected) {
+				t.Fatalf("unexpected result; got %v; want %v", result, resultExpected)
+			}
+			return
+		}
+		if result != resultExpected {
+			t.Fatalf("unexpected result; got %v; want %v", result, resultExpected)
+		}
+	}
+	f(nan, nil, nan)
+
+	// Small initial value
+	f(nan, []float64{1}, 1)
+	f(nan, []float64{10}, 10)
+	f(nan, []float64{100}, 100)
+	f(nan, []float64{1, 2, 3}, 3)
+	f(1, []float64{1, 2, 3}, 2)
+	f(nan, []float64{5, 6, 8}, 8)
+	f(2, []float64{5, 6, 8}, 6)
+
+	// Too big initial value must be skipped.
+	f(nan, []float64{1000}, 0)
+	f(nan, []float64{1000, 1001, 1002}, 2)
+
+	// Empty values
+	f(1, nil, 0)
+	f(100, nil, 0)
+}