tsdb

package
v2.0.0-alpha4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 27, 2021 License: Apache-2.0 Imports: 42 Imported by: 0

README

TSDB

GoDoc

This directory contains the Prometheus storage layer that is used in its 2.x releases.

A writeup of its design can be found here.

Based on the Gorilla TSDB white papers.

Video: Storing 16 Bytes at Scale from PromCon 2017.

See also the format documentation.

A series of blog posts explaining different components of TSDB:

Documentation

Overview

Package tsdb implements a time series storage for float64 sample data.

Index

Constants

View Source 
const (
	// WALMagic is a 4 byte number every WAL segment file starts with.
	WALMagic = uint32(0x43AF00EF)

	// WALFormatDefault is the version flag for the default outer segment file format.
	WALFormatDefault = byte(1)
)
View Source 
const (
	// Default duration of a block in milliseconds.
	DefaultBlockDuration = int64(2 * time.Hour / time.Millisecond)
)
View Source 
const (
	// DefaultStripeSize is the default number of entries to allocate in the stripeSeries hash map.
	DefaultStripeSize = 1 << 14
)

Variables

View Source 
var (
	// ErrInvalidSample is returned if an appended sample is not valid and can't
	// be ingested.
	ErrInvalidSample = errors.New("invalid sample")
	// ErrInvalidExemplar is returned if an appended exemplar is not valid and can't
	// be ingested.
	ErrInvalidExemplar = errors.New("invalid exemplar")
	// ErrAppenderClosed is returned if an appender has already be successfully
	// rolled back or committed.
	ErrAppenderClosed = errors.New("appender closed")
)
View Source 
var ErrClosed = errors.New("db already closed")

ErrClosed is returned when the db is closed.

View Source 
var ErrClosing = errors.New("block is closing")

ErrClosing is returned when a block is in the process of being closed.

View Source 
var ErrInvalidTimes = fmt.Errorf("max time is lesser than min time")
View Source 
var ErrNoSeriesAppended error = errors.New("no series appended, aborting")

ErrNoSeriesAppended is returned if the series count is zero while flushing blocks.

View Source 
var (
	// ErrNotReady is returned if the underlying storage is not ready yet.
	ErrNotReady = errors.New("TSDB not ready")
)

Functions

func BeyondSizeRetention 

func BeyondSizeRetention(db *DB, blocks []*Block) (deletable map[ulid.ULID]struct{})

BeyondSizeRetention returns those blocks which are beyond the size retention set in the db options.

func BeyondTimeRetention 

func BeyondTimeRetention(db *DB, blocks []*Block) (deletable map[ulid.ULID]struct{})

BeyondTimeRetention returns those blocks which are beyond the time retention set in the db options.

func CreateBlock 

func CreateBlock(series []storage.Series, dir string, chunkRange int64, logger log.Logger) (string, error)

CreateBlock creates a chunkrange block from the samples passed to it, and writes it to disk.

func ExponentialBlockRanges 

func ExponentialBlockRanges(minSize int64, steps, stepSize int) []int64

ExponentialBlockRanges returns the time ranges based on the stepSize.

func MigrateWAL 

func MigrateWAL(logger log.Logger, dir string) (err error)

MigrateWAL rewrites the deprecated write ahead log into the new format.

func NewBlockChunkQuerier 

func NewBlockChunkQuerier(b BlockReader, mint, maxt int64) (storage.ChunkQuerier, error)

NewBlockChunkQuerier returns a chunk querier against the block reader and requested min and max time range.

func NewBlockQuerier 

func NewBlockQuerier(b BlockReader, mint, maxt int64) (storage.Querier, error)

NewBlockQuerier returns a querier against the block reader and requested min and max time range.

func NewMergedStringIter 

func NewMergedStringIter(a index.StringIter, b index.StringIter) index.StringIter

NewMergedStringIter returns string iterator that allows to merge symbols on demand and stream result.

func PostingsForMatchers 

func PostingsForMatchers(ix IndexReader, ms ...*labels.Matcher) (index.Postings, error)

PostingsForMatchers assembles a single postings iterator against the index reader based on the given matchers. The resulting postings are not ordered by series.

Types

type Block 

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

Block represents a directory of time series data covering a continuous time range.

func OpenBlock 

func OpenBlock(logger log.Logger, dir string, pool chunkenc.Pool) (pb *Block, err error)

OpenBlock opens the block in the directory. It can be passed a chunk pool, which is used to instantiate chunk structs.

