Method starting

pkg/compactor/compactor.go:633–769

Start the compactor.

(ctx context.Context)

// Start the compactor.
func (c *Compactor) starting(ctx context.Context) error {
	begin := time.Now()
	defer func() {
		c.CompactorStartDurationSeconds.Set(time.Since(begin).Seconds())
		level.Info(c.logger).Log("msg", "compactor started", "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
	}()

	var err error

	// Create bucket client.
	c.bucketClient, err = c.bucketClientFactory(ctx)
	if err != nil {
		return errors.Wrap(err, "failed to create bucket client")
	}

	// Create blocks compactor dependencies.
	c.blocksCompactor, c.blocksPlannerFactory, err = c.blocksCompactorFactory(ctx, c.compactorCfg, c.logger, c.registerer)
	if err != nil {
		return errors.Wrap(err, "failed to initialize compactor dependencies")
	}

	// Wrap the bucket client to write block deletion marks in the global location too.
	c.bucketClient = bucketindex.BucketWithGlobalMarkers(c.bucketClient)

	cleanerBucketClient := c.bucketClient
	if c.compactorCfg.CleanerCachingBucketEnabled {
		cleanerBucketClient, err = cortex_tsdb.CreateCachingBucketForCompactor(c.storageCfg.BucketStore.MetadataCache, true, c.bucketClient, c.logger, extprom.WrapRegistererWith(prometheus.Labels{"component": "cleaner"}, c.registerer))
		if err != nil {
			return errors.Wrap(err, "create caching bucket for cleaner")
		}
	}

	if c.compactorCfg.CachingBucketEnabled {
		c.bucketClient, err = cortex_tsdb.CreateCachingBucketForCompactor(c.storageCfg.BucketStore.MetadataCache, false, c.bucketClient, c.logger, extprom.WrapRegistererWith(prometheus.Labels{"component": "compactor"}, c.registerer))
		if err != nil {
			return errors.Wrap(err, "create caching bucket for compactor")
		}
	}

	// Create the users scanner.
	c.usersScanner, err = users.NewScanner(c.storageCfg.UsersScanner, c.bucketClient, c.logger, extprom.WrapRegistererWith(prometheus.Labels{"component": "compactor"}, c.registerer))
	if err != nil {
		return errors.Wrap(err, "failed to create users scanner")
	}

	var cleanerRingLifecyclerID = "default-cleaner"
	// Initialize the compactors ring if sharding is enabled.
	if c.compactorCfg.ShardingEnabled {
		lifecyclerCfg := c.compactorCfg.ShardingRing.ToLifecyclerConfig()
		var delegate ring.LifecyclerDelegate
		delegate = &ring.DefaultLifecyclerDelegate{}
		if c.compactorCfg.ShardingRing.AutoForgetDelay > 0 {
			delegate = ring.NewLifecyclerAutoForgetDelegate(c.compactorCfg.ShardingRing.AutoForgetDelay, delegate, c.logger)
		}
		c.ringLifecycler, err = ring.NewLifecyclerWithDelegate(lifecyclerCfg, ring.NewNoopFlushTransferer(), "compactor", ringKey, true, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.registerer), delegate)
		if err != nil {
			return errors.Wrap(err, "unable to initialize compactor ring lifecycler")
		}