Method running

pkg/compactor/compactor.go:786–822

(ctx context.Context)

}

func (c *Compactor) running(ctx context.Context) error {
	// Ensure an initial cleanup occurred as first thing when running compactor.
	if err := services.StartAndAwaitRunning(ctx, c.blocksCleaner); err != nil {
		c.ringSubservices.StopAsync()
		return errors.Wrap(err, "failed to start the blocks cleaner")
	}

	if c.userIndexUpdater != nil {
		go c.userIndexUpdateLoop(ctx)
	}

	// Run an initial compaction before starting the interval.
	// Insert jitter right before compaction starts to avoid multiple starting compactor to be in sync
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(time.Duration(rand.Int63n(int64(float64(c.compactorCfg.CompactionInterval) * 0.1)))):
	}
	c.compactUsers(ctx)

	ticker := time.NewTicker(c.compactorCfg.CompactionInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			// Insert jitter right before compaction starts, so that there will always
			// have jitter even compaction time is longer than CompactionInterval
			time.Sleep(time.Duration(rand.Int63n(int64(float64(c.compactorCfg.CompactionInterval) * 0.1))))
			c.compactUsers(ctx)
		case <-ctx.Done():
			return nil
		case err := <-c.ringSubservicesWatcher.Chan():
			return errors.Wrap(err, "compactor subservice failed")
		}
	}
}

func (c *Compactor) compactUsers(ctx context.Context) {
	succeeded := false