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 = []BlockMeta{bm[0]}
		continuousPending = true
	)
	for _, b := range bm[1:] {
		var newPending []BlockMeta
		for _, p := range pending {
			if b.MinTime >= p.MaxTime {
				continuousPending = false
				continue
			}
			newPending = append(newPending, p)
		}
		pending = append(newPending, b)
		if len(newPending) == 0 {
			continue
		}
		if continuousPending && len(overlaps) > 0 {
			overlaps[len(overlaps)-1] = append(overlaps[len(overlaps)-1], b)
			continue
		}
		overlaps = append(overlaps, append(newPending, b))
		continuousPending = true
	}
	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
}
OverlappingBlocks returns all overlapping blocks from given meta files.
package tsdb
import (
	"bytes"
	"fmt"
	"io"
	"io/ioutil"
	"math"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/oklog/ulid"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/tsdb/chunkenc"
	"github.com/prometheus/tsdb/fileutil"
	"github.com/prometheus/tsdb/labels"
	"golang.org/x/sync/errgroup"
)
var DefaultOptions = &Options{
	WALFlushInterval:  5 * time.Second,
	RetentionDuration: 15 * 24 * 60 * 60 * 1000,  
	BlockRanges:       ExponentialBlockRanges(int64(2*time.Hour)/1e6, 3, 5),
	NoLockfile:        false,
}
type Options struct {
	WALFlushInterval time.Duration
	RetentionDuration uint64
	BlockRanges []int64
	NoLockfile bool
}
type Appender interface {
	Add(l labels.Labels, t int64, v float64) (uint64, error)
	AddFast(ref uint64, t int64, v float64) error
	Commit() error
	Rollback() error
}
type DB struct {
	dir   string
	lockf fileutil.Releaser
	logger    log.Logger
	metrics   *dbMetrics
	opts      *Options
	chunkPool chunkenc.Pool
	compactor Compactor
	mtx    sync.RWMutex
	blocks []*Block
	head *Head
	compactc chan struct{}
	donec    chan struct{}
	stopc    chan struct{}
	cmtx               sync.Mutex
	compactionsEnabled bool
}
type dbMetrics struct {
	loadedBlocks         prometheus.GaugeFunc
	reloads              prometheus.Counter
	reloadsFailed        prometheus.Counter
	compactionsTriggered prometheus.Counter
	cutoffs              prometheus.Counter
	cutoffsFailed        prometheus.Counter
	tombCleanTimer       prometheus.Histogram
}
func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
	m := &dbMetrics{}
	m.loadedBlocks = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
		Name: "prometheus_tsdb_blocks_loaded",
		Help: "Number of currently loaded data blocks",
	}, func() float64 {
		db.mtx.RLock()
		defer db.mtx.RUnlock()
		return float64(len(db.blocks))
	})
	m.reloads = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "prometheus_tsdb_reloads_total",
		Help: "Number of times the database reloaded block data from disk.",
	})
	m.reloadsFailed = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "prometheus_tsdb_reloads_failures_total",
		Help: "Number of times the database failed to reload block data from disk.",
	})
	m.compactionsTriggered = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "prometheus_tsdb_compactions_triggered_total",
		Help: "Total number of triggered compactions for the partition.",
	})
	m.cutoffs = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "prometheus_tsdb_retention_cutoffs_total",
		Help: "Number of times the database cut off block data from disk.",
	})
	m.cutoffsFailed = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "prometheus_tsdb_retention_cutoffs_failures_total",
		Help: "Number of times the database failed to cut off block data from disk.",
	})
	m.tombCleanTimer = prometheus.NewHistogram(prometheus.HistogramOpts{
		Name: "prometheus_tsdb_tombstone_cleanup_seconds",
		Help: "The time taken to recompact blocks to remove tombstones.",
	})
	if r != nil {
		r.MustRegister(
			m.loadedBlocks,
			m.reloads,
			m.reloadsFailed,
			m.cutoffs,
			m.cutoffsFailed,
			m.compactionsTriggered,
			m.tombCleanTimer,
		)
	}
	return m
}
func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db *DB, err error) {
	if err := os.MkdirAll(dir, 0777); err != nil {
		return nil, err
	}
	if l == nil {
		l = log.NewNopLogger()
	}
	if opts == nil {
		opts = DefaultOptions
	}
	if err := repairBadIndexVersion(l, dir); err != nil {
		return nil, err
	}
	db = &DB{
		dir:                dir,
		logger:             l,
		opts:               opts,
		compactc:           make(chan struct{}, 1),
		donec:              make(chan struct{}),
		stopc:              make(chan struct{}),
		compactionsEnabled: true,
		chunkPool:          chunkenc.NewPool(),
	}
	db.metrics = newDBMetrics(db, r)
	if !opts.NoLockfile {
		absdir, err := filepath.Abs(dir)
		if err != nil {
			return nil, err
		}
		lockf, _, err := fileutil.Flock(filepath.Join(absdir, "lock"))
		if err != nil {
			return nil, errors.Wrap(err, "lock DB directory")
		}
		db.lockf = lockf
	}
	db.compactor, err = NewLeveledCompactor(r, l, opts.BlockRanges, db.chunkPool)
	if err != nil {
		return nil, errors.Wrap(err, "create leveled compactor")
	}
	wal, err := OpenSegmentWAL(filepath.Join(dir, "wal"), l, opts.WALFlushInterval, r)
	if err != nil {
		return nil, err
	}
	db.head, err = NewHead(r, l, wal, opts.BlockRanges[0])
	if err != nil {
		return nil, err
	}
	if err := db.reload(); err != nil {
		return nil, err
	}
	if err := db.head.ReadWAL(); err != nil {
		return nil, errors.Wrap(err, "read WAL")
	}
	go db.run()
	return db, nil
}
func (db *DB) Dir() string {
	return db.dir
}
func (db *DB) run() {
	defer close(db.donec)
	backoff := time.Duration(0)
	for {
		select {
		case <-db.stopc:
			return
		case <-time.After(backoff):
		}
		select {
		case <-time.After(1 * time.Minute):
			select {
			case db.compactc <- struct{}{}:
			default:
			}
		case <-db.compactc:
			db.metrics.compactionsTriggered.Inc()
			_, err := db.compact()
			if err != nil {
				level.Error(db.logger).Log("msg", "compaction failed", "err", err)
				backoff = exponential(backoff, 1*time.Second, 1*time.Minute)
			} else {
				backoff = 0
			}
		case <-db.stopc:
			return
		}
	}
}
func (db *DB) beyondRetention(meta *BlockMeta) bool {
	if db.opts.RetentionDuration == 0 {
		return false
	}
	db.mtx.RLock()
	blocks := db.blocks[:]
	db.mtx.RUnlock()
	if len(blocks) == 0 {
		return false
	}
	last := blocks[len(db.blocks)-1]
	mint := last.Meta().MaxTime - int64(db.opts.RetentionDuration)
	return meta.MaxTime < mint
}
func (db *DB) Appender() Appender {
	return dbAppender{db: db, Appender: db.head.Appender()}
}
type dbAppender struct {
	Appender
	db *DB
}
func (a dbAppender) Commit() error {
	err := a.Appender.Commit()
	if a.db.head.MaxTime()-a.db.head.MinTime() > a.db.head.chunkRange/2*3 {
		select {
		case a.db.compactc <- struct{}{}:
		default:
		}
	}
	return err
}
func (db *DB) compact() (changes bool, err error) {
	db.cmtx.Lock()
	defer db.cmtx.Unlock()
	if !db.compactionsEnabled {
		return false, nil
	}
	for {
		select {
		case <-db.stopc:
			return changes, nil
		default:
		}
		if db.head.MaxTime()-db.head.MinTime() <= db.opts.BlockRanges[0]/2*3 {
			break
		}
		mint, maxt := rangeForTimestamp(db.head.MinTime(), db.opts.BlockRanges[0])
		head := &rangeHead{
			head: db.head,
			mint: mint,
			maxt: maxt - 1,
		}
		if _, err = db.compactor.Write(db.dir, head, mint, maxt, nil); err != nil {
			return changes, errors.Wrap(err, "persist head block")
		}
		changes = true
		runtime.GC()
		if err := db.reload(); err != nil {
			return changes, errors.Wrap(err, "reload blocks")
		}
		runtime.GC()
	}
	for {
		plan, err := db.compactor.Plan(db.dir)
		if err != nil {
			return changes, errors.Wrap(err, "plan compaction")
		}
		if len(plan) == 0 {
			break
		}
		select {
		case <-db.stopc:
			return changes, nil
		default:
		}
		if _, err := db.compactor.Compact(db.dir, plan...); err != nil {
			return changes, errors.Wrapf(err, "compact %s", plan)
		}
		changes = true
		runtime.GC()
		if err := db.reload(); err != nil {
			return changes, errors.Wrap(err, "reload blocks")
		}
		runtime.GC()
	}
	return changes, nil
}
func (db *DB) getBlock(id ulid.ULID) (*Block, bool) {
	for _, b := range db.blocks {
		if b.Meta().ULID == id {
			return b, true
		}
	}
	return nil, false
}
func (db *DB) reload() (err error) {
	defer func() {
		if err != nil {
			db.metrics.reloadsFailed.Inc()
		}
		db.metrics.reloads.Inc()
	}()
	dirs, err := blockDirs(db.dir)
	if err != nil {
		return errors.Wrap(err, "find blocks")
	}
	var (
		blocks     []*Block
		corrupted  = map[ulid.ULID]error{}
		opened     = map[ulid.ULID]struct{}{}
		deleteable = map[ulid.ULID]struct{}{}
	)
	for _, dir := range dirs {
		meta, err := readMetaFile(dir)
		if err != nil {
			level.Warn(db.logger).Log("msg", "read meta information", "err", err, "dir", dir)
			ulid, err2 := ulid.Parse(filepath.Base(dir))
			if err2 != nil {
				level.Error(db.logger).Log("msg", "not a block dir", "dir", dir)
				continue
			}
			corrupted[ulid] = err
			continue
		}
		if db.beyondRetention(meta) {
			deleteable[meta.ULID] = struct{}{}
			continue
		}
		for _, b := range meta.Compaction.Parents {
			deleteable[b.ULID] = struct{}{}
		}
	}
	for c, err := range corrupted {
		if _, ok := deleteable[c]; !ok {
			return errors.Wrapf(err, "unexpected corrupted block %s", c)
		}
	}
	for _, dir := range dirs {
		meta, err := readMetaFile(dir)
		if err != nil {
			return errors.Wrapf(err, "read meta information %s", dir)
		}
		if _, ok := deleteable[meta.ULID]; ok {
			continue
		}
		b, ok := db.getBlock(meta.ULID)
		if !ok {
			b, err = OpenBlock(dir, db.chunkPool)
			if err != nil {
				return errors.Wrapf(err, "open block %s", dir)
			}
		}
		blocks = append(blocks, b)
		opened[meta.ULID] = struct{}{}
	}
	sort.Slice(blocks, func(i, j int) bool {
		return blocks[i].Meta().MinTime < blocks[j].Meta().MinTime
	})
	if err := validateBlockSequence(blocks); err != nil {
		return errors.Wrap(err, "invalid block sequence")
	}
	db.mtx.Lock()
	oldBlocks := db.blocks
	db.blocks = blocks
	db.mtx.Unlock()
	for _, b := range oldBlocks {
		if _, ok := opened[b.Meta().ULID]; ok {
			continue
		}
		if err := b.Close(); err != nil {
			level.Warn(db.logger).Log("msg", "closing block failed", "err", err)
		}
	}
	for ulid := range deleteable {
		if err := os.RemoveAll(filepath.Join(db.dir, ulid.String())); err != nil {
			return errors.Wrapf(err, "delete obsolete block %s", ulid)
		}
	}
	if len(blocks) == 0 {
		return nil
	}
	maxt := blocks[len(blocks)-1].Meta().MaxTime
	return errors.Wrap(db.head.Truncate(maxt), "head truncate failed")
}
func validateBlockSequence(bs []*Block) error {
	if len(bs) <= 1 {
		return nil
	}
	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
}
type TimeRange struct {
	Min, Max int64
}
type Overlaps map[TimeRange][]BlockMeta
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(
				"<ulid: %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(
			"[mint: %d, maxt: %d, range: %s, blocks: %d]: %s",
			r.Min, r.Max,
			(time.Duration((r.Max-r.Min)/1000)*time.Second).String(),
			len(overlaps),
			strings.Join(groups, ", ")),
		)
	}
	return strings.Join(res, "\n")
}