Merge remote-tracking branch 'upstream/master' into dont-write-empty-…

…blocks

Signed-off-by: Ganesh Vernekar <cs15btech11018@iith.ac.in>

block.go

@@ -15,6 +15,7 @@
 package tsdb
 
 import (
+	"encoding/binary"
 	"encoding/json"
 	"io/ioutil"
 	"os"
@@ -252,6 +253,10 @@ type Block struct {
 	dir  string
 	meta BlockMeta
 
+	// Symbol Table Size in bytes.
+	// We maintain this variable to avoid recalculation everytime.
+	symbolTableSize uint64
+
 	chunkr     ChunkReader
 	indexr     IndexReader
 	tombstones TombstoneReader
@@ -279,12 +284,23 @@ func OpenBlock(dir string, pool chunkenc.Pool) (*Block, error) {
 		return nil, err
 	}
 
+	// Calculating symbol table size.
+	tmp := make([]byte, 8)
+	symTblSize := uint64(0)
+	for _, v := range ir.SymbolTable() {
+		// Size of varint length of the symbol.
+		symTblSize += uint64(binary.PutUvarint(tmp, uint64(len(v))))
+		// Size of the symbol.
+		symTblSize += uint64(len(v))
+	}
+
 	pb := &Block{
-		dir:        dir,
-		meta:       *meta,
-		chunkr:     cr,
-		indexr:     ir,
-		tombstones: tr,
+		dir:             dir,
+		meta:            *meta,
+		chunkr:          cr,
+		indexr:          ir,
+		tombstones:      tr,
+		symbolTableSize: symTblSize,
 	}
 	return pb, nil
 }
@@ -354,6 +370,11 @@ func (pb *Block) Tombstones() (TombstoneReader, error) {
 	return blockTombstoneReader{TombstoneReader: pb.tombstones, b: pb}, nil
 }
 
+// GetSymbolTableSize returns the Symbol Table Size in the index of this block.
+func (pb *Block) GetSymbolTableSize() uint64 {
+	return pb.symbolTableSize
+}
+
 func (pb *Block) setCompactionFailed() error {
 	pb.meta.Compaction.Failed = true
 	return writeMetaFile(pb.dir, &pb.meta)
@@ -487,10 +508,13 @@ Outer:
 func (pb *Block) CleanTombstones(dest string, c Compactor) (*ulid.ULID, error) {
 	numStones := 0
 
-	pb.tombstones.Iter(func(id uint64, ivs Intervals) error {
+	if err := pb.tombstones.Iter(func(id uint64, ivs Intervals) error {
 		numStones += len(ivs)
 		return nil
-	})
+	}); err != nil {
+		// This should never happen, as the iteration function only returns nil.
+		panic(err)
+	}
 	if numStones == 0 {
 		return nil, nil
 	}

checkpoint.go

@@ -109,6 +109,10 @@ func Checkpoint(logger log.Logger, w *wal.WAL, m, n int, keep func(id uint64) bo
 	stats := &CheckpointStats{}
 
 	var sr io.Reader
+	// We close everything explicitly because Windows needs files to be
+	// closed before being deleted. But we also have defer so that we close
+	// files if there is an error somewhere.
+	var closers []io.Closer
 	{
 		lastFn, k, err := LastCheckpoint(w.Dir())
 		if err != nil && err != ErrNotFound {
@@ -126,6 +130,7 @@ func Checkpoint(logger log.Logger, w *wal.WAL, m, n int, keep func(id uint64) bo
 				return nil, errors.Wrap(err, "open last checkpoint")
 			}
 			defer last.Close()
+			closers = append(closers, last)
 			sr = last
 		}
 
@@ -134,6 +139,7 @@ func Checkpoint(logger log.Logger, w *wal.WAL, m, n int, keep func(id uint64) bo
 			return nil, errors.Wrap(err, "create segment reader")
 		}
 		defer segsr.Close()
+		closers = append(closers, segsr)
 
 		if sr != nil {
 			sr = io.MultiReader(sr, segsr)
@@ -263,6 +269,9 @@ func Checkpoint(logger log.Logger, w *wal.WAL, m, n int, keep func(id uint64) bo
 	if err := fileutil.Replace(cpdirtmp, cpdir); err != nil {
 		return nil, errors.Wrap(err, "rename checkpoint directory")
 	}
+	if err := closeAll(closers...); err != nil {
+		return stats, errors.Wrap(err, "close opened files")
+	}
 	if err := w.Truncate(n + 1); err != nil {
 		// If truncating fails, we'll just try again at the next checkpoint.
 		// Leftover segments will just be ignored in the future if there's a checkpoint

cmd/tsdb/main.go

@@ -145,7 +145,9 @@ func (b *writeBenchmark) run() {
 		if err := b.storage.Close(); err != nil {
 			exitWithError(err)
 		}
-		b.stopProfiling()
+		if err := b.stopProfiling(); err != nil {
+			exitWithError(err)
+		}
 	})
 }
 
@@ -248,7 +250,9 @@ func (b *writeBenchmark) startProfiling() {
 	if err != nil {
 		exitWithError(fmt.Errorf("bench: could not create cpu profile: %v", err))
 	}
-	pprof.StartCPUProfile(b.cpuprof)
+	if err := pprof.StartCPUProfile(b.cpuprof); err != nil {
+		exitWithError(fmt.Errorf("bench: could not start CPU profile: %v", err))
+	}
 
 	// Start memory profiling.
 	b.memprof, err = os.Create(filepath.Join(b.outPath, "mem.prof"))
@@ -271,29 +275,36 @@ func (b *writeBenchmark) startProfiling() {
 	runtime.SetMutexProfileFraction(20)
 }
 
-func (b *writeBenchmark) stopProfiling() {
+func (b *writeBenchmark) stopProfiling() error {
 	if b.cpuprof != nil {
 		pprof.StopCPUProfile()
 		b.cpuprof.Close()
 		b.cpuprof = nil
 	}
 	if b.memprof != nil {
-		pprof.Lookup("heap").WriteTo(b.memprof, 0)
+		if err := pprof.Lookup("heap").WriteTo(b.memprof, 0); err != nil {
+			return fmt.Errorf("error writing mem profile: %v", err)
+		}
 		b.memprof.Close()
 		b.memprof = nil
 	}
 	if b.blockprof != nil {
-		pprof.Lookup("block").WriteTo(b.blockprof, 0)
+		if err := pprof.Lookup("block").WriteTo(b.blockprof, 0); err != nil {
+			return fmt.Errorf("error writing block profile: %v", err)
+		}
 		b.blockprof.Close()
 		b.blockprof = nil
 		runtime.SetBlockProfileRate(0)
 	}
 	if b.mtxprof != nil {
-		pprof.Lookup("mutex").WriteTo(b.mtxprof, 0)
+		if err := pprof.Lookup("mutex").WriteTo(b.mtxprof, 0); err != nil {
+			return fmt.Errorf("error writing mutex profile: %v", err)
+		}
 		b.mtxprof.Close()
 		b.mtxprof = nil
 		runtime.SetMutexProfileFraction(0)
 	}
+	return nil
 }
 
 func measureTime(stage string, f func()) time.Duration {

compact.go

@@ -99,7 +99,7 @@ func newCompactorMetrics(r prometheus.Registerer) *compactorMetrics {
 		Buckets: prometheus.ExponentialBuckets(1, 2, 10),
 	})
 	m.chunkSize = prometheus.NewHistogram(prometheus.HistogramOpts{
-		Name:    "prometheus_tsdb_compaction_chunk_size",
+		Name:    "prometheus_tsdb_compaction_chunk_size_bytes",
 		Help:    "Final size of chunks on their first compaction",
 		Buckets: prometheus.ExponentialBuckets(32, 1.5, 12),
 	})
@@ -109,7 +109,7 @@ func newCompactorMetrics(r prometheus.Registerer) *compactorMetrics {
 		Buckets: prometheus.ExponentialBuckets(4, 1.5, 12),
 	})
 	m.chunkRange = prometheus.NewHistogram(prometheus.HistogramOpts{
-		Name:    "prometheus_tsdb_compaction_chunk_range",
+		Name:    "prometheus_tsdb_compaction_chunk_range_seconds",
 		Help:    "Final time range of chunks on their first compaction",
 		Buckets: prometheus.ExponentialBuckets(100, 4, 10),
 	})
@@ -469,6 +469,7 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 	if err != nil {
 		return errors.Wrap(err, "open chunk writer")
 	}
+	defer chunkw.Close()
 	// Record written chunk sizes on level 1 compactions.
 	if meta.Compaction.Level == 1 {
 		chunkw = &instrumentedChunkWriter{
@@ -483,11 +484,15 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 	if err != nil {
 		return errors.Wrap(err, "open index writer")
 	}
+	defer indexw.Close()
 
 	if err := c.populateBlock(blocks, meta, indexw, chunkw); err != nil {
 		return errors.Wrap(err, "write compaction")
 	}
-
+	// We are explicitly closing them here to check for error even
+	// though these are covered under defer. This is because in Windows,
+	// you cannot delete these unless they are closed and the defer is to
+	// make sure they are closed if the function exits due to an error above.
 	if err = chunkw.Close(); err != nil {
 		return errors.Wrap(err, "close chunk writer")
 	}
@@ -651,7 +656,9 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
 		}
 
 		for _, chk := range chks {
-			c.chunkPool.Put(chk.Chunk)
+			if err := c.chunkPool.Put(chk.Chunk); err != nil {
+				return errors.Wrap(err, "put chunk")
+			}
 		}
 
 		for _, l := range lset {

db.go

@@ -119,11 +119,13 @@ type DB struct {
 
 type dbMetrics struct {
 	loadedBlocks         prometheus.GaugeFunc
+	symbolTableSize      prometheus.GaugeFunc
 	reloads              prometheus.Counter
 	reloadsFailed        prometheus.Counter
 	compactionsTriggered prometheus.Counter
 	cutoffs              prometheus.Counter
 	cutoffsFailed        prometheus.Counter
+	startTime            prometheus.GaugeFunc
 	tombCleanTimer       prometheus.Histogram
 }
 
@@ -138,6 +140,19 @@ func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
 		defer db.mtx.RUnlock()
 		return float64(len(db.blocks))
 	})
+	m.symbolTableSize = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "prometheus_tsdb_symbol_table_size_bytes",
+		Help: "Size of symbol table on disk (in bytes)",
+	}, func() float64 {
+		db.mtx.RLock()
+		blocks := db.blocks[:]
+		db.mtx.RUnlock()
+		symTblSize := uint64(0)
+		for _, b := range blocks {
+			symTblSize += b.GetSymbolTableSize()
+		}
+		return float64(symTblSize)
+	})
 	m.reloads = prometheus.NewCounter(prometheus.CounterOpts{
 		Name: "prometheus_tsdb_reloads_total",
 		Help: "Number of times the database reloaded block data from disk.",
@@ -158,6 +173,17 @@ func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
 		Name: "prometheus_tsdb_retention_cutoffs_failures_total",
 		Help: "Number of times the database failed to cut off block data from disk.",
 	})
+	m.startTime = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "prometheus_tsdb_lowest_timestamp",
+		Help: "Lowest timestamp value stored in the database.",
+	}, func() float64 {
+		db.mtx.RLock()
+		defer db.mtx.RUnlock()
+		if len(db.blocks) == 0 {
+			return float64(db.head.minTime)
+		}
+		return float64(db.blocks[0].meta.MinTime)
+	})
 	m.tombCleanTimer = prometheus.NewHistogram(prometheus.HistogramOpts{
 		Name: "prometheus_tsdb_tombstone_cleanup_seconds",
 		Help: "The time taken to recompact blocks to remove tombstones.",
@@ -166,11 +192,13 @@ func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
 	if r != nil {
 		r.MustRegister(
 			m.loadedBlocks,
+			m.symbolTableSize,
 			m.reloads,
 			m.reloadsFailed,
 			m.cutoffs,
 			m.cutoffsFailed,
 			m.compactionsTriggered,
+			m.startTime,
 			m.tombCleanTimer,
 		)
 	}
@@ -192,7 +220,7 @@ func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db
 	if err := repairBadIndexVersion(l, dir); err != nil {
 		return nil, err
 	}
-	// Migrate old WAL.
+	// Migrate old WAL if one exists.
 	if err := MigrateWAL(l, filepath.Join(dir, "wal")); err != nil {
 		return nil, errors.Wrap(err, "migrate WAL")
 	}
@@ -272,7 +300,7 @@ func (db *DB) run() {
 		case <-db.compactc:
 			db.metrics.compactionsTriggered.Inc()
 
-			_, err := db.compact()
+			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)
@@ -338,20 +366,20 @@ func (a dbAppender) Commit() error {
 // this is sufficient to reliably delete old data.
 // Old blocks are only deleted on reload based on the new block's parent information.
 // See DB.reload documentation for further information.
-func (db *DB) compact() (changes bool, err error) {
+func (db *DB) compact() (err error) {
 	db.cmtx.Lock()
 	defer db.cmtx.Unlock()
 
 	if !db.compactionsEnabled {
-		return false, nil
+		return nil
 	}
 
 	// Check whether we have pending head blocks that are ready to be persisted.
 	// They have the highest priority.
 	for {
 		select {
 		case <-db.stopc:
-			return changes, nil
+			return nil
 		default:
 		}
 		// The head has a compactable range if 1.5 level 0 ranges are between the oldest
@@ -374,14 +402,13 @@ func (db *DB) compact() (changes bool, err error) {
 			maxt: maxt - 1,
 		}
 		if _, err = db.compactor.Write(db.dir, head, mint, maxt, nil); err != nil {
-			return changes, errors.Wrap(err, "persist head block")
+			return errors.Wrap(err, "persist head block")
 		}
-		changes = true
 
 		runtime.GC()
 
 		if err := db.reload(); err != nil {
-			return changes, errors.Wrap(err, "reload blocks")
+			return errors.Wrap(err, "reload blocks")
 		}
 		db.mtx.RLock()
 		l := len(db.blocks)
@@ -408,31 +435,30 @@ func (db *DB) compact() (changes bool, err error) {
 	for {
 		plan, err := db.compactor.Plan(db.dir)
 		if err != nil {
-			return changes, errors.Wrap(err, "plan compaction")
+			return errors.Wrap(err, "plan compaction")
 		}
 		if len(plan) == 0 {
 			break
 		}
 
 		select {
 		case <-db.stopc:
-			return changes, nil
+			return nil
 		default:
 		}
 
 		if _, err := db.compactor.Compact(db.dir, plan...); err != nil {
-			return changes, errors.Wrapf(err, "compact %s", plan)
+			return errors.Wrapf(err, "compact %s", plan)
 		}
-		changes = true
 		runtime.GC()
 
 		if err := db.reload(); err != nil {
-			return changes, errors.Wrap(err, "reload blocks")
+			return errors.Wrap(err, "reload blocks")
 		}
 		runtime.GC()
 	}
 
-	return changes, nil
+	return nil
 }
 
 func (db *DB) getBlock(id ulid.ULID) (*Block, bool) {