func (*Block) Chunks 

func (pb *Block) Chunks() (ChunkReader, error)

Chunks returns a new ChunkReader against the block data.

func (*Block) CleanTombstones 

func (pb *Block) CleanTombstones(dest string, c Compactor) (*ulid.ULID, bool, error)

CleanTombstones will remove the tombstones and rewrite the block (only if there are any tombstones). If there was a rewrite, then it returns the ULID of the new block written, else nil. If the resultant block is empty (tombstones covered the whole block), then it deletes the new block and return nil UID. It returns a boolean indicating if the parent block can be deleted safely of not.

func (*Block) Close 

func (pb *Block) Close() error

Close closes the on-disk block. It blocks as long as there are readers reading from the block.

func (*Block) Delete 

func (pb *Block) Delete(mint, maxt int64, ms ...*labels.Matcher) error

Delete matching series between mint and maxt in the block.

func (*Block) Dir 

func (pb *Block) Dir() string

Dir returns the directory of the block.

func (*Block) GetSymbolTableSize 

func (pb *Block) GetSymbolTableSize() uint64

GetSymbolTableSize returns the Symbol Table Size in the index of this block.

func (*Block) Index 

func (pb *Block) Index() (IndexReader, error)

Index returns a new IndexReader against the block data.

func (*Block) LabelNames 

func (pb *Block) LabelNames() ([]string, error)

LabelNames returns all the unique label names present in the Block in sorted order.

func (*Block) MaxTime 

func (pb *Block) MaxTime() int64

MaxTime returns the max time of the meta.

func (*Block) Meta 

func (pb *Block) Meta() BlockMeta

Meta returns meta information about the block.

func (*Block) MinTime 

func (pb *Block) MinTime() int64

MinTime returns the min time of the meta.

func (*Block) OverlapsClosedInterval 

func (pb *Block) OverlapsClosedInterval(mint, maxt int64) bool

OverlapsClosedInterval returns true if the block overlaps [mint, maxt].

func (*Block) Size 

func (pb *Block) Size() int64

Size returns the number of bytes that the block takes up.

func (*Block) Snapshot 

func (pb *Block) Snapshot(dir string) error

Snapshot creates snapshot of the block into dir.

func (*Block) String 

func (pb *Block) String() string

func (*Block) Tombstones 

func (pb *Block) Tombstones() (tombstones.Reader, error)

Tombstones returns a new TombstoneReader against the block data.

type BlockDesc 

type BlockDesc struct {
	ULID    ulid.ULID `json:"ulid"`
	MinTime int64     `json:"minTime"`
	MaxTime int64     `json:"maxTime"`
}

BlockDesc describes a block by ULID and time range.

type BlockMeta 

type BlockMeta struct {
	// Unique identifier for the block and its contents. Changes on compaction.
	ULID ulid.ULID `json:"ulid"`

	// MinTime and MaxTime specify the time range all samples
	// in the block are in.
	MinTime int64 `json:"minTime"`
	MaxTime int64 `json:"maxTime"`

	// Stats about the contents of the block.
	Stats BlockStats `json:"stats,omitempty"`

	// Information on compactions the block was created from.
	Compaction BlockMetaCompaction `json:"compaction"`

	// Version of the index format.
	Version int `json:"version"`
}

BlockMeta provides meta information about a block.

func CompactBlockMetas 

func CompactBlockMetas(uid ulid.ULID, blocks ...*BlockMeta) *BlockMeta

CompactBlockMetas merges many block metas into one, combining it's source blocks together and adjusting compaction level.

type BlockMetaCompaction 

type BlockMetaCompaction struct {
	// Maximum number of compaction cycles any source block has
	// gone through.
	Level int `json:"level"`
	// ULIDs of all source head blocks that went into the block.
	Sources []ulid.ULID `json:"sources,omitempty"`
	// Indicates that during compaction it resulted in a block without any samples
	// so it should be deleted on the next reloadBlocks.
	Deletable bool `json:"deletable,omitempty"`
	// Short descriptions of the direct blocks that were used to create
	// this block.
	Parents []BlockDesc `json:"parents,omitempty"`
	Failed  bool        `json:"failed,omitempty"`
}

BlockMetaCompaction holds information about compactions a block went through.

type BlockReader 

