Open db in Read only mode

CHANGELOG.md

@@ -1,4 +1,5 @@
 ## master / unreleased
+ - [FEATURE] New `OpenReadOnly` API to allow opening a database in read only mode.
 
 ## 0.7.0
  - [CHANGE] tsdb now requires golang 1.12 or higher.

cmd/tsdb/main.go

@@ -62,9 +62,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{
@@ -74,17 +71,25 @@ func main() {
  }
  wb.run()
  case listCmd.FullCommand():
- db, err := tsdb.Open(*listPath, nil, nil, &safeDBOptions)
+ db, err := tsdb.OpenReadOnly(*listPath, nil, nil)
  if err != nil {
  exitWithError(err)
  }
- printBlocks(db.Blocks(), listCmdHumanReadable)
+ blocks, err := db.Blocks()
+ if err != nil {
+ exitWithError(err)
+ }
+ printBlocks(blocks, listCmdHumanReadable)
  case analyzeCmd.FullCommand():
- db, err := tsdb.Open(*analyzePath, nil, nil, &safeDBOptions)
+ db, err := tsdb.OpenReadOnly(*analyzePath, nil, nil)
+
+ if err != nil {
+ exitWithError(err)
+ }
+ blocks, err := db.Blocks()
  if err != nil {
  exitWithError(err)
  }
- blocks := db.Blocks()
  var block *tsdb.Block
  if *analyzeBlockID != "" {
  for _, b := range blocks {
@@ -101,7 +106,8 @@ func main() {
  }
  analyzeBlock(block, *analyzeLimit)
  case dumpCmd.FullCommand():
- db, err := tsdb.Open(*dumpPath, nil, nil, &safeDBOptions)
+ db, err := tsdb.OpenReadOnly(*dumpPath, nil, nil)
+
  if err != nil {
  exitWithError(err)
  }
@@ -548,7 +554,7 @@ 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)

db.go

@@ -153,6 +153,7 @@ type dbMetrics struct {
  tombCleanTimer prometheus.Histogram
  blocksBytes prometheus.Gauge
  sizeRetentionCount prometheus.Counter
+ walCorruptionsTotal prometheus.Counter
 }
 
 func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
@@ -222,6 +223,10 @@ func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
  Name: "prometheus_tsdb_size_retentions_total",
  Help: "The number of times that blocks were deleted because the maximum number of bytes was exceeded.",
  })
