(ctx context.Context)
| 3147 | } |
| 3148 | |
| 3149 | func (i *Ingester) shipBlocksLoop(ctx context.Context) error { |
| 3150 | // We add a slight jitter to make sure that if the head compaction interval and ship interval are set to the same |
| 3151 | // value they don't clash (if they both continuously run at the same exact time, the head compaction may not run |
| 3152 | // because can't successfully change the state). |
| 3153 | shipTicker := time.NewTicker(util.DurationWithJitter(i.cfg.BlocksStorageConfig.TSDB.ShipInterval, 0.01)) |
| 3154 | defer shipTicker.Stop() |
| 3155 | |
| 3156 | for { |
| 3157 | select { |
| 3158 | case <-shipTicker.C: |
| 3159 | i.shipBlocks(ctx, nil) |
| 3160 | |
| 3161 | case req := <-i.TSDBState.shipTrigger: |
| 3162 | i.shipBlocks(ctx, req.users) |
| 3163 | close(req.callback) // Notify back. |
| 3164 | |
| 3165 | case <-ctx.Done(): |
| 3166 | return nil |
| 3167 | } |
| 3168 | } |
| 3169 | } |
| 3170 | |
| 3171 | // shipBlocks runs shipping for all users. |
| 3172 | func (i *Ingester) shipBlocks(ctx context.Context, allowed *users.AllowedTenants) { |
nothing calls this directly
no test coverage detected