Method runActiveUserCleanup

pkg/compactor/blocks_cleaner.go:333–343

(ctx context.Context, jobChan <-chan *cleanerJob)

}

func (c *BlocksCleaner) runActiveUserCleanup(ctx context.Context, jobChan <-chan *cleanerJob) {
	for job := range jobChan {
		if job.timestamp < time.Now().Add(-c.cfg.CleanupInterval).Unix() {
			level.Warn(c.logger).Log("msg", "Active user cleaner job too old. Ignoring to get recent data")
			continue
		}
		err := c.cleanUpActiveUsers(ctx, job.users, false)

		c.checkRunError(activeStatus, err)
	}
}

func (c *BlocksCleaner) cleanUpActiveUsers(ctx context.Context, users []string, firstRun bool) error {
	level.Info(c.logger).Log("msg", "started blocks cleanup and maintenance for active users")