retention

Documentation

Index

• Variables
• func ExtractIntervalFromTableName(tableName string) model.Interval
• type ChunkClient
• type ChunkEntry
• type ChunkEntryIterator
  • func NewChunkIndexIterator(bucket *bbolt.Bucket, config chunk.PeriodConfig) (ChunkEntryIterator, error)
• type ChunkRef
  • func (c ChunkRef) String() string
• type ExpirationChecker
  • func NewExpirationChecker(limits Limits) ExpirationChecker
• type InvalidIndexKeyError
  • func (e InvalidIndexKeyError) Error() string
  • func (e InvalidIndexKeyError) Is(target error) bool
• type LabelSeriesRangeKey
  • func (l LabelSeriesRangeKey) String() string
• type Limits
• type Marker
  • func NewMarker(workingDirectory string, config storage.SchemaConfig, ...) (*Marker, error)
  • func (t *Marker) MarkForDelete(ctx context.Context, tableName, userID string, db *bbolt.DB, logger log.Logger) (bool, bool, error)
• type MarkerProcessor
• type MarkerStorageWriter
  • func NewMarkerStorageWriter(workingDir string) (MarkerStorageWriter, error)
• type SeriesCleaner
• type Sweeper
  • func NewSweeper(workingDir string, deleteClient ChunkClient, deleteWorkerCount int, ...) (*Sweeper, error)
  • func (s *Sweeper) Start()
  • func (s *Sweeper) Stop()
• type TableMarker
• type TenantsRetention
  • func NewTenantsRetention(l Limits) *TenantsRetention
  • func (tr *TenantsRetention) RetentionPeriodFor(userID string, lbs labels.Labels) time.Duration

Variables

var ErrInvalidIndexKey = errors.New("invalid index key")

Functions

func ExtractIntervalFromTableName

func ExtractIntervalFromTableName(tableName string) model.Interval

ExtractIntervalFromTableName gives back the time interval for which the table is expected to hold the chunks index.

Types

type ChunkClient

type ChunkClient interface {
	DeleteChunk(ctx context.Context, userID, chunkID string) error
	IsChunkNotFoundErr(err error) bool
}

type ChunkEntry

type ChunkEntry struct {
	ChunkRef
	Labels labels.Labels
}

type ChunkEntryIterator

type ChunkEntryIterator interface {
	Next() bool
	Entry() ChunkEntry
	// Delete deletes the current entry.
	Delete() error
	Err() error
}

func NewChunkIndexIterator

func NewChunkIndexIterator(bucket *bbolt.Bucket, config chunk.PeriodConfig) (ChunkEntryIterator, error)

type ChunkRef

type ChunkRef struct {
	UserID   []byte
	SeriesID []byte
	ChunkID  []byte
	From     model.Time
	Through  model.Time
}

func (ChunkRef) String

func (c ChunkRef) String() string

type ExpirationChecker

type ExpirationChecker interface {
	Expired(ref ChunkEntry, now model.Time) (bool, []model.Interval)
	IntervalMayHaveExpiredChunks(interval model.Interval, userID string) bool
	MarkPhaseStarted()
	MarkPhaseFailed()
	MarkPhaseFinished()
	DropFromIndex(ref ChunkEntry, tableEndTime model.Time, now model.Time) bool
}

func NewExpirationChecker

func NewExpirationChecker(limits Limits) ExpirationChecker

type InvalidIndexKeyError

type InvalidIndexKeyError struct {
	HashKey  string
	RangeKey string
}

func (InvalidIndexKeyError) Error

func (e InvalidIndexKeyError) Error() string

func (InvalidIndexKeyError) Is

func (e InvalidIndexKeyError) Is(target error) bool

type LabelSeriesRangeKey

type LabelSeriesRangeKey struct {
	SeriesID []byte
	UserID   []byte
	Name     []byte
}

func (LabelSeriesRangeKey) String

func (l LabelSeriesRangeKey) String() string

type Limits

type Limits interface {
	RetentionPeriod(userID string) time.Duration
	StreamRetention(userID string) []validation.StreamRetention
	AllByUserID() map[string]*validation.Limits
	DefaultLimits() *validation.Limits
}

type Marker

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

func NewMarker

func NewMarker(workingDirectory string, config storage.SchemaConfig, expiration ExpirationChecker, chunkClient chunk.Client, r prometheus.Registerer) (*Marker, error)

func (*Marker) MarkForDelete

func (t *Marker) MarkForDelete(ctx context.Context, tableName, userID string, db *bbolt.DB, logger log.Logger) (bool, bool, error)

MarkForDelete marks all chunks expired for a given table.

type MarkerProcessor

type MarkerProcessor interface {
	// Start starts parsing marks and calling deleteFunc for each.
	// If deleteFunc returns no error the mark is deleted from the storage.
	// Otherwise the mark will reappears in future iteration.
	Start(deleteFunc func(ctx context.Context, chunkId []byte) error)
	// Stop stops processing marks.
	Stop()
}

type MarkerStorageWriter

type MarkerStorageWriter interface {
	Put(chunkID []byte) error
	Count() int64
	Close() error
}

func NewMarkerStorageWriter

func NewMarkerStorageWriter(workingDir string) (MarkerStorageWriter, error)

type SeriesCleaner

type SeriesCleaner interface {
	Cleanup(userID []byte, lbls labels.Labels) error
}

type Sweeper

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

func NewSweeper

func NewSweeper(workingDir string, deleteClient ChunkClient, deleteWorkerCount int, minAgeDelete time.Duration, r prometheus.Registerer) (*Sweeper, error)

func (*Sweeper) Start

func (s *Sweeper) Start()

func (*Sweeper) Stop

func (s *Sweeper) Stop()

type TableMarker

type TableMarker interface {
	// MarkForDelete marks chunks to delete for a given table and returns if it's empty or modified.
	MarkForDelete(ctx context.Context, tableName, userID string, db *bbolt.DB, logger log.Logger) (bool, bool, error)
}

type TenantsRetention

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

func NewTenantsRetention

func NewTenantsRetention(l Limits) *TenantsRetention

func (*TenantsRetention) RetentionPeriodFor

func (tr *TenantsRetention) RetentionPeriodFor(userID string, lbs labels.Labels) time.Duration