compactor

type Chunk interface {
	GetFrom() model.Time
	GetThrough() model.Time
	GetFingerprint() uint64
	GetChecksum() uint32
	GetSize() uint32
	GetEntriesCount() uint32
}

type CompactedIndex interface {
	// IndexProcessor is used for applying custom retention and processing delete requests.
	retention.IndexProcessor
	// Cleanup should clean up all the state built during compaction.
	// It is typically called at the end or in case of an error.
	Cleanup()
	// ToIndexFile is used to convert the CompactedIndex to an IndexFile for uploading to the object store.
	// Once the IndexFile is uploaded using Index.Reader, the file is closed using Index.Close and removed from disk using Index.Path.
	ToIndexFile() (index.Index, error)
}

type Compactor struct {
	services.Service

	DeleteRequestsHandler     *deletion.DeleteRequestHandler
	DeleteRequestsGRPCHandler *deletion.GRPCRequestHandler

	JobQueue *jobqueue.Queue
	// contains filtered or unexported fields
}

type Config struct {
	WorkingDirectory               string                `yaml:"working_directory"`
	CompactionInterval             time.Duration         `yaml:"compaction_interval"`
	ApplyRetentionInterval         time.Duration         `yaml:"apply_retention_interval"`
	RetentionEnabled               bool                  `yaml:"retention_enabled"`
	RetentionDeleteDelay           time.Duration         `yaml:"retention_delete_delay"`
	RetentionDeleteWorkCount       int                   `yaml:"retention_delete_worker_count"`
	RetentionTableTimeout          time.Duration         `yaml:"retention_table_timeout"`
	RetentionBackoffConfig         backoff.Config        `yaml:"retention_backoff_config"`
	DeleteRequestStore             string                `yaml:"delete_request_store"`
	DeleteRequestStoreKeyPrefix    string                `yaml:"delete_request_store_key_prefix"`
	DeleteRequestStoreDBType       string                `yaml:"delete_request_store_db_type"`
	BackupDeleteRequestStoreDBType string                `yaml:"backup_delete_request_store_db_type"`
	DeleteBatchSize                int                   `yaml:"delete_batch_size"`
	DeleteRequestCancelPeriod      time.Duration         `yaml:"delete_request_cancel_period"`
	DeleteMaxInterval              time.Duration         `yaml:"delete_max_interval"`
	MaxCompactionParallelism       int                   `yaml:"max_compaction_parallelism"`
	UploadParallelism              int                   `yaml:"upload_parallelism"`
	CompactorRing                  lokiring.RingConfig   `` /* 210-byte string literal not displayed */
	RunOnce                        bool                  `yaml:"_" doc:"hidden"`
	TablesToCompact                int                   `yaml:"tables_to_compact"`
	SkipLatestNTables              int                   `yaml:"skip_latest_n_tables"`
	HorizontalScalingMode          string                `yaml:"horizontal_scaling_mode"`
	WorkerConfig                   jobqueue.WorkerConfig `yaml:"worker_config"`
	JobsConfig                     JobsConfig            `yaml:"jobs_config"`
}

type DeleteRequestResponse struct {
	RequestID string `json:"request_id"`
	StartTime int64  `json:"start_time"`
	EndTime   int64  `json:"end_time"`
	Query     string `json:"query"`
	Status    string `json:"status"`
	CreatedAt int64  `json:"created_at"`
	UserID    string `json:"user_id"`
}

type DeletionJobsConfig struct {
	DeletionManifestStorePrefix string        `yaml:"deletion_manifest_store_prefix"`
	ChunkProcessingConcurrency  int           `yaml:"chunk_processing_concurrency"`
	Timeout                     time.Duration `yaml:"timeout"`
	MaxRetries                  int           `yaml:"max_retries"`
}

type IndexCompactor interface {
	// NewTableCompactor returns a new TableCompactor for compacting a table.
	// commonIndexSet refers to common index files or in other words multi-tenant index.
	// existingUserIndexSet refers to existing user specific index files in the storage.
	// makeEmptyUserIndexSetFunc can be used for creating an empty indexSet for a user
	// who does not have an index for it in existingUserIndexSet.
	// periodConfig holds the PeriodConfig for the table.
	NewTableCompactor(
		ctx context.Context,
		commonIndexSet IndexSet,
		existingUserIndexSet map[string]IndexSet,
		makeEmptyUserIndexSetFunc MakeEmptyUserIndexSetFunc,
		periodConfig config.PeriodConfig,
	) TableCompactor

	// OpenCompactedIndexFile opens a compressed index file at given path.
	OpenCompactedIndexFile(
		ctx context.Context,
		path,
		tableName,
		userID,
		workingDir string,
		periodConfig config.PeriodConfig,
		logger log.Logger,
	) (
		CompactedIndex,
		error,
	)
}

type IndexFileConfig struct {
	CompressFile bool
}

type IndexRecords struct {
	Start, NumRecords int
}

type IndexSet interface {
	GetTableName() string
	ListSourceFiles() []storage.IndexFile
	GetSourceFile(indexFile storage.IndexFile) (string, error)
	GetLogger() log.Logger
	GetWorkingDir() string
	// SetCompactedIndex sets the CompactedIndex for upload/applying retention and making the compactor remove the source files.
	// CompactedIndex can be nil only in case of all the source files in common index set being compacted away to per tenant index.
	// It would return an error if the CompactedIndex is nil and removeSourceFiles is true in case of user index set since
	// compaction should either create new files or can be a noop if there is nothing to compact.
	// There is no need to call SetCompactedIndex if no changes were made to the index for this IndexSet.
	SetCompactedIndex(compactedIndex CompactedIndex, removeSourceFiles bool) error
}

type IndexesConfig struct {
	NumUnCompactedFiles, NumCompactedFiles int
}

type JobsConfig struct {
	Deletion DeletionJobsConfig `yaml:"deletion"`
}

type Limits interface {
	deletion.Limits
	retention.Limits
	DefaultLimits() *validation.Limits
}

type MakeEmptyUserIndexSetFunc func(userID string) (IndexSet, error)

type PerUserIndexesConfig struct {
	IndexesConfig
	NumUsers int
}

type TableCompactor interface {
	// CompactTable compacts the table.
	// After compaction is done successfully, it should set the new/updated CompactedIndex for relevant IndexSets.
	CompactTable() (err error)
}

type TablesManager interface {
	CompactTable(ctx context.Context, tableName string, applyRetention bool) error
	ApplyStorageUpdates(ctx context.Context, iterator deletion.StorageUpdatesIterator) error
	IterateTables(ctx context.Context, callback func(string, deletion.Table) error) (err error)
}