From 253be23c00eba5592e1673845a38f4fd08efff83 Mon Sep 17 00:00:00 2001
From: beorn7 <beorn@soundcloud.com>
Date: Tue, 13 Dec 2016 18:41:02 +0100
Subject: [PATCH] storage: Sanity-check number of loaded chunk descs

Two cases:

- An unarchived metric must have at least one chunk desc loaded upon
  unarchival. Otherwise, the file is gone or has size 0, which is an
  inconsistency (because the series is still indexed in the archive
  index). Hence, quarantining is triggered.

- If loading the chunk descs of a series with a known chunkDescsOffset
  (i.e. != -1), the number of chunks loaded must be equal to
  chunkDescsOffset. If not, there is a data corruption. An error is
  returned, which leads to qurantining.

In any case, there is a guard added to not access the 1st element of
an empty chunkDescs slice. (That's what triggered the crashes in issue
2249.)  A time series with unknown chunkDescsOffset and no chunks in
memory and no chunks on disk either could trigger that case. I would
assume such a "null series" doesn't exist, but it's not entirely
unthinkable and unreasonable to happen (perhaps in future uses of the
storage). (Create a series, and then something tries to preload chunks
before the first sample is added.)
---
 storage/local/series.go  | 10 +++++++++-
 storage/local/storage.go |  3 +++
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/storage/local/series.go b/storage/local/series.go
index 4acc40884..2e4c2b1f0 100644
--- a/storage/local/series.go
+++ b/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) {
diff --git a/storage/local/storage.go b/storage/local/storage.go
index fd571f8f7..8c9f54a4c 100644
--- a/storage/local/storage.go
+++ b/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