adds Start() method to WAL interface to delay checkpointing until aft…

…er replay (#3339)

pkg/ingester/flush_test.go

@@ -62,6 +62,7 @@ func TestChunkFlushingShutdown(t *testing.T) {
 type fullWAL struct{}
 
 func (fullWAL) Log(_ *WALRecord) error { return &os.PathError{Err: syscall.ENOSPC} }
+func (fullWAL) Start()                 {}
 func (fullWAL) Stop() error            { return nil }
 
 func Benchmark_FlushLoop(b *testing.B) {

pkg/ingester/ingester.go

@@ -285,6 +285,7 @@ func (i *Ingester) starting(ctx context.Context) error {
 		i.metrics.walReplayDuration.Set(elapsed.Seconds())
 		level.Info(util_log.Logger).Log("msg", "recovery finished", "time", elapsed.String())
 
+		i.wal.Start()
 	}
 
 	i.InitFlushQueues()

pkg/ingester/wal.go

@@ -52,6 +52,7 @@ func (cfg *WALConfig) RegisterFlags(f *flag.FlagSet) {
 
 // WAL interface allows us to have a no-op WAL when the WAL is disabled.
 type WAL interface {
+	Start()
 	// Log marshalls the records and writes it into the WAL.
 	Log(*WALRecord) error
 	// Stop stops all the WAL operations.
@@ -60,6 +61,7 @@ type WAL interface {
 
 type noopWAL struct{}
 
+func (noopWAL) Start()               {}
 func (noopWAL) Log(*WALRecord) error { return nil }
 func (noopWAL) Stop() error          { return nil }
 
@@ -92,9 +94,12 @@ func newWAL(cfg WALConfig, registerer prometheus.Registerer, metrics *ingesterMe
 		seriesIter: seriesIter,
 	}
 
+	return w, nil
+}
+
+func (w *walWrapper) Start() {
 	w.wait.Add(1)
 	go w.run()
-	return w, nil
 }
 
 func (w *walWrapper) Log(record *WALRecord) error {