Merge pull request #310 from Bplotka/overlap-detection-fix

db: Fixed validateBlockSequence, exported it and added tests.

db.go

@@ -19,16 +19,15 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"os"
 	"path/filepath"
 	"runtime"
 	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
-	"unsafe"
-
-	"golang.org/x/sync/errgroup"
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
@@ -39,6 +38,7 @@ import (
 	"github.com/prometheus/tsdb/chunkenc"
 	"github.com/prometheus/tsdb/fileutil"
 	"github.com/prometheus/tsdb/labels"
+	"golang.org/x/sync/errgroup"
 )
 
 // DefaultOptions used for the DB. They are sane for setups using
@@ -556,22 +556,123 @@ func (db *DB) reload(deleteable ...string) (err error) {
 	return errors.Wrap(db.head.Truncate(maxt), "head truncate failed")
 }
 
+// ValidateBlockSequence returns error if given block meta files indicate that some blocks overlaps within sequence.
 func validateBlockSequence(bs []*Block) error {
-	if len(bs) == 0 {
+	if len(bs) <= 1 {
 		return nil
 	}
-	sort.Slice(bs, func(i, j int) bool {
-		return bs[i].Meta().MinTime < bs[j].Meta().MinTime
-	})
-	prev := bs[0]
-	for _, b := range bs[1:] {
-		if b.Meta().MinTime < prev.Meta().MaxTime {
-			return errors.Errorf("block time ranges overlap (%d, %d)", b.Meta().MinTime, prev.Meta().MaxTime)
-		}
+
+	var metas []BlockMeta
+	for _, b := range bs {
+		metas = append(metas, b.meta)
 	}
+
+	overlaps := OverlappingBlocks(metas)
+	if len(overlaps) > 0 {
+		return errors.Errorf("block time ranges overlap: %s", overlaps)
+	}
+
 	return nil
 }
 
+// TimeRange specifies minTime and maxTime range.
+type TimeRange struct {
+	Min, Max int64
+}
+
+// Overlaps contains overlapping blocks aggregated by overlapping range.
+type Overlaps map[TimeRange][]BlockMeta
+
+// String returns human readable string form of overlapped blocks.
+func (o Overlaps) String() string {
+	var res []string
+	for r, overlaps := range o {
+		var groups []string
+		for _, m := range overlaps {
+			groups = append(groups, fmt.Sprintf(
+				"[id: %s mint: %d maxt: %d range: %s]",
+				m.ULID.String(),
+				m.MinTime,
+				m.MaxTime,
+				(time.Duration((m.MaxTime-m.MinTime)/1000)*time.Second).String(),
+			))
+		}
+		res = append(res, fmt.Sprintf("[%d %d]: <%s> ", r.Min, r.Max, strings.Join(groups, "")))
+	}
+	return strings.Join(res, "")
+}
+
+// OverlappingBlocks returns all overlapping blocks from given meta files.
+func OverlappingBlocks(bm []BlockMeta) Overlaps {
+	if len(bm) <= 1 {
+		return nil
+	}
+	sort.Slice(bm, func(i, j int) bool {
+		return bm[i].MinTime < bm[j].MinTime
+	})
+
+	var (
+		overlaps [][]BlockMeta
+
+		// pending contains not ended blocks in regards to "current" timestamp.
+		pending = []BlockMeta{bm[0]}
+		// continuousPending helps to aggregate same overlaps to single group.
+		continuousPending = true
+	)
+
+	// We have here blocks sorted by minTime. We iterate over each block and treat its minTime as our "current" timestamp.
+	// We check if any of the pending block finished (blocks that we have seen before, but their maxTime was still ahead current
+	// timestamp). If not, it means they overlap with our current block. In the same time current block is assumed pending.
+	for _, b := range bm[1:] {
+		var newPending []BlockMeta
+
+		for _, p := range pending {
+			// "b.MinTime" is our current time.
+			if b.MinTime >= p.MaxTime {
+				continuousPending = false
+				continue
+			}
+
+			// "p" overlaps with "b" and "p" is still pending.
+			newPending = append(newPending, p)
+		}
+
+		// Our block "b" is now pending.
+		pending = append(newPending, b)
+		if len(newPending) == 0 {
+			// No overlaps.
+			continue
+		}
+
+		if continuousPending && len(overlaps) > 0 {
+			overlaps[len(overlaps)-1] = append(overlaps[len(overlaps)-1], b)
+			continue
+		}
+		overlaps = append(overlaps, append(newPending, b))
+		// Start new pendings.
+		continuousPending = true
+	}
+
+	// Fetch the critical overlapped time range foreach overlap groups.
+	overlapGroups := Overlaps{}
+	for _, overlap := range overlaps {
+
+		minRange := TimeRange{Min: 0, Max: math.MaxInt64}
+		for _, b := range overlap {
+			if minRange.Max > b.MaxTime {
+				minRange.Max = b.MaxTime
+			}
+
+			if minRange.Min < b.MinTime {
+				minRange.Min = b.MinTime
+			}
+		}
+		overlapGroups[minRange] = overlap
+	}
+
+	return overlapGroups
+}
+
 func (db *DB) String() string {
 	return "HEAD"
 }
@@ -764,10 +865,6 @@ func intervalOverlap(amin, amax, bmin, bmax int64) bool {
 	return amin <= bmax && bmin <= amax
 }
 
-func intervalContains(min, max, t int64) bool {
-	return t >= min && t <= max
-}
-
 func isBlockDir(fi os.FileInfo) bool {
 	if !fi.IsDir() {
 		return false
@@ -866,9 +963,6 @@ func (es MultiError) Err() error {
 	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 {
 	var merr MultiError
 

db_test.go

@@ -892,3 +892,87 @@ func expandSeriesSet(ss SeriesSet) ([]labels.Labels, error) {
 
 	return result, ss.Err()
 }
+
+func TestOverlappingBlocksDetectsAllOverlaps(t *testing.T) {
+	// Create 10 blocks that does not overlap (0-10, 10-20, ..., 100-110) but in reverse order to ensure our algorithm
+	// will handle that.
+	var metas = make([]BlockMeta, 11)
+	for i := 10; i >= 0; i-- {
+		metas[i] = BlockMeta{MinTime: int64(i * 10), MaxTime: int64((i + 1) * 10)}
+	}
+
+	testutil.Assert(t, len(OverlappingBlocks(metas)) == 0, "we found unexpected overlaps")
+
+	// Add overlapping blocks.
+
+	// o1 overlaps with 10-20.
+	o1 := BlockMeta{MinTime: 15, MaxTime: 17}
+	testutil.Equals(t, Overlaps{
+		{Min: 15, Max: 17}: {metas[1], o1},
+	}, OverlappingBlocks(append(metas, o1)))
+
+	// o2 overlaps with 20-30 and 30-40.
+	o2 := BlockMeta{MinTime: 21, MaxTime: 31}
+	testutil.Equals(t, Overlaps{
+		{Min: 21, Max: 30}: {metas[2], o2},
+		{Min: 30, Max: 31}: {o2, metas[3]},
+	}, OverlappingBlocks(append(metas, o2)))
+
+	// o3a and o3b overlaps with 30-40 and each other.
+	o3a := BlockMeta{MinTime: 33, MaxTime: 39}
+	o3b := BlockMeta{MinTime: 34, MaxTime: 36}
+	testutil.Equals(t, Overlaps{
+		{Min: 34, Max: 36}: {metas[3], o3a, o3b},
+	}, OverlappingBlocks(append(metas, o3a, o3b)))
+
+	// o4 is 1:1 overlap with 50-60.
+	o4 := BlockMeta{MinTime: 50, MaxTime: 60}
+	testutil.Equals(t, Overlaps{
+		{Min: 50, Max: 60}: {metas[5], o4},
+	}, OverlappingBlocks(append(metas, o4)))
+
+	// o5 overlaps with 60-70, 70-80 and 80-90.
+	o5 := BlockMeta{MinTime: 61, MaxTime: 85}
+	testutil.Equals(t, Overlaps{
+		{Min: 61, Max: 70}: {metas[6], o5},
+		{Min: 70, Max: 80}: {o5, metas[7]},
+		{Min: 80, Max: 85}: {o5, metas[8]},
+	}, OverlappingBlocks(append(metas, o5)))
+
+	// o6a overlaps with 90-100, 100-110 and o6b, o6b overlaps with 90-100 and o6a.
+	o6a := BlockMeta{MinTime: 92, MaxTime: 105}
+	o6b := BlockMeta{MinTime: 94, MaxTime: 99}
+	testutil.Equals(t, Overlaps{
+		{Min: 94, Max: 99}:   {metas[9], o6a, o6b},
+		{Min: 100, Max: 105}: {o6a, metas[10]},
+	}, OverlappingBlocks(append(metas, o6a, o6b)))
+
+	// All together.
+	testutil.Equals(t, Overlaps{
+		{Min: 15, Max: 17}: {metas[1], o1},
+		{Min: 21, Max: 30}: {metas[2], o2}, {Min: 30, Max: 31}: {o2, metas[3]},
+		{Min: 34, Max: 36}: {metas[3], o3a, o3b},
+		{Min: 50, Max: 60}: {metas[5], o4},
+		{Min: 61, Max: 70}: {metas[6], o5}, {Min: 70, Max: 80}: {o5, metas[7]}, {Min: 80, Max: 85}: {o5, metas[8]},
+		{Min: 94, Max: 99}: {metas[9], o6a, o6b}, {Min: 100, Max: 105}: {o6a, metas[10]},
+	}, OverlappingBlocks(append(metas, o1, o2, o3a, o3b, o4, o5, o6a, o6b)))
+
+	// Additional case.
+	var nc1 []BlockMeta
+	nc1 = append(nc1, BlockMeta{MinTime: 1, MaxTime: 5})
+	nc1 = append(nc1, BlockMeta{MinTime: 2, MaxTime: 3})
+	nc1 = append(nc1, BlockMeta{MinTime: 2, MaxTime: 3})
+	nc1 = append(nc1, BlockMeta{MinTime: 2, MaxTime: 3})
+	nc1 = append(nc1, BlockMeta{MinTime: 2, MaxTime: 3})
+	nc1 = append(nc1, BlockMeta{MinTime: 2, MaxTime: 6})
+	nc1 = append(nc1, BlockMeta{MinTime: 3, MaxTime: 5})
+	nc1 = append(nc1, BlockMeta{MinTime: 5, MaxTime: 7})
+	nc1 = append(nc1, BlockMeta{MinTime: 7, MaxTime: 10})
+	nc1 = append(nc1, BlockMeta{MinTime: 8, MaxTime: 9})
+	testutil.Equals(t, Overlaps{
+		{Min: 2, Max: 3}: {nc1[0], nc1[1], nc1[2], nc1[3], nc1[4], nc1[5]}, // 1-5, 2-3, 2-3, 2-3, 2-3, 2,6
+		{Min: 3, Max: 5}: {nc1[0], nc1[5], nc1[6]},                         // 1-5, 2-6, 3-5
+		{Min: 5, Max: 6}: {nc1[5], nc1[7]},                                 // 2-6, 5-7
+		{Min: 8, Max: 9}: {nc1[8], nc1[9]},                                 // 7-10, 8-9
+	}, OverlappingBlocks(nc1))
+}