| 343 | } |
| 344 | |
| 345 | func (c *BlocksCleaner) cleanUpActiveUsers(ctx context.Context, users []string, firstRun bool) error { |
| 346 | level.Info(c.logger).Log("msg", "started blocks cleanup and maintenance for active users") |
| 347 | c.runsStarted.WithLabelValues(activeStatus).Inc() |
| 348 | |
| 349 | return concurrency.ForEachUser(ctx, users, c.cfg.CleanupConcurrency, func(ctx context.Context, userID string) error { |
| 350 | userLogger := util_log.WithUserID(userID, c.logger) |
| 351 | userBucket := bucket.NewUserBucketClient(userID, c.bucketClient, c.cfgProvider) |
| 352 | visitMarkerManager, isVisited, err := c.obtainVisitMarkerManager(ctx, userLogger, userBucket) |
| 353 | if err != nil { |
| 354 | return err |
| 355 | } |
| 356 | if isVisited { |
| 357 | return nil |
| 358 | } |
| 359 | errChan := make(chan error, 1) |
| 360 | go visitMarkerManager.HeartBeat(ctx, errChan, c.cleanerVisitMarkerFileUpdateInterval, true) |
| 361 | defer func() { |
| 362 | errChan <- nil |
| 363 | }() |
| 364 | return errors.Wrapf(c.cleanUser(ctx, userLogger, userBucket, userID, firstRun), "failed to delete blocks for user: %s", userID) |
| 365 | }) |
| 366 | } |
| 367 | |
| 368 | func (c *BlocksCleaner) runDeleteUserCleanup(ctx context.Context, jobChan chan *cleanerJob) { |
| 369 | for job := range jobChan { |