Improve on error handling and visit marker status (#178)

Signed-off-by: Alex Le <[email protected]>

Author: alexqyle

pkg/compactor/blocks_cleaner.go

@@ -301,6 +301,18 @@ func (c *BlocksCleaner) cleanUpActiveUsers(ctx context.Context, users []string,
 	err := concurrency.ForEachUser(ctx, users, c.cfg.CleanupConcurrency, func(ctx context.Context, userID string) error {
 		userLogger := util_log.WithUserID(userID, c.logger)
 		userBucket := bucket.NewUserBucketClient(userID, c.bucketClient, c.cfgProvider)
+		visitMarkerManager, err := c.obtainVisitMarkerManager(ctx, userLogger, userBucket)
+		if err != nil {
+			return err
+		}
+		if visitMarkerManager == nil {
+			return nil
+		}
+		errChan := make(chan error, 1)
+		go visitMarkerManager.HeartBeat(ctx, errChan, c.cleanerVisitMarkerFileUpdateInterval, true)
+		defer func() {
+			errChan <- nil
+		}()
 		return errors.Wrapf(c.cleanUser(ctx, userLogger, userBucket, userID, firstRun), "failed to delete blocks for user: %s", userID)
 	})
 
@@ -334,6 +346,18 @@ func (c *BlocksCleaner) cleanDeletedUsers(ctx context.Context, users []string) e
 	err := concurrency.ForEachUser(ctx, users, c.cfg.CleanupConcurrency, func(ctx context.Context, userID string) error {
 		userLogger := util_log.WithUserID(userID, c.logger)
 		userBucket := bucket.NewUserBucketClient(userID, c.bucketClient, c.cfgProvider)
+		visitMarkerManager, err := c.obtainVisitMarkerManager(ctx, userLogger, userBucket)
+		if err != nil {
+			return err
+		}
+		if visitMarkerManager == nil {
+			return nil
+		}
+		errChan := make(chan error, 1)
+		go visitMarkerManager.HeartBeat(ctx, errChan, c.cleanerVisitMarkerFileUpdateInterval, true)
+		defer func() {
+			errChan <- nil
+		}()
 		return errors.Wrapf(c.deleteUserMarkedForDeletion(ctx, userLogger, userBucket, userID), "failed to delete user marked for deletion: %s", userID)
 	})
 
@@ -382,6 +406,21 @@ func (c *BlocksCleaner) scanUsers(ctx context.Context) ([]string, []string, erro
 	return users, deleted, nil
 }
 
+func (c *BlocksCleaner) obtainVisitMarkerManager(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket) (*VisitMarkerManager, error) {
+	cleanerVisitMarker := NewCleanerVisitMarker(c.ringLifecyclerID)
+	visitMarkerManager := NewVisitMarkerManager(userBucket, userLogger, c.ringLifecyclerID, cleanerVisitMarker, c.CleanerVisitMarkerReadFailed, c.CleanerVisitMarkerWriteFailed)
+
+	existingCleanerVisitMarker := &CleanerVisitMarker{}
+	err := visitMarkerManager.ReadVisitMarker(ctx, existingCleanerVisitMarker)
+	if err != nil && !errors.Is(err, ErrorVisitMarkerNotFound) {
+		return nil, errors.Wrapf(err, "failed to read cleaner visit marker")
+	}
+	if errors.Is(err, ErrorVisitMarkerNotFound) || !existingCleanerVisitMarker.IsVisited(c.cleanerVisitMarkerTimeout) {
+		return visitMarkerManager, nil
+	}
+	return nil, nil
+}
+
 // Remove blocks and remaining data for tenant marked for deletion.
 func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string) error {
 

pkg/compactor/cleaner_visit_marker.go

@@ -45,6 +45,10 @@ func (b *CleanerVisitMarker) IsFailed() bool {
 	return b.Status == Failed
 }
 
+func (b *CleanerVisitMarker) IsInProgress() bool {
+	return b.Status == InProgress
+}
+
 func (b *CleanerVisitMarker) IsPending() bool {
 	return b.Status == Pending
 }
@@ -53,7 +57,13 @@ func (b *CleanerVisitMarker) GetVisitMarkerFilePath() string {
 	return GetCleanerVisitMarkerFilePath()
 }
 
-func (b *CleanerVisitMarker) MarkVisited(ownerIdentifier string) {
+func (b *CleanerVisitMarker) MarkInProgress(ownerIdentifier string) {
+	b.CompactorID = ownerIdentifier
+	b.Status = InProgress
+	b.VisitTime = time.Now().Unix()
+}
+
+func (b *CleanerVisitMarker) MarkPending(ownerIdentifier string) {
 	b.CompactorID = ownerIdentifier
 	b.Status = Pending
 	b.VisitTime = time.Now().Unix()

pkg/compactor/compactor.go

@@ -897,13 +897,23 @@ func (c *Compactor) compactUserWithRetries(ctx context.Context, userID string) e
 			return nil
 		}
 		if c.isCausedByPermissionDenied(lastErr) {
-			level.Warn(c.logger).Log("msg", "skipping compactUser due to PermissionDenied", "user", userID, "err", lastErr)
+			level.Warn(c.logger).Log("msg", "skipping compactUser due to PermissionDenied", "org_id", userID, "err", lastErr)
 			return nil
 		}
-
+		if compact.IsHaltError(lastErr) {
+			level.Error(c.logger).Log("msg", "compactor returned critical error", "org_id", userID, "err", lastErr)
+			c.compactorMetrics.compactionHaltErrors.WithLabelValues(userID).Inc()
+			return lastErr
+		}
+		c.compactorMetrics.compactionRetryErrors.WithLabelValues(userID).Inc()
 		retries.Wait()
 	}
 
+	err := errors.Unwrap(errors.Cause(lastErr))
+	if errors.Is(err, PlannerCompletedPartitionError) || errors.Is(err, PlannerVisitedPartitionError) {
+		return nil
+	}
+
 	return lastErr
 }
 

pkg/compactor/compactor_metrics.go

@@ -41,6 +41,8 @@ type compactorMetrics struct {
 	compactionDuration      *prometheus.GaugeVec
 	partitionGroupDuration  *prometheus.GaugeVec
 	blockGroupDuration      *prometheus.GaugeVec
+	compactionRetryErrors   *prometheus.CounterVec
+	compactionHaltErrors    *prometheus.CounterVec
 }
 
 const (
@@ -179,6 +181,14 @@ func newCompactorMetricsWithLabels(reg prometheus.Registerer, commonLabels []str
 		Name: "cortex_compact_block_group_duration_seconds",
 		Help: "Duration of sharding grouper in seconds",
 	}, commonLabels)
+	m.compactionRetryErrors = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+		Name: "cortex_compactor_compaction_retry_error_total",
+		Help: "Total number of retry errors from compactions.",
+	}, CommonLabels)
+	m.compactionHaltErrors = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+		Name: "cortex_compactor_compaction_halt_error_total",
+		Help: "Total number of halt errors from compactions.",
+	}, CommonLabels)
 
 	return &m
 }

