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.)
This commit is contained in:
beorn7 2016-12-13 18:41:02 +01:00
parent e714079cf2
commit 253be23c00
2 changed files with 12 additions and 1 deletions

View file

@ -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) {

View file

@ -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