Tsdb/wal rotate fix (#6423)

* rotate tsdb wal under mutex
Signed-off-by: Owen Diehl <ow.diehl@gmail.com>

* ensure we rotate and doublecheck conditions under lock in tsdb headmanager
Signed-off-by: Owen Diehl <ow.diehl@gmail.com>

pkg/storage/stores/tsdb/head_manager.go

@@ -231,6 +231,14 @@ func managerPerTenantDir(parent string) string {
 }
 
 func (m *HeadManager) Rotate(t time.Time) error {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
+
+	if m.activeHeads != nil && m.period.PeriodFor(t) == m.period.PeriodFor(m.activeHeads.start) {
+		// no-op, we've already rotated to the desired period
+		return nil
+	}
+
 	// create new wal
 	nextWALPath := walPath(m.dir, t)
 	nextWAL, err := newHeadWAL(m.log, nextWALPath, t)
@@ -255,12 +263,10 @@ func (m *HeadManager) Rotate(t time.Time) error {
 	}
 
 	stopPrev("previous cycle") // stop the previous wal if it hasn't been cleaned up yet
-	m.mtx.Lock()
 	m.prev = m.active
 	m.prevHeads = m.activeHeads
 	m.active = nextWAL
 	m.activeHeads = nextHeads
-	m.mtx.Unlock()
 	stopPrev("freshly rotated") // stop the newly rotated-out wal
 
 	// build tsdb from rotated-out period
@@ -292,10 +298,8 @@ func (m *HeadManager) Rotate(t time.Time) error {
 	}
 
 	// Now that the tsdbManager has the updated TSDBs, we can remove our references
-	m.mtx.Lock()
 	m.prevHeads = nil
 	m.prev = nil
-	m.mtx.Unlock()
 	return nil
 }