compactor

Documentation

Index

• Variables
• type BlocksCleaner
  • func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, ...) *BlocksCleaner
• type BlocksCleanerConfig
• type BlocksCompactorFactory
• type BlocksGrouperFactory
• type Compactor
  • func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, ...) (*Compactor, error)
  • func (c *Compactor) RingHandler(w http.ResponseWriter, req *http.Request)
• type Config
  • func (cfg *Config) RegisterFlags(f *flag.FlagSet)
  • func (cfg *Config) Validate() error
• type ConfigProvider
• type LabelRemoverFilter
  • func NewLabelRemoverFilter(labels []string) *LabelRemoverFilter
  • func (f *LabelRemoverFilter) Filter(_ context.Context, metas map[ulid.ULID]*metadata.Meta, _ *extprom.TxGaugeVec) error
• type RingConfig
  • func (cfg *RingConfig) RegisterFlags(f *flag.FlagSet)
  • func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig

Variables

var (
	RingOp = ring.NewOp([]ring.InstanceState{ring.ACTIVE}, nil)

	DefaultBlocksGrouperFactory = func(ctx context.Context, cfg Config, bkt objstore.Bucket, logger log.Logger, reg prometheus.Registerer, blocksMarkedForDeletion prometheus.Counter, garbageCollectedBlocks prometheus.Counter) compact.Grouper {
		return compact.NewDefaultGrouper(
			logger,
			bkt,
			false,
			true,
			reg,
			blocksMarkedForDeletion,
			garbageCollectedBlocks,
			prometheus.NewCounter(prometheus.CounterOpts{}),
			metadata.NoneFunc)
	}

	DefaultBlocksCompactorFactory = func(ctx context.Context, cfg Config, logger log.Logger, reg prometheus.Registerer) (compact.Compactor, compact.Planner, error) {
		compactor, err := tsdb.NewLeveledCompactor(ctx, reg, logger, cfg.BlockRanges.ToMilliseconds(), downsample.NewPool(), nil)
		if err != nil {
			return nil, nil, err
		}

		planner := compact.NewTSDBBasedPlanner(logger, cfg.BlockRanges.ToMilliseconds())
		return compactor, planner, nil
	}
)

Types

type BlocksCleaner

type BlocksCleaner struct {
	services.Service
	// contains filtered or unexported fields
}

func NewBlocksCleaner

func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, usersScanner *cortex_tsdb.UsersScanner, cfgProvider ConfigProvider, logger log.Logger, reg prometheus.Registerer) *BlocksCleaner

type BlocksCleanerConfig

type BlocksCleanerConfig struct {
	DeletionDelay                      time.Duration
	CleanupInterval                    time.Duration
	CleanupConcurrency                 int
	BlockDeletionMarksMigrationEnabled bool          // TODO Discuss whether we should remove it in Cortex 1.8.0 and document that upgrading to 1.7.0 before 1.8.0 is required.
	TenantCleanupDelay                 time.Duration // Delay before removing tenant deletion mark and "debug".
}

type BlocksCompactorFactory

type BlocksCompactorFactory func(
	ctx context.Context,
	cfg Config,
	logger log.Logger,
	reg prometheus.Registerer,
) (compact.Compactor, compact.Planner, error)

BlocksCompactorFactory builds and returns the compactor and planner to use to compact a tenant's blocks.

type BlocksGrouperFactory

type BlocksGrouperFactory func(
	ctx context.Context,
	cfg Config,
	bkt objstore.Bucket,
	logger log.Logger,
	reg prometheus.Registerer,
	blocksMarkedForDeletion prometheus.Counter,
	garbageCollectedBlocks prometheus.Counter,
) compact.Grouper

BlocksGrouperFactory builds and returns the grouper to use to compact a tenant's blocks.

type Compactor

type Compactor struct {
	services.Service
	// contains filtered or unexported fields
}

Compactor is a multi-tenant TSDB blocks compactor based on Thanos.

func NewCompactor

func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, cfgProvider ConfigProvider, logger log.Logger, registerer prometheus.Registerer) (*Compactor, error)

NewCompactor makes a new Compactor.

func (*Compactor) RingHandler

func (c *Compactor) RingHandler(w http.ResponseWriter, req *http.Request)

type Config