type BlockReader interface {
	// Index returns an IndexReader over the block's data.
	Index() (IndexReader, error)

	// Chunks returns a ChunkReader over the block's data.
	Chunks() (ChunkReader, error)

	// Tombstones returns a tombstones.Reader over the block's deleted data.
	Tombstones() (tombstones.Reader, error)

	// Meta provides meta information about the block reader.
	Meta() BlockMeta

	// Size returns the number of bytes that the block takes up on disk.
	Size() int64
}

BlockReader provides reading access to a data block.

type BlockStats 

type BlockStats struct {
	NumSamples    uint64 `json:"numSamples,omitempty"`
	NumSeries     uint64 `json:"numSeries,omitempty"`
	NumChunks     uint64 `json:"numChunks,omitempty"`
	NumTombstones uint64 `json:"numTombstones,omitempty"`
}

BlockStats contains stats about contents of a block.

type BlockWriter 

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

BlockWriter is a block writer that allows appending and flushing series to disk.

func NewBlockWriter 

func NewBlockWriter(logger log.Logger, dir string, blockSize int64) (*BlockWriter, error)

NewBlockWriter create a new block writer.

The returned writer accumulates all the series in the Head block until `Flush` is called.

Note that the writer will not check if the target directory exists or contains anything at all. It is the caller's responsibility to ensure that the resulting blocks do not overlap etc. Writer ensures the block flush is atomic (via rename).

func (*BlockWriter) Appender 

func (w *BlockWriter) Appender(ctx context.Context) storage.Appender

Appender returns a new appender on the database. Appender can't be called concurrently. However, the returned Appender can safely be used concurrently.

func (*BlockWriter) Close 

func (w *BlockWriter) Close() error

func (*BlockWriter) Flush 

func (w *BlockWriter) Flush(ctx context.Context) (ulid.ULID, error)

Flush implements the Writer interface. This is where actual block writing happens. After flush completes, no writes can be done.

type BlocksToDeleteFunc 

type BlocksToDeleteFunc func(blocks []*Block) map[ulid.ULID]struct{}

func DefaultBlocksToDelete 

func DefaultBlocksToDelete(db *DB) BlocksToDeleteFunc

DefaultBlocksToDelete returns a filter which decides time based and size based retention from the options of the db.

type ChunkReader 

type ChunkReader interface {
	// Chunk returns the series data chunk with the given reference.
	Chunk(ref uint64) (chunkenc.Chunk, error)

	// Close releases all underlying resources of the reader.
	Close() error
}

ChunkReader provides reading access of serialized time series data.

type ChunkWriter 

type ChunkWriter interface {
	// WriteChunks writes several chunks. The Chunk field of the ChunkMetas
	// must be populated.
	// After returning successfully, the Ref fields in the ChunkMetas
	// are set and can be used to retrieve the chunks from the written data.
	WriteChunks(chunks ...chunks.Meta) error

	// Close writes any required finalization and closes the resources
	// associated with the underlying writer.
	Close() error
}

ChunkWriter serializes a time block of chunked series data.

type CircularExemplarStorage 

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

func (*CircularExemplarStorage) AddExemplar 

func (ce *CircularExemplarStorage) AddExemplar(l labels.Labels, e exemplar.Exemplar) error

func (*CircularExemplarStorage) Appender 

func (ce *CircularExemplarStorage) Appender() *CircularExemplarStorage

func (*CircularExemplarStorage) ExemplarQuerier 

func (ce *CircularExemplarStorage) ExemplarQuerier(_ context.Context) (storage.ExemplarQuerier, error)

func (*CircularExemplarStorage) Querier 

func (ce *CircularExemplarStorage) Querier(_ context.Context) (storage.ExemplarQuerier, error)

func (*CircularExemplarStorage) Select 

func (ce *CircularExemplarStorage) Select(start, end int64, matchers ...[]*labels.Matcher) ([]exemplar.QueryResult, error)

Select returns exemplars for a given set of label matchers.

type Compactor 

type Compactor interface {
	// Plan returns a set of directories that can be compacted concurrently.
	// The directories can be overlapping.
	// Results returned when compactions are in progress are undefined.
	Plan(dir string) ([]string, error)

	// Write persists a Block into a directory.
	// No Block is written when resulting Block has 0 samples, and returns empty ulid.ULID{}.
	Write(dest string, b BlockReader, mint, maxt int64, parent *BlockMeta) (ulid.ULID, error)

	// Compact runs compaction against the provided directories. Must
	// only be called concurrently with results of Plan().
	// Can optionally pass a list of already open blocks,
	// to avoid having to reopen them.
	// When resulting Block has 0 samples
	//  * No block is written.
	//  * The source dirs are marked Deletable.
	//  * Returns empty ulid.ULID{}.
	Compact(dest string, dirs []string, open []*Block) (ulid.ULID, error)
}

