(userID string, isCleanUp bool)
| 1168 | } |
| 1169 | |
| 1170 | func (c *Compactor) ownUser(userID string, isCleanUp bool) (bool, error) { |
| 1171 | if !c.allowedTenants.IsAllowed(userID) { |
| 1172 | return false, nil |
| 1173 | } |
| 1174 | |
| 1175 | // Always owned if sharding is disabled |
| 1176 | if !c.compactorCfg.ShardingEnabled { |
| 1177 | return true, nil |
| 1178 | } |
| 1179 | |
| 1180 | // If we aren't cleaning up user blocks, and we are using shuffle-sharding, ownership is determined by a subring |
| 1181 | // Cleanup should only be owned by a single compactor, as there could be race conditions during block deletion |
| 1182 | if !isCleanUp && c.compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle { |
| 1183 | shardSize := c.getShardSizeForUser(userID) |
| 1184 | subRing := c.ring.ShuffleShard(userID, shardSize) |
| 1185 | |
| 1186 | rs, err := subRing.GetAllHealthy(RingOp) |
| 1187 | if err != nil { |
| 1188 | return false, err |
| 1189 | } |
| 1190 | |
| 1191 | return rs.Includes(c.ringLifecycler.Addr), nil |
| 1192 | } |
| 1193 | |
| 1194 | // Hash the user ID. |
| 1195 | hasher := fnv.New32a() |
| 1196 | _, _ = hasher.Write([]byte(userID)) |
| 1197 | userHash := hasher.Sum32() |
| 1198 | |
| 1199 | // Check whether this compactor instance owns the user. |
| 1200 | rs, err := c.ring.Get(userHash, RingOp, nil, nil, nil) |
| 1201 | if err != nil { |
| 1202 | return false, err |
| 1203 | } |
| 1204 | |
| 1205 | if len(rs.Instances) != 1 { |
| 1206 | return false, fmt.Errorf("unexpected number of compactors in the shard (expected 1, got %d)", len(rs.Instances)) |
| 1207 | } |
| 1208 | |
| 1209 | return rs.Instances[0].Addr == c.ringLifecycler.Addr, nil |
| 1210 | } |
| 1211 | |
| 1212 | func (c *Compactor) userIndexUpdateLoop(ctx context.Context) { |
| 1213 | // Hardcode ID to check which compactor owns updating user index. |
no test coverage detected