boltdb

Documentation

Index

• Constants
• func BoltDBToIndexFile(boltdbFile *bbolt.DB, name string) index.Index
• func OpenIndexFile(path string) (index.Index, error)
• func QueryBoltDB(ctx context.Context, db *bbolt.DB, userID []byte, queries []series_index.Query, ...) error
• type BoltDBIndexClient
• type Config
• type IndexCfg
  • func (cfg *IndexCfg) RegisterFlags(f *flag.FlagSet)
  • func (cfg *IndexCfg) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)
  • func (cfg *IndexCfg) Validate() error
• type IndexClient
  • func NewIndexClient(prefix string, cfg IndexCfg, storageClient client.ObjectClient, ...) (*IndexClient, error)
  • func (i *IndexClient) BatchWrite(ctx context.Context, batch series_index.WriteBatch) error
  • func (i *IndexClient) NewWriteBatch() series_index.WriteBatch
  • func (i *IndexClient) QueryPages(ctx context.Context, queries []series_index.Query, ...) error
  • func (i *IndexClient) Stop()
• type IndexFile
  • func (i *IndexFile) Close() error
  • func (i *IndexFile) GetBoltDB() *bbolt.DB
  • func (i *IndexFile) Name() string
  • func (i *IndexFile) Path() string
  • func (i *IndexFile) Reader() (io.ReadSeeker, error)
• type Querier
  • func NewQuerier(writer Writer, indexShipper Shipper) Querier
• type Shipper
• type Table
  • func LoadTable(path, uploader string, indexShipper Shipper, makePerTenantBuckets bool, ...) (*Table, error)
  • func NewTable(path, uploader string, indexShipper Shipper, makePerTenantBuckets bool) (*Table, error)
  • func (lt *Table) ForEach(_ context.Context, callback func(b *bbolt.DB) error) error
  • func (lt *Table) HandoverIndexesToShipper(force bool) error
  • func (lt *Table) Snapshot() error
  • func (lt *Table) Stop()
  • func (lt *Table) Write(ctx context.Context, writes local.TableWrites) error
• type TableManager
  • func NewTableManager(cfg Config, indexShipper Shipper, tableRange config.TableRange, ...) (*TableManager, error)
  • func (tm *TableManager) BatchWrite(ctx context.Context, batch index.WriteBatch) error
  • func (tm *TableManager) ForEach(ctx context.Context, tableName string, callback func(boltdb *bbolt.DB) error) error
  • func (tm *TableManager) Stop()
• type Writer

Constants

const (
	// create a new db sharded by time based on when write request is received
	ShardDBsByDuration = 15 * time.Minute
)

const TempFileSuffix = ".temp"

Functions

func BoltDBToIndexFile

func BoltDBToIndexFile(boltdbFile *bbolt.DB, name string) index.Index

nolint:revive

func OpenIndexFile

func OpenIndexFile(path string) (index.Index, error)

func QueryBoltDB

func QueryBoltDB(ctx context.Context, db *bbolt.DB, userID []byte, queries []series_index.Query, callback series_index.QueryPagesCallback) error

Types

type BoltDBIndexClient

type BoltDBIndexClient interface {
	QueryWithCursor(_ context.Context, c *bbolt.Cursor, query index.Query, callback index.QueryPagesCallback) error
	WriteToDB(ctx context.Context, db *bbolt.DB, bucketName []byte, writes local.TableWrites) error
}

nolint:revive

type Config

type Config struct {
	Uploader             string
	IndexDir             string
	DBRetainPeriod       time.Duration
	MakePerTenantBuckets bool
}

type IndexCfg

type IndexCfg struct {
	indexshipper.Config `yaml:",inline"`
	BuildPerTenantIndex bool `yaml:"build_per_tenant_index"`
}

func (*IndexCfg) RegisterFlags

func (cfg *IndexCfg) RegisterFlags(f *flag.FlagSet)

