Method cleanUser

pkg/compactor/blocks_cleaner.go:596–763

(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string, firstRun bool)

}

func (c *BlocksCleaner) cleanUser(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string, firstRun bool) (returnErr error) {
	c.blocksMarkedForDeletion.WithLabelValues(userID, reasonValueRetention)
	startTime := time.Now()

	bucketIndexDeleted := false

	level.Info(userLogger).Log("msg", "started blocks cleanup and maintenance")
	defer func() {
		if returnErr != nil {
			level.Warn(userLogger).Log("msg", "failed blocks cleanup and maintenance", "err", returnErr)
		} else {
			level.Info(userLogger).Log("msg", "completed blocks cleanup and maintenance", "duration", time.Since(startTime), "duration_ms", time.Since(startTime).Milliseconds())
		}
		c.tenantCleanDuration.WithLabelValues(userID).Set(time.Since(startTime).Seconds())
	}()

	if c.cfg.ShardingStrategy == util.ShardingStrategyShuffle && c.cfg.CompactionStrategy == util.CompactionStrategyPartitioning {
		begin := time.Now()
		c.cleanPartitionedGroupInfo(ctx, userBucket, userLogger, userID)
		level.Info(userLogger).Log("msg", "finish cleaning partitioned group info files", "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
	}

	// Migrate block deletion marks to the global markers location. This operation is a best-effort.
	if firstRun && c.cfg.BlockDeletionMarksMigrationEnabled {
		if err := bucketindex.MigrateBlockDeletionMarksToGlobalLocation(ctx, c.bucketClient, userID, c.cfgProvider); err != nil {
			level.Warn(userLogger).Log("msg", "failed to migrate block deletion marks to the global markers location", "err", err)
		} else {
			level.Info(userLogger).Log("msg", "migrated block deletion marks to the global markers location")
		}
	}

	// Reading bucket index sync stats
	idxs, err := bucketindex.ReadSyncStatus(ctx, c.bucketClient, userID, userLogger)

	if err != nil {
		level.Warn(userLogger).Log("msg", "error reading the bucket index status", "err", err)
		idxs = bucketindex.Status{Version: bucketindex.SyncStatusFileVersion, NonQueryableReason: bucketindex.Unknown}
	}

	idxs.Status = bucketindex.Ok
	idxs.SyncTime = time.Now().Unix()

	// Read the bucket index.
	begin := time.Now()
	idx, err := bucketindex.ReadIndex(ctx, c.bucketClient, userID, c.cfgProvider, c.logger)

	defer func() {
		if bucketIndexDeleted {
			level.Info(userLogger).Log("msg", "deleting bucket index sync status since bucket index is empty")
			if err := bucketindex.DeleteIndexSyncStatus(ctx, c.bucketClient, userID); err != nil {
				level.Warn(userLogger).Log("msg", "error deleting index sync status when index is empty", "err", err)
			}
			if err := c.deleteNonDataFiles(ctx, userLogger, userBucket); err != nil {
				level.Warn(userLogger).Log("msg", "error deleting non-data files", "err", err)
			}
		} else {
			bucketindex.WriteSyncStatus(ctx, c.bucketClient, userID, idxs, userLogger)
		}