Open db in Read only mode

CHANGELOG.md

@@ -1,4 +1,9 @@
-## Master / unreleased
+## master / unreleased
+
+ - [FEATURE] Added `DBReadOnly` to allow opening a database in read only mode.
+    - `DBReadOnly.Blocks()` exposes a slice of `BlockReader`s and 
+    for this the  interface is refactored to return the full block meta instead of 
+    just MinTime/MaxTime. Required to allow reading the ULID of a block.
 
 ## 0.9.1
 

block.go

@@ -138,11 +138,8 @@ type BlockReader interface {
 	// Tombstones returns a TombstoneReader over the block's deleted data.
 	Tombstones() (TombstoneReader, error)
 
-	// MinTime returns the min time of the block.
-	MinTime() int64
-
-	// MaxTime returns the max time of the block.
-	MaxTime() int64
+	// Meta provides meta information about the block reader.
+	Meta() BlockMeta
 }
 
 // Appendable defines an entity to which data can be appended.

block_test.go

@@ -175,8 +175,7 @@ func TestBlockSize(t *testing.T) {
 			testutil.Ok(t, blockInit.Close())
 		}()
 		expSizeInit = blockInit.Size()
-		actSizeInit, err := testutil.DirSize(blockInit.Dir())
-		testutil.Ok(t, err)
+		actSizeInit := testutil.DirSize(t, blockInit.Dir())
 		testutil.Equals(t, expSizeInit, actSizeInit)
 	}
 
@@ -185,7 +184,7 @@ func TestBlockSize(t *testing.T) {
 		testutil.Ok(t, blockInit.Delete(1, 10, labels.NewMustRegexpMatcher("", ".*")))
 		expAfterDelete := blockInit.Size()
 		testutil.Assert(t, expAfterDelete > expSizeInit, "after a delete the block size should be bigger as the tombstone file should grow %v > %v", expAfterDelete, expSizeInit)
-		actAfterDelete, err := testutil.DirSize(blockDirInit)
+		actAfterDelete := testutil.DirSize(t, blockDirInit)
 		testutil.Ok(t, err)
 		testutil.Equals(t, expAfterDelete, actAfterDelete, "after a delete reported block size doesn't match actual disk size")
 
@@ -199,8 +198,7 @@ func TestBlockSize(t *testing.T) {
 			testutil.Ok(t, blockAfterCompact.Close())
 		}()
 		expAfterCompact := blockAfterCompact.Size()
-		actAfterCompact, err := testutil.DirSize(blockAfterCompact.Dir())
-		testutil.Ok(t, err)
+		actAfterCompact := testutil.DirSize(t, blockAfterCompact.Dir())
 		testutil.Assert(t, actAfterDelete > actAfterCompact, "after a delete and compaction the block size should be smaller %v,%v", actAfterDelete, actAfterCompact)
 		testutil.Equals(t, expAfterCompact, actAfterCompact, "after a delete and compaction reported block size doesn't match actual disk size")
 	}

cmd/tsdb/main.go

@@ -61,9 +61,6 @@ func main() {
 		dumpMaxTime          = dumpCmd.Flag("max-time", "maximum timestamp to dump").Default(strconv.FormatInt(math.MaxInt64, 10)).Int64()
 	)
 
