Merge pull request #603 from prometheus/beorn7/storage-resilience

Increase resilience of the storage against data corruption.

main.go

@@ -59,8 +59,10 @@ var (
 
 	checkpointInterval         = flag.Duration("storage.local.checkpoint-interval", 5*time.Minute, "The period at which the in-memory metrics and the chunks not yet persisted to series files are checkpointed.")
 	checkpointDirtySeriesLimit = flag.Int("storage.local.checkpoint-dirty-series-limit", 5000, "If approx. that many time series are in a state that would require a recovery operation after a crash, a checkpoint is triggered, even if the checkpoint interval hasn't passed yet. A recovery operation requires a disk seek. The default limit intends to keep the recovery time below 1min even on spinning disks. With SSD, recovery is much faster, so you might want to increase this value in that case to avoid overly frequent checkpoints.")
+	seriesSyncStrategy         = flag.String("storage.local.series-sync-strategy", "adaptive", "When to sync series files after modification. Possible values: 'never', 'always', 'adaptive'. Sync'ing slows down storage performance but reduces the risk of data loss in case of an OS crash. With the 'adaptive' strategy, series files are sync'd for as long as the storage is not too much behind on chunk persistence.")
 
-	storageDirty = flag.Bool("storage.local.dirty", false, "If set, the local storage layer will perform crash recovery even if the last shutdown appears to be clean.")
+	storageDirty          = flag.Bool("storage.local.dirty", false, "If set, the local storage layer will perform crash recovery even if the last shutdown appears to be clean.")
+	storagePedanticChecks = flag.Bool("storage.local.pedantic-checks", false, "If set, a crash recovery will perform checks on each series file. This might take a very long time.")
 
 	printVersion = flag.Bool("version", false, "Print version information.")
 )
