tsdb: Add counter reset hint to histograms and support in WAL

Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>

model/histogram/float_histogram.go

@@ -27,6 +27,8 @@ import (
 // used to represent a histogram with integer counts and thus serves as a more
 // generalized representation.
 type FloatHistogram struct {
+	// Counter reset information.
+	CounterResetHint CounterResetHint
 	// Currently valid schema numbers are -4 <= n <= 8.  They are all for
 	// base-2 bucket schemas, where 1 is a bucket boundary in each case, and
 	// then each power of two is divided into 2^n logarithmic buckets.  Or

model/histogram/histogram.go

@@ -19,6 +19,17 @@ import (
 	"strings"
 )
 
+// CounterResetHint contains the known information about a counter reset,
+// or alternatively that we are dealing with a gauge histogram, where counter resets do not apply.
+type CounterResetHint byte
+
+const (
+	UnknownCounterReset CounterResetHint = iota // UnknownCounterReset means we cannot say if this histogram signals a counter reset or not.
+	CounterReset                                // CounterReset means there was definitely a counter reset starting from this histogram.
+	NotCounterReset                             // NotCounterReset means there was definitely no counter reset with this histogram.
+	GaugeType                                   // GaugeType means this is a gauge histogram, where counter resets do not happen.
+)
+
 // Histogram encodes a sparse, high-resolution histogram. See the design
 // document for full details:
 // https://docs.google.com/document/d/1cLNv3aufPZb3fNfaJgdaRBZsInZKKIHo9E6HinJVbpM/edit#
@@ -35,6 +46,8 @@ import (
 //
 // Which bucket indices are actually used is determined by the spans.
 type Histogram struct {
+	// Counter reset information.
+	CounterResetHint CounterResetHint
 	// Currently valid schema numbers are -4 <= n <= 8.  They are all for
 	// base-2 bucket schemas, where 1 is a bucket boundary in each case, and
 	// then each power of two is divided into 2^n logarithmic buckets.  Or
@@ -295,6 +308,7 @@ func (h *Histogram) ToFloat() *FloatHistogram {
 	}
 
 	return &FloatHistogram{
+		CounterResetHint: h.CounterResetHint,
 		Schema:           h.Schema,
 		ZeroThreshold:    h.ZeroThreshold,
 		ZeroCount:        float64(h.ZeroCount),

tsdb/record/record.go

@@ -441,6 +441,8 @@ func (d *Decoder) HistogramSamples(rec []byte, histograms []RefHistogramSample)
 			H:   &histogram.Histogram{},
 		}
 
+		rh.H.CounterResetHint = histogram.CounterResetHint(dec.Byte())
+
 		rh.H.Schema = int32(dec.Varint64())
 		rh.H.ZeroThreshold = math.Float64frombits(dec.Be64())
 
@@ -517,6 +519,8 @@ func (d *Decoder) FloatHistogramSamples(rec []byte, histograms []RefFloatHistogr
 			FH:  &histogram.FloatHistogram{},
 		}
 
+		rh.FH.CounterResetHint = histogram.CounterResetHint(dec.Byte())
+
 		rh.FH.Schema = int32(dec.Varint64())
 		rh.FH.ZeroThreshold = dec.Be64Float64()
 
@@ -715,6 +719,8 @@ func (e *Encoder) HistogramSamples(histograms []RefHistogramSample, b []byte) []
 		buf.PutVarint64(int64(h.Ref) - int64(first.Ref))
 		buf.PutVarint64(h.T - first.T)
 
+		buf.PutByte(byte(h.H.CounterResetHint))
+
 		buf.PutVarint64(int64(h.H.Schema))
 		buf.PutBE64(math.Float64bits(h.H.ZeroThreshold))
 
@@ -766,6 +772,8 @@ func (e *Encoder) FloatHistogramSamples(histograms []RefFloatHistogramSample, b
 		buf.PutVarint64(int64(h.Ref) - int64(first.Ref))
 		buf.PutVarint64(h.T - first.T)
 
+		buf.PutByte(byte(h.FH.CounterResetHint))
+
 		buf.PutVarint64(int64(h.FH.Schema))
 		buf.PutBEFloat64(h.FH.ZeroThreshold)
 

tsdb/record/record_test.go

@@ -165,6 +165,22 @@ func TestRecord_EncodeDecode(t *testing.T) {
 	decFloatHistograms, err := dec.FloatHistogramSamples(enc.FloatHistogramSamples(floatHistograms, nil), nil)
 	require.NoError(t, err)
 	require.Equal(t, floatHistograms, decFloatHistograms)
+
+	// Gauge ingeger histograms.
+	for i := range histograms {
+		histograms[i].H.CounterResetHint = histogram.GaugeType
+	}
+	decHistograms, err = dec.HistogramSamples(enc.HistogramSamples(histograms, nil), nil)
+	require.NoError(t, err)
+	require.Equal(t, histograms, decHistograms)
+
+	// Gauge float histograms.
+	for i := range floatHistograms {
+		floatHistograms[i].FH.CounterResetHint = histogram.GaugeType
+	}
+	decFloatHistograms, err = dec.FloatHistogramSamples(enc.FloatHistogramSamples(floatHistograms, nil), nil)
+	require.NoError(t, err)
+	require.Equal(t, floatHistograms, decFloatHistograms)
 }
 
 // TestRecord_Corrupted ensures that corrupted records return the correct error.