[ES-1365579] add visibility into compact progress

Signed-off-by: Yi Jin <[email protected]>

Author: jnyi

cmd/thanos/compact.go

@@ -176,6 +176,7 @@ func runCompact(
 ) (rerr error) {
 	deleteDelay := time.Duration(conf.deleteDelay)
 	compactMetrics := newCompactMetrics(reg, deleteDelay)
+	progressRegistry := compact.NewProgressRegistry(reg, logger)
 	downsampleMetrics := newDownsampleMetrics(reg)
 
 	httpProbe := prober.NewHTTP()
@@ -444,14 +445,16 @@ func runCompact(
 
 	var cleanMtx sync.Mutex
 	// TODO(GiedriusS): we could also apply retention policies here but the logic would be a bit more complex.
-	cleanPartialMarked := func() error {
+	cleanPartialMarked := func(progress *compact.Progress) error {
 		cleanMtx.Lock()
 		defer cleanMtx.Unlock()
-
+		defer progress.Idle()
+		progress.Set(compact.SyncMeta)
 		if err := sy.SyncMetas(ctx); err != nil {
 			return errors.Wrap(err, "syncing metas")
 		}
 
+		progress.Set(compact.CleanBlocks)
 		compact.BestEffortCleanAbortedPartialUploads(ctx, logger, sy.Partial(), insBkt, compactMetrics.partialUploadDeleteAttempts, compactMetrics.blocksCleaned, compactMetrics.blockCleanupFailures)
 		if err := blocksCleaner.DeleteMarkedBlocks(ctx); err != nil {
 			return errors.Wrap(err, "cleaning marked blocks")
@@ -461,19 +464,23 @@ func runCompact(
 		return nil
 	}
 
-	compactMainFn := func() error {
+	compactMainFn := func(progress *compact.Progress) error {
+		defer progress.Idle()
 		// this should happen before any compaction to remove unnecessary process on backlogs beyond retention.
 		if len(retentionByTenant) != 0 && len(sy.Metas()) == 0 {
 			level.Info(logger).Log("msg", "sync before tenant retention due to no blocks")
+			progress.Set(compact.SyncMeta)
 			if err := sy.SyncMetas(ctx); err != nil {
 				return errors.Wrap(err, "sync before tenant retention")
 			}
 		}
+
+		progress.Set(compact.ApplyRetention)
 		if err := compact.ApplyRetentionPolicyByTenant(ctx, logger, insBkt, sy.Metas(), retentionByTenant, compactMetrics.blocksMarked.WithLabelValues(metadata.DeletionMarkFilename, metadata.TenantRetentionExpired)); err != nil {
 			return errors.Wrap(err, "retention by tenant failed")
 		}
 
-		if err := compactor.Compact(ctx); err != nil {
+		if err := compactor.Compact(ctx, progress); err != nil {
 			return errors.Wrap(err, "whole compaction error")
 		}
 
@@ -482,10 +489,11 @@ func runCompact(
 			// We run two passes of this to ensure that the 1h downsampling is generated
 			// for 5m downsamplings created in the first run.
 			level.Info(logger).Log("msg", "start first pass of downsampling")
+			progress.Set(compact.SyncMeta)
 			if err := sy.SyncMetas(ctx); err != nil {
 				return errors.Wrap(err, "sync before first pass of downsampling")
 			}
-
+			progress.Set(compact.DownSampling)
 			filteredMetas := sy.Metas()
 			noDownsampleBlocks := noDownsampleMarkerFilter.NoDownsampleMarkedBlocks()
 			for ul := range noDownsampleBlocks {
@@ -514,6 +522,7 @@ func runCompact(
 			}
 
 			level.Info(logger).Log("msg", "start second pass of downsampling")
+			progress.Set(compact.SyncMeta)
 			if err := sy.SyncMetas(ctx); err != nil {
 				return errors.Wrap(err, "sync before second pass of downsampling")
 			}
@@ -547,27 +556,29 @@ func runCompact(
 		}
 
 		// TODO(bwplotka): Find a way to avoid syncing if no op was done.
+		progress.Set(compact.SyncMeta)
 		if err := sy.SyncMetas(ctx); err != nil {
 			return errors.Wrap(err, "sync before retention")
 		}
 
+		progress.Set(compact.ApplyRetention)
 		if err := compact.ApplyRetentionPolicyByResolution(ctx, logger, insBkt, sy.Metas(), retentionByResolution, compactMetrics.blocksMarked.WithLabelValues(metadata.DeletionMarkFilename, "")); err != nil {
 			return errors.Wrap(err, "retention failed")
 		}
 
-		return cleanPartialMarked()
+		return cleanPartialMarked(progress)
 	}
 
 	g.Add(func() error {
 		defer runutil.CloseWithLogOnErr(logger, insBkt, "bucket client")
 
 		if !conf.wait {
-			return compactMainFn()
+			return compactMainFn(progressRegistry.Get(compact.Main))
 		}
 
 		// --wait=true is specified.
 		return runutil.Repeat(conf.waitInterval, ctx.Done(), func() error {
-			err := compactMainFn()
+			err := compactMainFn(progressRegistry.Get(compact.Main))
 			if err == nil {
 				compactMetrics.iterations.Inc()
 				return nil
@@ -633,9 +644,11 @@ func runCompact(
 				// For /global state make sure to fetch periodically.
 				return runutil.Repeat(conf.blockViewerSyncBlockInterval, ctx.Done(), func() error {
 					return runutil.RetryWithLog(logger, time.Minute, ctx.Done(), func() error {
+						progress := progressRegistry.Get(compact.Web)
+						defer progress.Idle()
 						iterCtx, iterCancel := context.WithTimeout(ctx, conf.blockViewerSyncBlockTimeout)
 						defer iterCancel()
-
+						progress.Set(compact.SyncMeta)
 						_, _, err := f.Fetch(iterCtx)
 						return err
 					})
@@ -650,7 +663,7 @@ func runCompact(
 		if conf.cleanupBlocksInterval > 0 {
 			g.Add(func() error {
 				return runutil.Repeat(conf.cleanupBlocksInterval, ctx.Done(), func() error {
-					err := cleanPartialMarked()
+					err := cleanPartialMarked(progressRegistry.Get(compact.Cleanup))
 					if err != nil && compact.IsRetryError(err) {
 						// The RetryError signals that we hit an retriable error (transient error, no connection).
 						// You should alert on this being triggered too frequently.
@@ -678,7 +691,9 @@ func runCompact(
 				}
 
 				return runutil.Repeat(conf.progressCalculateInterval, ctx.Done(), func() error {
-
+					progress := progressRegistry.Get(compact.Calculate)
+					defer progress.Idle()
+					progress.Set(compact.SyncMeta)
 					if err := sy.SyncMetas(ctx); err != nil {
 						// The RetryError signals that we hit an retriable error (transient error, no connection).
 						// You should alert on this being triggered too frequently.
@@ -693,29 +708,34 @@ func runCompact(
 					}
 
 					metas := sy.Metas()
+					progress.Set(compact.Grouping)
 					groups, err := grouper.Groups(metas)
 					if err != nil {
 						return errors.Wrapf(err, "could not group metadata for compaction")
 					}
-
+					progress.Set(compact.CalculateProgress)
 					if err = ps.ProgressCalculate(ctx, groups); err != nil {
 						return errors.Wrapf(err, "could not calculate compaction progress")
 					}
 
+					progress.Set(compact.Grouping)
 					retGroups, err := grouper.Groups(metas)
 					if err != nil {
 						return errors.Wrapf(err, "could not group metadata for retention")
 					}
 
+					progress.Set(compact.CalculateProgress)
 					if err = rs.ProgressCalculate(ctx, retGroups); err != nil {
 						return errors.Wrapf(err, "could not calculate retention progress")
 					}
 
 					if !conf.disableDownsampling {
+						progress.Set(compact.Grouping)
 						groups, err = grouper.Groups(metas)
 						if err != nil {
 							return errors.Wrapf(err, "could not group metadata into downsample groups")
 						}
+						progress.Set(compact.CalculateProgress)
 						if err := ds.ProgressCalculate(ctx, groups); err != nil {
 							return errors.Wrapf(err, "could not calculate downsampling progress")
 						}

pkg/compact/compact.go

@@ -75,6 +75,8 @@ type SyncerMetrics struct {
 	GarbageCollectionFailures prometheus.Counter
 	GarbageCollectionDuration prometheus.Observer
 	BlocksMarkedForDeletion   prometheus.Counter
+	SyncMetas                 prometheus.Counter
+	SyncMetasDuration         prometheus.Observer
 }
 
 func NewSyncerMetrics(reg prometheus.Registerer, blocksMarkedForDeletion, garbageCollectedBlocks prometheus.Counter) *SyncerMetrics {
@@ -96,10 +98,22 @@ func NewSyncerMetrics(reg prometheus.Registerer, blocksMarkedForDeletion, garbag
 	})
 
 	m.BlocksMarkedForDeletion = blocksMarkedForDeletion
-
+	m.SyncMetas = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "thanos_compact_sync_metas_total",
+		Help: "Total number of synced block metas.",
+	})
+	m.SyncMetasDuration = promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
+		Name:    "thanos_compact_sync_metas_duration_seconds",
+		Help:    "Time it took to sync block metas.",
+		Buckets: []float64{10, 60, 180, 300, 450, 600, 1200, 1800, 3600, 7200, 10800, 21600},
+	})
 	return &m
 }
 
+func measureFuncDuration(duration prometheus.Observer, beginTime time.Time) {
+	duration.Observe(time.Since(beginTime).Seconds())
+}
+
 // NewMetaSyncer returns a new Syncer for the given Bucket and directory.
 // Blocks must be at least as old as the sync delay for being considered.
 func NewMetaSyncer(logger log.Logger, reg prometheus.Registerer, bkt objstore.Bucket, fetcher block.MetadataFetcher, duplicateBlocksFilter block.DeduplicateFilter, ignoreDeletionMarkFilter *block.IgnoreDeletionMarkFilter, blocksMarkedForDeletion, garbageCollectedBlocks prometheus.Counter, syncMetasTimeout time.Duration) (*Syncer, error) {
@@ -147,6 +161,8 @@ func UntilNextDownsampling(m *metadata.Meta) (time.Duration, error) {
 
 // SyncMetas synchronizes local state of block metas with what we have in the bucket.
 func (s *Syncer) SyncMetas(ctx context.Context) error {
+	s.metrics.SyncMetas.Inc()
+	defer measureFuncDuration(s.metrics.SyncMetasDuration, time.Now())
 	var cancel func() = func() {}
 	if s.syncMetasTimeout > 0 {
 		ctx, cancel = context.WithTimeout(ctx, s.syncMetasTimeout)
@@ -197,7 +213,8 @@ func (s *Syncer) Metas() map[ulid.ULID]*metadata.Meta {
 // block with a higher compaction level.
 // Call to SyncMetas function is required to populate duplicateIDs in duplicateBlocksFilter.
 func (s *Syncer) GarbageCollect(ctx context.Context) error {
-	begin := time.Now()
+	s.metrics.GarbageCollections.Inc()
+	defer measureFuncDuration(s.metrics.GarbageCollectionDuration, time.Now())
 
 	// Ignore filter exists before deduplicate filter.
 	deletionMarkMap := s.ignoreDeletionMarkFilter.DeletionMarkBlocks()
@@ -236,8 +253,6 @@ func (s *Syncer) GarbageCollect(ctx context.Context) error {
 		s.mtx.Unlock()
 		s.metrics.GarbageCollectedBlocks.Inc()
 	}
-	s.metrics.GarbageCollections.Inc()
-	s.metrics.GarbageCollectionDuration.Observe(time.Since(begin).Seconds())
 	return nil
 }
 
@@ -1443,7 +1458,7 @@ func NewBucketCompactorWithCheckerAndCallback(
 }
 
 // Compact runs compaction over bucket.
-func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
+func (c *BucketCompactor) Compact(ctx context.Context, progress *Progress) (rerr error) {
 	if c.concurrency == 0 {
 		level.Warn(c.logger).Log("msg", "compactor is disabled")
 		return nil
@@ -1462,6 +1477,7 @@ func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
 
 	// Loop over bucket and compact until there's no work left.
 	for {
+		SetProgress(progress, Initializing)
 		var (
 			wg                     sync.WaitGroup
 			workCtx, workCtxCancel = context.WithCancel(ctx)
@@ -1521,17 +1537,20 @@ func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
 		}
 
 		level.Info(c.logger).Log("msg", "start sync of metas")
+		SetProgress(progress, SyncMeta)
 		if err := c.sy.SyncMetas(ctx); err != nil {
 			return errors.Wrap(err, "sync")
 		}
 
 		level.Info(c.logger).Log("msg", "start of GC")
+		SetProgress(progress, GarbageCollect)
 		// Blocks that were compacted are garbage collected after each Compaction.
 		// However if compactor crashes we need to resolve those on startup.
 		if err := c.sy.GarbageCollect(ctx); err != nil {
 			return errors.Wrap(err, "garbage")
 		}
 
+		SetProgress(progress, Grouping)
 		groups, err := c.grouper.Groups(c.sy.Metas())
 		if err != nil {
 			return errors.Wrap(err, "build compaction groups")
@@ -1544,12 +1563,13 @@ func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
 			}
 		}
 
+		SetProgress(progress, CleanBlocks)
 		if err := runutil.DeleteAll(c.compactDir, ignoreDirs...); err != nil {
 			level.Warn(c.logger).Log("msg", "failed deleting non-compaction group directories/files, some disk space usage might have leaked. Continuing", "err", err, "dir", c.compactDir)
 		}
 
 		level.Info(c.logger).Log("msg", "start of compactions")
-
+		SetProgress(progress, Compacting)
 		// Send all groups found during this pass to the compaction workers.
 		var groupErrs errutil.MultiError
 	groupLoop:

pkg/compact/compact_e2e_test.go

@@ -221,7 +221,7 @@ func testGroupCompactE2e(t *testing.T, mergeFunc storage.VerticalChunkSeriesMerg
 		testutil.Ok(t, err)
 
 		// Compaction on empty should not fail.
-		testutil.Ok(t, bComp.Compact(ctx))
+		testutil.Ok(t, bComp.Compact(ctx, nil))
 		testutil.Equals(t, 0.0, promtest.ToFloat64(sy.metrics.GarbageCollectedBlocks))
 		testutil.Equals(t, 0.0, promtest.ToFloat64(sy.metrics.BlocksMarkedForDeletion))
 		testutil.Equals(t, 0.0, promtest.ToFloat64(sy.metrics.GarbageCollectionFailures))
@@ -314,7 +314,7 @@ func testGroupCompactE2e(t *testing.T, mergeFunc storage.VerticalChunkSeriesMerg
 		groupKey1 := metas[0].Thanos.GroupKey()
 		groupKey2 := metas[6].Thanos.GroupKey()
 
-		testutil.Ok(t, bComp.Compact(ctx))
+		testutil.Ok(t, bComp.Compact(ctx, nil))
 		testutil.Equals(t, 2.0, promtest.ToFloat64(sy.metrics.GarbageCollectedBlocks))
 		testutil.Equals(t, 2.0, promtest.ToFloat64(sy.metrics.BlocksMarkedForDeletion))
 		testutil.Equals(t, 0.0, promtest.ToFloat64(sy.metrics.GarbageCollectionFailures))

pkg/compact/progress.go

@@ -0,0 +1,85 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package compact
+
+import (
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+type CompactorState int64
+
+// Use a gauge to track the state of a compactor pod.
+// DO not inject new state in the middle of the enum otherwise it will break the old state syntax.
+const (
+	Initializing CompactorState = iota
+	Idle
+	SyncMeta
+	ApplyRetention
+	Compacting
+	GarbageCollect
+	DownSampling
+	CleanBlocks
+	Grouping
+	CalculateProgress
+)
+
+// Register all the compactor important threads here.
+const (
+	Main      = "main"      // Main compaction goroutine
+	Web       = "web"       // Web server goroutine
+	Cleanup   = "cleanup"   // Cleanup goroutine
+	Calculate = "Calculate" // Calculate calculation compactor progress goroutine
+)
+
+type Progress struct {
+	state  prometheus.Gauge
+	logger log.Logger
+}
+
+func (p *Progress) Set(state CompactorState) {
+	level.Info(p.logger).Log("msg", "Setting compactor progress state", "state", state)
+	p.state.Set(float64(state))
+}
+
+func (p *Progress) Idle() {
+	p.Set(Idle)
+}
+
+type ProgressRegistry struct {
+	*prometheus.GaugeVec
+	logger log.Logger
+}
+
+func NewProgressRegistry(reg *prometheus.Registry, logger log.Logger) *ProgressRegistry {
+	registry := promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{
+		Name: "thanos_compactor_progress_state",
+		Help: "The compaction progress state of the compactor.",
+	}, []string{"thread"})
+	level.Info(logger).Log("msg", "Registering compactor progress state gauge")
+	registry.WithLabelValues(Main).Set(float64(Initializing))
+	registry.WithLabelValues(Web).Set(float64(Initializing))
+	registry.WithLabelValues(Cleanup).Set(float64(Initializing))
+	return &ProgressRegistry{
+		registry,
+		logger,
+	}
+}
+
+func (pr *ProgressRegistry) Get(thread string) *Progress {
+	return &Progress{
+		state:  pr.WithLabelValues(thread),
+		logger: pr.logger,
+	}
+}
+
+// SetProgress is a helper function to set the progress state of a compactor and can be no-op if the progress is nil.
+func SetProgress(progress *Progress, state CompactorState) {
+	if progress == nil {
+		return
+	}
+	progress.Set(state)
+}