uploads

Documentation

Index

• Constants
• type BoltDBIndexClient
• type Config
• type StorageClient
• type Table
  • func LoadTable(path, uploader string, storageClient StorageClient, ...) (*Table, error)
  • func NewTable(path, uploader string, storageClient StorageClient, ...) (*Table, error)
  • func (lt *Table) Cleanup(dbRetainPeriod time.Duration) error
  • func (lt *Table) MultiQueries(ctx context.Context, queries []index.Query, callback index.QueryPagesCallback) error
  • func (lt *Table) RemoveDB(name string) error
  • func (lt *Table) RemoveSnapshotDB(name string) error
  • func (lt *Table) Snapshot() error
  • func (lt *Table) Stop()
  • func (lt *Table) Upload(ctx context.Context, force bool) error
  • func (lt *Table) Write(ctx context.Context, writes local.TableWrites) error
• type TableManager
  • func NewTableManager(cfg Config, boltIndexClient BoltDBIndexClient, storageClient StorageClient, ...) (*TableManager, error)
  • func (tm *TableManager) BatchWrite(ctx context.Context, batch index.WriteBatch) error
  • func (tm *TableManager) QueryPages(ctx context.Context, queries []index.Query, callback index.QueryPagesCallback) error
  • func (tm *TableManager) Stop()

Constants

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

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
}

type Config

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

type StorageClient

type StorageClient interface {
	PutFile(ctx context.Context, tableName, fileName string, file io.ReadSeeker) error
}

type Table

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

Table is a collection of multiple dbs created for a same table by the ingester. All the public methods are concurrency safe and take care of mutexes to avoid any data race.

func LoadTable

func LoadTable(path, uploader string, storageClient StorageClient, boltdbIndexClient BoltDBIndexClient,
	makePerTenantBuckets bool, metrics *metrics,
) (*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, storageClient StorageClient, boltdbIndexClient BoltDBIndexClient, makePerTenantBuckets bool) (*Table, error)

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

func (*Table) Cleanup

func (lt *Table) Cleanup(dbRetainPeriod time.Duration) error

Cleanup removes dbs which are already uploaded and have not been modified for period longer than dbRetainPeriod. This is to avoid keeping all the files forever in the ingesters.

func (*Table) MultiQueries

func (lt *Table) MultiQueries(ctx context.Context, queries []index.Query, callback index.QueryPagesCallback) error

MultiQueries runs multiple queries without having to take lock multiple times for each query.

func (*Table) RemoveDB

func (lt *Table) RemoveDB(name string) error

RemoveDB closes the db and removes the file locally.

func (*Table) RemoveSnapshotDB

func (lt *Table) RemoveSnapshotDB(name string) error

func (*Table) Snapshot

func (lt *Table) Snapshot() error

func (*Table) Stop

func (lt *Table) Stop()

Stop closes all the open dbs.

func (*Table) Upload

func (lt *Table) Upload(ctx context.Context, force bool) error

Upload uploads all the dbs which are never uploaded or have been modified since the last batch was uploaded.

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, boltIndexClient BoltDBIndexClient, storageClient StorageClient, registerer prometheus.Registerer) (*TableManager, error)

func (*TableManager) BatchWrite

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

func (*TableManager) QueryPages

func (tm *TableManager) QueryPages(ctx context.Context, queries []index.Query, callback index.QueryPagesCallback) error

func (*TableManager) Stop

func (tm *TableManager) Stop()