Compactor provides compaction against an underlying storage of time series data.

type DB 

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

DB handles reads and writes of time series falling into a hashed partition of a seriedb.

func Open 

func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db *DB, err error)

Open returns a new DB in the given directory. If options are empty, DefaultOptions will be used.

func (*DB) Appender 

func (db *DB) Appender(ctx context.Context) storage.Appender

Appender opens a new appender against the database.

func (*DB) Blocks 

func (db *DB) Blocks() []*Block

Blocks returns the databases persisted blocks.

func (*DB) ChunkQuerier 

func (db *DB) ChunkQuerier(_ context.Context, mint, maxt int64) (storage.ChunkQuerier, error)

ChunkQuerier returns a new chunk querier over the data partition for the given time range.

func (*DB) CleanTombstones 

func (db *DB) CleanTombstones() (err error)

CleanTombstones re-writes any blocks with tombstones.

func (*DB) Close 

func (db *DB) Close() error

Close the partition.

func (*DB) Compact 

func (db *DB) Compact() (returnErr error)

Compact data if possible. After successful compaction blocks are reloaded which will also delete the blocks that fall out of the retention window. Old blocks are only deleted on reloadBlocks based on the new block's parent information. See DB.reloadBlocks documentation for further information.

func (*DB) CompactHead 

func (db *DB) CompactHead(head *RangeHead) error

CompactHead compacts the given RangeHead.

func (*DB) Delete 

func (db *DB) Delete(mint, maxt int64, ms ...*labels.Matcher) error

Delete implements deletion of metrics. It only has atomicity guarantees on a per-block basis.

func (*DB) Dir 

func (db *DB) Dir() string

Dir returns the directory of the database.

func (*DB) DisableCompactions 

func (db *DB) DisableCompactions()

DisableCompactions disables auto compactions.

func (*DB) EnableCompactions 

func (db *DB) EnableCompactions()

EnableCompactions enables auto compactions.

func (*DB) ExemplarQuerier 

func (db *DB) ExemplarQuerier(ctx context.Context) (storage.ExemplarQuerier, error)

func (*DB) Head 

func (db *DB) Head() *Head

Head returns the databases's head.

func (*DB) Querier 

func (db *DB) Querier(_ context.Context, mint, maxt int64) (storage.Querier, error)

Querier returns a new querier over the data partition for the given time range.

func (*DB) Snapshot 

func (db *DB) Snapshot(dir string, withHead bool) error

Snapshot writes the current data to the directory. If withHead is set to true it will create a new block containing all data that's currently in the memory buffer/WAL.

func (*DB) StartTime 

func (db *DB) StartTime() (int64, error)

StartTime implements the Storage interface.

func (*DB) String 

func (db *DB) String() string

type DBReadOnly 

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

DBReadOnly provides APIs for read only operations on a database. Current implementation doesn't support concurrency so all API calls should happen in the same go routine.

func OpenDBReadOnly 

func OpenDBReadOnly(dir string, l log.Logger) (*DBReadOnly, error)

OpenDBReadOnly opens DB in the given directory for read only operations.

func (*DBReadOnly) Blocks 

func (db *DBReadOnly) Blocks() ([]BlockReader, error)

Blocks returns a slice of block readers for persisted blocks.

func (*DBReadOnly) ChunkQuerier 

func (db *DBReadOnly) ChunkQuerier(ctx context.Context, mint, maxt int64) (storage.ChunkQuerier, error)

ChunkQuerier loads blocks and the wal and returns a new chunk querier over the data partition for the given time range. Current implementation doesn't support multiple ChunkQueriers.

func (*DBReadOnly) Close 

func (db *DBReadOnly) Close() error

Close all block readers.

func (*DBReadOnly) FlushWAL 

func (db *DBReadOnly) FlushWAL(dir string) (returnErr error)

FlushWAL creates a new block containing all data that's currently in the memory buffer/WAL. Samples that are in existing blocks will not be written to the new block. Note that if the read only database is running concurrently with a writable database then writing the WAL to the database directory can race.

func (*DBReadOnly) Querier 

func (db *DBReadOnly) Querier(ctx context.Context, mint, maxt int64) (storage.Querier, error)

