db: Added tests for validateBlockSequence to confirm a bug.

(That's why test fails)

Signed-off-by: Bartek Plotka <bwplotka@gmail.com>
This commit is contained in:
Bartek Plotka 2018-03-28 15:50:52 +01:00
parent 195bc0d286
commit a9b28a6aa0
2 changed files with 71 additions and 14 deletions

41
db.go
View file

@ -26,7 +26,6 @@ import (
"strconv" "strconv"
"sync" "sync"
"time" "time"
"unsafe"
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
@ -560,14 +559,35 @@ func validateBlockSequence(bs []*Block) error {
if len(bs) == 0 { if len(bs) == 0 {
return nil return nil
} }
sort.Slice(bs, func(i, j int) bool {
return bs[i].Meta().MinTime < bs[j].Meta().MinTime var metas []BlockMeta
for _, b := range bs {
metas = append(metas, b.meta)
}
overlappedBlocks := ValidateBlockSequence(metas)
if len(overlappedBlocks) == 0 {
return nil
}
return errors.Errorf("block time ranges overlap (%v)", overlappedBlocks)
}
// ValidateBlockSequence returns error if given block meta files indicate that some blocks overlaps within sequence.
func ValidateBlockSequence(bm []BlockMeta) [][]BlockMeta {
if len(bm) == 0 {
return nil
}
sort.Slice(bm, func(i, j int) bool {
return bm[i].MinTime < bm[j].MinTime
}) })
prev := bs[0]
for _, b := range bs[1:] { prev := bm[0]
if b.Meta().MinTime < prev.Meta().MaxTime { for _, b := range bm[1:] {
return errors.Errorf("block time ranges overlap (%d, %d)", b.Meta().MinTime, prev.Meta().MaxTime) if b.MinTime < prev.MaxTime {
return [][]BlockMeta{{b, prev}}
} }
//prev = b
} }
return nil return nil
} }
@ -767,10 +787,6 @@ func intervalOverlap(amin, amax, bmin, bmax int64) bool {
return amin <= bmax && bmin <= amax return amin <= bmax && bmin <= amax
} }
func intervalContains(min, max, t int64) bool {
return t >= min && t <= max
}
func isBlockDir(fi os.FileInfo) bool { func isBlockDir(fi os.FileInfo) bool {
if !fi.IsDir() { if !fi.IsDir() {
return false return false
@ -869,9 +885,6 @@ func (es MultiError) Err() error {
return es return es
} }
func yoloString(b []byte) string { return *((*string)(unsafe.Pointer(&b))) }
func yoloBytes(s string) []byte { return *((*[]byte)(unsafe.Pointer(&s))) }
func closeAll(cs ...io.Closer) error { func closeAll(cs ...io.Closer) error {
var merr MultiError var merr MultiError

View file

@ -892,3 +892,47 @@ func expandSeriesSet(ss SeriesSet) ([]labels.Labels, error) {
return result, ss.Err() return result, ss.Err()
} }
func TestValidateBlockSequenceDetectsAllOverlaps(t *testing.T) {
var metas []BlockMeta
// Create 10 blocks that does not overlap (0-10, 10-20, ..., 90-100)
for i := 0; i < 10; i++ {
metas = append(metas, BlockMeta{MinTime: int64(i * 10), MaxTime: int64((i + 1) * 10)})
}
overlappedBlocks := ValidateBlockSequence(metas)
testutil.Assert(t, len(overlappedBlocks) == 0, "we found unexpected overlaps")
// Add overlaping blocks.
// o1 overlaps with 10-20.
o1 := BlockMeta{MinTime: 15, MaxTime: 17}
overlappedBlocks = ValidateBlockSequence(append(metas, o1))
expectedOverlaps := [][]BlockMeta{
{metas[1], o1},
}
testutil.Equals(t, expectedOverlaps, overlappedBlocks)
//// o2 overlaps with 20-30 and 30-40.
//o2 := BlockMeta{MinTime: 21, MaxTime: 31}
//
//// o3a and o3b overlaps with 30-40 and each other.
//o3a := BlockMeta{MinTime: 33, MaxTime: 39}
//o3b := BlockMeta{MinTime: 34, MaxTime: 36}
//
//// o4 is 1:1 overlap with 50-60
//o4 := BlockMeta{MinTime: 50, MaxTime: 60}
//
//// o5 overlaps with 50-60, 60-70 and 70,80
//o5 := BlockMeta{MinTime: 60, MaxTime: 80}
//
//expectedOverlaps := [][]block.Meta{
// {metas[1], o1},
// {metas[2], o2},
// {metas[3], o2},
// {metas[3], o3, o3b},
//}
}