cleanup tmp file if file renaming failed

Signed-off-by: zhulongcheng <[email protected]>

block.go

@@ -29,6 +29,7 @@ import (
 	"github.com/prometheus/tsdb/chunkenc"
 	"github.com/prometheus/tsdb/chunks"
 	tsdb_errors "github.com/prometheus/tsdb/errors"
+	"github.com/prometheus/tsdb/fileutil"
 	"github.com/prometheus/tsdb/index"
 	"github.com/prometheus/tsdb/labels"
 )
@@ -236,6 +237,9 @@ func writeMetaFile(dir string, meta *BlockMeta) error {
 	// Make any changes to the file appear atomic.
 	path := filepath.Join(dir, metaFilename)
 	tmp := path + ".tmp"
+	defer func() {
+		os.RemoveAll(tmp)
+	}()
 
 	f, err := os.Create(tmp)
 	if err != nil {
@@ -259,7 +263,12 @@ func writeMetaFile(dir string, meta *BlockMeta) error {
 	if err := f.Close(); err != nil {
 		return err
 	}
-	return renameFile(tmp, path)
+
+	if err := fileutil.Rename(tmp, path); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // Block represents a directory of time series data covering a continuous time range.

compact.go

@@ -639,7 +639,7 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 	df = nil
 
 	// Block successfully written, make visible and remove old ones.
-	if err := renameFile(tmp, dir); err != nil {
+	if err := fileutil.Rename(tmp, dir); err != nil {
 		return errors.Wrap(err, "rename block dir")
 	}
 
@@ -1013,24 +1013,3 @@ func (c *compactionMerger) Err() error {
 func (c *compactionMerger) At() (labels.Labels, []chunks.Meta, Intervals) {
 	return c.l, c.c, c.intervals
 }
-
-func renameFile(from, to string) error {
-	if err := os.RemoveAll(to); err != nil {
-		return err
-	}
-	if err := os.Rename(from, to); err != nil {
-		return err
-	}
-
-	// Directory was renamed; sync parent dir to persist rename.
-	pdir, err := fileutil.OpenDir(filepath.Dir(to))
-	if err != nil {
-		return err
-	}
-
-	if err = pdir.Sync(); err != nil {
-		pdir.Close()
-		return err
-	}
-	return pdir.Close()
-}

repair.go

@@ -23,6 +23,8 @@ import (
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	"github.com/pkg/errors"
+	tsdb_errors "github.com/prometheus/tsdb/errors"
+	"github.com/prometheus/tsdb/fileutil"
 )
 
 // repairBadIndexVersion repairs an issue in index and meta.json persistence introduced in
@@ -38,6 +40,16 @@ func repairBadIndexVersion(logger log.Logger, dir string) error {
 	wrapErr := func(err error, d string) error {
 		return errors.Wrapf(err, "block dir: %q", d)
 	}
+
+	tmpFiles := make([]string, 0, len(dir))
+	defer func() {
+		for _, tmp := range tmpFiles {
+			if err := os.RemoveAll(tmp); err != nil {
+				level.Error(logger).Log("msg", "remove tmp file", "err", err.Error())
+			}
+		}
+	}()
+
 	for _, d := range dirs {
 		meta, err := readBogusMetaFile(d)
 		if err != nil {
@@ -63,27 +75,36 @@ func repairBadIndexVersion(logger log.Logger, dir string) error {
 		if err != nil {
 			return wrapErr(err, d)
 		}
+		tmpFiles = append(tmpFiles, repl.Name())
+
 		broken, err := os.Open(filepath.Join(d, indexFilename))
 		if err != nil {
 			return wrapErr(err, d)
 		}
 		if _, err := io.Copy(repl, broken); err != nil {
 			return wrapErr(err, d)
 		}
+
+		var merr tsdb_errors.MultiError
+
 		// Set the 5th byte to 2 to indicate the correct file format version.
 		if _, err := repl.WriteAt([]byte{2}, 4); err != nil {
-			return wrapErr(err, d)
+			merr.Add(wrapErr(err, d))
+			merr.Add(wrapErr(repl.Close(), d))
+			return merr.Err()
 		}
 		if err := repl.Sync(); err != nil {
-			return wrapErr(err, d)
+			merr.Add(wrapErr(err, d))
+			merr.Add(wrapErr(repl.Close(), d))
+			return merr.Err()
 		}
 		if err := repl.Close(); err != nil {
 			return wrapErr(err, d)
 		}
 		if err := broken.Close(); err != nil {
 			return wrapErr(err, d)
 		}
-		if err := renameFile(repl.Name(), broken.Name()); err != nil {
+		if err := fileutil.Replace(repl.Name(), broken.Name()); err != nil {
 			return wrapErr(err, d)
 		}
 		// Reset version of meta.json to 1.

tombstones.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/tsdb/encoding"
 	tsdb_errors "github.com/prometheus/tsdb/errors"
+	"github.com/prometheus/tsdb/fileutil"
 )
 
 const tombstoneFilename = "tombstones"
@@ -64,6 +65,7 @@ func writeTombstoneFile(dir string, tr TombstoneReader) error {
 		if f != nil {
 			f.Close()
 		}
+		os.RemoveAll(tmp)
 	}()
 
 	buf := encoding.Encbuf{B: make([]byte, 3*binary.MaxVarintLen64)}
@@ -111,7 +113,11 @@ func writeTombstoneFile(dir string, tr TombstoneReader) error {
 		return err
 	}
 	f = nil
-	return renameFile(tmp, path)
+
+	if err := fileutil.Replace(tmp, path); err != nil {
+		return err
+	}
+	return nil
 }
 
 // Stone holds the information on the posting and time-range

wal.go

@@ -338,6 +338,10 @@ func (w *SegmentWAL) Truncate(mint int64, keep func(uint64) bool) error {
 	if err != nil {
 		return errors.Wrap(err, "create compaction segment")
 	}
+	defer func() {
+		os.RemoveAll(f.Name())
+	}()
+
 	var (
 		csf          = newSegmentFile(f)
 		crc32        = newCRC32()
@@ -389,9 +393,10 @@ func (w *SegmentWAL) Truncate(mint int64, keep func(uint64) bool) error {
 	csf.Close()
 
 	candidates[0].Close() // need close before remove on platform windows
-	if err := renameFile(csf.Name(), candidates[0].Name()); err != nil {
+	if err := fileutil.Replace(csf.Name(), candidates[0].Name()); err != nil {
 		return errors.Wrap(err, "rename compaction segment")
 	}
+
 	for _, f := range candidates[1:] {
 		f.Close() // need close before remove on platform windows
 		if err := os.RemoveAll(f.Name()); err != nil {