Remove old temporary checkpoints (#2726)

If an ingester crashes while doing a checkpoint, which is likely as
checkpoints are almost always happening, then a checkpoint like
checkpoint.123456.tmp will be left on disk. When the next checkpoint
succeeds, the old .tmp checkpoints are not deleted even though it will
never be used. After enough crashes it is possible to fill the entire
disk with .tmp checkpoints, bringing down an ingester.

Signed-off-by: Chris Marchbanks <[email protected]>

Author: csmarchbanks

CHANGELOG.md

@@ -139,6 +139,7 @@
 * [BUGFIX] Cassandra: fixed an edge case leading to an invalid CQL query when querying the index on a Cassandra store. #2639
 * [BUGFIX] Ingester: increment series per metric when recovering from WAL or transfer. #2674
 * [BUGFIX] Fixed `wrong number of arguments for 'mget' command` Redis error when a query has no chunks to lookup from storage. #2700
+* [BUGFIX] Ingester: Automatically remove old tmp checkpoints, fixing a potential disk space leak after an ingester crashes.
 
 ## 1.1.0 / 2020-05-21
 

pkg/ingester/wal.go

@@ -8,9 +8,9 @@ import (
 	"math"
 	"os"
 	"path/filepath"
+	"regexp"
 	"runtime"
 	"strconv"
-	"strings"
 	"sync"
 	"time"
 
@@ -413,16 +413,13 @@ func lastCheckpoint(dir string) (string, int, error) {
 	for i := 0; i < len(dirs); i++ {
 		di := dirs[i]
 
-		if !strings.HasPrefix(di.Name(), checkpointPrefix) {
+		idx, err := checkpointIndex(di.Name(), false)
+		if err != nil {
 			continue
 		}
 		if !di.IsDir() {
 			return "", -1, fmt.Errorf("checkpoint %s is not a directory", di.Name())
 		}
-		idx, err := strconv.Atoi(di.Name()[len(checkpointPrefix):])
-		if err != nil {
-			continue
-		}
 		if idx > maxIdx {
 			checkpointDir = di.Name()
 			maxIdx = idx
@@ -450,10 +447,7 @@ func (w *walWrapper) deleteCheckpoints(maxIndex int) (err error) {
 		return err
 	}
 	for _, fi := range files {
-		if !strings.HasPrefix(fi.Name(), checkpointPrefix) {
-			continue
-		}
-		index, err := strconv.Atoi(fi.Name()[len(checkpointPrefix):])
+		index, err := checkpointIndex(fi.Name(), true)
 		if err != nil || index >= maxIndex {
 			continue
 		}
@@ -464,6 +458,23 @@ func (w *walWrapper) deleteCheckpoints(maxIndex int) (err error) {
 	return errs.Err()
 }
 
+var checkpointRe = regexp.MustCompile("^" + regexp.QuoteMeta(checkpointPrefix) + "(\\d+)(\\.tmp)?$")
+
+// checkpointIndex returns the index of a given checkpoint file. It handles
+// both regular and temporary checkpoints according to the includeTmp flag. If
+// the file is not a checkpoint it returns an error.
+func checkpointIndex(filename string, includeTmp bool) (int, error) {
+	result := checkpointRe.FindStringSubmatch(filename)
+	if len(result) < 2 {
+		return 0, errors.New("file is not a checkpoint")
+	}
+	// Filter out temporary checkpoints if desired.
+	if !includeTmp && len(result) == 3 && result[2] != "" {
+		return 0, errors.New("temporary checkpoint")
+	}
+	return strconv.Atoi(result[1])
+}
+
 // checkpointSeries write the chunks of the series to the checkpoint.
 func (w *walWrapper) checkpointSeries(userID string, fp model.Fingerprint, series *memorySeries, wireChunks []client.Chunk, b []byte) ([]client.Chunk, []byte, error) {
 	var err error

pkg/ingester/wal_test.go

@@ -285,6 +285,66 @@ func TestMigrationToTypedRecord(t *testing.T) {
 	require.Equal(t, checkpointRecord, newCheckpointRecordDecoded)
 }
 
+func TestCheckpointIndex(t *testing.T) {
+	tcs := []struct {
+		filename    string
+		includeTmp  bool
+		index       int
+		shouldError bool
+	}{
+		{
+			filename:    "checkpoint.123456",
+			includeTmp:  false,
+			index:       123456,
+			shouldError: false,
+		},
+		{
+			filename:    "checkpoint.123456",
+			includeTmp:  true,
+			index:       123456,
+			shouldError: false,
+		},
+		{
+			filename:    "checkpoint.123456.tmp",
+			includeTmp:  true,
+			index:       123456,
+			shouldError: false,
+		},
+		{
+			filename:    "checkpoint.123456.tmp",
+			includeTmp:  false,
+			shouldError: true,
+		},
+		{
+			filename:    "not-checkpoint.123456.tmp",
+			includeTmp:  true,
+			shouldError: true,
+		},
+		{
+			filename:    "checkpoint.123456.tmp2",
+			shouldError: true,
+		},
+		{
+			filename:    "checkpoints123456",
+			shouldError: true,
+		},
+		{
+			filename:    "012345",
+			shouldError: true,
+		},
+	}
+	for _, tc := range tcs {
+		index, err := checkpointIndex(tc.filename, tc.includeTmp)
+		if tc.shouldError {
+			require.Error(t, err, "filename: %s, includeTmp: %t", tc.filename, tc.includeTmp)
+			continue
+		}
+
+		require.NoError(t, err, "filename: %s, includeTmp: %t", tc.filename, tc.includeTmp)
+		require.Equal(t, tc.index, index)
+	}
+}
+
 func BenchmarkWALReplay(b *testing.B) {
 	dirname, err := ioutil.TempDir("", "cortex-wal")
 	require.NoError(b, err)