archive

Documentation

Index

• Variables
• func NewQueryThreadsOptimizer(maxThreads int, sortDesc bool, lowBlockNum uint64, ...) *queryThreadsOptimizer
• func TestNewClient(t *testing.T, searchEngine *ArchiveBackend) (pb.BackendClient, func())
• type ArchiveBackend
  • func NewBackend(pool *IndexPool, dmeshClient dmeshClient.Client, searchPeer *dmesh.SearchPeer, ...) *ArchiveBackend
  • func (b *ArchiveBackend) Check(ctx context.Context, in *pbhealth.HealthCheckRequest) (*pbhealth.HealthCheckResponse, error)
  • func (b *ArchiveBackend) GetHeadInfo(ctx context.Context, r *pbhead.HeadInfoRequest) (*pbhead.HeadInfoResponse, error)
  • func (b *ArchiveBackend) Launch()
  • func (b *ArchiveBackend) SetMaxQueryThreads(threads int)
  • func (b *ArchiveBackend) StreamHeadInfo(r *pbhead.HeadInfoRequest, ...) error
  • func (b *ArchiveBackend) StreamMatches(req *pbsearch.BackendRequest, stream pbsearch.Backend_StreamMatchesServer) error
  • func (b *ArchiveBackend) WarmUpArchive(ctx context.Context, low, high uint64, bquery *search.BleveQuery) error
  • func (b *ArchiveBackend) WarmupWithQuery(query string, low, high uint64) error
• type IndexPool
  • func NewIndexPool(indexesPath string, readOnlyIndexesPaths []string, shardSize uint64, ...) (*IndexPool, error)
  • func (p *IndexPool) AppendReadIndexes(idx ...*search.ShardIndex)
  • func (p *IndexPool) CleanOnDiskIndexes(startBlock, stopBlock uint64) error
  • func (p *IndexPool) CloseIndexes() (err error)
  • func (p *IndexPool) GetIndexIterator(lowBlockNum, highBlockNum uint64, sortDesc bool) (*indexIterator, error)
  • func (p *IndexPool) GetLowestServeableBlockNum() uint64
  • func (p *IndexPool) IsReady() bool
  • func (p *IndexPool) LastReadOnlyIndexedBlock() uint64
  • func (p *IndexPool) LastReadOnlyIndexedBlockID() string
  • func (p *IndexPool) LowestServeableBlockNumAbove(blockNum uint64) uint64
  • func (p *IndexPool) PollRemoteIndices(startBlockNum, stopBlockNum uint64)
  • func (p *IndexPool) ScanOnDiskIndexes(startBlock uint64) error
  • func (p *IndexPool) SetLowestServeableBlockNum(startBlockNum uint64) error
  • func (p *IndexPool) SetReady() error
  • func (p *IndexPool) SyncFromStorage(startBlock, stopBlock uint64, maxIndexes int, parallelDownloads int) error
  • func (p *IndexPool) Upload(baseIndex uint64, indexPath string) (err error)
• type Truncator
  • func NewTruncator(indexPool *IndexPool, blockCount uint64) *Truncator
  • func (t *Truncator) Launch()

Variables

var LivenessQuery *search.BleveQuery

Functions

func NewQueryThreadsOptimizer

func NewQueryThreadsOptimizer(maxThreads int, sortDesc bool, lowBlockNum uint64, llerrgroup *llerrgroup.Group) *queryThreadsOptimizer

func TestNewClient

func TestNewClient(t *testing.T, searchEngine *ArchiveBackend) (pb.BackendClient, func())

Types

type ArchiveBackend

type ArchiveBackend struct {
	*shutter.Shutter

	Pool       *IndexPool
	SearchPeer *dmesh.SearchPeer

	MaxQueryThreads int
	// contains filtered or unexported fields
}

Search is the top-level object, embodying the rest.

func NewBackend

func NewBackend(
	pool *IndexPool,
	dmeshClient dmeshClient.Client,
	searchPeer *dmesh.SearchPeer,
	grpcListenAddr string,
	httpListenAddr string,
	shutdownDelay time.Duration,
) *ArchiveBackend

func (*ArchiveBackend) Check

func (b *ArchiveBackend) Check(ctx context.Context, in *pbhealth.HealthCheckRequest) (*pbhealth.HealthCheckResponse, error)

GRPC health check endpoint

func (*ArchiveBackend) GetHeadInfo

