Make SafeDelete faster for large blocks; TSDB block repair case

pkg/block/index.go

@@ -495,12 +495,16 @@ func Repair(logger log.Logger, dir string, id ulid.ULID, source metadata.SourceT
 	resmeta.Stats = tsdb.BlockStats{} // reset stats
 	resmeta.Thanos.Source = source    // update source
 
-	if err := rewrite(indexr, chunkr, indexw, chunkw, &resmeta, ignoreChkFns); err != nil {
+	if err := rewrite(logger, indexr, chunkr, indexw, chunkw, &resmeta, ignoreChkFns); err != nil {
 		return resid, errors.Wrap(err, "rewrite block")
 	}
 	if err := metadata.Write(logger, resdir, &resmeta); err != nil {
 		return resid, err
 	}
+	// TSDB may rewrite metadata in bdir
+	if err := metadata.Write(logger, bdir, meta); err != nil {
+		return resid, err
+	}
 	return resid, nil
 }
 
@@ -595,6 +599,7 @@ type seriesRepair struct {
 // rewrite writes all data from the readers back into the writers while cleaning
 // up mis-ordered and duplicated chunks.
 func rewrite(
+	logger log.Logger,
 	indexr tsdb.IndexReader, chunkr tsdb.ChunkReader,
 	indexw tsdb.IndexWriter, chunkw tsdb.ChunkWriter,
 	meta *metadata.Meta,
@@ -665,8 +670,16 @@ func rewrite(
 		return labels.Compare(series[i].lset, series[j].lset) < 0
 	})
 
+	lastSet := labels.Labels{}
 	// Build a new TSDB block.
 	for _, s := range series {
+		if labels.Compare(lastSet, s.lset) == 0 {
+			level.Warn(logger).Log("msg",
+				"dropping duplicate series in tsdb block found",
+				"labelset", fmt.Sprintf("%s", s.lset),
+			)
+			continue
+		}
 		if err := chunkw.WriteChunks(s.chks...); err != nil {
 			return errors.Wrap(err, "write chunks")
 		}
@@ -691,6 +704,7 @@ func rewrite(
 		}
 		postings.Add(i, s.lset)
 		i++
+		lastSet = s.lset
 	}
 
 	s := make([]string, 0, 256)

pkg/verifier/duplicated_compaction.go

@@ -80,7 +80,7 @@ func DuplicatedCompactionIssue(ctx context.Context, logger log.Logger, bkt objst
 	}
 
 	for i, id := range toKill {
-		if err := SafeDelete(ctx, logger, bkt, backupBkt, id); err != nil {
+		if err := SafeDelete(ctx, logger, bkt, backupBkt, id, ""); err != nil {
 			return err
 		}
 		level.Info(logger).Log("msg", "Removed duplicated block", "id", id, "to-be-removed", len(toKill)-(i+1), "removed", i+1, "issue", DuplicatedCompactionIssueID)

pkg/verifier/index_issue.go

@@ -117,7 +117,7 @@ func IndexIssue(ctx context.Context, logger log.Logger, bkt objstore.Bucket, bac
 		}
 
 		level.Info(logger).Log("msg", "safe deleting broken block", "id", id, "issue", IndexIssueID)
-		if err := SafeDelete(ctx, logger, bkt, backupBkt, id); err != nil {
+		if err := SafeDelete(ctx, logger, bkt, backupBkt, id, tmpdir); err != nil {
 			return errors.Wrapf(err, "safe deleting old block %s failed", id)
 		}
 		level.Info(logger).Log("msg", "all good, continuing", "id", id, "issue", IndexIssueID)

pkg/verifier/safe_delete.go

@@ -2,6 +2,7 @@ package verifier
 
 import (
 	"context"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -14,10 +15,13 @@ import (
 	"github.com/pkg/errors"
 )
 
-// SafeDelete moves block to backup bucket and if succeeded, removes it from source bucket.
-// It returns error if block dir already exists in backup bucket (blocks should be immutable) or any
-// of the operation fails.
-func SafeDelete(ctx context.Context, logger log.Logger, bkt objstore.Bucket, backupBkt objstore.Bucket, id ulid.ULID) error {
+// SafeDelete moves block to backup bucket and if succeeded, removes it from
+// source bucket.  It returns error if block dir already exists in backup
+// bucket (blocks should be immutable) or any of the operation fails.  If
+// the block has already been downloaded it must exist in tempdir, else it
+// will be downloaded to be copied to the backup bucket.  tempdir set to the
+// zero value forces download.
+func SafeDelete(ctx context.Context, logger log.Logger, bkt objstore.Bucket, backupBkt objstore.Bucket, id ulid.ULID, tempdir string) error {
 	foundDir := false
 	err := backupBkt.Iter(ctx, id.String(), func(name string) error {
 		foundDir = true
@@ -31,23 +35,33 @@ func SafeDelete(ctx context.Context, logger log.Logger, bkt objstore.Bucket, bac
 		return errors.Errorf("%s dir seems to exists in backup bucket. Remove this block manually if you are sure it is safe to do", id)
 	}
 
-	tempdir, err := ioutil.TempDir("", "safe-delete")
-	if err != nil {
-		return err
+	if tempdir == "" {
+		// tempdir unspecified, create one
+		tempdir, err = ioutil.TempDir("", fmt.Sprintf("safe-delete-%s", id.String()))
+		if err != nil {
+			return err
+		}
 	}
 	dir := filepath.Join(tempdir, id.String())
-	err = os.Mkdir(dir, 0755)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err := os.RemoveAll(tempdir); err != nil {
-			level.Warn(logger).Log("msg", "failed to delete dir", "dir", tempdir, "err", err)
+
+	if _, err := os.Stat(dir); err != nil {
+		// TSDB block not already present, download it
+		err = os.Mkdir(dir, 0755)
+		if err != nil {
+			return err
 		}
-	}()
+		defer func() {
+			if err := os.RemoveAll(tempdir); err != nil {
+				level.Warn(logger).Log("msg", "failed to delete dir", "dir", tempdir, "err", err)
+			}
+		}()
 
-	if err := block.Download(ctx, logger, bkt, id, dir); err != nil {
-		return errors.Wrap(err, "download from source")
+		if err := block.Download(ctx, logger, bkt, id, dir); err != nil {
+			return errors.Wrap(err, "download from source")
+		}
+	} else {
+		level.Info(logger).Log("msg", "using previously downloaded tsdb block",
+			"id", id.String())
 	}
 
 	if err := block.Upload(ctx, logger, backupBkt, dir); err != nil {