RepoMirrors
/
prometheus
mirror of https://github.com/prometheus/prometheus
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix interjections at the end 

Signed-off-by: beorn7 <beorn@grafana.com>
This commit is contained in:
beorn7 2021-07-05 23:01:39 +02:00
parent a979d8e1dc
commit dc1c744169
2 changed files with 181 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
tsdb/chunkenc/histo_meta.go
View File

@ -186,6 +186,7 @@ func compareSpans(a, b []histogram.Span) ([]Interjection, bool) {
return interjections, false
} else if !aok && bok { // a misses a value that is in b. forward b and recompare
inter.num++
inter.pos++
bv, bok = bi.Next()
continue
} else { // both iterators ran out. we're done
@ -199,16 +200,18 @@ func compareSpans(a, b []histogram.Span) ([]Interjection, bool) {
return interjections, true
}
// caller is responsible for making sure len(in) and len(out) are appropriate for the provided interjections!
// interject merges 'in' with the provided interjections and writes them into
// 'out', which must already have the appropriate length.
func interject(in, out []int64, interjections []Interjection) []int64 {
var j int // position in out
var v int64 // the last value seen
var interj int // the next interjection to process
var j int // Position in out.
var v int64 // The last value seen.
var interj int // The next interjection to process.
for i, d := range in {
if interj < len(interjections) && i == interjections[interj].pos {
// we have an interjection!
// add interjection.num new delta values such as their bucket values equate 0
// We have an interjection!
// Add interjection.num new delta values such that their
// bucket values equate 0.
out[j] = int64(-v)
j++
for x := 1; x < interjections[interj].num; x++ {
@ -217,19 +220,35 @@ func interject(in, out []int64, interjections []Interjection) []int64 {
}
interj++
// now save the value from the input. the delta value we should save is
// the original delta value + the last value of the point before the interjection (to undo the delta that was introduced by the interjection)
// Now save the value from the input. The delta value we
// should save is the original delta value + the last
// value of the point before the interjection (to undo
// the delta that was introduced by the interjection).
out[j] = d + v
j++
v = d + v
continue
}
// if there was no interjection, the original delta is still valid
// If there was no interjection, the original delta is still
// valid.
out[j] = d
j++
v += d
}
switch interj {
case len(interjections):
// All interjections processed. Nothing more to do.
case len(interjections) - 1:
// One more interjection to process at the end.
out[j] = int64(-v)
j++
for x := 1; x < interjections[interj].num; x++ {
out[j] = 0
j++
}
default:
panic("unprocessed interjections left")
}
return out
}

158
tsdb/chunkenc/histo_meta_test.go
View File

@ -108,21 +108,134 @@ func TestBucketIterator(t *testing.T) {
}
func TestInterjection(t *testing.T) {
// this tests the scenario as described in compareSpans's comments
a := []histogram.Span{
scenarios := []struct {
description string
spansA, spansB []histogram.Span
valid bool
interjections []Interjection
bucketsIn, bucketsOut []int64
}{
{
description: "single prepend at the beginning",
spansA: []histogram.Span{
{Offset: -10, Length: 3},
},
spansB: []histogram.Span{
{Offset: -11, Length: 4},
},
valid: true,
interjections: []Interjection{
{
pos: 0,
num: 1,
},
},
bucketsIn: []int64{6, -3, 0},
bucketsOut: []int64{0, 6, -3, 0},
},
{
description: "single append at the end",
spansA: []histogram.Span{
{Offset: -10, Length: 3},
},
spansB: []histogram.Span{
{Offset: -10, Length: 4},
},
valid: true,
interjections: []Interjection{
{
pos: 3,
num: 1,
},
},
bucketsIn: []int64{6, -3, 0},
bucketsOut: []int64{6, -3, 0, -3},
},
{
description: "double prepend at the beginning",
spansA: []histogram.Span{
{Offset: -10, Length: 3},
},
spansB: []histogram.Span{
{Offset: -12, Length: 5},
},
valid: true,
interjections: []Interjection{
{
pos: 0,
num: 2,
},
},
bucketsIn: []int64{6, -3, 0},
bucketsOut: []int64{0, 0, 6, -3, 0},
},
{
description: "double append at the end",
spansA: []histogram.Span{
{Offset: -10, Length: 3},
},
spansB: []histogram.Span{
{Offset: -10, Length: 5},
},
valid: true,
interjections: []Interjection{
{
pos: 3,
num: 2,
},
},
bucketsIn: []int64{6, -3, 0},
bucketsOut: []int64{6, -3, 0, -3, 0},
},
{
description: "single removal of bucket at the start",
spansA: []histogram.Span{
{Offset: -10, Length: 4},
},
spansB: []histogram.Span{
{Offset: -9, Length: 3},
},
valid: false,
},
{
description: "single removal of bucket in the middle",
spansA: []histogram.Span{
{Offset: -10, Length: 4},
},
spansB: []histogram.Span{
{Offset: -10, Length: 2},
{Offset: 1, Length: 1},
},
valid: false,
},
{
description: "single removal of bucket at the end",
spansA: []histogram.Span{
{Offset: -10, Length: 4},
},
spansB: []histogram.Span{
{Offset: -10, Length: 3},
},
valid: false,
},
// TODO(beorn7): Add more scenarios.
{
description: "as described in doc comment",
spansA: []histogram.Span{
{Offset: 0, Length: 2},
{Offset: 2, Length: 1},
{Offset: 3, Length: 2},
{Offset: 3, Length: 1},
{Offset: 1, Length: 1},
}
b := []histogram.Span{
},
spansB: []histogram.Span{
{Offset: 0, Length: 3},
{Offset: 1, Length: 1},
{Offset: 1, Length: 4},
{Offset: 3, Length: 3},
}
interj := []Interjection{
},
valid: true,
interjections: []Interjection{
{
pos: 2,
num: 1,
@ -135,25 +248,26 @@ func TestInterjection(t *testing.T) {
pos: 6,
num: 1,
},
}
testCompareSpans(a, b, interj, t)
testInterject(interj, t)
},
bucketsIn: []int64{6, -3, 0, -1, 2, 1, -4},
bucketsOut: []int64{6, -3, -3, 3, -3, 0, 2, 2, 1, -5, 1},
},
}
func testCompareSpans(a, b []histogram.Span, exp []Interjection, t *testing.T) {
got, ok := compareSpans(a, b)
require.Equal(t, true, ok)
require.Equal(t, exp, got)
for _, s := range scenarios {
t.Run(s.description, func(t *testing.T) {
interjections, valid := compareSpans(s.spansA, s.spansB)
if !s.valid {
require.False(t, valid, "compareScan unexpectedly returned true")
return
}
require.True(t, valid, "compareScan unexpectedly returned false")
require.Equal(t, s.interjections, interjections)
func testInterject(interjections []Interjection, t *testing.T) {
// this tests the scenario as described in compareSpans's comments
// original deltas that represent these counts : 6, 3, 3, 2, 4, 5, 1
a := []int64{6, -3, 0, -1, 2, 1, -4}
// modified deltas to represent the interjected counts: 6, 3, 0, 3, 0, 0, 2, 4, 5, 0, 1
exp := []int64{6, -3, -3, 3, -3, 0, 2, 2, 1, -5, 1}
b := make([]int64, len(a)+4)
interject(a, b, interjections)
require.Equal(t, exp, b)
gotBuckets := make([]int64, len(s.bucketsOut))
interject(s.bucketsIn, gotBuckets, interjections)
require.Equal(t, s.bucketsOut, gotBuckets)
})
}
}