Querier loads the blocks and wal and returns a new querier over the data partition for the given time range. Current implementation doesn't support multiple Queriers.

type DeletedIterator 

type DeletedIterator struct {
	// Iter is an Iterator to be wrapped.
	Iter chunkenc.Iterator
	// Intervals are the deletion intervals.
	Intervals tombstones.Intervals
}

DeletedIterator wraps chunk Iterator and makes sure any deleted metrics are not returned.

func (*DeletedIterator) At 

func (it *DeletedIterator) At() (int64, float64)

func (*DeletedIterator) Err 

func (it *DeletedIterator) Err() error

func (*DeletedIterator) Next 

func (it *DeletedIterator) Next() bool

func (*DeletedIterator) Seek 

func (it *DeletedIterator) Seek(t int64) bool

type ExemplarStorage 

type ExemplarStorage interface {
	storage.ExemplarQueryable
	AddExemplar(labels.Labels, exemplar.Exemplar) error
}

func NewCircularExemplarStorage 

func NewCircularExemplarStorage(len int, reg prometheus.Registerer) (ExemplarStorage, error)

NewCircularExemplarStorage creates an circular in memory exemplar storage. If we assume the average case 95 bytes per exemplar we can fit 5651272 exemplars in 1GB of extra memory, accounting for the fact that this is heap allocated space. If len < 1, then the exemplar storage is disabled. 

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

Head handles reads and writes of time series data within a time window.

func NewHead 

func NewHead(r prometheus.Registerer, l log.Logger, wal *wal.WAL, opts *HeadOptions) (*Head, error)

NewHead opens the head block in dir.

func (*Head) Appender 

func (h *Head) Appender(_ context.Context) storage.Appender

Appender returns a new Appender on the database.

func (*Head) Chunks 

func (h *Head) Chunks() (ChunkReader, error)

Chunks returns a ChunkReader against the block.

func (*Head) Close 

func (h *Head) Close() error

Close flushes the WAL and closes the head.

func (*Head) Delete 

func (h *Head) Delete(mint, maxt int64, ms ...*labels.Matcher) error

Delete all samples in the range of [mint, maxt] for series that satisfy the given label matchers.

func (*Head) ExemplarAppender 

func (h *Head) ExemplarAppender() storage.ExemplarAppender

func (*Head) ExemplarQuerier 

func (h *Head) ExemplarQuerier(ctx context.Context) (storage.ExemplarQuerier, error)

func (*Head) Index 

func (h *Head) Index() (IndexReader, error)

Index returns an IndexReader against the block.

func (*Head) Init 

func (h *Head) Init(minValidTime int64) error

Init loads data from the write ahead log and prepares the head for writes. It should be called before using an appender so that it limits the ingested samples to the head min valid time.

func (*Head) MaxTime 

func (h *Head) MaxTime() int64

MaxTime returns the highest timestamp seen in data of the head.

func (*Head) Meta 

func (h *Head) Meta() BlockMeta

Meta returns meta information about the head. The head is dynamic so will return dynamic results.

func (*Head) MinTime 

func (h *Head) MinTime() int64

MinTime returns the lowest time bound on visible data in the head.

func (*Head) NumSeries 

func (h *Head) NumSeries() uint64

NumSeries returns the number of active series in the head.

func (*Head) PostingsCardinalityStats 

func (h *Head) PostingsCardinalityStats(statsByLabelName string) *index.PostingsStats

PostingsCardinalityStats returns top 10 highest cardinality stats By label and value names.

func (*Head) SetMinValidTime 

func (h *Head) SetMinValidTime(minValidTime int64)

SetMinValidTime sets the minimum timestamp the head can ingest.

func (*Head) Size 

func (h *Head) Size() int64

func (*Head) Stats 

func (h *Head) Stats(statsByLabelName string) *Stats

Stats returns important current HEAD statistics. Note that it is expensive to calculate these.

func (*Head) String 

func (h *Head) String() string

String returns an human readable representation of the TSDB head. It's important to keep this function in order to avoid the struct dump when the head is stringified in errors or logs.

func (*Head) Tombstones 

func (h *Head) Tombstones() (tombstones.Reader, error)

Tombstones returns a new reader over the head's tombstones

func (*Head) Truncate 

func (h *Head) Truncate(mint int64) (err error)

Truncate removes old data before mint from the head and WAL.

type HeadOptions 

