Make SafeDelete faster for large blocks; TSDB block repair case

pkg/block/index.go

@@ -495,12 +495,18 @@ 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.
+	// TODO: This is not needed in newer TSDB code. See
+	// https://github.com/prometheus/tsdb/pull/637
+	if err := metadata.Write(logger, bdir, meta); err != nil {
+		return resid, err
+	}
 	return resid, nil
 }
 
@@ -595,6 +601,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 +672,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 +712,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 := BackupAndDelete(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 := BackupAndDeleteDownloaded(ctx, logger, filepath.Join(tmpdir, id.String()), bkt, backupBkt, id); 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,29 +15,35 @@ 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 {
+// TSDBBlockExistsInBucket checks to see if a given TSDB block ID exists in a
+// bucket. If so, true is returned. An error is returned on failure and in
+// such case the boolean result has no meaning.
+func TSDBBlockExistsInBucket(ctx context.Context, bkt objstore.Bucket, id ulid.ULID) (bool, error) {
 	foundDir := false
-	err := backupBkt.Iter(ctx, id.String(), func(name string) error {
+	err := bkt.Iter(ctx, id.String(), func(name string) error {
 		foundDir = true
 		return nil
 	})
+
+	return foundDir, err
+}
+
+// BackupAndDelete 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.
+func BackupAndDelete(ctx context.Context, logger log.Logger, bkt, backupBkt objstore.Bucket, id ulid.ULID) error {
+	// Does this TSDB block exist in backupBkt already?
+	found, err := TSDBBlockExistsInBucket(ctx, backupBkt, id)
 	if err != nil {
 		return err
 	}
-
-	if foundDir {
+	if found {
 		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)
+	// Create a tempdir to locally store TSDB block.
+	tempdir, err := ioutil.TempDir("", fmt.Sprintf("safe-delete-%s", id.String()))
 	if err != nil {
 		return err
 	}
@@ -46,18 +53,72 @@ func SafeDelete(ctx context.Context, logger log.Logger, bkt objstore.Bucket, bac
 		}
 	}()
 
+	// Download the TSDB block.
+	dir := filepath.Join(tempdir, id.String())
 	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 {
-		return errors.Wrap(err, "upload to backup")
+	// Backup the block.
+	if err := backupDownloaded(ctx, logger, dir, backupBkt, id); err != nil {
+		return err
+	}
+
+	// Block uploaded, so we are ok to remove from src bucket.
+	level.Info(logger).Log("msg", "Deleting block", "id", id.String())
+	if err := block.Delete(ctx, bkt, id); err != nil {
+		return errors.Wrap(err, "delete from source")
+	}
+
+	return nil
+}
+
+// BackupAndDeleteDownloaded works much like BackupAndDelete in that it will
+// move a TSDB block from a bucket to a backup bucket. The bdir parameter
+// points to the location on disk where the TSDB block was previously
+// downloaded allowing this function to avoid downloading the TSDB block from
+// the source bucket again. An error is returned if any operation fails.
+func BackupAndDeleteDownloaded(ctx context.Context, logger log.Logger, bdir string, bkt, backupBkt objstore.Bucket, id ulid.ULID) error {
+	// Does this TSDB block exist in backupBkt already?
+	found, err := TSDBBlockExistsInBucket(ctx, backupBkt, id)
+	if err != nil {
+		return err
+	}
+	if found {
+		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)
+	}
+
+	// Backup the block.
+	if err := backupDownloaded(ctx, logger, bdir, backupBkt, id); err != nil {
+		return err
 	}
 
 	// Block uploaded, so we are ok to remove from src bucket.
+	level.Info(logger).Log("msg", "Deleting block", "id", id.String())
 	if err := block.Delete(ctx, bkt, id); err != nil {
 		return errors.Wrap(err, "delete from source")
 	}
 
 	return nil
 }
+
+// backupDownloaded is a helper function that uploads a TSDB block
+// found on disk to the given bucket. An error is returned if any operation
+// fails.
+func backupDownloaded(ctx context.Context, logger log.Logger, bdir string, backupBkt objstore.Bucket, id ulid.ULID) error {
+	// Safety checks.
+	if _, err := os.Stat(filepath.Join(bdir, "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")
+	}
+
+	// Upload the on disk TSDB block.
+	level.Info(logger).Log("msg", "Uploading block to backup bucket", "id", id.String())
+	if err := block.Upload(ctx, logger, backupBkt, bdir); err != nil {
+		return errors.Wrap(err, "upload to backup")
+	}
+
+	return nil
+}