func (b *ArchiveBackend) GetHeadInfo(ctx context.Context, r *pbhead.HeadInfoRequest) (*pbhead.HeadInfoResponse, error)

func (*ArchiveBackend) Launch

func (b *ArchiveBackend) Launch()

func (*ArchiveBackend) SetMaxQueryThreads

func (b *ArchiveBackend) SetMaxQueryThreads(threads int)

func (*ArchiveBackend) StreamHeadInfo

func (b *ArchiveBackend) StreamHeadInfo(r *pbhead.HeadInfoRequest, stream pbhead.StreamingHeadInfo_StreamHeadInfoServer) error

headinfo.StreamingHeadInfo gRPC implementation

func (*ArchiveBackend) StreamMatches

func (b *ArchiveBackend) StreamMatches(req *pbsearch.BackendRequest, stream pbsearch.Backend_StreamMatchesServer) error

Archive.StreamMatches gRPC implementation

func (*ArchiveBackend) WarmUpArchive

func (b *ArchiveBackend) WarmUpArchive(ctx context.Context, low, high uint64, bquery *search.BleveQuery) error

func (*ArchiveBackend) WarmupWithQuery

func (b *ArchiveBackend) WarmupWithQuery(query string, low, high uint64) error

type IndexPool

type IndexPool struct {
	ReadOnlyIndexesPaths []string // list of paths where to load on start
	IndexesPath          string   //local path where indices are stored on disk
	ShardSize            uint64

	SearchPeer *dmesh.SearchPeer

	LowestServeableBlockNum uint64

	ReadPool        []*search.ShardIndex
	PerQueryThreads int // Each end-user query will parallelize sub-queries on 15K+ indices
	// contains filtered or unexported fields
}

func NewIndexPool

func NewIndexPool(indexesPath string, readOnlyIndexesPaths []string, shardSize uint64, indexesStore dstore.Store, cache roarcache.Cache, dmeshClient dmeshClient.Client, searchPeer *dmesh.SearchPeer) (*IndexPool, error)

func (*IndexPool) AppendReadIndexes

func (p *IndexPool) AppendReadIndexes(idx ...*search.ShardIndex)

func (*IndexPool) CleanOnDiskIndexes

func (p *IndexPool) CleanOnDiskIndexes(startBlock, stopBlock uint64) error

func (*IndexPool) CloseIndexes

func (p *IndexPool) CloseIndexes() (err error)

func (*IndexPool) GetIndexIterator

func (p *IndexPool) GetIndexIterator(lowBlockNum, highBlockNum uint64, sortDesc bool) (*indexIterator, error)

func (*IndexPool) GetLowestServeableBlockNum

func (p *IndexPool) GetLowestServeableBlockNum() uint64

func (*IndexPool) IsReady

func (p *IndexPool) IsReady() bool

func (*IndexPool) LastReadOnlyIndexedBlock

func (p *IndexPool) LastReadOnlyIndexedBlock() uint64

LastReadOnlyIndexedBlock returns the block inclusively (999)

func (*IndexPool) LastReadOnlyIndexedBlockID

func (p *IndexPool) LastReadOnlyIndexedBlockID() string

func (*IndexPool) LowestServeableBlockNumAbove

func (p *IndexPool) LowestServeableBlockNumAbove(blockNum uint64) uint64

func (*IndexPool) PollRemoteIndices

func (p *IndexPool) PollRemoteIndices(startBlockNum, stopBlockNum uint64)

func (*IndexPool) ScanOnDiskIndexes

func (p *IndexPool) ScanOnDiskIndexes(startBlock uint64) error

func (*IndexPool) SetLowestServeableBlockNum

func (p *IndexPool) SetLowestServeableBlockNum(startBlockNum uint64) error

func (*IndexPool) SetReady

func (p *IndexPool) SetReady() error

SetReady marks the process as ready, meaning it has crossed the "close to real-time" threshold.

func (*IndexPool) SyncFromStorage

func (p *IndexPool) SyncFromStorage(startBlock, stopBlock uint64, maxIndexes int, parallelDownloads int) error

func (*IndexPool) Upload

func (p *IndexPool) Upload(baseIndex uint64, indexPath string) (err error)

type Truncator

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

func NewTruncator

func NewTruncator(indexPool *IndexPool, blockCount uint64) *Truncator

func (*Truncator) Launch

func (t *Truncator) Launch()