creates a single tsdb for entire wal recovery (#6203)

* creates a single tsdb for entire wal recovery

* tsdb wal replay builds TSDBs with unrounded timestamps to avoid clobbering

pkg/storage/stores/tsdb/head_manager.go

@@ -175,52 +175,36 @@ func (m *HeadManager) Start() error {
 		}
 	}
 
-	now := time.Now()
-	curPeriod := m.period.PeriodFor(now)
-
 	walsByPeriod, err := walsByPeriod(m.dir, m.period)
 	if err != nil {
 		return err
 	}
 	level.Info(m.log).Log("msg", "loaded wals by period", "groups", len(walsByPeriod))
 
-	m.activeHeads = newTenantHeads(now, m.shards, m.metrics, m.log)
-
 	// Load the shipper with any previously built TSDBs
 	if err := m.tsdbManager.Start(); err != nil {
 		return errors.Wrap(err, "failed to start tsdb manager")
 	}
 
-	// Build any old WALs into TSDBs for the shipper
+	// Build any old WALs into a TSDB for the shipper
+	var allWALs []WALIdentifier
 	for _, group := range walsByPeriod {
-		if group.period < curPeriod {
-			if err := m.tsdbManager.BuildFromWALs(
-				m.period.TimeForPeriod(group.period),
-				group.wals,
-			); err != nil {
-				return errors.Wrap(err, "building tsdb")
-			}
-			// Now that we've built tsdbs of this data, we can safely remove the WALs
-			if err := m.removeWALGroup(group); err != nil {
-				return errors.Wrapf(err, "removing wals for period %d", group.period)
-			}
-		}
+		allWALs = append(allWALs, group.wals...)
+	}
 
-		if group.period >= curPeriod {
-			if err := recoverHead(m.dir, m.activeHeads, group.wals); err != nil {
-				return errors.Wrap(err, "recovering tsdb head from wal")
-			}
-		}
+	now := time.Now()
+	if err := m.tsdbManager.BuildFromWALs(
+		now,
+		allWALs,
+	); err != nil {
+		return errors.Wrap(err, "building tsdb")
 	}
 
-	nextWALPath := walPath(m.dir, now)
-	nextWAL, err := newHeadWAL(m.log, nextWALPath, now)
-	if err != nil {
-		return errors.Wrapf(err, "creating tsdb wal: %s", nextWALPath)
+	if err := os.RemoveAll(managerWalDir(m.dir)); err != nil {
+		return errors.New("cleaning (removing) wal dir")
 	}
-	m.active = nextWAL
 
-	return nil
+	return m.Rotate(now)
 }
 
 func managerRequiredDirs(parent string) []string {

pkg/storage/stores/tsdb/head_manager_test.go

@@ -19,10 +19,22 @@ import (
 	"github.com/grafana/loki/pkg/storage/stores/tsdb/index"
 )
 
-type noopTSDBManager struct{ NoopIndex }
+type noopTSDBManager struct {
+	dir string
+	*tenantHeads
+}
 
-func (noopTSDBManager) BuildFromWALs(_ time.Time, _ []WALIdentifier) error { return nil }
-func (noopTSDBManager) Start() error                                       { return nil }
+func newNoopTSDBManager(dir string) noopTSDBManager {
+	return noopTSDBManager{
+		dir:         dir,
+		tenantHeads: newTenantHeads(time.Now(), defaultHeadManagerStripeSize, NewMetrics(nil), log.NewNopLogger()),
+	}
+}
+
+func (m noopTSDBManager) BuildFromWALs(_ time.Time, wals []WALIdentifier) error {
+	return recoverHead(m.dir, m.tenantHeads, wals)
+}
+func (m noopTSDBManager) Start() error { return nil }
 
 func chunkMetasToChunkRefs(user string, fp uint64, xs index.ChunkMetas) (res []ChunkRef) {
 	for _, x := range xs {
@@ -154,7 +166,7 @@ func Test_HeadManager_RecoverHead(t *testing.T) {
 		},
 	}
 
-	mgr := NewHeadManager(log.NewNopLogger(), dir, nil, noopTSDBManager{})
+	mgr := NewHeadManager(log.NewNopLogger(), dir, nil, newNoopTSDBManager(dir))
 	// This bit is normally handled by the Start() fn, but we're testing a smaller surface area
 	// so ensure our dirs exist
 	for _, d := range managerRequiredDirs(dir) {
@@ -237,7 +249,7 @@ func Test_HeadManager_Lifecycle(t *testing.T) {
 		},
 	}
 
-	mgr := NewHeadManager(log.NewNopLogger(), dir, nil, noopTSDBManager{})
+	mgr := NewHeadManager(log.NewNopLogger(), dir, nil, newNoopTSDBManager(dir))
 	w, err := newHeadWAL(log.NewNopLogger(), walPath(mgr.dir, curPeriod), curPeriod)
 	require.Nil(t, err)