type HeadOptions struct {
	ChunkRange int64
	// ChunkDirRoot is the parent directory of the chunks directory.
	ChunkDirRoot         string
	ChunkPool            chunkenc.Pool
	ChunkWriteBufferSize int
	// StripeSize sets the number of entries in the hash map, it must be a power of 2.
	// A larger StripeSize will allocate more memory up-front, but will increase performance when handling a large number of series.
	// A smaller StripeSize reduces the memory allocated, but can decrease performance with large number of series.
	StripeSize     int
	SeriesCallback SeriesLifecycleCallback
	NumExemplars   int
}

HeadOptions are parameters for the Head block.

func DefaultHeadOptions 

func DefaultHeadOptions() *HeadOptions

type IndexReader 

type IndexReader interface {
	// Symbols return an iterator over sorted string symbols that may occur in
	// series' labels and indices. It is not safe to use the returned strings
	// beyond the lifetime of the index reader.
	Symbols() index.StringIter

	// SortedLabelValues returns sorted possible label values.
	SortedLabelValues(name string, matchers ...*labels.Matcher) ([]string, error)

	// LabelValues returns possible label values which may not be sorted.
	LabelValues(name string, matchers ...*labels.Matcher) ([]string, error)

	// Postings returns the postings list iterator for the label pairs.
	// The Postings here contain the offsets to the series inside the index.
	// Found IDs are not strictly required to point to a valid Series, e.g.
	// during background garbage collections. Input values must be sorted.
	Postings(name string, values ...string) (index.Postings, error)

	// SortedPostings returns a postings list that is reordered to be sorted
	// by the label set of the underlying series.
	SortedPostings(index.Postings) index.Postings

	// Series populates the given labels and chunk metas for the series identified
	// by the reference.
	// Returns storage.ErrNotFound if the ref does not resolve to a known series.
	Series(ref uint64, lset *labels.Labels, chks *[]chunks.Meta) error

	// LabelNames returns all the unique label names present in the index in sorted order.
	LabelNames() ([]string, error)

	// LabelValueFor returns label value for the given label name in the series referred to by ID.
	// If the series couldn't be found or the series doesn't have the requested label a
	// storage.ErrNotFound is returned as error.
	LabelValueFor(id uint64, label string) (string, error)

	// Close releases the underlying resources of the reader.
	Close() error
}

IndexReader provides reading access of serialized index data.

type IndexWriter 

type IndexWriter interface {
	// AddSymbols registers all string symbols that are encountered in series
	// and other indices. Symbols must be added in sorted order.
	AddSymbol(sym string) error

	// AddSeries populates the index writer with a series and its offsets
	// of chunks that the index can reference.
	// Implementations may require series to be insert in strictly increasing order by
	// their labels. The reference numbers are used to resolve entries in postings lists
	// that are added later.
	AddSeries(ref uint64, l labels.Labels, chunks ...chunks.Meta) error

	// Close writes any finalization and closes the resources associated with
	// the underlying writer.
	Close() error
}

IndexWriter serializes the index for a block of series data. The methods must be called in the order they are specified in.

type LeveledCompactor 

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

LeveledCompactor implements the Compactor interface.

func NewLeveledCompactor 

func NewLeveledCompactor(ctx context.Context, r prometheus.Registerer, l log.Logger, ranges []int64, pool chunkenc.Pool) (*LeveledCompactor, error)

NewLeveledCompactor returns a LeveledCompactor.

func NewLeveledCompactorWithChunkSize 

func NewLeveledCompactorWithChunkSize(ctx context.Context, r prometheus.Registerer, l log.Logger, ranges []int64, pool chunkenc.Pool, maxBlockChunkSegmentSize int64) (*LeveledCompactor, error)

func (*LeveledCompactor) Compact 

func (c *LeveledCompactor) Compact(dest string, dirs []string, open []*Block) (uid ulid.ULID, err error)

Compact creates a new block in the compactor's directory from the blocks in the provided directories.

func (*LeveledCompactor) Plan 

func (c *LeveledCompactor) Plan(dir string) ([]string, error)

Plan returns a list of compactable blocks in the provided directory.

func (*LeveledCompactor) Write 

func (c *LeveledCompactor) Write(dest string, b BlockReader, mint, maxt int64, parent *BlockMeta) (ulid.ULID, error)

type Options 

