Method loop

pkg/compactor/blocks_cleaner.go:231–303

(ctx context.Context)

}

func (c *BlocksCleaner) loop(ctx context.Context) error {
	t := time.NewTicker(c.cfg.CleanupInterval)
	defer t.Stop()

	usersChan := make(chan *cleanerJob)
	deleteChan := make(chan *cleanerJob)
	defer close(usersChan)
	defer close(deleteChan)

	go func() {
		c.runActiveUserCleanup(ctx, usersChan)
	}()
	go func() {
		c.runDeleteUserCleanup(ctx, deleteChan)
	}()
	var metricsChan chan *cleanerJob
	if c.cfg.ShardingStrategy == util.ShardingStrategyShuffle &&
		c.cfg.CompactionStrategy == util.CompactionStrategyPartitioning {
		metricsChan = make(chan *cleanerJob)
		defer close(metricsChan)
		go func() {
			c.runEmitPartitionMetricsWorker(ctx, metricsChan)
		}()
	}

	for {
		select {
		case <-t.C:
			activeUsers, deletedUsers, err := c.scanUsers(ctx)
			if err != nil {
				level.Error(c.logger).Log("msg", "failed to scan users blocks cleanup and maintenance", "err", err.Error())
				c.runsFailed.WithLabelValues(deletedStatus).Inc()
				c.runsFailed.WithLabelValues(activeStatus).Inc()
				continue
			}
			cleanJobTimestamp := time.Now().Unix()

			select {
			case usersChan <- &cleanerJob{
				users:     activeUsers,
				timestamp: cleanJobTimestamp,
			}:
			default:
				level.Warn(c.logger).Log("msg", "unable to push cleaning job to usersChan")
				c.enqueueJobFailed.WithLabelValues(activeStatus).Inc()
			}

			select {
			case deleteChan <- &cleanerJob{
				users:     deletedUsers,
				timestamp: cleanJobTimestamp,
			}:
			default:
				level.Warn(c.logger).Log("msg", "unable to push deletion job to deleteChan")
				c.enqueueJobFailed.WithLabelValues(deletedStatus).Inc()
			}

			if metricsChan != nil {