bucketindex

Documentation

Index

• Constants
• Variables
• func DeleteIndex(ctx context.Context, bkt objstore.Bucket, userID string, ...) error
• func WriteIndex(ctx context.Context, bkt objstore.Bucket, userID string, ...) error
• type Block
  • func BlockFromMeta(meta block.Meta) *Block
  • func (m *Block) GetUploadedAt() time.Time
  • func (m *Block) String() string
  • func (m *Block) Within(minT, maxT model.Time) bool
• type BlockDeletionMark
  • func DeletionMarkFromBlockMarker(mark *block.DeletionMark) *BlockDeletionMark
  • func (m *BlockDeletionMark) BlockDeletionMark() *block.DeletionMark
  • func (m *BlockDeletionMark) GetDeletionTime() time.Time
• type BlockDeletionMarks
  • func (s BlockDeletionMarks) Clone() BlockDeletionMarks
  • func (s BlockDeletionMarks) GetULIDs() []ulid.ULID
• type Blocks
  • func (s Blocks) GetULIDs() []ulid.ULID
  • func (s Blocks) String() string
• type Index
  • func ReadIndex(ctx context.Context, bkt objstore.Bucket, userID string, ...) (*Index, error)
  • func (idx *Index) GetUpdatedAt() time.Time
  • func (idx *Index) RemoveBlock(id ulid.ULID)
• type Loader
  • func NewLoader(cfg LoaderConfig, bucketClient objstore.Bucket, ...) *Loader
  • func (l *Loader) GetIndex(ctx context.Context, userID string) (*Index, error)
• type LoaderConfig
• type Updater
  • func NewUpdater(bkt objstore.Bucket, userID string, cfgProvider objstore.TenantConfigProvider, ...) *Updater
  • func (w *Updater) UpdateIndex(ctx context.Context, old *Index) (*Index, map[ulid.ULID]error, error)

Constants

const (
	IndexFilename           = "bucket-index.json"
	IndexCompressedFilename = IndexFilename + ".gz"
	IndexVersion1           = 1
	IndexVersion2           = 2 // Added CompactorShardID field.
)

Variables

var (
	ErrIndexNotFound  = errors.New("bucket index not found")
	ErrIndexCorrupted = errors.New("bucket index corrupted")
)

var (
	ErrBlockMetaNotFound          = block.ErrorSyncMetaNotFound
	ErrBlockMetaCorrupted         = block.ErrorSyncMetaCorrupted
	ErrBlockDeletionMarkNotFound  = errors.New("block deletion mark not found")
	ErrBlockDeletionMarkCorrupted = errors.New("block deletion mark corrupted")
)

Functions

func DeleteIndex

func DeleteIndex(ctx context.Context, bkt objstore.Bucket, userID string, cfgProvider objstore.TenantConfigProvider) error

DeleteIndex deletes the bucket index from the storage. No error is returned if the index does not exist.

func WriteIndex

func WriteIndex(ctx context.Context, bkt objstore.Bucket, userID string, cfgProvider objstore.TenantConfigProvider, idx *Index) error

WriteIndex uploads the provided index to the storage.

Types

type Block

type Block struct {
	// Block ID.
	ID ulid.ULID `json:"block_id"`

	// MinTime and MaxTime specify the time range all samples in the block are in (millis precision).
	MinTime model.Time `json:"min_time"`
	MaxTime model.Time `json:"max_time"`

	// UploadedAt is a unix timestamp (seconds precision) of when the block has been completed to be uploaded
	// to the storage.
	UploadedAt int64 `json:"uploaded_at"`

	// Block's compactor shard ID, copied from tsdb.CompactorShardIDExternalLabel label.
	CompactorShardID string `json:"compactor_shard_id,omitempty"`
}

Block holds the information about a block in the index.

func BlockFromMeta

func BlockFromMeta(meta block.Meta) *Block

func (*Block) GetUploadedAt

func (m *Block) GetUploadedAt() time.Time

func (*Block) String

func (m *Block) String() string

func (*Block) Within

func (m *Block) Within(minT, maxT model.Time) bool

Within returns whether the block contains samples within the provided range. Input minT and maxT are both inclusive.