-	safeDBOptions := *tsdb.DefaultOptions
-	safeDBOptions.RetentionDuration = 0
-
 	switch kingpin.MustParse(cli.Parse(os.Args[1:])) {
 	case benchWriteCmd.FullCommand():
 		wb := &writeBenchmark{
@@ -73,18 +70,35 @@ func main() {
 		}
 		wb.run()
 	case listCmd.FullCommand():
-		db, err := tsdb.Open(*listPath, nil, nil, &safeDBOptions)
+		db, err := tsdb.OpenDBReadOnly(*listPath, nil)
+		if err != nil {
+			exitWithError(err)
+		}
+		defer func() {
+			if err := db.Close(); err != nil {
+				exitWithError(err)
+			}
+		}()
+		blocks, err := db.Blocks()
 		if err != nil {
 			exitWithError(err)
 		}
-		printBlocks(db.Blocks(), listCmdHumanReadable)
+		printBlocks(blocks, listCmdHumanReadable)
 	case analyzeCmd.FullCommand():
-		db, err := tsdb.Open(*analyzePath, nil, nil, &safeDBOptions)
+		db, err := tsdb.OpenDBReadOnly(*analyzePath, nil)
 		if err != nil {
 			exitWithError(err)
 		}
-		blocks := db.Blocks()
-		var block *tsdb.Block
+		defer func() {
+			if err := db.Close(); err != nil {
+				exitWithError(err)
+			}
+		}()
+		blocks, err := db.Blocks()
+		if err != nil {
+			exitWithError(err)
+		}
+		var block tsdb.BlockReader
 		if *analyzeBlockID != "" {
 			for _, b := range blocks {
 				if b.Meta().ULID.String() == *analyzeBlockID {
@@ -100,10 +114,15 @@ func main() {
 		}
 		analyzeBlock(block, *analyzeLimit)
 	case dumpCmd.FullCommand():
-		db, err := tsdb.Open(*dumpPath, nil, nil, &safeDBOptions)
+		db, err := tsdb.OpenDBReadOnly(*dumpPath, nil)
 		if err != nil {
 			exitWithError(err)
 		}
+		defer func() {
+			if err := db.Close(); err != nil {
+				exitWithError(err)
+			}
+		}()
 		dumpSamples(db, *dumpMinTime, *dumpMaxTime)
 	}
 }
@@ -390,7 +409,7 @@ func exitWithError(err error) {
 	os.Exit(1)
 }
 
-func printBlocks(blocks []*tsdb.Block, humanReadable *bool) {
+func printBlocks(blocks []tsdb.BlockReader, humanReadable *bool) {
 	tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
 	defer tw.Flush()
 
@@ -417,9 +436,9 @@ func getFormatedTime(timestamp int64, humanReadable *bool) string {
 	return strconv.FormatInt(timestamp, 10)
 }
 
-func analyzeBlock(b *tsdb.Block, limit int) {
-	fmt.Printf("Block path: %s\n", b.Dir())
+func analyzeBlock(b tsdb.BlockReader, limit int) {
 	meta := b.Meta()
+	fmt.Printf("Block ID: %s\n", meta.ULID)
 	// Presume 1ms resolution that Prometheus uses.
 	fmt.Printf("Duration: %s\n", (time.Duration(meta.MaxTime-meta.MinTime) * 1e6).String())
 	fmt.Printf("Series: %d\n", meta.Stats.NumSeries)
@@ -570,11 +589,16 @@ func analyzeBlock(b *tsdb.Block, limit int) {
 	printInfo(postingInfos)
 }
 
-func dumpSamples(db *tsdb.DB, mint, maxt int64) {
+func dumpSamples(db *tsdb.DBReadOnly, mint, maxt int64) {
 	q, err := db.Querier(mint, maxt)
 	if err != nil {
 		exitWithError(err)
 	}
+	defer func() {
+		if err := q.Close(); err != nil {
+			exitWithError(err)
+		}
+	}()
 
 	ss, err := q.Select(labels.NewMustRegexpMatcher("", ".*"))
 	if err != nil {

compact.go

@@ -662,7 +662,7 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
 	}()
 	c.metrics.populatingBlocks.Set(1)
 
-	globalMaxt := blocks[0].MaxTime()
+	globalMaxt := blocks[0].Meta().MaxTime
 	for i, b := range blocks {
 		select {
 		case <-c.ctx.Done():
@@ -671,13 +671,13 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
 		}
 
 		if !overlapping {
-			if i > 0 && b.MinTime() < globalMaxt {
+			if i > 0 && b.Meta().MinTime < globalMaxt {
 				c.metrics.overlappingBlocks.Inc()
 				overlapping = true
 				level.Warn(c.logger).Log("msg", "found overlapping blocks during compaction", "ulid", meta.ULID)
 			}
-			if b.MaxTime() > globalMaxt {
-				globalMaxt = b.MaxTime()
+			if b.Meta().MaxTime > globalMaxt {
+				globalMaxt = b.Meta().MaxTime
 			}
 		}
 

compact_test.go

@@ -458,8 +458,7 @@ type erringBReader struct{}
 func (erringBReader) Index() (IndexReader, error)          { return nil, errors.New("index") }
 func (erringBReader) Chunks() (ChunkReader, error)         { return nil, errors.New("chunks") }
 func (erringBReader) Tombstones() (TombstoneReader, error) { return nil, errors.New("tombstones") }
-func (erringBReader) MinTime() int64                       { return 0 }
-func (erringBReader) MaxTime() int64                       { return 0 }
+func (erringBReader) Meta() BlockMeta                      { return BlockMeta{} }
 
 type nopChunkWriter struct{}