cmd/thanos/receive: reduce WAL replays at startup

cmd/thanos/receive.go

@@ -230,30 +230,22 @@ func runReceive(
 			defer close(dbReady)
 			defer close(uploadC)
 
+			// Before actually starting, we need to make sure the
+			// WAL is flushed. The WAL is flushed after the
+			// hashring is loaded.
 			db := receive.NewFlushableStorage(
 				dataDir,
 				log.With(logger, "component", "tsdb"),
 				reg,
 				tsdbCfg,
 			)
 
-			// Before actually starting, we need to make sure the
-			// WAL is flushed. The WAL is flushed after the
-			// hashring ring is loaded.
-			if err := db.Open(); err != nil {
-				return errors.Wrap(err, "opening storage")
-			}
-
 			// Before quitting, ensure the WAL is flushed and the DB is closed.
 			defer func() {
 				if err := db.Flush(); err != nil {
 					level.Warn(logger).Log("err", err, "msg", "failed to flush storage")
 					return
 				}
-				if err := db.Close(); err != nil {
-					level.Warn(logger).Log("err", err, "msg", "failed to close storage")
-					return
-				}
 			}()
 
 			for {
@@ -267,6 +259,9 @@ func runReceive(
 					if err := db.Flush(); err != nil {
 						return errors.Wrap(err, "flushing storage")
 					}
+					if err := db.Open(); err != nil {
+						return errors.Wrap(err, "opening storage")
+					}
 					if upload {
 						uploadC <- struct{}{}
 						<-uploadDone

pkg/receive/tsdb.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/storage/tsdb"
@@ -69,32 +70,32 @@ func (f *FlushableStorage) open() error {
 }
 
 // Flush temporarily stops the storage and flushes the WAL to blocks.
-// Note: this operation leaves the storage in the same state it was in.
+// Note: this operation leaves the storage closed.
 func (f *FlushableStorage) Flush() error {
+	_, err := os.Stat(filepath.Join(f.path, "wal"))
+	if os.IsNotExist(err) {
+		level.Info(f.l).Log("msg", "No WAL was found for flushing; ignoring.")
+		return nil
+	}
+	if err != nil {
+		return errors.Wrap(err, "stating WAL")
+	}
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	var reopen bool
 	if !f.stopped {
 		if err := f.DB.Close(); err != nil {
 			return errors.Wrap(err, "stopping storage")
 		}
 		f.stopped = true
-		reopen = true
 	}
-	ro, err := promtsdb.OpenDBReadOnly(f.Dir(), f.l)
+	ro, err := promtsdb.OpenDBReadOnly(f.path, f.l)
 	if err != nil {
 		return errors.Wrap(err, "opening read-only DB")
 	}
-	if err := ro.FlushWAL(f.Dir()); err != nil {
+	if err := ro.FlushWAL(f.path); err != nil {
 		return errors.Wrap(err, "flushing WAL")
 	}
-	if err := os.RemoveAll(filepath.Join(f.Dir(), "wal")); err != nil {
-		return errors.Wrap(err, "removing stale WAL")
-	}
-	if reopen {
-		return errors.Wrap(f.open(), "re-starting storage")
-	}
-	return nil
+	return errors.Wrap(os.RemoveAll(filepath.Join(f.path, "wal")), "removing stale WAL")
 }
 
 // Close stops the storage.