type Config struct {
	BlockRanges           cortex_tsdb.DurationList `yaml:"block_ranges"`
	BlockSyncConcurrency  int                      `yaml:"block_sync_concurrency"`
	MetaSyncConcurrency   int                      `yaml:"meta_sync_concurrency"`
	ConsistencyDelay      time.Duration            `yaml:"consistency_delay"`
	DataDir               string                   `yaml:"data_dir"`
	CompactionInterval    time.Duration            `yaml:"compaction_interval"`
	CompactionRetries     int                      `yaml:"compaction_retries"`
	CompactionConcurrency int                      `yaml:"compaction_concurrency"`
	CleanupInterval       time.Duration            `yaml:"cleanup_interval"`
	CleanupConcurrency    int                      `yaml:"cleanup_concurrency"`
	DeletionDelay         time.Duration            `yaml:"deletion_delay"`
	TenantCleanupDelay    time.Duration            `yaml:"tenant_cleanup_delay"`

	// Whether the migration of block deletion marks to the global markers location is enabled.
	BlockDeletionMarksMigrationEnabled bool `yaml:"block_deletion_marks_migration_enabled"`

	EnabledTenants  flagext.StringSliceCSV `yaml:"enabled_tenants"`
	DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"`

	// Compactors sharding.
	ShardingEnabled bool       `yaml:"sharding_enabled"`
	ShardingRing    RingConfig `yaml:"sharding_ring"`

	// Allow downstream projects to customise the blocks compactor.
	BlocksGrouperFactory   BlocksGrouperFactory   `yaml:"-"`
	BlocksCompactorFactory BlocksCompactorFactory `yaml:"-"`
	// contains filtered or unexported fields
}

Config holds the Compactor config.

func (*Config) RegisterFlags

func (cfg *Config) RegisterFlags(f *flag.FlagSet)

RegisterFlags registers the Compactor flags.

func (*Config) Validate

func (cfg *Config) Validate() error

type ConfigProvider

type ConfigProvider interface {
	bucket.TenantConfigProvider
	CompactorBlocksRetentionPeriod(user string) time.Duration
}

ConfigProvider defines the per-tenant config provider for the Compactor.

type LabelRemoverFilter

type LabelRemoverFilter struct {
	// contains filtered or unexported fields
}

func NewLabelRemoverFilter

func NewLabelRemoverFilter(labels []string) *LabelRemoverFilter

NewLabelRemoverFilter creates a LabelRemoverFilter.

func (*LabelRemoverFilter) Filter

func (f *LabelRemoverFilter) Filter(_ context.Context, metas map[ulid.ULID]*metadata.Meta, _ *extprom.TxGaugeVec) error

Filter modifies external labels of existing blocks, removing given labels from the metadata of blocks that have it.

type RingConfig

type RingConfig struct {
	KVStore          kv.Config     `yaml:"kvstore"`
	HeartbeatPeriod  time.Duration `yaml:"heartbeat_period"`
	HeartbeatTimeout time.Duration `yaml:"heartbeat_timeout"`

	// Wait ring stability.
	WaitStabilityMinDuration time.Duration `yaml:"wait_stability_min_duration"`
	WaitStabilityMaxDuration time.Duration `yaml:"wait_stability_max_duration"`

	// Instance details
	InstanceID             string   `yaml:"instance_id" doc:"hidden"`
	InstanceInterfaceNames []string `yaml:"instance_interface_names"`
	InstancePort           int      `yaml:"instance_port" doc:"hidden"`
	InstanceAddr           string   `yaml:"instance_addr" doc:"hidden"`

	// Injected internally
	ListenPort int `yaml:"-"`

	WaitActiveInstanceTimeout time.Duration `yaml:"wait_active_instance_timeout"`

	ObservePeriod time.Duration `yaml:"-"`
}

RingConfig masks the ring lifecycler config which contains many options not really required by the compactors ring. This config is used to strip down the config to the minimum, and avoid confusion to the user.

func (*RingConfig) RegisterFlags

func (cfg *RingConfig) RegisterFlags(f *flag.FlagSet)

RegisterFlags adds the flags required to config this to the given FlagSet

func (*RingConfig) ToLifecyclerConfig

func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig

ToLifecyclerConfig returns a LifecyclerConfig based on the compactor ring config.