+ m.walCorruptionsTotal = prometheus.NewCounter(prometheus.CounterOpts{
+ Name: "prometheus_tsdb_wal_corruptions_total",
+ Help: "Total number of WAL corruptions.",
+ })
 
  if r != nil {
  r.MustRegister(
@@ -240,6 +245,107 @@ func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
  return m
 }
 
+// DBReadOnly provides APIs for read only operations on a database.
+type DBReadOnly struct {
+ logger log.Logger
+ dir string
+ registerer prometheus.Registerer
+}
+
+// OpenReadOnly returns a new DB in the given directory only for read operations.
+func OpenReadOnly(dir string, l log.Logger, r prometheus.Registerer) (*DBReadOnly, error) {
+ if _, err := os.Stat(dir); err != nil {
+ return nil, err
+ }
+
+ if l == nil {
+ l = log.NewNopLogger()
+ }
+
+ db := &DBReadOnly{
+ logger: l,
+ dir: dir,
+ registerer: r,
+ }
+
+ return db, nil
+}
+
+// Querier loads the wal and returns a new querier over the data partition for the given time range.
+// A goroutine must not handle more than one open Querier.
+func (dbRead *DBReadOnly) Querier(mint, maxt int64) (Querier, error) {
+ head, err := NewHead(dbRead.registerer, dbRead.logger, nil, 1)
+ if err != nil {
+ return nil, err
+ }
+
+ blocks, err := dbRead.Blocks()
+ if err != nil {
+ return nil, err
+ }
+
+ db := &DB{
+ dir: dbRead.dir,
+ logger: dbRead.logger,
+ head: head,
+ blocks: blocks,
+ }
+
+ // Set the min valid time for the ingested wal samples
+ // to be no lower than the maxt of the last block.
+ minValidTime := int64(math.MinInt64)
+ if len(blocks) > 0 {
+ minValidTime = blocks[len(blocks)-1].Meta().MaxTime
+ }
+
+ if err := db.head.Init(minValidTime); err != nil {
+ return nil, errors.Wrap(err, "read WAL")
+ }
+
+ return db.Querier(mint, maxt)
+}
+
+// Blocks returns the databases persisted blocks.
+func (dbRead *DBReadOnly) Blocks() ([]*Block, error) {
+ db := &DB{
+ dir: dbRead.dir,
+ logger: dbRead.logger,
+ }
+ db.metrics = newDBMetrics(db, dbRead.registerer)
+
+ loadable, corrupted, err := db.openBlocks()
+ if err != nil {
+ return nil, err
+ }
+
+ // Corrupted blocks that have been replaced by parents can be safely ignored.
+ for _, block := range loadable {
+ for _, b := range block.Meta().Compaction.Parents {
+ delete(corrupted, b.ULID)
+ }
+ }
+ if len(corrupted) > 0 {
+ return nil, fmt.Errorf("unexpected corrupted block:%v", corrupted)
+ }
+
+ if len(loadable) == 0 {
+ return nil, errors.New("no blocks found")
+ }
+
+ sort.Slice(loadable, func(i, j int) bool {
+ return loadable[i].Meta().MinTime < loadable[j].Meta().MinTime
+ })
+
+ blockMetas := make([]BlockMeta, 0, len(loadable))
+ for _, b := range loadable {
+ blockMetas = append(blockMetas, b.Meta())
+ }
+ if overlaps := OverlappingBlocks(blockMetas); len(overlaps) > 0 {
+ level.Warn(dbRead.logger).Log("msg", "overlapping blocks found during opening", "detail", overlaps.String())
+ }
+ return loadable, nil
+}
+
 // Open returns a new DB in the given directory.
 func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db *DB, err error) {
  if err := os.MkdirAll(dir, 0777); err != nil {
@@ -322,8 +428,17 @@ func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db
  minValidTime = blocks[len(blocks)-1].Meta().MaxTime
  }
 
- if err := db.head.Init(minValidTime); err != nil {
- return nil, errors.Wrap(err, "read WAL")
+ if initErr := db.head.Init(minValidTime); initErr != nil {
+ err := errors.Cause(initErr) // So that we can pick up errors even when wrapped.
+ if _, ok := err.(*wal.CorruptionErr); ok {
+ level.Warn(db.logger).Log("msg", "encountered WAL corruption error, attempting repair", "err", err)
+ db.metrics.walCorruptionsTotal.Inc()
+ if err := wlog.Repair(err); err != nil {
+ return nil, errors.Wrap(err, "repair corrupted WAL")
+ }
+ } else {
+ return nil, errors.Wrap(initErr, "read WAL")
+ }
  }
 
  go db.run()

head.go

@@ -95,7 +95,6 @@ type headMetrics struct {
  maxTime prometheus.GaugeFunc
  samplesAppended prometheus.Counter
  walTruncateDuration prometheus.Summary
- walCorruptionsTotal prometheus.Counter
  headTruncateFail prometheus.Counter
  headTruncateTotal prometheus.Counter
  checkpointDeleteFail prometheus.Counter
@@ -159,10 +158,6 @@ func newHeadMetrics(h *Head, r prometheus.Registerer) *headMetrics {
  Name: "prometheus_tsdb_wal_truncate_duration_seconds",
  Help: "Duration of WAL truncation.",
  })
- m.walCorruptionsTotal = prometheus.NewCounter(prometheus.CounterOpts{
- Name: "prometheus_tsdb_wal_corruptions_total",
- Help: "Total number of WAL corruptions.",
- })
  m.samplesAppended = prometheus.NewCounter(prometheus.CounterOpts{
  Name: "prometheus_tsdb_head_samples_appended_total",
  Help: "Total number of appended samples.",
@@ -206,7 +201,6 @@ func newHeadMetrics(h *Head, r prometheus.Registerer) *headMetrics {
  m.maxTime,
  m.gcDuration,
  m.walTruncateDuration,
- m.walCorruptionsTotal,
  m.samplesAppended,
  m.headTruncateFail,
  m.headTruncateTotal,
@@ -312,7 +306,7 @@ func (h *Head) updateMinMaxTime(mint, maxt int64) {
  }
 }
 
-func (h *Head) loadWAL(r *wal.Reader) error {
+func (h *Head) loadWAL(r *wal.Reader) (err error) {
  // Track number of samples that referenced a series we don't know about
  // for error reporting.
  var unknownRefs uint64
@@ -328,6 +322,18 @@ func (h *Head) loadWAL(r *wal.Reader) error {
  )
  wg.Add(n)
 
+ defer func() {
+ // For CorruptionErr ensure to terminate all workers before exiting.
+ if _, ok := err.(*wal.CorruptionErr); ok {
+ for i := 0; i < n; i++ {
+ close(inputs[i])
+ for range outputs[i] {
+ }
+ }
+ wg.Wait()
+ }
+ }()
+
  for i := 0; i < n; i++ {
  outputs[i] = make(chan []RefSample, 300)
  inputs[i] = make(chan []RefSample, 300)
@@ -345,9 +351,12 @@ func (h *Head) loadWAL(r *wal.Reader) error {
  samples []RefSample
  tstones []Stone
  allStones = newMemTombstones()
- err error
  )
- defer allStones.Close()
+ defer func() {
+ if err := allStones.Close(); err != nil {
+ level.Warn(h.logger).Log("msg", "closing memTombstones during wal read", "err", err)
+ }
+ }()
  for r.Next() {
  series, samples, tstones = series[:0], samples[:0], tstones[:0]
  rec := r.Record()
@@ -432,9 +441,6 @@ func (h *Head) loadWAL(r *wal.Reader) error {
  }
  }
  }
- if r.Err() != nil {
- return errors.Wrap(r.Err(), "read records")
- }
 
  // Signal termination to each worker and wait for it to close its output channel.
  for i := 0; i < n; i++ {
@@ -444,6 +450,10 @@ func (h *Head) loadWAL(r *wal.Reader) error {
  }
  wg.Wait()
 
+ if r.Err() != nil {
+ return errors.Wrap(r.Err(), "read records")
+ }
+
  if err := allStones.Iter(func(ref uint64, dranges Intervals) error {
  return h.chunkRewrite(ref, dranges)
  }); err != nil {
@@ -488,23 +498,19 @@ func (h *Head) Init(minValidTime int64) error {
  startFrom++
  }
 
- // Backfill segments from the last checkpoint onwards
+ // Backfill segments from the last checkpoint onwards.
  sr, err := wal.NewSegmentsRangeReader(wal.SegmentRange{Dir: h.wal.Dir(), First: startFrom, Last: -1})
  if err != nil {
  return errors.Wrap(err, "open WAL segments")
  }
-
+ defer func() {
+ if err := sr.Close(); err != nil {
+ level.Warn(h.logger).Log("msg", "error while closing the wal segments reader", "err", err)
+ }
+ }()
  err = h.loadWAL(wal.NewReader(sr))
- sr.Close() // Close the reader so that if there was an error the repair can remove the corrupted file under Windows.
- if err == nil {
- return nil
- }
- level.Warn(h.logger).Log("msg", "encountered WAL error, attempting repair", "err", err)
- h.metrics.walCorruptionsTotal.Inc()
- if err := h.wal.Repair(err); err != nil {
- return errors.Wrap(err, "repair corrupted WAL")
- }
- return nil
+
+ return err
 }
 
 // Truncate removes old data before mint from the head.