Increase resilience of the storage against data corruption - step 1.
Step 1: Admit the problem by turning the various "panic"s into logged errors, followed by marking the persistence as dirty.
This commit is contained in:
parent
da7c0461c6
commit
3d8d8928be
|
@ -809,7 +809,8 @@ func (s *memorySeriesStorage) writeMemorySeries(
|
||||||
series.dropChunks(beforeTime)
|
series.dropChunks(beforeTime)
|
||||||
if len(series.chunkDescs) == 0 { // All chunks dropped from memory series.
|
if len(series.chunkDescs) == 0 { // All chunks dropped from memory series.
|
||||||
if !allDroppedFromPersistence {
|
if !allDroppedFromPersistence {
|
||||||
panic("all chunks dropped from memory but chunks left in persistence")
|
glog.Errorf("All chunks dropped from memory but chunks left in persistence for fingerprint %v, series %v.", fp, series)
|
||||||
|
s.persistence.setDirty(true)
|
||||||
}
|
}
|
||||||
s.fpToSeries.del(fp)
|
s.fpToSeries.del(fp)
|
||||||
s.numSeries.Dec()
|
s.numSeries.Dec()
|
||||||
|
@ -823,7 +824,9 @@ func (s *memorySeriesStorage) writeMemorySeries(
|
||||||
} else {
|
} else {
|
||||||
series.chunkDescsOffset -= numDroppedFromPersistence
|
series.chunkDescsOffset -= numDroppedFromPersistence
|
||||||
if series.chunkDescsOffset < 0 {
|
if series.chunkDescsOffset < 0 {
|
||||||
panic("dropped more chunks from persistence than from memory")
|
glog.Errorf("Dropped more chunks from persistence than from memory for fingerprint %v, series %v.", fp, series)
|
||||||
|
s.persistence.setDirty(true)
|
||||||
|
series.chunkDescsOffset = -1 // Makes sure it will be looked at during crash recovery.
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return false
|
return false
|
||||||
|
|
Loading…
Reference in New Issue