retention

Documentation

Index

• func ExtractIntervalFromTableName(tableName string) model.Interval
• type ChunkClient
• type ChunkEntry
• type ChunkEntryCallback
• type ChunkIterator
• type ChunkRef
  • func (c ChunkRef) String() string
• type ExpirationChecker
  • func NeverExpiringExpirationChecker(limits Limits) ExpirationChecker
  • func NewExpirationChecker(limits Limits) ExpirationChecker
• type IndexProcessor
• type IntervalFilter
• type Limits
• type Marker
  • func NewMarker(workingDirectory string, expiration ExpirationChecker, ...) (*Marker, error)
  • func (t *Marker) MarkForDelete(ctx context.Context, tableName, userID string, indexProcessor IndexProcessor, ...) (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

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 ChunkEntryCallback

type ChunkEntryCallback func(ChunkEntry) (deleteChunk bool, err error)

type ChunkIterator

type ChunkIterator interface {
	ForEachChunk(ctx context.Context, callback ChunkEntryCallback) 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, []IntervalFilter)
	IntervalMayHaveExpiredChunks(interval model.Interval, userID string) bool
	MarkPhaseStarted()
	MarkPhaseFailed()
	MarkPhaseTimedOut()
	MarkPhaseFinished()
	DropFromIndex(ref ChunkEntry, tableEndTime model.Time, now model.Time) bool
}

func NeverExpiringExpirationChecker

func NeverExpiringExpirationChecker(limits Limits) ExpirationChecker

NeverExpiringExpirationChecker returns an expiration checker that never expires anything

func NewExpirationChecker

func NewExpirationChecker(limits Limits) ExpirationChecker

type IndexProcessor

type IndexProcessor interface {
	ChunkIterator

	SeriesCleaner
	// contains filtered or unexported methods
}

type IntervalFilter

type IntervalFilter struct {
	Interval model.Interval
	Filter   filter.Func
}

IntervalFilter contains the interval to delete and the function that filters lines. These will be applied to a chunk.

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, expiration ExpirationChecker, markTimeout time.Duration, chunkClient client.Client, r prometheus.Registerer) (*Marker, error)

func (*Marker) MarkForDelete

func (t *Marker) MarkForDelete(ctx context.Context, tableName, userID string, indexProcessor IndexProcessor, 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 {
	// CleanupSeries is for cleaning up the series that do have any chunks left in the index.
	// It would only be called for the series that have all their chunks deleted without adding new ones.
	CleanupSeries(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, indexProcessor IndexProcessor, 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