compact: accept malformed index

cmd/thanos/compact.go

@@ -66,6 +66,9 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application, name stri
 
 	haltOnError := cmd.Flag("debug.halt-on-error", "Halt the process if a critical compaction error is detected.").
 		Hidden().Default("true").Bool()
+	acceptMalformed := cmd.Flag("debug.accept-malformed-index",
+		"Compaction index verification will ignore out of order label names.").
+		Hidden().Default("false").Bool()
 
 	httpAddr := regHTTPAddrFlag(cmd)
 
@@ -102,6 +105,7 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application, name stri
 			objStoreConfig,
 			time.Duration(*syncDelay),
 			*haltOnError,
+			*acceptMalformed,
 			*wait,
 			map[compact.ResolutionLevel]time.Duration{
 				compact.ResolutionLevelRaw: time.Duration(*retentionRaw),
@@ -125,6 +129,7 @@ func runCompact(
 	objStoreConfig *pathOrContent,
 	syncDelay time.Duration,
 	haltOnError bool,
+	acceptMalformed bool,
 	wait bool,
 	retentionByResolution map[compact.ResolutionLevel]time.Duration,
 	component string,
@@ -206,7 +211,7 @@ func runCompact(
 
 	ctx, cancel := context.WithCancel(context.Background())
 	f := func() error {
-		if err := compactor.Compact(ctx); err != nil {
+		if err := compactor.Compact(ctx, acceptMalformed); err != nil {
 			return errors.Wrap(err, "compaction failed")
 		}
 		level.Info(logger).Log("msg", "compaction iterations done")

pkg/block/index.go

@@ -16,6 +16,7 @@ import (
 	"github.com/prometheus/tsdb/fileutil"
 
 	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
 	"github.com/improbable-eng/thanos/pkg/runutil"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
@@ -248,6 +249,20 @@ type Stats struct {
 	// Specifically we mean here chunks with minTime == block.maxTime and maxTime > block.MaxTime. These are
 	// are segregated into separate counters. These chunks are safe to be deleted, since they are duplicated across 2 blocks.
 	Issue347OutsideChunks int
+	// OutOfOrderLabels represents the number of postings that contained out
+	// of order labels, a bug present in Prometheus 2.8.0 and below.
+	OutOfOrderLabels int
+}
+
+// PrometheusIssue5372Err returns an error if statsd indicates postings with out
+// of order labels.  This is introduced by Prometheus Issue #5372 in 2.8.0 and
+// below.
+func (i Stats) PrometheusIssue5372Err() error {
+	if i.OutOfOrderLabels > 0 {
+		return errors.Errorf("index contains %d postings with out of order labels",
+			i.OutOfOrderLabels)
+	}
+	return nil
 }
 
 // Issue347OutsideChunksErr returns error if stats indicates issue347 block issue, that is repaired explicitly before compaction (on plan block).
@@ -301,6 +316,10 @@ func (i Stats) AnyErr() error {
 		errMsg = append(errMsg, err.Error())
 	}
 
+	if err := i.PrometheusIssue5372Err(); err != nil {
+		errMsg = append(errMsg, err.Error())
+	}
+
 	if len(errMsg) > 0 {
 		return errors.New(strings.Join(errMsg, ", "))
 	}
@@ -348,7 +367,12 @@ func GatherIndexIssueStats(logger log.Logger, fn string, minTime int64, maxTime
 		l0 := lset[0]
 		for _, l := range lset[1:] {
 			if l.Name < l0.Name {
-				return stats, errors.Errorf("out-of-order label set %s for series %d", lset, id)
+				stats.OutOfOrderLabels++
+				level.Warn(logger).Log("msg",
+					"out-of-order label set: known bug in Prometheus 2.8.0 and below",
+					"labelset", fmt.Sprintf("%s", lset),
+					"series", fmt.Sprintf("%d", id),
+				)
 			}
 			l0 = l
 		}

pkg/compact/compact.go

@@ -513,7 +513,7 @@ func (cg *Group) Resolution() int64 {
 
 // Compact plans and runs a single compaction against the group. The compacted result
 // is uploaded into the bucket the blocks were retrieved from.
-func (cg *Group) Compact(ctx context.Context, dir string, comp tsdb.Compactor) (bool, ulid.ULID, error) {
+func (cg *Group) Compact(ctx context.Context, dir string, comp tsdb.Compactor, ignoreMalformedIndex bool) (bool, ulid.ULID, error) {
 	subDir := filepath.Join(dir, cg.Key())
 
 	if err := os.RemoveAll(subDir); err != nil {
@@ -523,7 +523,7 @@ func (cg *Group) Compact(ctx context.Context, dir string, comp tsdb.Compactor) (
 		return false, ulid.ULID{}, errors.Wrap(err, "create compaction group dir")
 	}
 
-	shouldRerun, compID, err := cg.compact(ctx, subDir, comp)
+	shouldRerun, compID, err := cg.compact(ctx, subDir, comp, ignoreMalformedIndex)
 	if err != nil {
 		cg.compactionFailures.Inc()
 	}
@@ -688,7 +688,7 @@ func RepairIssue347(ctx context.Context, logger log.Logger, bkt objstore.Bucket,
 	return nil
 }
 
-func (cg *Group) compact(ctx context.Context, dir string, comp tsdb.Compactor) (shouldRerun bool, compID ulid.ULID, err error) {
+func (cg *Group) compact(ctx context.Context, dir string, comp tsdb.Compactor, ignoreMalformed bool) (shouldRerun bool, compID ulid.ULID, err error) {
 	cg.mtx.Lock()
 	defer cg.mtx.Unlock()
 
@@ -769,6 +769,10 @@ func (cg *Group) compact(ctx context.Context, dir string, comp tsdb.Compactor) (
 		if err := stats.Issue347OutsideChunksErr(); err != nil {
 			return false, ulid.ULID{}, issue347Error(errors.Wrapf(err, "invalid, but reparable block %s", pdir), meta.ULID)
 		}
+
+		if err := stats.PrometheusIssue5372Err(); !ignoreMalformed && err != nil {
+			return false, ulid.ULID{}, errors.Wrapf(err, "block id %s", id)
+		}
 	}
 	level.Debug(cg.logger).Log("msg", "downloaded and verified blocks",
 		"blocks", fmt.Sprintf("%v", plan), "duration", time.Since(begin))
@@ -816,7 +820,7 @@ func (cg *Group) compact(ctx context.Context, dir string, comp tsdb.Compactor) (
 	}
 
 	// Ensure the output block is valid.
-	if err := block.VerifyIndex(cg.logger, filepath.Join(bdir, block.IndexFilename), newMeta.MinTime, newMeta.MaxTime); err != nil {
+	if err := block.VerifyIndex(cg.logger, filepath.Join(bdir, block.IndexFilename), newMeta.MinTime, newMeta.MaxTime); !ignoreMalformed && err != nil {
 		return false, ulid.ULID{}, halt(errors.Wrapf(err, "invalid result block %s", bdir))
 	}
 
@@ -886,7 +890,7 @@ func NewBucketCompactor(logger log.Logger, sy *Syncer, comp tsdb.Compactor, comp
 }
 
 // Compact runs compaction over bucket.
-func (c *BucketCompactor) Compact(ctx context.Context) error {
+func (c *BucketCompactor) Compact(ctx context.Context, IgnoreMalformedIndex bool) error {
 	// Loop over bucket and compact until there's no work left.
 	for {
 		// Clean up the compaction temporary directory at the beginning of every compaction loop.
@@ -914,7 +918,7 @@ func (c *BucketCompactor) Compact(ctx context.Context) error {
 		}
 		finishedAllGroups := true
 		for _, g := range groups {
-			shouldRerunGroup, _, err := g.Compact(ctx, c.compactDir, c.comp)
+			shouldRerunGroup, _, err := g.Compact(ctx, c.compactDir, c.comp, IgnoreMalformedIndex)
 			if err == nil {
 				if shouldRerunGroup {
 					finishedAllGroups = false

pkg/compact/compact_e2e_test.go

@@ -253,7 +253,7 @@ func TestGroup_Compact_e2e(t *testing.T) {
 		comp, err := tsdb.NewLeveledCompactor(nil, log.NewLogfmtLogger(os.Stderr), []int64{1000, 3000}, nil)
 		testutil.Ok(t, err)
 
-		shouldRerun, id, err := g.Compact(ctx, dir, comp)
+		shouldRerun, id, err := g.Compact(ctx, dir, comp, false)
 		testutil.Ok(t, err)
 		testutil.Assert(t, !shouldRerun, "group should be empty, but compactor did a compaction and told us to rerun")
 
@@ -262,7 +262,7 @@ func TestGroup_Compact_e2e(t *testing.T) {
 			testutil.Ok(t, g.Add(m))
 		}
 
-		shouldRerun, id, err = g.Compact(ctx, dir, comp)
+		shouldRerun, id, err = g.Compact(ctx, dir, comp, false)
 		testutil.Ok(t, err)
 		testutil.Assert(t, shouldRerun, "there should be compactible data, but the compactor reported there was not")