Method compactUserWithRetries

pkg/compactor/compactor.go:957–995

(ctx context.Context, userID string)

}

func (c *Compactor) compactUserWithRetries(ctx context.Context, userID string) error {
	var lastErr error

	retries := backoff.New(ctx, backoff.Config{
		MinBackoff: c.compactorCfg.retryMinBackoff,
		MaxBackoff: c.compactorCfg.retryMaxBackoff,
		MaxRetries: c.compactorCfg.CompactionRetries,
	})

	for retries.Ongoing() {
		lastErr = c.compactUser(ctx, userID)
		if lastErr == nil {
			return nil
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if c.isCausedByPermissionDenied(lastErr) {
			level.Warn(c.logger).Log("msg", "skipping compactUser due to PermissionDenied", "user", userID, "err", lastErr)
			c.compactorMetrics.compactionErrorsCount.WithLabelValues(userID, unauthorizedError).Inc()
			return nil
		}
		if compact.IsHaltError(lastErr) {
			level.Error(c.logger).Log("msg", "compactor returned critical error", "user", userID, "err", lastErr)
			c.compactorMetrics.compactionErrorsCount.WithLabelValues(userID, haltError).Inc()
			return lastErr
		}
		c.compactorMetrics.compactionErrorsCount.WithLabelValues(userID, retriableError).Inc()

		retries.Wait()
	}

	err := errors.Unwrap(errors.Cause(lastErr))
	if errors.Is(err, plannerCompletedPartitionError) || errors.Is(err, plannerVisitedPartitionError) {
		return nil
	}

	return lastErr
}

func (c *Compactor) compactUser(ctx context.Context, userID string) error {
	bucket := bucket.NewUserBucketClient(userID, c.bucketClient, c.limits)