2017-01-02 09:34:55 +00:00
|
|
|
package tsdb
|
|
|
|
|
|
|
|
import (
|
2017-03-20 09:41:43 +00:00
|
|
|
"fmt"
|
2017-02-27 09:46:15 +00:00
|
|
|
"math/rand"
|
2017-01-02 09:34:55 +00:00
|
|
|
"os"
|
2017-01-02 13:41:13 +00:00
|
|
|
"path/filepath"
|
2017-01-03 14:43:26 +00:00
|
|
|
"time"
|
2017-01-02 09:34:55 +00:00
|
|
|
|
2017-01-02 13:41:13 +00:00
|
|
|
"github.com/coreos/etcd/pkg/fileutil"
|
2017-01-02 09:34:55 +00:00
|
|
|
"github.com/fabxc/tsdb/labels"
|
2017-03-20 09:41:43 +00:00
|
|
|
"github.com/go-kit/kit/log"
|
2017-02-27 09:46:15 +00:00
|
|
|
"github.com/oklog/ulid"
|
2017-01-03 14:43:26 +00:00
|
|
|
"github.com/pkg/errors"
|
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
2017-01-02 09:34:55 +00:00
|
|
|
)
|
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
// Compactor provides compaction against an underlying storage
|
|
|
|
// of time series data.
|
2017-03-02 08:13:29 +00:00
|
|
|
type Compactor interface {
|
2017-03-02 13:32:09 +00:00
|
|
|
// Plan returns a set of non-overlapping directories that can
|
|
|
|
// be compacted concurrently.
|
|
|
|
// Results returned when compactions are in progress are undefined.
|
|
|
|
Plan(dir string) ([][]string, error)
|
2017-03-02 08:13:29 +00:00
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
// Write persists a Block into a directory.
|
|
|
|
Write(dir string, b Block) error
|
2017-03-02 08:13:29 +00:00
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
// Compact runs compaction against the provided directories. Must
|
|
|
|
// only be called concurrently with results of Plan().
|
2017-03-02 08:13:29 +00:00
|
|
|
Compact(dirs ...string) error
|
|
|
|
}
|
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
// compactor implements the Compactor interface.
|
2017-01-02 09:34:55 +00:00
|
|
|
type compactor struct {
|
2017-01-03 14:43:26 +00:00
|
|
|
metrics *compactorMetrics
|
2017-03-20 09:41:43 +00:00
|
|
|
logger log.Logger
|
2017-01-18 05:18:32 +00:00
|
|
|
opts *compactorOptions
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
|
2017-01-03 14:43:26 +00:00
|
|
|
type compactorMetrics struct {
|
2017-01-06 11:37:28 +00:00
|
|
|
ran prometheus.Counter
|
|
|
|
failed prometheus.Counter
|
|
|
|
duration prometheus.Histogram
|
2017-01-03 14:43:26 +00:00
|
|
|
}
|
|
|
|
|
2017-01-06 10:40:09 +00:00
|
|
|
func newCompactorMetrics(r prometheus.Registerer) *compactorMetrics {
|
2017-01-03 14:43:26 +00:00
|
|
|
m := &compactorMetrics{}
|
|
|
|
|
|
|
|
m.ran = prometheus.NewCounter(prometheus.CounterOpts{
|
2017-01-06 10:40:09 +00:00
|
|
|
Name: "tsdb_compactions_total",
|
|
|
|
Help: "Total number of compactions that were executed for the partition.",
|
2017-01-03 14:43:26 +00:00
|
|
|
})
|
|
|
|
m.failed = prometheus.NewCounter(prometheus.CounterOpts{
|
2017-01-06 10:40:09 +00:00
|
|
|
Name: "tsdb_compactions_failed_total",
|
|
|
|
Help: "Total number of compactions that failed for the partition.",
|
2017-01-03 14:43:26 +00:00
|
|
|
})
|
|
|
|
m.duration = prometheus.NewHistogram(prometheus.HistogramOpts{
|
2017-01-06 10:40:09 +00:00
|
|
|
Name: "tsdb_compaction_duration",
|
|
|
|
Help: "Duration of compaction runs.",
|
2017-01-03 14:43:26 +00:00
|
|
|
})
|
|
|
|
|
2017-01-06 10:40:09 +00:00
|
|
|
if r != nil {
|
|
|
|
r.MustRegister(
|
|
|
|
m.ran,
|
|
|
|
m.failed,
|
|
|
|
m.duration,
|
|
|
|
)
|
|
|
|
}
|
2017-01-03 14:43:26 +00:00
|
|
|
return m
|
|
|
|
}
|
|
|
|
|
2017-01-18 05:18:32 +00:00
|
|
|
type compactorOptions struct {
|
|
|
|
maxBlockRange uint64
|
2017-01-06 12:53:05 +00:00
|
|
|
}
|
|
|
|
|
2017-03-20 09:41:43 +00:00
|
|
|
func newCompactor(r prometheus.Registerer, l log.Logger, opts *compactorOptions) *compactor {
|
2017-01-18 05:18:32 +00:00
|
|
|
return &compactor{
|
|
|
|
opts: opts,
|
2017-03-20 09:41:43 +00:00
|
|
|
logger: l,
|
2017-01-09 18:14:21 +00:00
|
|
|
metrics: newCompactorMetrics(r),
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-01-20 06:58:19 +00:00
|
|
|
type compactionInfo struct {
|
2017-02-28 07:40:51 +00:00
|
|
|
seq int
|
2017-01-20 06:58:19 +00:00
|
|
|
generation int
|
|
|
|
mint, maxt int64
|
|
|
|
}
|
|
|
|
|
2017-02-23 09:50:22 +00:00
|
|
|
const compactionBlocksLen = 3
|
2017-01-30 08:42:38 +00:00
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
func (c *compactor) Plan(dir string) ([][]string, error) {
|
|
|
|
dirs, err := blockDirs(dir)
|
2017-03-02 08:13:29 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2017-01-03 14:43:26 +00:00
|
|
|
}
|
2017-01-18 05:18:32 +00:00
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
var bs []*BlockMeta
|
|
|
|
|
|
|
|
for _, dir := range dirs {
|
|
|
|
meta, err := readMetaFile(dir)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2017-01-06 12:53:05 +00:00
|
|
|
}
|
2017-03-02 08:13:29 +00:00
|
|
|
if meta.Compaction.Generation > 0 {
|
|
|
|
bs = append(bs, meta)
|
2017-01-30 08:42:38 +00:00
|
|
|
}
|
2017-03-02 08:13:29 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
if len(bs) == 0 {
|
|
|
|
return nil, nil
|
|
|
|
}
|
2017-01-30 08:42:38 +00:00
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
sliceDirs := func(i, j int) [][]string {
|
|
|
|
var res []string
|
|
|
|
for k := i; k < j; k++ {
|
|
|
|
res = append(res, dirs[k])
|
|
|
|
}
|
|
|
|
return [][]string{res}
|
2017-01-06 12:53:05 +00:00
|
|
|
}
|
2017-01-03 14:43:26 +00:00
|
|
|
|
2017-01-30 08:42:38 +00:00
|
|
|
// Then we care about compacting multiple blocks, starting with the oldest.
|
2017-03-02 14:10:13 +00:00
|
|
|
for i := 0; i < len(bs)-compactionBlocksLen+1; i++ {
|
2017-02-28 07:40:51 +00:00
|
|
|
if c.match(bs[i : i+3]) {
|
2017-03-02 08:13:29 +00:00
|
|
|
return sliceDirs(i, i+compactionBlocksLen), nil
|
2017-01-03 14:43:26 +00:00
|
|
|
}
|
|
|
|
}
|
2017-01-30 08:42:38 +00:00
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
return nil, nil
|
2017-01-04 20:11:15 +00:00
|
|
|
}
|
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
func (c *compactor) match(bs []*BlockMeta) bool {
|
|
|
|
g := bs[0].Compaction.Generation
|
2017-01-19 18:45:52 +00:00
|
|
|
|
2017-01-20 06:58:19 +00:00
|
|
|
for _, b := range bs {
|
2017-03-02 08:13:29 +00:00
|
|
|
if b.Compaction.Generation != g {
|
2017-01-19 18:45:52 +00:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
}
|
2017-03-02 08:13:29 +00:00
|
|
|
return uint64(bs[len(bs)-1].MaxTime-bs[0].MinTime) <= c.opts.maxBlockRange
|
2017-01-03 09:09:20 +00:00
|
|
|
}
|
|
|
|
|
2017-02-27 09:46:15 +00:00
|
|
|
var entropy = rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
|
|
|
2017-01-19 10:22:47 +00:00
|
|
|
func mergeBlockMetas(blocks ...Block) (res BlockMeta) {
|
2017-01-29 07:11:47 +00:00
|
|
|
m0 := blocks[0].Meta()
|
|
|
|
|
|
|
|
res.Sequence = m0.Sequence
|
|
|
|
res.MinTime = m0.MinTime
|
2017-01-19 10:22:47 +00:00
|
|
|
res.MaxTime = blocks[len(blocks)-1].Meta().MaxTime
|
2017-02-27 09:46:15 +00:00
|
|
|
res.ULID = ulid.MustNew(ulid.Now(), entropy)
|
2017-01-20 06:58:19 +00:00
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
res.Compaction.Generation = m0.Compaction.Generation + 1
|
2017-01-03 14:43:26 +00:00
|
|
|
|
|
|
|
for _, b := range blocks {
|
2017-01-19 10:22:47 +00:00
|
|
|
res.Stats.NumSamples += b.Meta().Stats.NumSamples
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
2017-01-03 14:43:26 +00:00
|
|
|
return res
|
|
|
|
}
|
2017-01-02 09:34:55 +00:00
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
func (c *compactor) Compact(dirs ...string) (err error) {
|
|
|
|
var blocks []Block
|
2017-03-02 13:32:09 +00:00
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
for _, d := range dirs {
|
|
|
|
b, err := newPersistedBlock(d)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2017-03-06 08:33:55 +00:00
|
|
|
defer b.Close()
|
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
blocks = append(blocks, b)
|
|
|
|
}
|
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
return c.write(dirs[0], blocks...)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *compactor) Write(dir string, b Block) error {
|
|
|
|
return c.write(dir, b)
|
2017-03-02 08:13:29 +00:00
|
|
|
}
|
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
// write creates a new block that is the union of the provided blocks into dir.
|
|
|
|
// It cleans up all files of the old blocks after completing successfully.
|
|
|
|
func (c *compactor) write(dir string, blocks ...Block) (err error) {
|
2017-03-20 09:41:43 +00:00
|
|
|
c.logger.Log("msg", "compact blocks", "blocks", fmt.Sprintf("%v", blocks))
|
|
|
|
|
2017-03-02 08:13:29 +00:00
|
|
|
defer func(t time.Time) {
|
2017-01-06 11:37:28 +00:00
|
|
|
if err != nil {
|
|
|
|
c.metrics.failed.Inc()
|
|
|
|
}
|
2017-03-02 08:13:29 +00:00
|
|
|
c.metrics.duration.Observe(time.Since(t).Seconds())
|
|
|
|
}(time.Now())
|
2017-01-03 14:43:26 +00:00
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
tmp := dir + ".tmp"
|
|
|
|
|
|
|
|
if err = os.RemoveAll(tmp); err != nil {
|
2017-02-19 15:04:37 +00:00
|
|
|
return err
|
2017-01-03 14:43:26 +00:00
|
|
|
}
|
2017-02-19 15:04:37 +00:00
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
if err = os.MkdirAll(tmp, 0777); err != nil {
|
2017-01-02 09:34:55 +00:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2017-03-02 13:35:06 +00:00
|
|
|
// Populate chunk and index files into temporary directory with
|
|
|
|
// data of all blocks.
|
2017-03-02 13:32:09 +00:00
|
|
|
chunkw, err := newChunkWriter(chunkDir(tmp))
|
2017-02-23 09:50:22 +00:00
|
|
|
if err != nil {
|
|
|
|
return errors.Wrap(err, "open chunk writer")
|
|
|
|
}
|
2017-03-02 13:32:09 +00:00
|
|
|
indexw, err := newIndexWriter(tmp)
|
2017-02-25 06:24:20 +00:00
|
|
|
if err != nil {
|
|
|
|
return errors.Wrap(err, "open index writer")
|
|
|
|
}
|
2017-01-03 14:43:26 +00:00
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
meta, err := c.populate(blocks, indexw, chunkw)
|
|
|
|
if err != nil {
|
2017-01-03 14:43:26 +00:00
|
|
|
return errors.Wrap(err, "write compaction")
|
|
|
|
}
|
2017-03-02 13:32:09 +00:00
|
|
|
if err = writeMetaFile(tmp, meta); err != nil {
|
|
|
|
return errors.Wrap(err, "write merged meta")
|
|
|
|
}
|
2017-01-03 14:43:26 +00:00
|
|
|
|
2017-01-06 11:37:28 +00:00
|
|
|
if err = chunkw.Close(); err != nil {
|
2017-01-03 14:43:26 +00:00
|
|
|
return errors.Wrap(err, "close chunk writer")
|
|
|
|
}
|
2017-01-06 11:37:28 +00:00
|
|
|
if err = indexw.Close(); err != nil {
|
2017-01-03 14:43:26 +00:00
|
|
|
return errors.Wrap(err, "close index writer")
|
|
|
|
}
|
2017-03-02 13:32:09 +00:00
|
|
|
|
2017-03-02 13:35:06 +00:00
|
|
|
// Block successfully written, make visible and remove old ones.
|
2017-03-02 13:32:09 +00:00
|
|
|
if err := renameFile(tmp, dir); err != nil {
|
|
|
|
return errors.Wrap(err, "rename block dir")
|
|
|
|
}
|
|
|
|
for _, b := range blocks[1:] {
|
|
|
|
if err := os.RemoveAll(b.Dir()); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
2017-03-02 13:35:06 +00:00
|
|
|
// Properly sync parent dir to ensure changes are visible.
|
|
|
|
df, err := fileutil.OpenDir(dir)
|
|
|
|
if err != nil {
|
|
|
|
return errors.Wrap(err, "sync block dir")
|
|
|
|
}
|
|
|
|
if err := fileutil.Fsync(df); err != nil {
|
|
|
|
return errors.Wrap(err, "sync block dir")
|
|
|
|
}
|
2017-03-02 13:32:09 +00:00
|
|
|
|
2017-01-06 11:37:28 +00:00
|
|
|
return nil
|
2017-01-03 14:43:26 +00:00
|
|
|
}
|
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
// populate fills the index and chunk writers with new data gathered as the union
|
|
|
|
// of the provided blocks. It returns meta information for the new block.
|
|
|
|
func (c *compactor) populate(blocks []Block, indexw IndexWriter, chunkw ChunkWriter) (*BlockMeta, error) {
|
2017-01-03 14:43:26 +00:00
|
|
|
var set compactionSet
|
2017-01-18 05:18:32 +00:00
|
|
|
|
2017-01-03 14:43:26 +00:00
|
|
|
for i, b := range blocks {
|
2017-01-10 14:28:22 +00:00
|
|
|
all, err := b.Index().Postings("", "")
|
2017-01-03 14:43:26 +00:00
|
|
|
if err != nil {
|
2017-03-02 13:32:09 +00:00
|
|
|
return nil, err
|
2017-01-03 14:43:26 +00:00
|
|
|
}
|
2017-01-05 14:13:01 +00:00
|
|
|
// TODO(fabxc): find more transparent way of handling this.
|
2017-01-10 14:28:22 +00:00
|
|
|
if hb, ok := b.(*headBlock); ok {
|
2017-01-05 14:13:01 +00:00
|
|
|
all = hb.remapPostings(all)
|
|
|
|
}
|
2017-02-23 09:50:22 +00:00
|
|
|
s := newCompactionSeriesSet(b.Index(), b.Chunks(), all)
|
2017-01-03 14:43:26 +00:00
|
|
|
|
|
|
|
if i == 0 {
|
|
|
|
set = s
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
set, err = newCompactionMerger(set, s)
|
|
|
|
if err != nil {
|
2017-03-02 13:32:09 +00:00
|
|
|
return nil, err
|
2017-01-03 14:43:26 +00:00
|
|
|
}
|
2017-01-02 10:12:28 +00:00
|
|
|
}
|
|
|
|
|
2017-01-02 09:34:55 +00:00
|
|
|
// We fully rebuild the postings list index from merged series.
|
|
|
|
var (
|
|
|
|
postings = &memPostings{m: make(map[term][]uint32, 512)}
|
|
|
|
values = map[string]stringset{}
|
|
|
|
i = uint32(0)
|
2017-01-19 10:22:47 +00:00
|
|
|
meta = mergeBlockMetas(blocks...)
|
2017-01-02 09:34:55 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
for set.Next() {
|
|
|
|
lset, chunks := set.At()
|
2017-02-19 10:27:31 +00:00
|
|
|
if err := chunkw.WriteChunks(chunks...); err != nil {
|
2017-03-02 13:32:09 +00:00
|
|
|
return nil, err
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
|
2017-02-19 10:27:31 +00:00
|
|
|
indexw.AddSeries(i, lset, chunks...)
|
|
|
|
|
2017-01-19 10:22:47 +00:00
|
|
|
meta.Stats.NumChunks += uint64(len(chunks))
|
|
|
|
meta.Stats.NumSeries++
|
2017-01-02 09:34:55 +00:00
|
|
|
|
|
|
|
for _, l := range lset {
|
|
|
|
valset, ok := values[l.Name]
|
|
|
|
if !ok {
|
|
|
|
valset = stringset{}
|
|
|
|
values[l.Name] = valset
|
|
|
|
}
|
|
|
|
valset.set(l.Value)
|
|
|
|
|
|
|
|
postings.add(i, term{name: l.Name, value: l.Value})
|
|
|
|
}
|
|
|
|
i++
|
|
|
|
}
|
|
|
|
if set.Err() != nil {
|
2017-03-02 13:32:09 +00:00
|
|
|
return nil, set.Err()
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
s := make([]string, 0, 256)
|
|
|
|
for n, v := range values {
|
|
|
|
s = s[:0]
|
|
|
|
|
|
|
|
for x := range v {
|
|
|
|
s = append(s, x)
|
|
|
|
}
|
2017-01-02 15:58:47 +00:00
|
|
|
if err := indexw.WriteLabelIndex([]string{n}, s); err != nil {
|
2017-03-02 13:32:09 +00:00
|
|
|
return nil, err
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
for t := range postings.m {
|
2017-01-02 15:58:47 +00:00
|
|
|
if err := indexw.WritePostings(t.name, t.value, postings.get(t)); err != nil {
|
2017-03-02 13:32:09 +00:00
|
|
|
return nil, err
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
// Write a postings list containing all series.
|
|
|
|
all := make([]uint32, i)
|
|
|
|
for i := range all {
|
|
|
|
all[i] = uint32(i)
|
|
|
|
}
|
2017-01-02 15:58:47 +00:00
|
|
|
if err := indexw.WritePostings("", "", newListPostings(all)); err != nil {
|
2017-03-02 13:32:09 +00:00
|
|
|
return nil, err
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
2017-01-19 13:01:38 +00:00
|
|
|
|
2017-03-02 13:32:09 +00:00
|
|
|
return &meta, nil
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
|
2017-01-03 14:43:26 +00:00
|
|
|
type compactionSet interface {
|
|
|
|
Next() bool
|
2017-03-14 14:40:16 +00:00
|
|
|
At() (labels.Labels, []*ChunkMeta)
|
2017-01-03 14:43:26 +00:00
|
|
|
Err() error
|
|
|
|
}
|
|
|
|
|
2017-01-02 09:34:55 +00:00
|
|
|
type compactionSeriesSet struct {
|
|
|
|
p Postings
|
|
|
|
index IndexReader
|
2017-02-23 09:50:22 +00:00
|
|
|
chunks ChunkReader
|
2017-01-02 09:34:55 +00:00
|
|
|
|
|
|
|
l labels.Labels
|
2017-03-14 14:40:16 +00:00
|
|
|
c []*ChunkMeta
|
2017-01-02 09:34:55 +00:00
|
|
|
err error
|
|
|
|
}
|
|
|
|
|
2017-02-23 09:50:22 +00:00
|
|
|
func newCompactionSeriesSet(i IndexReader, c ChunkReader, p Postings) *compactionSeriesSet {
|
2017-01-02 09:34:55 +00:00
|
|
|
return &compactionSeriesSet{
|
|
|
|
index: i,
|
2017-02-23 09:50:22 +00:00
|
|
|
chunks: c,
|
2017-01-02 09:34:55 +00:00
|
|
|
p: p,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *compactionSeriesSet) Next() bool {
|
|
|
|
if !c.p.Next() {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2017-01-02 12:27:52 +00:00
|
|
|
c.l, c.c, c.err = c.index.Series(c.p.At())
|
2017-01-02 09:34:55 +00:00
|
|
|
if c.err != nil {
|
|
|
|
return false
|
|
|
|
}
|
2017-03-14 14:40:16 +00:00
|
|
|
for _, chk := range c.c {
|
2017-02-23 09:50:22 +00:00
|
|
|
chk.Chunk, c.err = c.chunks.Chunk(chk.Ref)
|
2017-01-02 09:34:55 +00:00
|
|
|
if c.err != nil {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *compactionSeriesSet) Err() error {
|
|
|
|
if c.err != nil {
|
|
|
|
return c.err
|
|
|
|
}
|
|
|
|
return c.p.Err()
|
|
|
|
}
|
|
|
|
|
2017-03-14 14:40:16 +00:00
|
|
|
func (c *compactionSeriesSet) At() (labels.Labels, []*ChunkMeta) {
|
2017-01-02 09:34:55 +00:00
|
|
|
return c.l, c.c
|
|
|
|
}
|
|
|
|
|
|
|
|
type compactionMerger struct {
|
2017-01-03 14:43:26 +00:00
|
|
|
a, b compactionSet
|
2017-01-02 09:34:55 +00:00
|
|
|
|
2017-01-04 20:11:15 +00:00
|
|
|
aok, bok bool
|
|
|
|
l labels.Labels
|
2017-03-14 14:40:16 +00:00
|
|
|
c []*ChunkMeta
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
type compactionSeries struct {
|
|
|
|
labels labels.Labels
|
2017-03-14 14:40:16 +00:00
|
|
|
chunks []*ChunkMeta
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
|
2017-01-03 14:43:26 +00:00
|
|
|
func newCompactionMerger(a, b compactionSet) (*compactionMerger, error) {
|
2017-01-02 09:34:55 +00:00
|
|
|
c := &compactionMerger{
|
|
|
|
a: a,
|
|
|
|
b: b,
|
|
|
|
}
|
|
|
|
// Initialize first elements of both sets as Next() needs
|
|
|
|
// one element look-ahead.
|
2017-01-04 20:11:15 +00:00
|
|
|
c.aok = c.a.Next()
|
|
|
|
c.bok = c.b.Next()
|
2017-01-02 09:34:55 +00:00
|
|
|
|
|
|
|
return c, c.Err()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *compactionMerger) compare() int {
|
2017-01-04 20:11:15 +00:00
|
|
|
if !c.aok {
|
2017-01-02 09:34:55 +00:00
|
|
|
return 1
|
|
|
|
}
|
2017-01-04 20:11:15 +00:00
|
|
|
if !c.bok {
|
2017-01-02 09:34:55 +00:00
|
|
|
return -1
|
|
|
|
}
|
|
|
|
a, _ := c.a.At()
|
|
|
|
b, _ := c.b.At()
|
|
|
|
return labels.Compare(a, b)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *compactionMerger) Next() bool {
|
2017-01-04 20:11:15 +00:00
|
|
|
if !c.aok && !c.bok || c.Err() != nil {
|
2017-01-02 09:34:55 +00:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
|
|
|
d := c.compare()
|
|
|
|
// Both sets contain the current series. Chain them into a single one.
|
|
|
|
if d > 0 {
|
|
|
|
c.l, c.c = c.b.At()
|
2017-01-04 20:11:15 +00:00
|
|
|
c.bok = c.b.Next()
|
2017-01-02 09:34:55 +00:00
|
|
|
} else if d < 0 {
|
|
|
|
c.l, c.c = c.a.At()
|
2017-01-04 20:11:15 +00:00
|
|
|
c.aok = c.a.Next()
|
2017-01-02 09:34:55 +00:00
|
|
|
} else {
|
|
|
|
l, ca := c.a.At()
|
|
|
|
_, cb := c.b.At()
|
|
|
|
|
|
|
|
c.l = l
|
|
|
|
c.c = append(ca, cb...)
|
|
|
|
|
2017-01-04 20:11:15 +00:00
|
|
|
c.aok = c.a.Next()
|
|
|
|
c.bok = c.b.Next()
|
2017-01-02 09:34:55 +00:00
|
|
|
}
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *compactionMerger) Err() error {
|
|
|
|
if c.a.Err() != nil {
|
|
|
|
return c.a.Err()
|
|
|
|
}
|
|
|
|
return c.b.Err()
|
|
|
|
}
|
|
|
|
|
2017-03-14 14:40:16 +00:00
|
|
|
func (c *compactionMerger) At() (labels.Labels, []*ChunkMeta) {
|
2017-01-02 09:34:55 +00:00
|
|
|
return c.l, c.c
|
|
|
|
}
|
2017-01-02 13:41:13 +00:00
|
|
|
|
2017-03-01 16:19:57 +00:00
|
|
|
func renameFile(from, to string) error {
|
2017-01-02 13:41:13 +00:00
|
|
|
if err := os.RemoveAll(to); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := os.Rename(from, to); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Directory was renamed; sync parent dir to persist rename.
|
|
|
|
pdir, err := fileutil.OpenDir(filepath.Dir(to))
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err = fileutil.Fsync(pdir); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err = pdir.Close(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|