uploads

Documentation

Index

• type BlugeDBIndexClient
• 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() error
  • func (lt *Table) MultiQueries(ctx context.Context, queries []bluge_db.IndexQuery, ...) error
  • func (lt *Table) RemoveDB(name string) error
  • func (lt *Table) Stop()
  • func (lt *Table) Upload(ctx context.Context, force bool) error
  • func (lt *Table) Write(ctx context.Context, writes bluge_db.TableWrites) error
• type TableManager
  • func NewTableManager(cfg Config, storageClient StorageClient, registerer prometheus.Registerer) (*TableManager, error)
  • func (tm *TableManager) BatchWrite(ctx context.Context, batch chunk.WriteBatch) error
  • func (tm *TableManager) QueryPages(ctx context.Context, queries []bluge_db.IndexQuery, ...) error
  • func (tm *TableManager) Stop()

Types

type BlugeDBIndexClient

type BlugeDBIndexClient interface {
	QueryDB(ctx context.Context, query chunk.IndexQuery, callback func(chunk.IndexQuery, chunk.ReadBatch) (shouldContinue bool)) error
	WriteToDB(ctx context.Context, writes bluge_db.TableWrites) error
}

type Config

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

type StorageClient

type StorageClient interface {
	PutObject(ctx context.Context, objectKey string, object 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) (*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) (*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() 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. 清理已经上传的 并且最后修改时间超过dbRetainPeriod的db

func (*Table) MultiQueries

func (lt *Table) MultiQueries(ctx context.Context, queries []bluge_db.IndexQuery, callback segment.StoredFieldVisitor) 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) 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 bluge_db.TableWrites) error

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

type TableManager

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

实现数据的写入，将客户端发送的数据写入本地（以blugedb格式数据存储），然后将数据压缩返送往objectstore服务器端。数据是根据时间段，定时上传的， 因为是定时上传所以，objectstore数据会有一段的时间差，所以objectstore加uploader数据才是完整的数据，如果要实现完整查询需要donwloder下载完整 objectstore数据实现查询，并利用uploader查询接口查询uploader尚未上传的数据，两者的并集才是完整的查询。

func NewTableManager

func NewTableManager(cfg Config, storageClient StorageClient, registerer prometheus.Registerer) (*TableManager, error)

func (*TableManager) BatchWrite

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

func (*TableManager) QueryPages

func (tm *TableManager) QueryPages(ctx context.Context, queries []bluge_db.IndexQuery, callback segment.StoredFieldVisitor) error

func (*TableManager) Stop

func (tm *TableManager) Stop()