Function NewCompactor

pkg/compactor/compactor.go:465–512

NewCompactor makes a new Compactor.

(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, logger log.Logger, registerer prometheus.Registerer, limits *validation.Overrides, ingestionReplicationFactor int)

// NewCompactor makes a new Compactor.
func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, logger log.Logger, registerer prometheus.Registerer, limits *validation.Overrides, ingestionReplicationFactor int) (*Compactor, error) {
	bucketClientFactory := func(ctx context.Context) (objstore.InstrumentedBucket, error) {
		return bucket.NewClient(ctx, storageCfg.Bucket, nil, "compactor", logger, registerer)
	}

	blocksGrouperFactory := compactorCfg.BlocksGrouperFactory
	if blocksGrouperFactory == nil {
		if compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
			blocksGrouperFactory = ShuffleShardingGrouperFactory
		} else {
			blocksGrouperFactory = DefaultBlocksGrouperFactory
		}
	}

	blocksCompactorFactory := compactorCfg.BlocksCompactorFactory
	if blocksCompactorFactory == nil {
		if compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
			blocksCompactorFactory = ShuffleShardingBlocksCompactorFactory
		} else {
			blocksCompactorFactory = DefaultBlocksCompactorFactory
		}
	}

	var blockDeletableCheckerFactory BlockDeletableCheckerFactory
	if compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle && compactorCfg.CompactionStrategy == util.CompactionStrategyPartitioning {
		blockDeletableCheckerFactory = PartitionCompactionBlockDeletableCheckerFactory
	} else {
		blockDeletableCheckerFactory = DefaultBlockDeletableCheckerFactory
	}

	var compactionLifecycleCallbackFactory CompactionLifecycleCallbackFactory
	if compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle && compactorCfg.CompactionStrategy == util.CompactionStrategyPartitioning {
		compactionLifecycleCallbackFactory = ShardedCompactionLifecycleCallbackFactory
	} else {
		compactionLifecycleCallbackFactory = DefaultCompactionLifecycleCallbackFactory
	}

	if ingestionReplicationFactor <= 0 {
		ingestionReplicationFactor = 1
	}

	cortexCompactor, err := newCompactor(compactorCfg, storageCfg, logger, registerer, bucketClientFactory, blocksGrouperFactory, blocksCompactorFactory, blockDeletableCheckerFactory, compactionLifecycleCallbackFactory, limits, ingestionReplicationFactor)
	if err != nil {
		return nil, errors.Wrap(err, "failed to create Cortex blocks compactor")
	}

	return cortexCompactor, nil
}

func newCompactor(
	compactorCfg Config,