diff --git a/storage/local/storage.go b/storage/local/storage.go index aa0745e98..63eb94698 100644 --- a/storage/local/storage.go +++ b/storage/local/storage.go @@ -1433,6 +1433,11 @@ func (s *MemorySeriesStorage) maintainMemorySeries( // contains no chunks after dropping old chunks, it is purged entirely. In that // case, the method returns true. // +// If a persist error is encountered, the series is queued for quarantine. In +// that case, the method returns true, too, because the series should not be +// processed anymore (even if it will only be gone for real once quarantining +// has been completed). +// // The caller must have locked the fp. func (s *MemorySeriesStorage) writeMemorySeries( fp model.Fingerprint, series *memorySeries, beforeTime model.Time, @@ -1474,7 +1479,7 @@ func (s *MemorySeriesStorage) writeMemorySeries( var offset int offset, persistErr = s.persistence.persistChunks(fp, chunks) if persistErr != nil { - return false + return true } if series.chunkDescsOffset == -1 { // This is the first chunk persisted for a newly created @@ -1488,10 +1493,10 @@ func (s *MemorySeriesStorage) writeMemorySeries( newFirstTime, offset, numDroppedFromPersistence, allDroppedFromPersistence, persistErr := s.persistence.dropAndPersistChunks(fp, beforeTime, chunks) if persistErr != nil { - return false + return true } if persistErr = series.dropChunks(beforeTime); persistErr != nil { - return false + return true } if len(series.chunkDescs) == 0 && allDroppedFromPersistence { // All chunks dropped from both memory and persistence. Delete the series for good.