type BlockDeletionMark

type BlockDeletionMark struct {
	// Block ID.
	ID ulid.ULID `json:"block_id"`

	// DeletionTime is a unix timestamp (seconds precision) of when the block was marked to be deleted.
	DeletionTime int64 `json:"deletion_time"`
}

BlockDeletionMark holds the information about a block's deletion mark in the index.

func DeletionMarkFromBlockMarker

func DeletionMarkFromBlockMarker(mark *block.DeletionMark) *BlockDeletionMark

func (*BlockDeletionMark) BlockDeletionMark

func (m *BlockDeletionMark) BlockDeletionMark() *block.DeletionMark

BlockDeletionMark returns the block deletion mark.

func (*BlockDeletionMark) GetDeletionTime

func (m *BlockDeletionMark) GetDeletionTime() time.Time

type BlockDeletionMarks

type BlockDeletionMarks []*BlockDeletionMark

BlockDeletionMarks holds a set of block deletion marks in the index. No ordering guaranteed.

func (BlockDeletionMarks) Clone

func (s BlockDeletionMarks) Clone() BlockDeletionMarks

func (BlockDeletionMarks) GetULIDs

func (s BlockDeletionMarks) GetULIDs() []ulid.ULID

type Blocks

type Blocks []*Block

Blocks holds a set of blocks in the index. No ordering guaranteed.

func (Blocks) GetULIDs

func (s Blocks) GetULIDs() []ulid.ULID

func (Blocks) String

func (s Blocks) String() string

type Index

type Index struct {
	// Version of the index format.
	Version int `json:"version"`

	// List of complete blocks (partial blocks are excluded from the index).
	Blocks Blocks `json:"blocks"`

	// List of block deletion marks.
	BlockDeletionMarks BlockDeletionMarks `json:"block_deletion_marks"`

	// UpdatedAt is a unix timestamp (seconds precision) of when the index has been updated
	// (written in the storage) the last time.
	UpdatedAt int64 `json:"updated_at"`
}

Index contains all known blocks and markers of a tenant.

func ReadIndex

func ReadIndex(ctx context.Context, bkt objstore.Bucket, userID string, cfgProvider objstore.TenantConfigProvider, logger log.Logger) (*Index, error)

ReadIndex reads, parses and returns a bucket index from the bucket. ReadIndex has a one-minute timeout for completing the read against the bucket. One minute is hard-coded to a reasonably high value to protect against operations that can take unbounded time.

func (*Index) GetUpdatedAt

func (idx *Index) GetUpdatedAt() time.Time

func (*Index) RemoveBlock

func (idx *Index) RemoveBlock(id ulid.ULID)

RemoveBlock removes block and its deletion mark (if any) from index.

type Loader

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

Loader is responsible to lazy load bucket indexes and, once loaded for the first time, keep them updated in background. Loaded indexes are automatically offloaded once the idle timeout expires.

func NewLoader

func NewLoader(cfg LoaderConfig, bucketClient objstore.Bucket, cfgProvider objstore.TenantConfigProvider, logger log.Logger, reg prometheus.Registerer) *Loader

NewLoader makes a new Loader.

func (*Loader) GetIndex

func (l *Loader) GetIndex(ctx context.Context, userID string) (*Index, error)

GetIndex returns the bucket index for the given user. It returns the in-memory cached index if available, or load it from the bucket otherwise.

type LoaderConfig

type LoaderConfig struct {
	CheckInterval         time.Duration
	UpdateOnStaleInterval time.Duration
	UpdateOnErrorInterval time.Duration
	IdleTimeout           time.Duration
}

type Updater

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

Updater is responsible to generate an update in-memory bucket index.

func NewUpdater

func NewUpdater(bkt objstore.Bucket, userID string, cfgProvider objstore.TenantConfigProvider, logger log.Logger) *Updater

func (*Updater) UpdateIndex

func (w *Updater) UpdateIndex(ctx context.Context, old *Index) (*Index, map[ulid.ULID]error, error)

UpdateIndex generates the bucket index and returns it, without storing it to the storage. If the old index is not passed in input, then the bucket index will be generated from scratch.