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,22 @@ func rewrite(
 		return labels.Compare(series[i].lset, series[j].lset) < 0
 	})
 
+	lastSet := labels.Labels{}
 	// Build a new TSDB block.
 	for _, s := range series {
+		// The TSDB library will throw an error if we add a series with
+		// identical labels as the last series.  This means that we have
+		// discovered a duplicate time series in the old block.  We drop
+		// all duplicate series preserving the first one.
+		// TODO: Add metric to count dropped series if repair becomes a daemon
+		// rather than a batch job.
+		if labels.Compare(lastSet, s.lset) == 0 {
+			level.Warn(logger).Log("msg",
+				"dropping duplicate series in tsdb block found",
+				"labelset", s.lset.String(),
+			)
+			continue
+		}
 		if err := chunkw.WriteChunks(s.chks...); err != nil {
 			return errors.Wrap(err, "write chunks")
 		}
@@ -691,6 +710,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, false, ""); 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, true, 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 a TSDB block to a backup bucket and, on success, removes
+// it from the source bucket.  It returns error if block dir already exists in
+// the backup bucket (blocks should be immutable) or if any of the operations
+// fail.  When useExistingTempdir is true do not download the block but use the
+// previously downloaded block found in tempdir to avoid repeated downloads.
+// The tempdir value is ignored unless useExistingTempdir is set to true.
+func SafeDelete(ctx context.Context, logger log.Logger, bkt objstore.Bucket, backupBkt objstore.Bucket, id ulid.ULID, useExistingTempdir bool, tempdir string) error {
 	foundDir := false
 	err := backupBkt.Iter(ctx, id.String(), func(name string) error {
 		foundDir = true
@@ -31,23 +35,48 @@ 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
-	}
-	dir := filepath.Join(tempdir, id.String())
-	err = os.Mkdir(dir, 0755)
-	if err != nil {
-		return err
+	if useExistingTempdir && tempdir == "" {
+		return errors.New("instructed to use existing tempdir but no tempdir provided")
 	}
-	defer func() {
-		if err := os.RemoveAll(tempdir); err != nil {
-			level.Warn(logger).Log("msg", "failed to delete dir", "dir", tempdir, "err", err)
+
+	var dir string
+	if useExistingTempdir {
+		dir = filepath.Join(tempdir, id.String())
+		if _, err := os.Stat(filepath.Join(dir, "meta.json")); err != nil {
+			// If there is any error stat'ing meta.json inside the TSDB block
+			// then declare the existing block as bad and refuse to upload it.
+			// TODO: Make this check more robust.
+			return errors.Wrap(err, "existing tsdb block is invalid")
+		}
+		level.Info(logger).Log("msg", "using previously downloaded tsdb block",
+			"id", id.String())
+	} else {
+		// tempdir unspecified, create one
+		tempdir, err = ioutil.TempDir("", fmt.Sprintf("safe-delete-%s", id.String()))
+		if err != nil {
+			return err
 		}
-	}()
+		// We manage this tempdir so we clean it up on exit.
+		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")
+		dir = filepath.Join(tempdir, id.String())
+		if _, err := os.Stat(dir); os.IsNotExist(err) {
+			// TSDB block not already present, create it.
+			err = os.Mkdir(dir, 0755)
+			if err != nil {
+				return errors.Wrap(err, "creating tsdb block tempdir")
+			}
+		} else if err != nil {
+			// Error calling Stat() and something is really wrong.
+			return errors.Wrap(err, "stat call on directory")
+		}
+		if err := block.Download(ctx, logger, bkt, id, dir); err != nil {
+			return errors.Wrap(err, "download from source")
+		}
 	}
 
 	if err := block.Upload(ctx, logger, backupBkt, dir); err != nil {