@@ -87,6 +89,18 @@ func NewPrometheus() *prometheus {
 
 	notificationHandler := notification.NewNotificationHandler(*alertmanagerURL, *notificationQueueCapacity)
 
+	var syncStrategy local.SyncStrategy
+	switch *seriesSyncStrategy {
+	case "never":
+		syncStrategy = local.Never
+	case "always":
+		syncStrategy = local.Always
+	case "adaptive":
+		syncStrategy = local.Adaptive
+	default:
+		glog.Fatalf("Invalid flag value for 'storage.local.series-sync-strategy': %s", *seriesSyncStrategy)
+	}
+
 	o := &local.MemorySeriesStorageOptions{
 		MemoryChunks:               *numMemoryChunks,
 		MaxChunksToPersist:         *maxChunksToPersist,
@@ -94,7 +108,9 @@ func NewPrometheus() *prometheus {
 		PersistenceRetentionPeriod: *persistenceRetentionPeriod,
 		CheckpointInterval:         *checkpointInterval,
 		CheckpointDirtySeriesLimit: *checkpointDirtySeriesLimit,
-		Dirty: *storageDirty,
+		Dirty:          *storageDirty,
+		PedanticChecks: *storagePedanticChecks,
+		SyncStrategy:   syncStrategy,
 	}
 	memStorage, err := local.NewMemorySeriesStorage(o)
 	if err != nil {

storage/local/crashrecovery.go

@@ -193,6 +193,7 @@ func (p *persistence) sanitizeSeries(
 
 	bytesToTrim := fi.Size() % int64(chunkLen+chunkHeaderLen)
 	chunksInFile := int(fi.Size()) / (chunkLen + chunkHeaderLen)
+	modTime := fi.ModTime()
 	if bytesToTrim != 0 {
 		glog.Warningf(
 			"Truncating file %s to exactly %d chunks, trimming %d extraneous bytes.",
@@ -221,7 +222,11 @@ func (p *persistence) sanitizeSeries(
 		if s == nil {
 			panic("fingerprint mapped to nil pointer")
 		}
-		if bytesToTrim == 0 && s.chunkDescsOffset != -1 && chunksInFile == s.chunkDescsOffset+s.persistWatermark {
+		if !p.pedanticChecks &&
+			bytesToTrim == 0 &&
+			s.chunkDescsOffset != -1 &&
+			chunksInFile == s.chunkDescsOffset+s.persistWatermark &&
+			modTime.Equal(s.modTime) {
 			// Everything is consistent. We are good.
 			return fp, true
 		}
@@ -238,8 +243,9 @@ func (p *persistence) sanitizeSeries(
 				s.metric, fp, chunksInFile,
 			)
 			s.chunkDescs = nil
-			s.chunkDescsOffset = -1
+			s.chunkDescsOffset = chunksInFile
 			s.persistWatermark = 0
+			s.modTime = modTime
 			return fp, true
 		}
 		// This is the tricky one: We have chunks from heads.db, but
@@ -265,6 +271,7 @@ func (p *persistence) sanitizeSeries(
 		}
 		s.persistWatermark = len(cds)
 		s.chunkDescsOffset = 0
+		s.modTime = modTime
 
 		lastTime := cds[len(cds)-1].lastTime()
 		keepIdx := -1

storage/local/instrumentation.go

@@ -72,6 +72,12 @@ const (
 	// Op-types for chunkOps and chunkDescOps.
 	evict = "evict"
 	load  = "load"
+
+	seriesLocationLabel = "location"
+
+	// Maintenance types for maintainSeriesDuration.
+	maintainInMemory = "memory"
+	maintainArchived = "archived"
 )
 
 func init() {

storage/local/persistence.go

@@ -111,18 +111,21 @@ type persistence struct {
 	indexingQueueLength   prometheus.Gauge
 	indexingQueueCapacity prometheus.Metric
 	indexingBatchSizes    prometheus.Summary
-	indexingBatchLatency  prometheus.Summary
+	indexingBatchDuration prometheus.Summary
 	checkpointDuration    prometheus.Gauge
 
-	dirtyMtx      sync.Mutex     // Protects dirty and becameDirty.
-	dirty         bool           // true if persistence was started in dirty state.
-	becameDirty   bool           // true if an inconsistency came up during runtime.
-	dirtyFileName string         // The file used for locking and to mark dirty state.
-	fLock         flock.Releaser // The file lock to protect against concurrent usage.
+	dirtyMtx       sync.Mutex     // Protects dirty and becameDirty.
+	dirty          bool           // true if persistence was started in dirty state.
+	becameDirty    bool           // true if an inconsistency came up during runtime.
+	pedanticChecks bool           // true if crash recovery should check each series.
+	dirtyFileName  string         // The file used for locking and to mark dirty state.
+	fLock          flock.Releaser // The file lock to protect against concurrent usage.
+
+	shouldSync syncStrategy
 }
 
 // newPersistence returns a newly allocated persistence backed by local disk storage, ready to use.
-func newPersistence(basePath string, dirty bool) (*persistence, error) {
+func newPersistence(basePath string, dirty, pedanticChecks bool, shouldSync syncStrategy) (*persistence, error) {
 	dirtyPath := filepath.Join(basePath, dirtyFileName)
 	versionPath := filepath.Join(basePath, versionFileName)
 
@@ -211,12 +214,12 @@ func newPersistence(basePath string, dirty bool) (*persistence, error) {
 				Help:      "Quantiles for indexing batch sizes (number of metrics per batch).",
 			},
 		),
-		indexingBatchLatency: prometheus.NewSummary(
+		indexingBatchDuration: prometheus.NewSummary(
 			prometheus.SummaryOpts{
 				Namespace: namespace,
 				Subsystem: subsystem,
-				Name:      "indexing_batch_latency_milliseconds",
-				Help:      "Quantiles for batch indexing latencies in milliseconds.",
+				Name:      "indexing_batch_duration_milliseconds",
+				Help:      "Quantiles for batch indexing duration in milliseconds.",
 			},
 		),
 		checkpointDuration: prometheus.NewGauge(prometheus.GaugeOpts{
@@ -225,9 +228,11 @@ func newPersistence(basePath string, dirty bool) (*persistence, error) {
 			Name:      "checkpoint_duration_milliseconds",
 			Help:      "The duration (in milliseconds) it took to checkpoint in-memory metrics and head chunks.",
 		}),
-		dirty:         dirty,
-		dirtyFileName: dirtyPath,
-		fLock:         fLock,
+		dirty:          dirty,
+		pedanticChecks: pedanticChecks,
+		dirtyFileName:  dirtyPath,
+		fLock:          fLock,
+		shouldSync:     shouldSync,
 	}
 
 	if p.dirty {
@@ -259,7 +264,7 @@ func (p *persistence) Describe(ch chan<- *prometheus.Desc) {
 	ch <- p.indexingQueueLength.Desc()
 	ch <- p.indexingQueueCapacity.Desc()
 	p.indexingBatchSizes.Describe(ch)
-	p.indexingBatchLatency.Describe(ch)
+	p.indexingBatchDuration.Describe(ch)
 	ch <- p.checkpointDuration.Desc()
 }
 
@@ -270,7 +275,7 @@ func (p *persistence) Collect(ch chan<- prometheus.Metric) {
 	ch <- p.indexingQueueLength
 	ch <- p.indexingQueueCapacity
 	p.indexingBatchSizes.Collect(ch)
-	p.indexingBatchLatency.Collect(ch)
+	p.indexingBatchDuration.Collect(ch)
 	ch <- p.checkpointDuration
 }
 
@@ -340,7 +345,7 @@ func (p *persistence) persistChunks(fp clientmodel.Fingerprint, chunks []chunk)
 	if err != nil {
 		return -1, err
 	}
-	defer f.Close()
+	defer p.closeChunkFile(f)
 
 	if err := writeChunks(f, chunks); err != nil {
 		return -1, err
@@ -477,7 +482,11 @@ func (p *persistence) loadChunkDescs(fp clientmodel.Fingerprint, beforeTime clie
 //
 // (4.4) The varint-encoded persistWatermark. (Missing in v1.)
 //
-// (4.5) The varint-encoded chunkDescsOffset.
+// (4.5) The modification time of the series file as nanoseconds elapsed since
+// January 1, 1970 UTC. -1 if the modification time is unknown or no series file
+// exists yet. (Missing in v1.)
+//
+// (4.6) The varint-encoded chunkDescsOffset.
 //
 // (4.6) The varint-encoded savedFirstTime.
 //
@@ -569,6 +578,15 @@ func (p *persistence) checkpointSeriesMapAndHeads(fingerprintToSeries *seriesMap
 			if _, err = codable.EncodeVarint(w, int64(m.series.persistWatermark)); err != nil {
 				return
 			}
+			if m.series.modTime.IsZero() {
+				if _, err = codable.EncodeVarint(w, -1); err != nil {
+					return
+				}
+			} else {
+				if _, err = codable.EncodeVarint(w, m.series.modTime.UnixNano()); err != nil {
+					return
+				}
+			}
 			if _, err = codable.EncodeVarint(w, int64(m.series.chunkDescsOffset)); err != nil {
 				return
 			}
@@ -706,6 +724,7 @@ func (p *persistence) loadSeriesMapAndHeads() (sm *seriesMap, chunksToPersist in
 			return sm, chunksToPersist, nil
 		}
 		var persistWatermark int64
+		var modTime time.Time
 		if version != headsFormatLegacyVersion {
 			// persistWatermark only present in v2.
 			persistWatermark, err = binary.ReadVarint(r)
@@ -714,6 +733,15 @@ func (p *persistence) loadSeriesMapAndHeads() (sm *seriesMap, chunksToPersist in
 				p.dirty = true
 				return sm, chunksToPersist, nil
 			}
+			modTimeNano, err := binary.ReadVarint(r)
+			if err != nil {
+				glog.Warning("Could not decode modification time:", err)
+				p.dirty = true
+				return sm, chunksToPersist, nil
+			}
+			if modTimeNano != -1 {
+				modTime = time.Unix(0, modTimeNano)
+			}
 		}
 		chunkDescsOffset, err := binary.ReadVarint(r)
 		if err != nil {
@@ -784,6 +812,7 @@ func (p *persistence) loadSeriesMapAndHeads() (sm *seriesMap, chunksToPersist in
 			metric:           clientmodel.Metric(metric),
 			chunkDescs:       chunkDescs,
 			persistWatermark: int(persistWatermark),
+			modTime:          modTime,
 			chunkDescsOffset: int(chunkDescsOffset),
 			savedFirstTime:   clientmodel.Timestamp(savedFirstTime),
 			headChunkClosed:  persistWatermark >= numChunkDescs,
@@ -921,7 +950,7 @@ func (p *persistence) dropAndPersistChunks(
 		return
 	}
 	defer func() {
-		temp.Close()
+		p.closeChunkFile(temp)
 		if err == nil {
 			err = os.Rename(p.tempFileNameForFingerprint(fp), p.fileNameForFingerprint(fp))
 		}
@@ -962,6 +991,17 @@ func (p *persistence) deleteSeriesFile(fp clientmodel.Fingerprint) (int, error)
 	return numChunks, nil
 }
 
+// getSeriesFileModTime returns the modification time of the series file
+// belonging to the provided fingerprint. In case of an error, the zero value of
+// time.Time is returned.
+func (p *persistence) getSeriesFileModTime(fp clientmodel.Fingerprint) time.Time {
+	var modTime time.Time
+	if fi, err := os.Stat(p.fileNameForFingerprint(fp)); err == nil {
+		return fi.ModTime()
+	}
+	return modTime
+}
+
 // indexMetric queues the given metric for addition to the indexes needed by
 // getFingerprintsForLabelPair, getLabelValuesForLabelName, and
 // getFingerprintsModifiedBefore.  If the queue is full, this method blocks
@@ -1195,6 +1235,19 @@ func (p *persistence) openChunkFileForWriting(fp clientmodel.Fingerprint) (*os.F
 	// would still be detected.
 }
 
+// closeChunkFile first syncs the provided file if mandated so by the sync
+// strategy. Then it closes the file. Errors are logged.
+func (p *persistence) closeChunkFile(f *os.File) {
+	if p.shouldSync() {
+		if err := f.Sync(); err != nil {
+			glog.Error("Error syncing file:", err)
+		}
+	}
+	if err := f.Close(); err != nil {
+		glog.Error("Error closing chunk file:", err)
+	}
+}
+
 func (p *persistence) openChunkFileForReading(fp clientmodel.Fingerprint) (*os.File, error) {
 	return os.Open(p.fileNameForFingerprint(fp))
 }
@@ -1217,7 +1270,9 @@ func (p *persistence) processIndexingQueue() {
 	commitBatch := func() {
 		p.indexingBatchSizes.Observe(float64(batchSize))
 		defer func(begin time.Time) {
-			p.indexingBatchLatency.Observe(float64(time.Since(begin) / time.Millisecond))
+			p.indexingBatchDuration.Observe(
+				float64(time.Since(begin)) / float64(time.Millisecond),
+			)
 		}(time.Now())
 
 		if err := p.labelPairToFingerprints.IndexBatch(pairToFPs); err != nil {

storage/local/persistence_test.go

@@ -36,7 +36,7 @@ var (
 func newTestPersistence(t *testing.T, encoding chunkEncoding) (*persistence, test.Closer) {
 	*defaultChunkEncoding = int(encoding)
 	dir := test.NewTemporaryDirectory("test_persistence", t)
-	p, err := newPersistence(dir.Path(), false)
+	p, err := newPersistence(dir.Path(), false, false, func() bool { return false })
 	if err != nil {
 		dir.Close()
 		t.Fatal(err)

storage/local/series.go

@@ -143,6 +143,9 @@ type memorySeries struct {
 	// points to a non-persisted chunk. If all chunks are persisted, then
 	// persistWatermark == len(chunkDescs).
 	persistWatermark int
+	// The modification time of the series file. The zero value of time.Time
+	// is used to mark an unknown modification time.
+	modTime time.Time
 	// The chunkDescs in memory might not have all the chunkDescs for the
 	// chunks that are persisted to disk. The missing chunkDescs are all
 	// contiguous and at the tail end. chunkDescsOffset is the index of the

storage/local/storage.go

@@ -36,6 +36,12 @@ const (
 	fpMaxSweepTime    = 6 * time.Hour
 
 	maxEvictInterval = time.Minute
+
+	// If numChunskToPersist is this percentage of maxChunksToPersist, we
+	// consider the storage in "graceful degradation mode", i.e. we do not
+	// checkpoint anymore based on the dirty series count, and we do not
+	// sync series files anymore if using the adaptive sync strategy.
+	percentChunksToPersistForDegradation = 80
 )
 
 var (
@@ -56,6 +62,21 @@ type evictRequest struct {
 	evict bool
 }
 
+// SyncStrategy is an enum to select a sync strategy for series files.
+type SyncStrategy int
+
+// Possible values for SyncStrategy.
+const (
+	_ SyncStrategy = iota
+	Never
+	Always
+	Adaptive
+)
+
+// A syncStrategy is a function that returns whether series files should be
+// synced or not. It does not need to be goroutine safe.
+type syncStrategy func() bool
+
 type memorySeriesStorage struct {
 	fpLocker   *fingerprintLocker
 	fpToSeries *seriesMap
@@ -68,11 +89,10 @@ type memorySeriesStorage struct {
 
 	numChunksToPersist int64 // The number of chunks waiting for persistence.
 	maxChunksToPersist int   // If numChunksToPersist reaches this threshold, ingestion will stall.
+	degraded           bool
 
 	persistence *persistence
 
-	countPersistedHeadChunks chan struct{}
-
 	evictList                   *list.List
 	evictRequests               chan evictRequest
 	evictStopping, evictStopped chan struct{}
@@ -82,6 +102,7 @@ type memorySeriesStorage struct {
 	seriesOps                   *prometheus.CounterVec
 	ingestedSamplesCount        prometheus.Counter
 	invalidPreloadRequestsCount prometheus.Counter
+	maintainSeriesDuration      *prometheus.SummaryVec
 }
 
 // MemorySeriesStorageOptions contains options needed by
@@ -95,32 +116,15 @@ type MemorySeriesStorageOptions struct {
 	CheckpointInterval         time.Duration // How often to checkpoint the series map and head chunks.
 	CheckpointDirtySeriesLimit int           // How many dirty series will trigger an early checkpoint.
 	Dirty                      bool          // Force the storage to consider itself dirty on startup.
+	PedanticChecks             bool          // If dirty, perform crash-recovery checks on each series file.
+	SyncStrategy               SyncStrategy  // Which sync strategy to apply to series files.
 }
 
 // NewMemorySeriesStorage returns a newly allocated Storage. Storage.Serve still
 // has to be called to start the storage.
 func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) (Storage, error) {
-	p, err := newPersistence(o.PersistenceStoragePath, o.Dirty)
-	if err != nil {
-		return nil, err
-	}
-	glog.Info("Loading series map and head chunks...")
-	fpToSeries, numChunksToPersist, err := p.loadSeriesMapAndHeads()
-	if err != nil {
-		return nil, err
-	}
-	glog.Infof("%d series loaded.", fpToSeries.length())
-	numSeries := prometheus.NewGauge(prometheus.GaugeOpts{
-		Namespace: namespace,
-		Subsystem: subsystem,
-		Name:      "memory_series",
-		Help:      "The current number of series in memory.",
-	})
-	numSeries.Set(float64(fpToSeries.length()))
-
 	s := &memorySeriesStorage{
-		fpLocker:   newFingerprintLocker(1024),
-		fpToSeries: fpToSeries,
+		fpLocker: newFingerprintLocker(1024),
 
 		loopStopping:               make(chan struct{}),
 		loopStopped:                make(chan struct{}),
@@ -130,10 +134,6 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) (Storage, error) {
 		checkpointDirtySeriesLimit: o.CheckpointDirtySeriesLimit,
 
 		maxChunksToPersist: o.MaxChunksToPersist,
-		numChunksToPersist: numChunksToPersist,
-		persistence:        p,
-
-		countPersistedHeadChunks: make(chan struct{}, 100),
 
 		evictList:     list.New(),
 		evictRequests: make(chan evictRequest, evictRequestsCap),
@@ -146,7 +146,12 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) (Storage, error) {
 			Name:      "persist_errors_total",
 			Help:      "The total number of errors while persisting chunks.",
 		}),
-		numSeries: numSeries,
+		numSeries: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "memory_series",
+			Help:      "The current number of series in memory.",
+		}),
 		seriesOps: prometheus.NewCounterVec(
 			prometheus.CounterOpts{
 				Namespace: namespace,
@@ -168,8 +173,43 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) (Storage, error) {
 			Name:      "invalid_preload_requests_total",
 			Help:      "The total number of preload requests referring to a non-existent series. This is an indication of outdated label indexes.",
 		}),
+		maintainSeriesDuration: prometheus.NewSummaryVec(
+			prometheus.SummaryOpts{
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "maintain_series_duration_milliseconds",
+				Help:      "The duration (in milliseconds) it took to perform maintenance on a series.",
+			},
+			[]string{seriesLocationLabel},
+		),
 	}
 
+	var syncStrategy syncStrategy
+	switch o.SyncStrategy {
+	case Never:
+		syncStrategy = func() bool { return false }
+	case Always:
+		syncStrategy = func() bool { return true }
+	case Adaptive:
+		syncStrategy = func() bool { return !s.isDegraded() }
+	default:
+		panic("unknown sync strategy")
+	}
+
+	p, err := newPersistence(o.PersistenceStoragePath, o.Dirty, o.PedanticChecks, syncStrategy)
+	if err != nil {
+		return nil, err
+	}
+	s.persistence = p
+
+	glog.Info("Loading series map and head chunks...")
+	s.fpToSeries, s.numChunksToPersist, err = p.loadSeriesMapAndHeads()
+	if err != nil {
+		return nil, err
+	}
+	glog.Infof("%d series loaded.", s.fpToSeries.length())
+	s.numSeries.Set(float64(s.fpToSeries.length()))
+
 	return s, nil
 }
 
@@ -629,20 +669,14 @@ loop:
 		case fp := <-memoryFingerprints:
 			if s.maintainMemorySeries(fp, clientmodel.TimestampFromTime(time.Now()).Add(-s.dropAfter)) {
 				dirtySeriesCount++
-				// Check if we have enough "dirty" series so
-				// that we need an early checkpoint.  However,
-				// if we are already at 90% capacity of the
-				// persist queue, creating a checkpoint would be
-				// counterproductive, as it would slow down
-				// chunk persisting even more, while in a
-				// situation like that, where we are clearly
-				// lacking speed of disk maintenance, the best
-				// we can do for crash recovery is to work
-				// through the persist queue as quickly as
-				// possible. So only checkpoint if the persist
-				// queue is at most 90% full.
-				if dirtySeriesCount >= s.checkpointDirtySeriesLimit &&
-					s.getNumChunksToPersist() < s.maxChunksToPersist*9/10 {
+				// Check if we have enough "dirty" series so that we need an early checkpoint.
+				// However, if we are already behind persisting chunks, creating a checkpoint
+				// would be counterproductive, as it would slow down chunk persisting even more,
+				// while in a situation like that, where we are clearly lacking speed of disk
+				// maintenance, the best we can do for crash recovery is to persist chunks as
+				// quickly as possible. So only checkpoint if the storage is not in "graceful
+				// degratadion mode".
+				if dirtySeriesCount >= s.checkpointDirtySeriesLimit && !s.isDegraded() {
 					checkpointTimer.Reset(0)
 				}
 			}
@@ -691,6 +725,12 @@ loop:
 func (s *memorySeriesStorage) maintainMemorySeries(
 	fp clientmodel.Fingerprint, beforeTime clientmodel.Timestamp,
 ) (becameDirty bool) {
+	defer func(begin time.Time) {
+		s.maintainSeriesDuration.WithLabelValues(maintainInMemory).Observe(
+			float64(time.Since(begin)) / float64(time.Millisecond),
+		)
+	}(time.Now())
+
 	s.fpLocker.Lock(fp)
 	defer s.fpLocker.Unlock(fp)
 
@@ -773,6 +813,7 @@ func (s *memorySeriesStorage) writeMemorySeries(
 		}
 		s.incNumChunksToPersist(-len(cds))
 		chunkOps.WithLabelValues(persistAndUnpin).Add(float64(len(cds)))
+		series.modTime = s.persistence.getSeriesFileModTime(fp)
 	}()
 
 	// Get the actual chunks from underneath the chunkDescs.
@@ -809,7 +850,8 @@ func (s *memorySeriesStorage) writeMemorySeries(
 	series.dropChunks(beforeTime)
 	if len(series.chunkDescs) == 0 { // All chunks dropped from memory series.
 		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.numSeries.Dec()
@@ -823,7 +865,9 @@ func (s *memorySeriesStorage) writeMemorySeries(
 	} else {
 		series.chunkDescsOffset -= numDroppedFromPersistence
 		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
@@ -832,6 +876,12 @@ func (s *memorySeriesStorage) writeMemorySeries(
 // maintainArchivedSeries drops chunks older than beforeTime from an archived
 // series. If the series contains no chunks after that, it is purged entirely.
 func (s *memorySeriesStorage) maintainArchivedSeries(fp clientmodel.Fingerprint, beforeTime clientmodel.Timestamp) {
+	defer func(begin time.Time) {
+		s.maintainSeriesDuration.WithLabelValues(maintainArchived).Observe(
+			float64(time.Since(begin)) / float64(time.Millisecond),
+		)
+	}(time.Now())
+
 	s.fpLocker.Lock(fp)
 	defer s.fpLocker.Unlock(fp)
 
@@ -883,6 +933,28 @@ func (s *memorySeriesStorage) incNumChunksToPersist(by int) {
 	atomic.AddInt64(&s.numChunksToPersist, int64(by))
 }
 
+// isDegraded returns whether the storage is in "graceful degradation mode",
+// which is the case if the number of chunks waiting for persistence has reached
+// a percentage of maxChunksToPersist that exceeds
+// percentChunksToPersistForDegradation. The method is not goroutine safe (but
+// only ever called from the goroutine dealing with series maintenance).
+// Changes of degradation mode are logged.
+func (s *memorySeriesStorage) isDegraded() bool {
+	nowDegraded := s.getNumChunksToPersist() > s.maxChunksToPersist*percentChunksToPersistForDegradation/100
+	if s.degraded && !nowDegraded {
+		glog.Warning("Storage has left graceful degradation mode. Things are back to normal.")
+	} else if !s.degraded && nowDegraded {
+		glog.Warningf(
+			"%d chunks waiting for persistence (%d%% of the allowed maximum %d). Storage is now in graceful degradation mode. Series files are not synced anymore if following the adaptive strategy. Checkpoints are not performed more often than every %v.",
+			s.getNumChunksToPersist(),
+			s.getNumChunksToPersist()*100/s.maxChunksToPersist,
+			s.maxChunksToPersist,
+			s.checkpointInterval)
+	}
+	s.degraded = nowDegraded
+	return s.degraded
+}
+
 // Describe implements prometheus.Collector.
 func (s *memorySeriesStorage) Describe(ch chan<- *prometheus.Desc) {
 	s.persistence.Describe(ch)
@@ -894,8 +966,8 @@ func (s *memorySeriesStorage) Describe(ch chan<- *prometheus.Desc) {
 	s.seriesOps.Describe(ch)
 	ch <- s.ingestedSamplesCount.Desc()
 	ch <- s.invalidPreloadRequestsCount.Desc()
-
 	ch <- numMemChunksDesc
+	s.maintainSeriesDuration.Describe(ch)
 }
 
 // Collect implements prometheus.Collector.
@@ -917,10 +989,10 @@ func (s *memorySeriesStorage) Collect(ch chan<- prometheus.Metric) {
 	s.seriesOps.Collect(ch)
 	ch <- s.ingestedSamplesCount
 	ch <- s.invalidPreloadRequestsCount
-
 	ch <- prometheus.MustNewConstMetric(
 		numMemChunksDesc,
 		prometheus.GaugeValue,
 		float64(atomic.LoadInt64(&numMemChunks)),
 	)
+	s.maintainSeriesDuration.Collect(ch)
 }

storage/local/storage_test.go

@@ -161,6 +161,7 @@ func TestLoop(t *testing.T) {
 		PersistenceRetentionPeriod: 24 * 7 * time.Hour,
 		PersistenceStoragePath:     directory.Path(),
 		CheckpointInterval:         250 * time.Millisecond,
+		SyncStrategy:               Adaptive,
 	}
 	storage, err := NewMemorySeriesStorage(o)
 	if err != nil {
@@ -673,6 +674,7 @@ func benchmarkFuzz(b *testing.B, encoding chunkEncoding) {
 		PersistenceRetentionPeriod: time.Hour,
 		PersistenceStoragePath:     directory.Path(),
 		CheckpointInterval:         time.Second,
+		SyncStrategy:               Adaptive,
 	}
 	s, err := NewMemorySeriesStorage(o)
 	if err != nil {

storage/local/test_helpers.go

@@ -46,6 +46,7 @@ func NewTestStorage(t test.T, encoding chunkEncoding) (Storage, test.Closer) {
 		PersistenceRetentionPeriod: 24 * time.Hour * 365 * 100, // Enough to never trigger purging.
 		PersistenceStoragePath:     directory.Path(),
 		CheckpointInterval:         time.Hour,
+		SyncStrategy:               Adaptive,
 	}
 	storage, err := NewMemorySeriesStorage(o)
 	if err != nil {