index

Documentation

Index

• Variables
• type Config
  • func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)
• type Index
  • func NewIndex(store Store, logger log.Logger, cfg *Config) *Index
  • func (i *Index) CreatePartitionKey(blockId string) PartitionKey
  • func (i *Index) FindBlock(shardNum uint32, tenant string, blockId string) *metastorev1.BlockMeta
  • func (i *Index) FindBlocksInRange(start, end int64, tenants map[string]struct{}) ([]*metastorev1.BlockMeta, error)
  • func (i *Index) FindPartitionMetas(blockId string) []*PartitionMeta
  • func (i *Index) ForEachPartition(ctx context.Context, fn func(meta *PartitionMeta) error) error
  • func (i *Index) InsertBlock(b *metastorev1.BlockMeta)
  • func (i *Index) LoadPartitions()
  • func (i *Index) ReplaceBlocks(sources []string, sourceShard uint32, sourceTenant string, ...)
• type PartitionKey
  • func (k PartitionKey) Parse() (t time.Time, d time.Duration, err error)
• type PartitionMeta
  • func (m *PartitionMeta) AddTenant(tenant string)
  • func (m *PartitionMeta) EndTime() time.Time
  • func (m *PartitionMeta) HasTenant(tenant string) bool
  • func (m *PartitionMeta) StartTime() time.Time
• type Store

Variables

var DefaultConfig = Config{
	PartitionDuration:     24 * time.Hour,
	PartitionCacheSize:    7,
	QueryLookaroundPeriod: time.Hour,
}

Types

type Config

type Config struct {
	PartitionDuration     time.Duration `yaml:"partition_duration"`
	PartitionCacheSize    int           `yaml:"partition_cache_size"`
	QueryLookaroundPeriod time.Duration `yaml:"query_lookaround_period"`
}

func (*Config) RegisterFlagsWithPrefix

func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)

type Index

type Index struct {
	Config *Config
	// contains filtered or unexported fields
}

func NewIndex

func NewIndex(store Store, logger log.Logger, cfg *Config) *Index

NewIndex initializes a new metastore index.

The index provides access to block metadata. The data is partitioned by time, shard and tenant. Partition identifiers contain the time period referenced by partitions, e.g., "20240923T16.1h" refers to a partition for the 1-hour period between 2024-09-23T16:00:00.000Z and 2024-09-23T16:59:59.999Z.

Partitions are mostly transparent for the end user, though PartitionMeta is at times used externally. Partition durations are configurable (at application level).

The index requires a backing Store for loading data in memory. Data is loaded directly via LoadPartitions() or when looking up blocks with FindBlock() or FindBlocksInRange().

func (*Index) CreatePartitionKey

func (i *Index) CreatePartitionKey(blockId string) PartitionKey

CreatePartitionKey creates a partition key for a block. It is meant to be used for newly inserted blocks, as it relies on the index's currently configured partition duration to create the key.

Note: Using this for existing blocks following a partition duration change can produce the wrong key. Callers should verify that the returned partition actually contains the block.

func (*Index) FindBlock

func (i *Index) FindBlock(shardNum uint32, tenant string, blockId string) *metastorev1.BlockMeta

FindBlock tries to retrieve an existing block from the index. It will load the corresponding partition if it is not already loaded. Returns nil if the block cannot be found.

func (*Index) FindBlocksInRange

func (i *Index) FindBlocksInRange(start, end int64, tenants map[string]struct{}) ([]*metastorev1.BlockMeta, error)

FindBlocksInRange retrieves all blocks that might contain data for the given time range and tenants.

It is not enough to scan for partition keys that fall in the given time interval. Partitions are built on top of block identifiers which refer to the moment a block was created and not to the timestamps of the profiles contained within the block (min_time, max_time). This method works around this by including blocks from adjacent partitions.

func (*Index) FindPartitionMetas

func (i *Index) FindPartitionMetas(blockId string) []*PartitionMeta

func (*Index) ForEachPartition

func (i *Index) ForEachPartition(ctx context.Context, fn func(meta *PartitionMeta) error) error

ForEachPartition executes the given function concurrently for each partition. It will be called for all partitions, regardless if they are fully loaded in memory or not.

func (*Index) InsertBlock

func (i *Index) InsertBlock(b *metastorev1.BlockMeta)

InsertBlock is the primary way for adding blocks to the index.

func (*Index) LoadPartitions

func (i *Index) LoadPartitions()

LoadPartitions reads all partitions from the backing store and loads the recent ones in memory.

func (*Index) ReplaceBlocks

func (i *Index) ReplaceBlocks(sources []string, sourceShard uint32, sourceTenant string, replacements []*metastorev1.BlockMeta)

ReplaceBlocks removes source blocks from the index and inserts replacement blocks into the index. The intended usage is for block compaction. The replacement blocks could be added to the same or a different partition.

type PartitionKey

type PartitionKey string

func (PartitionKey) Parse

func (k PartitionKey) Parse() (t time.Time, d time.Duration, err error)

type PartitionMeta

type PartitionMeta struct {
	Key      PartitionKey
	Ts       time.Time
	Duration time.Duration
	Tenants  []string
	// contains filtered or unexported fields
}

func (*PartitionMeta) AddTenant

func (m *PartitionMeta) AddTenant(tenant string)

func (*PartitionMeta) EndTime

func (m *PartitionMeta) EndTime() time.Time

func (*PartitionMeta) HasTenant

func (m *PartitionMeta) HasTenant(tenant string) bool

func (*PartitionMeta) StartTime

func (m *PartitionMeta) StartTime() time.Time

type Store

type Store interface {
	ListPartitions() []PartitionKey
	ListShards(p PartitionKey) []uint32
	ListTenants(p PartitionKey, shard uint32) []string
	ListBlocks(p PartitionKey, shard uint32, tenant string) []*metastorev1.BlockMeta
}