mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2024-12-25 05:34:05 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Fix more counterResetInAnyBucket edgecases

Browse source 
Case a) empty span is at the beginning of the spans.
Case b) two consequtive empty spans with positive offsets.

Signed-off-by: György Krajcsovits <gyorgy.krajcsovits@grafana.com>
This commit is contained in:
György Krajcsovits 2023-09-13 12:59:42 +02:00
parent 4419399e4e
commit b2fa4d910a
5 changed files with 386 additions and 113 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
tsdb/chunkenc/float_histogram.go
View file

@ -339,11 +339,16 @@ func counterResetInAnyFloatBucket(oldBuckets []xorValue, newBuckets []float64, o
return false
}
oldSpanSliceIdx, newSpanSliceIdx := 0, 0 // Index for the span slices.
oldInsideSpanIdx, newInsideSpanIdx := uint32(0), uint32(0) // Index inside a span.
oldIdx, newIdx := oldSpans[0].Offset, newSpans[0].Offset
var (
oldSpanSliceIdx, newSpanSliceIdx int = -1, -1 // Index for the span slices. Starts at -1 to indicate that the first non empty span is not yet found.
oldInsideSpanIdx, newInsideSpanIdx uint32 // Index inside a span.
oldIdx, newIdx int32 // Index inside a bucket slice.
oldBucketSliceIdx, newBucketSliceIdx int // Index inside bucket slice.
)
oldBucketSliceIdx, newBucketSliceIdx := 0, 0 // Index inside bucket slice.
// Find first non empty spans.
oldSpanSliceIdx, oldIdx = nextNonEmptySpanSliceIdx(oldSpanSliceIdx, oldIdx, oldSpans)
newSpanSliceIdx, newIdx = nextNonEmptySpanSliceIdx(newSpanSliceIdx, newIdx, newSpans)
oldVal, newVal := oldBuckets[0].value, newBuckets[0]
// Since we assume that new spans won't have missing buckets, there will never be a case
@ -359,13 +364,12 @@ func counterResetInAnyFloatBucket(oldBuckets []xorValue, newBuckets []float64, o
// Moving ahead old bucket and span by 1 index.
if oldInsideSpanIdx+1 >= oldSpans[oldSpanSliceIdx].Length {
// Current span is over.
oldSpanSliceIdx = nextNonEmptySpanSliceIdx(oldSpanSliceIdx, oldSpans)
oldSpanSliceIdx, oldIdx = nextNonEmptySpanSliceIdx(oldSpanSliceIdx, oldIdx, oldSpans)
oldInsideSpanIdx = 0
if oldSpanSliceIdx >= len(oldSpans) {
// All old spans are over.
break
}
oldIdx += 1 + oldSpans[oldSpanSliceIdx].Offset
} else {
oldInsideSpanIdx++
oldIdx++
@ -378,14 +382,13 @@ func counterResetInAnyFloatBucket(oldBuckets []xorValue, newBuckets []float64, o
// Moving ahead new bucket and span by 1 index.
if newInsideSpanIdx+1 >= newSpans[newSpanSliceIdx].Length {
// Current span is over.
newSpanSliceIdx = nextNonEmptySpanSliceIdx(newSpanSliceIdx, newSpans)
newSpanSliceIdx, newIdx = nextNonEmptySpanSliceIdx(newSpanSliceIdx, newIdx, newSpans)
newInsideSpanIdx = 0
if newSpanSliceIdx >= len(newSpans) {
// All new spans are over.
// This should not happen, old spans above should catch this first.
panic("new spans over before old spans in counterReset")
}
newIdx += 1 + newSpans[newSpanSliceIdx].Offset
} else {
newInsideSpanIdx++
newIdx++

225
tsdb/chunkenc/float_histogram_test.go
View file

@ -550,62 +550,193 @@ func assertRecodedFloatHistogramChunkOnAppend(t *testing.T, prevChunk Chunk, hAp
}
func TestFloatHistogramChunkAppendableWithEmptySpan(t *testing.T) {
h1 := &histogram.FloatHistogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
tests := map[string]struct {
h1 *histogram.FloatHistogram
h2 *histogram.FloatHistogram
}{
"empty span in old and new histogram": {
h1: &histogram.FloatHistogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
},
PositiveBuckets: []float64{1, 2, 1, 1, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 2, 1, 2, 2, 2, 2},
},
h2: &histogram.FloatHistogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
},
PositiveBuckets: []float64{1, 3, 1, 2, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 4, 2, 7, 5, 5, 2},
},
},
PositiveBuckets: []float64{1, 2, 1, 1, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
"empty span in old histogram": {
h1: &histogram.FloatHistogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 1, Length: 0}, // This span will disappear.
{Offset: 2, Length: 4},
{Offset: 0, Length: 3},
},
PositiveBuckets: []float64{1, 2, 1, 1, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 2, 1, 2, 2, 2, 2},
},
h2: &histogram.FloatHistogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 3, Length: 4},
{Offset: 0, Length: 3},
},
PositiveBuckets: []float64{1, 3, 1, 2, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 4, 2, 7, 5, 5, 2},
},
},
NegativeBuckets: []float64{1, 2, 1, 2, 2, 2, 2},
}
h2 := &histogram.FloatHistogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
"empty span in new histogram": {
h1: &histogram.FloatHistogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 3, Length: 3},
},
PositiveBuckets: []float64{1, 2, 1, 1, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 2, 1, 2, 2, 2, 2},
},
h2: &histogram.FloatHistogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 1, Length: 0}, // This span is new.
{Offset: 2, Length: 3},
},
PositiveBuckets: []float64{1, 3, 1, 2, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 4, 2, 7, 5, 5, 2},
},
},
PositiveBuckets: []float64{1, 3, 1, 2, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
"two empty spans mixing offsets": {
h1: &histogram.FloatHistogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 1, Length: 0},
{Offset: 3, Length: 0},
{Offset: 4, Length: 3},
},
PositiveBuckets: []float64{1, 2, 1, 1, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 2, 1, 2, 2, 2, 2},
},
h2: &histogram.FloatHistogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 3, Length: 0},
{Offset: 1, Length: 0},
{Offset: 4, Length: 3},
},
PositiveBuckets: []float64{1, 3, 1, 2, 1, 1, 1},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []float64{1, 4, 2, 7, 5, 5, 2},
},
},
NegativeBuckets: []float64{1, 4, 2, 7, 5, 5, 2},
}
c := Chunk(NewFloatHistogramChunk())
for name, tc := range tests {
t.Run(name, func(t *testing.T) {
c := Chunk(NewFloatHistogramChunk())
// Create fresh appender and add the first histogram.
app, err := c.Appender()
require.NoError(t, err)
require.Equal(t, 0, c.NumSamples())
// Create fresh appender and add the first histogram.
app, err := c.Appender()
require.NoError(t, err)
require.Equal(t, 0, c.NumSamples())
_, _, _, err = app.AppendFloatHistogram(nil, 1, h1, true)
require.NoError(t, err)
require.Equal(t, 1, c.NumSamples())
hApp, _ := app.(*FloatHistogramAppender)
_, _, _, err = app.AppendFloatHistogram(nil, 1, tc.h1, true)
require.NoError(t, err)
require.Equal(t, 1, c.NumSamples())
hApp, _ := app.(*FloatHistogramAppender)
pI, nI, okToAppend, counterReset := hApp.appendable(h2)
require.Empty(t, pI)
require.Empty(t, nI)
require.True(t, okToAppend)
require.False(t, counterReset)
pI, nI, okToAppend, counterReset := hApp.appendable(tc.h2)
require.Empty(t, pI)
require.Empty(t, nI)
require.True(t, okToAppend)
require.False(t, counterReset)
})
}
}
func TestFloatHistogramChunkAppendableGauge(t *testing.T) {

19
tsdb/chunkenc/histogram.go
View file

@ -359,11 +359,16 @@ func counterResetInAnyBucket(oldBuckets, newBuckets []int64, oldSpans, newSpans
return false
}
oldSpanSliceIdx, newSpanSliceIdx := 0, 0 // Index for the span slices.
oldInsideSpanIdx, newInsideSpanIdx := uint32(0), uint32(0) // Index inside a span.
oldIdx, newIdx := oldSpans[0].Offset, newSpans[0].Offset
var (
oldSpanSliceIdx, newSpanSliceIdx int = -1, -1 // Index for the span slices. Starts at -1 to indicate that the first non empty span is not yet found.
oldInsideSpanIdx, newInsideSpanIdx uint32 // Index inside a span.
oldIdx, newIdx int32 // Index inside a bucket slice.
oldBucketSliceIdx, newBucketSliceIdx int // Index inside bucket slice.
)
oldBucketSliceIdx, newBucketSliceIdx := 0, 0 // Index inside bucket slice.
// Find first non empty spans.
oldSpanSliceIdx, oldIdx = nextNonEmptySpanSliceIdx(oldSpanSliceIdx, oldIdx, oldSpans)
newSpanSliceIdx, newIdx = nextNonEmptySpanSliceIdx(newSpanSliceIdx, newIdx, newSpans)
oldVal, newVal := oldBuckets[0], newBuckets[0]
// Since we assume that new spans won't have missing buckets, there will never be a case
@ -379,13 +384,12 @@ func counterResetInAnyBucket(oldBuckets, newBuckets []int64, oldSpans, newSpans
// Moving ahead old bucket and span by 1 index.
if oldInsideSpanIdx+1 >= oldSpans[oldSpanSliceIdx].Length {
// Current span is over.
oldSpanSliceIdx = nextNonEmptySpanSliceIdx(oldSpanSliceIdx, oldSpans)
oldSpanSliceIdx, oldIdx = nextNonEmptySpanSliceIdx(oldSpanSliceIdx, oldIdx, oldSpans)
oldInsideSpanIdx = 0
if oldSpanSliceIdx >= len(oldSpans) {
// All old spans are over.
break
}
oldIdx += 1 + oldSpans[oldSpanSliceIdx].Offset
} else {
oldInsideSpanIdx++
oldIdx++
@ -398,14 +402,13 @@ func counterResetInAnyBucket(oldBuckets, newBuckets []int64, oldSpans, newSpans
// Moving ahead new bucket and span by 1 index.
if newInsideSpanIdx+1 >= newSpans[newSpanSliceIdx].Length {
// Current span is over.
newSpanSliceIdx = nextNonEmptySpanSliceIdx(newSpanSliceIdx, newSpans)
newSpanSliceIdx, newIdx = nextNonEmptySpanSliceIdx(newSpanSliceIdx, newIdx, newSpans)
newInsideSpanIdx = 0
if newSpanSliceIdx >= len(newSpans) {
// All new spans are over.
// This should not happen, old spans above should catch this first.
panic("new spans over before old spans in counterReset")
}
newIdx += 1 + newSpans[newSpanSliceIdx].Offset
} else {
newInsideSpanIdx++
newIdx++

11
tsdb/chunkenc/histogram_meta.go
View file

@ -489,8 +489,13 @@ func counterResetHint(crh CounterResetHeader, numRead uint16) histogram.CounterR
}
// Handle pathological case of empty span when advancing span idx.
func nextNonEmptySpanSliceIdx(idx int, spans []histogram.Span) (newIdx int) {
for idx++; idx < len(spans) && spans[idx].Length == 0; idx++ {
// Call it with idx==-1 to find the first non empty span.
func nextNonEmptySpanSliceIdx(idx int, bucketIdx int32, spans []histogram.Span) (newIdx int, newBucketIdx int32) {
for idx++; idx < len(spans); idx++ {
if spans[idx].Length > 0 {
return idx, bucketIdx + spans[idx].Offset + 1
}
bucketIdx += spans[idx].Offset
}
return idx
return idx, 0
}

225
tsdb/chunkenc/histogram_test.go
View file

@ -573,62 +573,193 @@ func assertSampleCount(t *testing.T, c Chunk, exp int64, vtype ValueType) {
}
func TestHistogramChunkAppendableWithEmptySpan(t *testing.T) {
h1 := &histogram.Histogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
tests := map[string]struct {
h1 *histogram.Histogram
h2 *histogram.Histogram
}{
"empty span in old and new histogram": {
h1: &histogram.Histogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
},
PositiveBuckets: []int64{1, 1, -1, 0, 0, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 1, -1, 1, 0, 0, 0},
},
h2: &histogram.Histogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
},
PositiveBuckets: []int64{1, 2, -2, 1, -1, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 3, -2, 5, -2, 0, -3},
},
},
PositiveBuckets: []int64{1, 1, -1, 0, 0, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
"empty span in old histogram": {
h1: &histogram.Histogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 1, Length: 0}, // This span will disappear.
{Offset: 2, Length: 4},
{Offset: 0, Length: 3},
},
PositiveBuckets: []int64{1, 1, -1, 0, 0, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 1, -1, 1, 0, 0, 0},
},
h2: &histogram.Histogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 3, Length: 4},
{Offset: 0, Length: 3},
},
PositiveBuckets: []int64{1, 2, -2, 1, -1, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 3, -2, 5, -2, 0, -3},
},
},
NegativeBuckets: []int64{1, 1, -1, 1, 0, 0, 0},
}
h2 := &histogram.Histogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 0, Length: 0},
{Offset: 0, Length: 3},
"empty span in new histogram": {
h1: &histogram.Histogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 3, Length: 3},
},
PositiveBuckets: []int64{1, 1, -1, 0, 0, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 1, -1, 1, 0, 0, 0},
},
h2: &histogram.Histogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 1, Length: 0}, // This span is new.
{Offset: 2, Length: 3},
},
PositiveBuckets: []int64{1, 2, -2, 1, -1, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 3, -2, 5, -2, 0, -3},
},
},
PositiveBuckets: []int64{1, 2, -2, 1, -1, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
"two empty spans mixing offsets": {
h1: &histogram.Histogram{
Schema: 0,
Count: 21,
Sum: 1234.5,
ZeroThreshold: 0.001,
ZeroCount: 4,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 1, Length: 0},
{Offset: 3, Length: 0},
{Offset: 4, Length: 3},
},
PositiveBuckets: []int64{1, 1, -1, 0, 0, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 1, -1, 1, 0, 0, 0},
},
h2: &histogram.Histogram{
Schema: 0,
Count: 37,
Sum: 2345.6,
ZeroThreshold: 0.001,
ZeroCount: 5,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 4},
{Offset: 3, Length: 0},
{Offset: 1, Length: 0},
{Offset: 4, Length: 3},
},
PositiveBuckets: []int64{1, 2, -2, 1, -1, 0, 0},
NegativeSpans: []histogram.Span{
{Offset: 1, Length: 4},
{Offset: 2, Length: 0},
{Offset: 2, Length: 3},
},
NegativeBuckets: []int64{1, 3, -2, 5, -2, 0, -3},
},
},
NegativeBuckets: []int64{1, 3, -2, 5, -2, 0, -3},
}
c := Chunk(NewHistogramChunk())
for name, tc := range tests {
t.Run(name, func(t *testing.T) {
c := Chunk(NewHistogramChunk())
// Create fresh appender and add the first histogram.
app, err := c.Appender()
require.NoError(t, err)
require.Equal(t, 0, c.NumSamples())
// Create fresh appender and add the first histogram.
app, err := c.Appender()
require.NoError(t, err)
require.Equal(t, 0, c.NumSamples())
_, _, _, err = app.AppendHistogram(nil, 1, h1, true)
require.NoError(t, err)
require.Equal(t, 1, c.NumSamples())
hApp, _ := app.(*HistogramAppender)
_, _, _, err = app.AppendHistogram(nil, 1, tc.h1, true)
require.NoError(t, err)
require.Equal(t, 1, c.NumSamples())
hApp, _ := app.(*HistogramAppender)
pI, nI, okToAppend, counterReset := hApp.appendable(h2)
require.Empty(t, pI)
require.Empty(t, nI)
require.True(t, okToAppend)
require.False(t, counterReset)
pI, nI, okToAppend, counterReset := hApp.appendable(tc.h2)
require.Empty(t, pI)
require.Empty(t, nI)
require.True(t, okToAppend)
require.False(t, counterReset)
})
}
}
func TestAtFloatHistogram(t *testing.T) {