From c5fa0b90c3d2ea0e750e71df60e49b06706e1948 Mon Sep 17 00:00:00 2001
From: beorn7 <beorn@soundcloud.com>
Date: Thu, 9 Apr 2015 15:57:11 +0200
Subject: [PATCH] Fix the case where a series in memory has 0 chunks, but
 chunks on disk.

This is actually completely normal for a freshly unarchived series.

Test added to expose.
---
 storage/local/crashrecovery.go |  2 +-
 storage/local/storage.go       |  7 ++--
 storage/local/storage_test.go  | 58 ++++++++++++++++++++++++++++++++++
 3 files changed, 61 insertions(+), 6 deletions(-)

diff --git a/storage/local/crashrecovery.go b/storage/local/crashrecovery.go
index aaa22802f..104766d16 100644
--- a/storage/local/crashrecovery.go
+++ b/storage/local/crashrecovery.go
@@ -410,7 +410,7 @@ func (p *persistence) cleanUpArchiveIndexes(
 			return err
 		}
 		if !deleted {
-			glog.Errorf("Fingerprint %s to be deleted from archivedFingerprintToTimeRange not found. This should never happen.", fp)
+			glog.Errorf("Fingerprint %v to be deleted from archivedFingerprintToTimeRange not found. This should never happen.", fp)
 		}
 		return nil
 	}); err != nil {
diff --git a/storage/local/storage.go b/storage/local/storage.go
index 1e8082fe6..e741dae61 100644
--- a/storage/local/storage.go
+++ b/storage/local/storage.go
@@ -861,11 +861,8 @@ func (s *memorySeriesStorage) writeMemorySeries(
 		return false
 	}
 	series.dropChunks(beforeTime)
-	if len(series.chunkDescs) == 0 { // All chunks dropped from memory series.
-		if !allDroppedFromPersistence {
-			glog.Errorf("All chunks dropped from memory but chunks left in persistence for fingerprint %v, series %v.", fp, series)
-			s.persistence.setDirty(true)
-		}
+	if len(series.chunkDescs) == 0 && allDroppedFromPersistence {
+		// All chunks dropped from both memory and persistence. Delete the series for good.
 		s.fpToSeries.del(fp)
 		s.numSeries.Dec()
 		s.seriesOps.WithLabelValues(memoryPurge).Inc()
diff --git a/storage/local/storage_test.go b/storage/local/storage_test.go
index 564d0dc5c..2f53b77ee 100644
--- a/storage/local/storage_test.go
+++ b/storage/local/storage_test.go
@@ -579,6 +579,64 @@ func testEvictAndPurgeSeries(t *testing.T, encoding chunkEncoding) {
 	if archived {
 		t.Fatal("archived series not dropped")
 	}
+
+	// Recreate series.
+	for _, sample := range samples {
+		s.Append(sample)
+	}
+	s.WaitForIndexing()
+
+	series, ok = ms.fpToSeries.get(fp)
+	if !ok {
+		t.Fatal("could not find series")
+	}
+
+	// Persist head chunk so we can safely archive.
+	series.headChunkClosed = true
+	ms.maintainMemorySeries(fp, clientmodel.Earliest)
+
+	// Archive metrics.
+	ms.fpToSeries.del(fp)
+	if err := ms.persistence.archiveMetric(
+		fp, series.metric, series.firstTime(), series.head().lastTime(),
+	); err != nil {
+		t.Fatal(err)
+	}
+
+	archived, _, _, err = ms.persistence.hasArchivedMetric(fp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !archived {
+		t.Fatal("not archived")
+	}
+
+	// Unarchive metrics.
+	ms.getOrCreateSeries(fp, clientmodel.Metric{})
+
+	series, ok = ms.fpToSeries.get(fp)
+	if !ok {
+		t.Fatal("could not find series")
+	}
+	archived, _, _, err = ms.persistence.hasArchivedMetric(fp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if archived {
+		t.Fatal("archived")
+	}
+	fmt.Println(series.headChunkClosed, len(series.chunkDescs))
+
+	// This will archive again, but must not drop it completely, despite the
+	// memorySeries being empty.
+	ms.maintainMemorySeries(fp, 1000)
+	archived, _, _, err = ms.persistence.hasArchivedMetric(fp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !archived {
+		t.Fatal("series purged completely")
+	}
 }
 
 func TestEvictAndPurgeSeriesChunkType0(t *testing.T) {