From 0f2ecde0e60c22cffad4af66a6ea1a5ffcefa78b Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@gmail.com>
Date: Sat, 20 Jul 2019 14:02:51 +0300
Subject: [PATCH] lib/encoding: improve gauge series detection

- Series with negative values are always gauges
- Counters may only have increasing values with possible counter resets

This should improve compression ratio for gauge series which
were previously mistakenly detected as counters.
---
 lib/encoding/encoding.go      | 50 +++++++++++++++++++----------------
 lib/encoding/encoding_test.go | 42 +++++++++++++++--------------
 2 files changed, 49 insertions(+), 43 deletions(-)

diff --git a/lib/encoding/encoding.go b/lib/encoding/encoding.go
index 3b44c8981..e93baaabe 100644
--- a/lib/encoding/encoding.go
+++ b/lib/encoding/encoding.go
@@ -296,7 +296,7 @@ func isDeltaConst(a []int64) bool {
 // i.e. arbitrary changing values.
 //
 // It is OK if a few gauges aren't detected (i.e. detected as counters),
-// since misdetected counters as gauges are much worse condition.
+// since misdetected counters as gauges leads to worser compression ratio.
 func isGauge(a []int64) bool {
 	// Check all the items in a, since a part of items may lead
 	// to incorrect gauge detection.
@@ -305,32 +305,36 @@ func isGauge(a []int64) bool {
 		return false
 	}
 
-	extremes := 0
-	plus := a[0] <= a[1]
-	v1 := a[1]
-	for _, v2 := range a[2:] {
-		if plus {
-			if v2 < v1 {
-				extremes++
-				plus = false
-			}
-		} else {
-			if v2 > v1 {
-				extremes++
-				plus = true
-			}
-		}
-		v1 = v2
+	resets := 0
+	vPrev := a[0]
+	if vPrev < 0 {
+		// Counter values cannot be negative.
+		return true
 	}
-	if extremes <= 2 {
-		// Probably counter reset.
+	for _, v := range a[1:] {
+		if v < vPrev {
+			if v < 0 {
+				// Counter values cannot be negative.
+				return true
+			}
+			if v > (vPrev >> 3) {
+				// Decreasing sequence detected.
+				// This is a gauge.
+				return true
+			}
+			// Possible counter reset.
+			resets++
+		}
+		vPrev = v
+	}
+	if resets <= 2 {
+		// Counter with a few resets.
 		return false
 	}
 
-	// A few extremes may indicate counter resets.
-	// Let it be a gauge if extremes exceed len(a)/32,
-	// otherwise assume counter reset.
-	return extremes > (len(a) >> 5)
+	// Let it be a gauge if resets exceeds len(a)/8,
+	// otherwise assume counter.
+	return resets > (len(a) >> 3)
 }
 
 func getCompressLevel(itemsCount int) int {
diff --git a/lib/encoding/encoding_test.go b/lib/encoding/encoding_test.go
index 836ed8b85..9a76b27fa 100644
--- a/lib/encoding/encoding_test.go
+++ b/lib/encoding/encoding_test.go
@@ -43,27 +43,29 @@ func TestIsDeltaConst(t *testing.T) {
 }
 
 func TestIsGauge(t *testing.T) {
-	testIsGauge(t, []int64{}, false)
-	testIsGauge(t, []int64{0}, false)
-	testIsGauge(t, []int64{1, 2}, false)
-	testIsGauge(t, []int64{0, 1, 2, 3, 4, 5}, false)
-	testIsGauge(t, []int64{0, -1, -2, -3, -4}, false)
-	testIsGauge(t, []int64{0, 0, 0, 0, 0, 0, 0}, false)
-	testIsGauge(t, []int64{1, 1, 1, 1, 1}, false)
-	testIsGauge(t, []int64{1, 1, 2, 2, 2, 2}, false)
-	testIsGauge(t, []int64{1, 5, 2, 3}, false) // a single counter reset
-	testIsGauge(t, []int64{1, 5, 2, 3, 2}, true)
-	testIsGauge(t, []int64{-1, -5, -2, -3}, false) // a single counter reset
-	testIsGauge(t, []int64{-1, -5, -2, -3, -2}, true)
-}
-
-func testIsGauge(t *testing.T, a []int64, okExpected bool) {
-	t.Helper()
-
-	ok := isGauge(a)
-	if ok != okExpected {
-		t.Fatalf("unexpected result for isGauge(%d); got %v; expecting %v", a, ok, okExpected)
+	f := func(a []int64, okExpected bool) {
+		t.Helper()
+		ok := isGauge(a)
+		if ok != okExpected {
+			t.Fatalf("unexpected result for isGauge(%d); got %v; expecting %v", a, ok, okExpected)
+		}
 	}
+	f([]int64{}, false)
+	f([]int64{0}, false)
+	f([]int64{1, 2}, false)
+	f([]int64{0, 1, 2, 3, 4, 5}, false)
+	f([]int64{0, -1, -2, -3, -4}, true)
+	f([]int64{0, 0, 0, 0, 0, 0, 0}, false)
+	f([]int64{1, 1, 1, 1, 1}, false)
+	f([]int64{1, 1, 2, 2, 2, 2}, false)
+	f([]int64{1, 17, 2, 3}, false) // a single counter reset
+	f([]int64{1, 5, 2, 3}, true)
+	f([]int64{1, 5, 2, 3, 2}, true)
+	f([]int64{-1, -5, -2, -3}, true)
+	f([]int64{-1, -5, -2, -3, -2}, true)
+	f([]int64{5, 6, 4, 3, 2}, true)
+	f([]int64{4, 5, 6, 5, 4, 3, 2}, true)
+	f([]int64{1064, 1132, 1083, 1062, 856, 747}, true)
 }
 
 func TestEnsureNonDecreasingSequence(t *testing.T) {