compact: Added support for no-compact markers in planner.

cmd/thanos/compact.go

@@ -220,6 +220,7 @@ func runCompact(
 		component,
 	)
 	api := blocksAPI.NewBlocksAPI(logger, conf.label, flagsMap)
+	noCompactMarkerFilter := compact.NewGatherNoCompactionMarkFilter(logger, bkt)
 	var sy *compact.Syncer
 	{
 		// Make sure all compactor meta syncs are done through Syncer.SyncMeta for readability.
@@ -229,6 +230,7 @@ func runCompact(
 				block.NewConsistencyDelayMetaFilter(logger, conf.consistencyDelay, extprom.WrapRegistererWithPrefix("thanos_", reg)),
 				ignoreDeletionMarkFilter,
 				duplicateBlocksFilter,
+				noCompactMarkerFilter,
 			}, []block.MetadataModifier{block.NewReplicaLabelRemover(logger, conf.dedupReplicaLabels)},
 		)
 		cf.UpdateOnChange(func(blocks []metadata.Meta, err error) {
@@ -280,7 +282,7 @@ func runCompact(
 
 	grouper := compact.NewDefaultGrouper(logger, bkt, conf.acceptMalformedIndex, enableVerticalCompaction, reg, blocksMarkedForDeletion, garbageCollectedBlocks)
 	blocksCleaner := compact.NewBlocksCleaner(logger, bkt, ignoreDeletionMarkFilter, deleteDelay, blocksCleaned, blockCleanupFailures)
-	compactor, err := compact.NewBucketCompactor(logger, sy, grouper, compact.NewTSDBBasedPlanner(logger, levels), comp, compactDir, bkt, conf.compactionConcurrency)
+	compactor, err := compact.NewBucketCompactor(logger, sy, grouper, compact.NewPlanner(logger, levels, noCompactMarkerFilter), comp, compactDir, bkt, conf.compactionConcurrency)
 	if err != nil {
 		cancel()
 		return errors.Wrap(err, "create bucket compactor")

pkg/block/fetcher.go

@@ -71,6 +71,9 @@ const (
 	// but don't have a replacement block yet.
 	markedForDeletionMeta = "marked-for-deletion"
 
+	// MarkedForNoCompactionMeta is label for blocks which are loaded but also marked for no compaction. This label is also counted in `loaded` label metric.
+	MarkedForNoCompactionMeta = "marked-for-no-compact"
+
 	// Modified label values.
 	replicaRemovedMeta = "replica-label-removed"
 )
@@ -111,6 +114,7 @@ func newFetcherMetrics(reg prometheus.Registerer) *fetcherMetrics {
 		[]string{timeExcludedMeta},
 		[]string{duplicateMeta},
 		[]string{markedForDeletionMeta},
+		[]string{MarkedForNoCompactionMeta},
 	)
 	m.modified = extprom.NewTxGaugeVec(
 		reg,
@@ -782,19 +786,20 @@ func (f *IgnoreDeletionMarkFilter) Filter(ctx context.Context, metas map[ulid.UL
 	f.deletionMarkMap = make(map[ulid.ULID]*metadata.DeletionMark)
 
 	for id := range metas {
-		deletionMark, err := metadata.ReadDeletionMark(ctx, f.bkt, f.logger, id.String())
-		if err == metadata.ErrorDeletionMarkNotFound {
-			continue
-		}
-		if errors.Cause(err) == metadata.ErrorUnmarshalDeletionMark {
-			level.Warn(f.logger).Log("msg", "found partial deletion-mark.json; if we will see it happening often for the same block, consider manually deleting deletion-mark.json from the object storage", "block", id, "err", err)
-			continue
-		}
-		if err != nil {
+		m := &metadata.DeletionMark{}
+		if err := metadata.ReadMarker(ctx, f.logger, f.bkt, id.String(), m); err != nil {
+			if errors.Cause(err) == metadata.ErrorMarkerNotFound {
+				continue
+			}
+			if errors.Cause(err) == metadata.ErrorUnmarshalMarker {
+				level.Warn(f.logger).Log("msg", "found partial deletion-mark.json; if we will see it happening often for the same block, consider manually deleting deletion-mark.json from the object storage", "block", id, "err", err)
+				continue
+			}
 			return err
 		}
-		f.deletionMarkMap[id] = deletionMark
-		if time.Since(time.Unix(deletionMark.DeletionTime, 0)).Seconds() > f.delay.Seconds() {
+
+		f.deletionMarkMap[id] = m
+		if time.Since(time.Unix(m.DeletionTime, 0)).Seconds() > f.delay.Seconds() {
 			synced.WithLabelValues(markedForDeletionMeta).Inc()
 			delete(metas, id)
 		}

pkg/block/metadata/markers.go

@@ -0,0 +1,115 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package metadata
+
+import (
+	"context"
+	"encoding/json"
+	"io/ioutil"
+	"path"
+
+	"github.com/go-kit/kit/log"
+	"github.com/oklog/ulid"
+	"github.com/pkg/errors"
+	"github.com/thanos-io/thanos/pkg/objstore"
+	"github.com/thanos-io/thanos/pkg/runutil"
+)
+
+const (
+	// DeletionMarkFilename is the known json filename for optional file storing details about when block is marked for deletion.
+	// If such file is present in block dir, it means the block is meant to be deleted after certain delay.
+	DeletionMarkFilename = "deletion-mark.json"
+	// NoCompactMarkFilename is the known json filename for optional file storing details about why block has to be excluded from compaction.
+	// If such file is present in block dir, it means the block has to excluded from compaction (both vertical and horizontal) or rewrite (e.g deletions).
+	NoCompactMarkFilename = "no-compact-mark.json"
+
+	// DeletionMarkVersion1 is the version of deletion-mark file supported by Thanos.
+	DeletionMarkVersion1 = 1
+	// NoCompactMarkVersion1 is the version of no-compact-mark file supported by Thanos.
+	NoCompactMarkVersion1 = 1
+)
+
+var (
+	// ErrorMarkerNotFound is the error when marker file is not found.
+	ErrorMarkerNotFound = errors.New("marker not found")
+	// ErrorUnmarshalMarker is the error when unmarshalling marker JSON file.
+	// This error can occur because marker has been partially uploaded to block storage
+	// or the marker file is not a valid json file.
+	ErrorUnmarshalMarker = errors.New("unmarshal marker JSON")
+)
+
+type Marker interface {
+	markerFilename() string
+}
+
+// DeletionMark stores block id and when block was marked for deletion.
+type DeletionMark struct {
+	// ID of the tsdb block.
+	ID ulid.ULID `json:"id"`
+	// Version of the file.
+	Version int `json:"version"`
+
+	// DeletionTime is a unix timestamp of when the block was marked to be deleted.
+	DeletionTime int64 `json:"deletion_time"`
+}
+
+func (m *DeletionMark) markerFilename() string { return DeletionMarkFilename }
+
+// NoCompactReason is a reason for a block to be excluded from compaction.
+type NoCompactReason string
+
+const (
+	// ManualNoCompactReason is a custom reason of excluding from compaction that should be added when no-compact mark is added for unknown/user specified reason.
+	ManualNoCompactReason NoCompactReason = "manual"
+	// IndexSizeExceedingNoCompactReason is a reason of index being too big (for example exceeding 64GB limit: https://github.com/thanos-io/thanos/issues/1424)
+	// This reason can be ignored when vertical block sharding will be implemented.
+	IndexSizeExceedingNoCompactReason = "index-size-exceeding"
+)
+
+// NoCompactMark marker stores reason of block being excluded from compaction if needed.
+type NoCompactMark struct {
+	// ID of the tsdb block.
+	ID ulid.ULID `json:"id"`
+	// Version of the file.
+	Version int `json:"version"`
+
+	Reason NoCompactReason `json:"reason"`
+	// Details is a human readable string giving details of reason.
+	Details string `json:"details"`
+}
+
+func (n *NoCompactMark) markerFilename() string { return NoCompactMarkFilename }
+
+// ReadMarker reads the given mark file from <dir>/<marker filename>.json in bucket.
+func ReadMarker(ctx context.Context, logger log.Logger, bkt objstore.InstrumentedBucketReader, dir string, marker Marker) error {
+	markerFile := path.Join(dir, marker.markerFilename())
+	r, err := bkt.ReaderWithExpectedErrs(bkt.IsObjNotFoundErr).Get(ctx, markerFile)
+	if err != nil {
+		if bkt.IsObjNotFoundErr(err) {
+			return ErrorMarkerNotFound
+		}
+		return errors.Wrapf(err, "get file: %s", markerFile)
+	}
+	defer runutil.CloseWithLogOnErr(logger, r, "close bkt marker reader")
+
+	metaContent, err := ioutil.ReadAll(r)
+	if err != nil {
+		return errors.Wrapf(err, "read file: %s", markerFile)
+	}
+
+	if err := json.Unmarshal(metaContent, marker); err != nil {
+		return errors.Wrapf(ErrorUnmarshalMarker, "file: %s; err: %v", markerFile, err.Error())
+	}
+	switch marker.markerFilename() {
+	case NoCompactMarkFilename:
+		if version := marker.(*NoCompactMark).Version; version != NoCompactMarkVersion1 {
+			return errors.Errorf("unexpected no-compact-mark file version %d, expected %d", version, NoCompactMarkVersion1)
+		}
+	case DeletionMarkFilename:
+		if version := marker.(*DeletionMark).Version; version != DeletionMarkVersion1 {
+			return errors.Errorf("unexpected deletion-mark file version %d, expected %d", version, DeletionMarkVersion1)
+		}
+	}
+	return nil
+}