(ctx context.Context)
| 3254 | } |
| 3255 | |
| 3256 | func (i *Ingester) compactionLoop(ctx context.Context) error { |
| 3257 | infoFunc := func() (int, int) { |
| 3258 | if i.cfg.LifecyclerConfig.RingConfig.ZoneAwarenessEnabled { |
| 3259 | zones := i.lifecycler.Zones() |
| 3260 | if len(zones) != 0 { |
| 3261 | return slices.Index(zones, i.lifecycler.Zone), len(zones) |
| 3262 | } |
| 3263 | } |
| 3264 | |
| 3265 | // Lets create the slot based on the hash id |
| 3266 | i := int(client.HashAdd32(client.HashNew32(), i.lifecycler.ID) % 10) |
| 3267 | return i, 10 |
| 3268 | } |
| 3269 | ticker := util.NewSlottedTicker(infoFunc, i.cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval, 1) |
| 3270 | defer ticker.Stop() |
| 3271 | |
| 3272 | for ctx.Err() == nil { |
| 3273 | select { |
| 3274 | case <-ticker.C: |
| 3275 | i.compactBlocks(ctx, false, nil) |
| 3276 | |
| 3277 | case req := <-i.TSDBState.forceCompactTrigger: |
| 3278 | i.compactBlocks(ctx, true, req.users) |
| 3279 | close(req.callback) // Notify back. |
| 3280 | |
| 3281 | case <-ctx.Done(): |
| 3282 | return nil |
| 3283 | } |
| 3284 | } |
| 3285 | return nil |
| 3286 | } |
| 3287 | |
| 3288 | // Compacts all compactable blocks. Force flag will force compaction even if head is not compactable yet. |
| 3289 | func (i *Ingester) compactBlocks(ctx context.Context, force bool, allowed *users.AllowedTenants) { |
nothing calls this directly
no test coverage detected