RegisterFlags registers flags.

func (*IndexCfg) RegisterFlagsWithPrefix

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

func (*IndexCfg) Validate

func (cfg *IndexCfg) Validate() error

type IndexClient

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

func NewIndexClient

func NewIndexClient(prefix string, cfg IndexCfg, storageClient client.ObjectClient, limits downloads.Limits,
	tenantFilter downloads.TenantFilter, tableRange config.TableRange, registerer prometheus.Registerer, logger log.Logger) (*IndexClient, error)

New creates a shipper for syncing local objects with a store

func (*IndexClient) BatchWrite

func (i *IndexClient) BatchWrite(ctx context.Context, batch series_index.WriteBatch) error

func (*IndexClient) NewWriteBatch

func (i *IndexClient) NewWriteBatch() series_index.WriteBatch

func (*IndexClient) QueryPages

func (i *IndexClient) QueryPages(ctx context.Context, queries []series_index.Query, callback series_index.QueryPagesCallback) error

func (*IndexClient) Stop

func (i *IndexClient) Stop()

type IndexFile

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

func (*IndexFile) Close

func (i *IndexFile) Close() error

func (*IndexFile) GetBoltDB

func (i *IndexFile) GetBoltDB() *bbolt.DB

func (*IndexFile) Name

func (i *IndexFile) Name() string

func (*IndexFile) Path

func (i *IndexFile) Path() string

func (*IndexFile) Reader

func (i *IndexFile) Reader() (io.ReadSeeker, error)

type Querier

type Querier interface {
	QueryPages(ctx context.Context, queries []index.Query, callback index.QueryPagesCallback) error
}

func NewQuerier

func NewQuerier(writer Writer, indexShipper Shipper) Querier

type Shipper

type Shipper interface {
	AddIndex(tableName, userID string, index shipperindex.Index) error
	ForEach(ctx context.Context, tableName, userID string, callback shipperindex.ForEachIndexCallback) error
}

type Table

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

Table is a collection of multiple index files created for a same table by the ingester. It is used on the write path for writing the index. All the public methods are concurrency safe and take care of mutexes to avoid any data race.

func LoadTable

func LoadTable(path, uploader string, indexShipper Shipper, makePerTenantBuckets bool, metrics *tableManagerMetrics) (*Table, error)

LoadTable loads local dbs belonging to the table and creates a new Table with references to dbs if there are any otherwise it doesn't create a table

func NewTable

func NewTable(path, uploader string, indexShipper Shipper, makePerTenantBuckets bool) (*Table, error)

NewTable create a new Table without looking for any existing local dbs belonging to the table.

func (*Table) ForEach

func (lt *Table) ForEach(_ context.Context, callback func(b *bbolt.DB) error) error

func (*Table) HandoverIndexesToShipper

func (lt *Table) HandoverIndexesToShipper(force bool) error

HandoverIndexesToShipper hands over the inactive dbs to shipper for uploading

func (*Table) Snapshot

func (lt *Table) Snapshot() error

func (*Table) Stop

func (lt *Table) Stop()

Stop closes all the open dbs.

func (*Table) Write

func (lt *Table) Write(ctx context.Context, writes local.TableWrites) error

Write writes to a db locally with write time set to now.

type TableManager

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

func NewTableManager

func NewTableManager(cfg Config, indexShipper Shipper, tableRange config.TableRange, registerer prometheus.Registerer, logger log.Logger) (*TableManager, error)

func (*TableManager) BatchWrite

func (tm *TableManager) BatchWrite(ctx context.Context, batch index.WriteBatch) error

func (*TableManager) ForEach

func (tm *TableManager) ForEach(ctx context.Context, tableName string, callback func(boltdb *bbolt.DB) error) error

func (*TableManager) Stop

func (tm *TableManager) Stop()

type Writer

type Writer interface {
	ForEach(ctx context.Context, tableName string, callback func(b *bbolt.DB) error) error
}