pkg/compactor/partition_compaction_grouper.go

@@ -658,7 +658,7 @@ func (g *PartitionCompactionGrouper) pickPartitionCompactionJob(partitionCompact
 		level.Info(partitionedGroupLogger).Log("msg", "found compactable group for user", "group", partitionedGroup.String())
 		begin := time.Now()
 
-		visitMarkerManager.MarkVisited(g.ctx)
+		visitMarkerManager.MarkPending(g.ctx)
 		level.Info(partitionedGroupLogger).Log("msg", "marked partition visited in group", "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds(), "group", partitionedGroup.String())
 
 		resolution := partitionedGroup.blocks[0].Thanos.Downsample.Resolution
@@ -711,9 +711,7 @@ func (g *PartitionCompactionGrouper) pickPartitionCompactionJob(partitionCompact
 
 		outGroups = append(outGroups, thanosGroup)
 		level.Debug(partitionedGroupLogger).Log("msg", "added partition to compaction groups")
-		// Grouper holds additional groups for compaction. In case, it lost
-		// competition for the first group it claimed.
-		if len(outGroups) >= g.compactionConcurrency+int(math.Min(math.Ceil(float64(g.compactionConcurrency)*0.1), 5)) {
+		if len(outGroups) >= g.compactionConcurrency {
 			break
 		}
 	}

pkg/compactor/partition_visit_marker.go

@@ -51,8 +51,8 @@ func (b *PartitionVisitMarker) IsVisited(partitionVisitMarkerTimeout time.Durati
 	return b.IsCompleted() || (partitionID == b.PartitionID && !b.IsExpired(partitionVisitMarkerTimeout))
 }
 
-func (b *PartitionVisitMarker) IsVisitedByCompactor(partitionVisitMarkerTimeout time.Duration, partitionID int, compactorID string) bool {
-	return b.CompactorID == compactorID && b.IsVisited(partitionVisitMarkerTimeout, partitionID)
+func (b *PartitionVisitMarker) IsPendingByCompactor(partitionVisitMarkerTimeout time.Duration, partitionID int, compactorID string) bool {
+	return b.CompactorID == compactorID && partitionID == b.PartitionID && b.IsPending() && !b.IsExpired(partitionVisitMarkerTimeout)
 }
 
 func (b *PartitionVisitMarker) IsCompleted() bool {
@@ -75,12 +75,18 @@ func (b *PartitionVisitMarker) GetVisitMarkerFilePath() string {
 	return GetPartitionVisitMarkerFilePath(b.PartitionedGroupID, b.PartitionID)
 }
 
-func (b *PartitionVisitMarker) MarkVisited(ownerIdentifier string) {
+func (b *PartitionVisitMarker) MarkInProgress(ownerIdentifier string) {
 	b.CompactorID = ownerIdentifier
 	b.Status = InProgress
 	b.VisitTime = time.Now().Unix()
 }
 
+func (b *PartitionVisitMarker) MarkPending(ownerIdentifier string) {
+	b.CompactorID = ownerIdentifier
+	b.Status = Pending
+	b.VisitTime = time.Now().Unix()
+}
+
 func (b *PartitionVisitMarker) MarkCompleted(ownerIdentifier string) {
 	b.CompactorID = ownerIdentifier
 	b.Status = Completed
@@ -95,15 +101,15 @@ func (b *PartitionVisitMarker) MarkFailed(ownerIdentifier string) {
 
 func (b *PartitionVisitMarker) LogInfo() []string {
 	return []string{
-		"partitioned_group_id",
+		"visit_marker_partitioned_group_id",
 		fmt.Sprintf("%d", b.PartitionedGroupID),
-		"partition_id",
+		"visit_marker_partition_id",
 		fmt.Sprintf("%d", b.PartitionID),
-		"compactor_id",
+		"visit_marker_compactor_id",
 		b.CompactorID,
-		"status",
+		"visit_marker_status",
 		string(b.Status),
-		"visit_time",
+		"visit_marker_visit_time",
 		time.Unix(b.VisitTime, 0).String(),
 	}
 }

pkg/compactor/shuffle_sharding_grouper.go

@@ -251,7 +251,7 @@ func (g *ShuffleShardingGrouper) Groups(blocks map[ulid.ULID]*metadata.Meta) (re
 		level.Info(partitionedGroupLogger).Log("msg", "found compactable group for user", "group", partitionedGroup.String())
 		begin := time.Now()
 
-		visitMarkerManager.MarkVisited(g.ctx)
+		visitMarkerManager.MarkPending(g.ctx)
 		level.Info(partitionedGroupLogger).Log("msg", "marked partition visited in group", "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds(), "group", partitionedGroup.String())
 
 		resolution := partitionedGroup.blocks[0].Thanos.Downsample.Resolution
@@ -304,9 +304,7 @@ func (g *ShuffleShardingGrouper) Groups(blocks map[ulid.ULID]*metadata.Meta) (re
 
 		outGroups = append(outGroups, thanosGroup)
 		level.Debug(partitionedGroupLogger).Log("msg", "added partition to compaction groups")
-		// Grouper holds additional groups for compaction. In case, it lost
-		// competition for the first group it claimed.
-		if len(outGroups) >= g.compactionConcurrency+int(math.Min(math.Ceil(float64(g.compactionConcurrency)*0.1), 5)) {
+		if len(outGroups) >= g.compactionConcurrency {
 			break
 		}
 	}

pkg/compactor/shuffle_sharding_grouper_test.go

@@ -207,7 +207,6 @@ func TestShuffleShardingGrouper_Groups(t *testing.T) {
 			blocks:      map[ulid.ULID]*metadata.Meta{block1hto2hExt1Ulid: blocks[block1hto2hExt1Ulid], block3hto4hExt1Ulid: blocks[block3hto4hExt1Ulid], block0hto1hExt1Ulid: blocks[block0hto1hExt1Ulid], block2hto3hExt1Ulid: blocks[block2hto3hExt1Ulid]},
 			expected: [][]ulid.ULID{
 				{block1hto2hExt1Ulid, block0hto1hExt1Ulid},
-				{block3hto4hExt1Ulid, block2hto3hExt1Ulid},
 			},
 			visitedPartitions: []struct {
 				partitionedGroupID uint32
@@ -239,7 +238,6 @@ func TestShuffleShardingGrouper_Groups(t *testing.T) {
 			expected: [][]ulid.ULID{
 				{block1hto2hExt2Ulid, block0hto1hExt2Ulid},
 				{block1hto2hExt1Ulid, block0hto1hExt1Ulid},
-				{block3hto4hExt1Ulid, block2hto3hExt1Ulid},
 			},
 		},
 		"test should skip block with no compact marker": {

pkg/compactor/shuffle_sharding_planner.go

@@ -14,6 +14,11 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 )
 
+var (
+	PlannerCompletedPartitionError = errors.New("got completed partition")
+	PlannerVisitedPartitionError   = errors.New("got partition visited by other compactor")
+)
+
 type ShuffleShardingPlanner struct {
 	ctx                                    context.Context
 	bkt                                    objstore.InstrumentedBucket
@@ -104,12 +109,12 @@ func (p *ShuffleShardingPlanner) PlanWithPartition(_ context.Context, metasByMin
 		if existingPartitionVisitMarker.IsCompleted() {
 			p.compactorMetrics.compactionsNotPlanned.WithLabelValues(p.userID, cortexMetaExtensions.TimeRangeStr()).Inc()
 			level.Warn(p.logger).Log("msg", "partition is in completed status", "partitioned_group_id", partitionedGroupID, "partition_id", partitionID, "compactor_id", p.ringLifecyclerID, existingPartitionVisitMarker.LogInfo())
-			return nil, nil
+			return nil, PlannerCompletedPartitionError
 		}
-		if !existingPartitionVisitMarker.IsVisitedByCompactor(p.partitionVisitMarkerTimeout, partitionID, p.ringLifecyclerID) {
+		if !existingPartitionVisitMarker.IsPendingByCompactor(p.partitionVisitMarkerTimeout, partitionID, p.ringLifecyclerID) {
 			p.compactorMetrics.compactionsNotPlanned.WithLabelValues(p.userID, cortexMetaExtensions.TimeRangeStr()).Inc()
 			level.Warn(p.logger).Log("msg", "partition is not visited by current compactor", "partitioned_group_id", partitionedGroupID, "partition_id", partitionID, "compactor_id", p.ringLifecyclerID, existingPartitionVisitMarker.LogInfo())
-			return nil, nil
+			return nil, PlannerVisitedPartitionError
 		}
 	}
 

pkg/compactor/shuffle_sharding_planner_test.go

@@ -241,7 +241,7 @@ func TestShuffleShardingPlanner_Plan(t *testing.T) {
 				isExpired:   false,
 				compactorID: otherCompactor,
 			},
-			expected: []*metadata.Meta{},
+			expectedErr: PlannerVisitedPartitionError,
 		},
 		"test should not compact if visit marker file is expired": {
 			ranges: []int64{2 * time.Hour.Milliseconds()},
@@ -265,7 +265,7 @@ func TestShuffleShardingPlanner_Plan(t *testing.T) {
 				isExpired:   true,
 				compactorID: currentCompactor,
 			},
-			expected: []*metadata.Meta{},
+			expectedErr: PlannerVisitedPartitionError,
 		},
 	}
 
@@ -285,6 +285,7 @@ func TestShuffleShardingPlanner_Plan(t *testing.T) {
 				PartitionedGroupID: partitionedGroupID,
 				PartitionID:        partitionID,
 				VisitTime:          expireTime.Unix(),
+				Status:             Pending,
 				Version:            PartitionVisitMarkerVersion1,
 			}
 			visitMarkerFileContent, _ := json.Marshal(visitMarker)