Merge pull request #2277 from prometheus/beorn7/storage2

storage: Sanity-check number of loaded chunk descs

storage/local/series.go

@@ -462,10 +462,18 @@ func (s *memorySeries) preloadChunksForRange(
 		if err != nil {
 			return nopIter, err
 		}
+		if s.chunkDescsOffset != -1 && len(cds) != s.chunkDescsOffset {
+			return nopIter, fmt.Errorf(
+				"unexpected number of chunk descs loaded for fingerprint %v: expected %d, got %d",
+				fp, s.chunkDescsOffset, len(cds),
+			)
+		}
 		s.chunkDescs = append(cds, s.chunkDescs...)
 		s.chunkDescsOffset = 0
 		s.persistWatermark += len(cds)
-		firstChunkDescTime = s.chunkDescs[0].FirstTime()
+		if len(s.chunkDescs) > 0 {
+			firstChunkDescTime = s.chunkDescs[0].FirstTime()
+		}
 	}
 
 	if len(s.chunkDescs) == 0 || through.Before(firstChunkDescTime) {

storage/local/storage.go

@@ -938,6 +938,9 @@ func (s *MemorySeriesStorage) getOrCreateSeries(fp model.Fingerprint, m model.Me
 			// while (which is confusing as it makes the series
 			// appear as archived or purged).
 			cds, err = s.loadChunkDescs(fp, 0)
+			if err == nil && len(cds) == 0 {
+				err = fmt.Errorf("unarchived fingerprint %v (metric %v) has no chunks on disk", fp, m)
+			}
 			if err != nil {
 				s.quarantineSeries(fp, m, err)
 				return nil, err