| 822 | } |
| 823 | |
| 824 | func (c *Compactor) compactUsers(ctx context.Context) { |
| 825 | succeeded := false |
| 826 | interrupted := false |
| 827 | compactionErrorCount := 0 |
| 828 | |
| 829 | c.CompactionRunsStarted.Inc() |
| 830 | |
| 831 | defer func() { |
| 832 | if succeeded && compactionErrorCount == 0 { |
| 833 | c.CompactionRunsCompleted.Inc() |
| 834 | c.CompactionRunsLastSuccess.SetToCurrentTime() |
| 835 | } else if interrupted { |
| 836 | c.CompactionRunsInterrupted.Inc() |
| 837 | } else { |
| 838 | c.CompactionRunsFailed.Inc() |
| 839 | } |
| 840 | |
| 841 | // Reset progress metrics once done. |
| 842 | c.CompactionRunDiscoveredTenants.Set(0) |
| 843 | c.CompactionRunSkippedTenants.Set(0) |
| 844 | c.CompactionRunSucceededTenants.Set(0) |
| 845 | c.CompactionRunFailedTenants.Set(0) |
| 846 | }() |
| 847 | |
| 848 | level.Info(c.logger).Log("msg", "discovering users from bucket") |
| 849 | userIDs, err := c.discoverUsersWithRetries(ctx) |
| 850 | if err != nil { |
| 851 | level.Error(c.logger).Log("msg", "failed to discover users from bucket", "err", err) |
| 852 | return |
| 853 | } |
| 854 | |
| 855 | level.Info(c.logger).Log("msg", "discovered users from bucket", "users", len(userIDs)) |
| 856 | c.CompactionRunDiscoveredTenants.Set(float64(len(userIDs))) |
| 857 | |
| 858 | // When starting multiple compactor replicas nearly at the same time, running in a cluster with |
| 859 | // a large number of tenants, we may end up in a situation where the 1st user is compacted by |
| 860 | // multiple replicas at the same time. Shuffling users helps reduce the likelihood this will happen. |
| 861 | rand.Shuffle(len(userIDs), func(i, j int) { |
| 862 | userIDs[i], userIDs[j] = userIDs[j], userIDs[i] |
| 863 | }) |
| 864 | |
| 865 | // Keep track of users owned by this shard, so that we can delete the local files for all other users. |
| 866 | ownedUsers := map[string]struct{}{} |
| 867 | for _, userID := range userIDs { |
| 868 | // Ensure the context has not been canceled (ie. compactor shutdown has been triggered). |
| 869 | if ctx.Err() != nil { |
| 870 | interrupted = true |
| 871 | level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "user", userID) |
| 872 | return |
| 873 | } |
| 874 | |
| 875 | // Ensure the user ID belongs to our shard. |
| 876 | if owned, err := c.ownUserForCompaction(userID); err != nil { |
| 877 | c.CompactionRunSkippedTenants.Inc() |
| 878 | level.Warn(c.logger).Log("msg", "unable to check if user is owned by this shard", "user", userID, "err", err) |
| 879 | continue |
| 880 | } else if !owned { |
| 881 | c.CompactionRunSkippedTenants.Inc() |