type Options struct {
	// Segments (wal files) max size.
	// WALSegmentSize = 0, segment size is default size.
	// WALSegmentSize > 0, segment size is WALSegmentSize.
	// WALSegmentSize < 0, wal is disabled.
	WALSegmentSize int

	// MaxBlockChunkSegmentSize is the max size of block chunk segment files.
	// MaxBlockChunkSegmentSize = 0, chunk segment size is default size.
	// MaxBlockChunkSegmentSize > 0, chunk segment size is MaxBlockChunkSegmentSize.
	MaxBlockChunkSegmentSize int64

	// Duration of persisted data to keep.
	// Unit agnostic as long as unit is consistent with MinBlockDuration and MaxBlockDuration.
	// Typically it is in milliseconds.
	RetentionDuration int64

	// Maximum number of bytes in blocks to be retained.
	// 0 or less means disabled.
	// NOTE: For proper storage calculations need to consider
	// the size of the WAL folder which is not added when calculating
	// the current size of the database.
	MaxBytes int64

	// NoLockfile disables creation and consideration of a lock file.
	NoLockfile bool

	// Overlapping blocks are allowed if AllowOverlappingBlocks is true.
	// This in-turn enables vertical compaction and vertical query merge.
	AllowOverlappingBlocks bool

	// WALCompression will turn on Snappy compression for records on the WAL.
	WALCompression bool

	// StripeSize is the size in entries of the series hash map. Reducing the size will save memory but impact performance.
	StripeSize int

	// The timestamp range of head blocks after which they get persisted.
	// It's the minimum duration of any persisted block.
	// Unit agnostic as long as unit is consistent with RetentionDuration and MaxBlockDuration.
	// Typically it is in milliseconds.
	MinBlockDuration int64

	// The maximum timestamp range of compacted blocks.
	// Unit agnostic as long as unit is consistent with MinBlockDuration and RetentionDuration.
	// Typically it is in milliseconds.
	MaxBlockDuration int64

	// HeadChunksWriteBufferSize configures the write buffer size used by the head chunks mapper.
	HeadChunksWriteBufferSize int

	// SeriesLifecycleCallback specifies a list of callbacks that will be called during a lifecycle of a series.
	// It is always a no-op in Prometheus and mainly meant for external users who import TSDB.
	SeriesLifecycleCallback SeriesLifecycleCallback

	// BlocksToDelete is a function which returns the blocks which can be deleted.
	// It is always the default time and size based retention in Prometheus and
	// mainly meant for external users who import TSDB.
	BlocksToDelete BlocksToDeleteFunc

	// MaxExemplars sets the size, in # of exemplars stored, of the single circular buffer used to store exemplars in memory.
	// See tsdb/exemplar.go, specifically the CircularExemplarStorage struct and it's constructor NewCircularExemplarStorage.
	MaxExemplars int
}

Options of the DB storage.

func DefaultOptions 

func DefaultOptions() *Options

DefaultOptions used for the DB. They are sane for setups using millisecond precision timestamps.

type Overlaps 

type Overlaps map[TimeRange][]BlockMeta

Overlaps contains overlapping blocks aggregated by overlapping range.

func OverlappingBlocks 

func OverlappingBlocks(bm []BlockMeta) Overlaps

OverlappingBlocks returns all overlapping blocks from given meta files.

func (Overlaps) String 

func (o Overlaps) String() string

String returns human readable string form of overlapped blocks.

type RangeHead 

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

func NewRangeHead 

func NewRangeHead(head *Head, mint, maxt int64) *RangeHead

NewRangeHead returns a *RangeHead.

func (*RangeHead) BlockMaxTime 

func (h *RangeHead) BlockMaxTime() int64

BlockMaxTime returns the max time of the potential block created from this head. It's different to MaxTime as we need to add +1 millisecond to block maxt because block intervals are half-open: [b.MinTime, b.MaxTime). Block intervals are always +1 than the total samples it includes.

func (*RangeHead) Chunks 

func (h *RangeHead) Chunks() (ChunkReader, error)

func (*RangeHead) Index 

func (h *RangeHead) Index() (IndexReader, error)

func (*RangeHead) MaxTime 

func (h *RangeHead) MaxTime() int64

MaxTime returns the max time of actual data fetch-able from the head. This controls the chunks time range which is closed [b.MinTime, b.MaxTime].

func (*RangeHead) Meta 

func (h *RangeHead) Meta() BlockMeta

func (*RangeHead) MinTime 

func (h *RangeHead) MinTime() int64

func (*RangeHead) NumSeries 

func (h *RangeHead) NumSeries() uint64

func (*RangeHead) Size 

func (h *RangeHead) Size() int64

func (*RangeHead) String 

func (h *RangeHead) String() string

String returns an human readable representation of the range head. It's important to keep this function in order to avoid the struct dump when the head is stringified in errors or logs.

func (*RangeHead) Tombstones 

func (h *RangeHead) Tombstones() (tombstones.Reader, error)

type SegmentWAL 

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

SegmentWAL is a write ahead log for series data.

DEPRECATED: use wal pkg combined with the record coders instead.

func OpenSegmentWAL 

func OpenSegmentWAL(dir string, logger log.Logger, flushInterval time.Duration, r prometheus.Registerer) (*SegmentWAL, error)

OpenSegmentWAL opens or creates a write ahead log in the given directory. The WAL must be read completely before new data is written.

func (*SegmentWAL) Close 

func (w *SegmentWAL) Close() error

Close syncs all data and closes the underlying resources.

func (*SegmentWAL) LogDeletes 

func (w *SegmentWAL) LogDeletes(stones []tombstones.Stone) error

LogDeletes write a batch of new deletes to the log.

func (*SegmentWAL) LogSamples 

func (w *SegmentWAL) LogSamples(samples []record.RefSample) error

LogSamples writes a batch of new samples to the log.

func (*SegmentWAL) LogSeries 

func (w *SegmentWAL) LogSeries(series []record.RefSeries) error

LogSeries writes a batch of new series labels to the log. The series have to be ordered.

func (*SegmentWAL) Reader 

func (w *SegmentWAL) Reader() WALReader

Reader returns a new reader over the write ahead log data. It must be completely consumed before writing to the WAL.

func (*SegmentWAL) Sync 

func (w *SegmentWAL) Sync() error

Sync flushes the changes to disk.

func (*SegmentWAL) Truncate 

func (w *SegmentWAL) Truncate(mint int64, keep func(uint64) bool) error

Truncate deletes the values prior to mint and the series which the keep function does not indicate to preserve.

type SeriesLifecycleCallback 

type SeriesLifecycleCallback interface {
	// PreCreation is called before creating a series to indicate if the series can be created.
	// A non nil error means the series should not be created.
	PreCreation(labels.Labels) error
	// PostCreation is called after creating a series to indicate a creation of series.
	PostCreation(labels.Labels)
	// PostDeletion is called after deletion of series.
	PostDeletion(...labels.Labels)
}

SeriesLifecycleCallback specifies a list of callbacks that will be called during a lifecycle of a series. It is always a no-op in Prometheus and mainly meant for external users who import TSDB. All the callbacks should be safe to be called concurrently. It is up to the user to implement soft or hard consistency by making the callbacks atomic or non-atomic. Atomic callbacks can cause degradation performance.

type Stats 

type Stats struct {
	NumSeries         uint64
	MinTime, MaxTime  int64
	IndexPostingStats *index.PostingsStats
}

type TimeRange 

type TimeRange struct {
	Min, Max int64
}

TimeRange specifies minTime and maxTime range.

type WAL 

type WAL interface {
	Reader() WALReader
	LogSeries([]record.RefSeries) error
	LogSamples([]record.RefSample) error
	LogDeletes([]tombstones.Stone) error
	Truncate(mint int64, keep func(uint64) bool) error
	Close() error
}

WAL is a write ahead log that can log new series labels and samples. It must be completely read before new entries are logged.

DEPRECATED: use wal pkg combined with the record codex instead.

type WALEntryType 

type WALEntryType uint8

WALEntryType indicates what data a WAL entry contains. 

const (
	WALEntrySymbols WALEntryType = 1
	WALEntrySeries  WALEntryType = 2
	WALEntrySamples WALEntryType = 3
	WALEntryDeletes WALEntryType = 4
)

Entry types in a segment file.

type WALReader 

type WALReader interface {
	Read(
		seriesf func([]record.RefSeries),
		samplesf func([]record.RefSample),
		deletesf func([]tombstones.Stone),
	) error
}

WALReader reads entries from a WAL.

Source Files

View all Source files

Directories

Path Synopsis
Package fileutil provides utility methods used when dealing with the filesystem in tsdb.
 Package fileutil provides utility methods used when dealing with the filesystem in tsdb.
Package goversion enforces the go version supported by the tsdb module.
 Package goversion enforces the go version supported by the tsdb module.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.