indexer

Documentation

Overview

Package indexer defines Tendermint's block and transaction event indexing logic.

Tendermint supports two primary means of block and transaction event indexing:

1. A key-value sink via an embedded database with a proprietary query language. 2. A Postgres-based sink.

An ABCI application can emit events during block and transaction execution in the form

<abci.Event.Type>.<abci.EventAttributeKey>=<abci.EventAttributeValue>

for example "transfer.amount=10000".

An operator can enable one or both of the supported indexing sinks via the 'tx-index.indexer' Tendermint configuration.

Example:

[tx-index]
indexer = ["kv", "psql"]

If an operator wants to completely disable indexing, they may simply just provide the "null" sink option in the configuration. All other sinks will be ignored if "null" is provided.

If indexing is enabled, the indexer.Service will iterate over all enabled sinks and invoke block and transaction indexing via the appropriate IndexBlockEvents and IndexTxEvents methods.

Note, the "kv" sink is considered deprecated and its query functionality is very limited, but does allow users to directly query for block and transaction events against Tendermint's RPC. Instead, operators are encouraged to use the "psql" indexing sink when more complex queries are required and for reliability purposes as PostgreSQL can scale.

Prior to starting Tendermint with the "psql" indexing sink enabled, operators must ensure the following:

1. The "psql" indexing sink is provided in Tendermint's configuration. 2. A 'tx-index.psql-conn' value is provided that contains the PostgreSQL connection URI. 3. The block and transaction event schemas have been created in the PostgreSQL database.

Tendermint provides the block and transaction event schemas in the following path: state/indexer/sink/psql/schema.sql

To create the schema in a PostgreSQL database, perform the schema query manually or invoke schema creation via the CLI:

$ psql <flags> -f state/indexer/sink/psql/schema.sql

The "psql" indexing sink prohibits queries via RPC. When using a PostgreSQL sink, queries can and should be made directly against the database using SQL.

The following are some example SQL queries against the database schema:

* Query for all transaction events for a given transaction hash:

SELECT * FROM tx_events WHERE hash = '3E7D1F...';

* Query for all transaction events for a given block height:

SELECT * FROM tx_events WHERE height = 25;

* Query for transaction events that have a given type (i.e. value wildcard):

SELECT * FROM tx_events WHERE key LIKE '%transfer.recipient%';

Note that if a complete abci.TxResult is needed, you will need to join "tx_events" with "tx_results" via a foreign key, to obtain contains the raw protobuf-encoded abci.TxResult.

Index

• Constants
• Variables
• func DeduplicateBatch(ops []*abci.TxResult, sink EventSink) ([]*abci.TxResult, error)
• func IndexingEnabled(sinks []EventSink) bool
• func IsRangeOperation(op syntax.Token) bool
• func KVSinkEnabled(sinks []EventSink) bool
• type Batch
  • func NewBatch(n int64) *Batch
  • func (b *Batch) Add(result *abci.TxResult) error
  • func (b *Batch) Size() int
• type BlockIndexer
• type EventSink
• type EventSinkType
• type Metrics
  • func NopMetrics() *Metrics
  • func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics
• type QueryRange
  • func (qr QueryRange) AnyBound() interface{}
  • func (qr QueryRange) LowerBoundValue() interface{}
  • func (qr QueryRange) UpperBoundValue() interface{}
• type QueryRanges
  • func LookForRanges(conditions []syntax.Condition) (ranges QueryRanges, indexes []int)
• type Service
  • func NewService(args ServiceArgs) *Service
  • func (is *Service) OnStart(ctx context.Context) error
  • func (is *Service) OnStop()
• type ServiceArgs
• type TxIndexer

Constants

const MetricsSubsystem = "indexer"

MetricsSubsystem is a the subsystem label for the indexer package.

Variables

var ErrorEmptyHash = errors.New("transaction hash cannot be empty")

ErrorEmptyHash indicates empty hash

Functions

func DeduplicateBatch

func DeduplicateBatch(ops []*abci.TxResult, sink EventSink) ([]*abci.TxResult, error)

DeduplicateBatch consider the case of duplicate txs. if the current one under investigation is NOT OK, then we need to check whether there's a previously indexed tx. SKIP the current tx if the previously indexed record is found and successful.

func IndexingEnabled

func IndexingEnabled(sinks []EventSink) bool

IndexingEnabled returns the given eventSinks is supporting the indexing services.

func IsRangeOperation

func IsRangeOperation(op syntax.Token) bool

IsRangeOperation returns a boolean signifying if a query Operator is a range operation or not.

func KVSinkEnabled

func KVSinkEnabled(sinks []EventSink) bool

KVSinkEnabled returns the given eventSinks is containing KVEventSink.

Types

type Batch

type Batch struct {
	Ops     []*abci.TxResult
	Pending int64
}

Batch groups together multiple Index operations to be performed at the same time. NOTE: Batch is NOT thread-safe and must not be modified after starting its execution.

func NewBatch

func NewBatch(n int64) *Batch

NewBatch creates a new Batch.

func (*Batch) Add

func (b *Batch) Add(result *abci.TxResult) error

Add or update an entry for the given result.Index.

func (*Batch) Size

func (b *Batch) Size() int

Size returns the total number of operations inside the batch.

type BlockIndexer

type BlockIndexer interface {
	// Has returns true if the given height has been indexed. An error is returned
	// upon database query failure.
	Has(height int64) (bool, error)

	// Index indexes FinalizeBlock events for a given block by its height.
	Index(types.EventDataNewBlockHeader) error

	// Search performs a query for block heights that match a given FinalizeBlock
	// event search criteria.
	Search(ctx context.Context, q *query.Query) ([]int64, error)
}

BlockIndexer defines an interface contract for indexing block events.

type EventSink

type EventSink interface {

	// IndexBlockEvents indexes the blockheader.
	IndexBlockEvents(types.EventDataNewBlockHeader) error

	// IndexTxEvents indexes the given result of transactions. To call it with multi transactions,
	// must guarantee the index of given transactions are in order.
	IndexTxEvents([]*abci.TxResult) error

	// SearchBlockEvents provides the block search by given query conditions. This function only
	// supported by the kvEventSink.
	SearchBlockEvents(context.Context, *query.Query) ([]int64, error)

	// SearchTxEvents provides the transaction search by given query conditions. This function only
	// supported by the kvEventSink.
	SearchTxEvents(context.Context, *query.Query) ([]*abci.TxResult, error)

	// GetTxByHash provides the transaction search by given transaction hash. This function only
	// supported by the kvEventSink.
	GetTxByHash([]byte) (*abci.TxResult, error)

	// HasBlock provides the transaction search by given transaction hash. This function only
	// supported by the kvEventSink.
	HasBlock(int64) (bool, error)

	// Type checks the eventsink structure type.
	Type() EventSinkType

	// Stop will close the data store connection, if the eventsink supports it.
	Stop() error
}

EventSink interface is defined the APIs for the IndexerService to interact with the data store, including the block/transaction indexing and the search functions.

The IndexerService will accept a list of one or more EventSink types. During the OnStart method it will call the appropriate APIs on each EventSink to index both block and transaction events.

type EventSinkType

type EventSinkType string

const (
	NULL EventSinkType = "null"
	KV   EventSinkType = "kv"
	PSQL EventSinkType = "psql"
)

type Metrics

type Metrics struct {
	// Latency for indexing block events.
	BlockEventsSeconds metrics.Histogram

	// Latency for indexing transaction events.
	TxEventsSeconds metrics.Histogram

	// Number of complete blocks indexed.
	BlocksIndexed metrics.Counter

	// Number of transactions indexed.
	TransactionsIndexed metrics.Counter
}

Metrics contains metrics exposed by this package.

func NopMetrics

func NopMetrics() *Metrics

func PrometheusMetrics

func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics

type QueryRange

type QueryRange struct {
	LowerBound        interface{} // int || time.Time
	UpperBound        interface{} // int || time.Time
	Key               string
	IncludeLowerBound bool
	IncludeUpperBound bool
}

QueryRange defines a range within a query condition.

func (QueryRange) AnyBound

func (qr QueryRange) AnyBound() interface{}

AnyBound returns either the lower bound if non-nil, otherwise the upper bound.

func (QueryRange) LowerBoundValue

func (qr QueryRange) LowerBoundValue() interface{}

LowerBoundValue returns the value for the lower bound. If the lower bound is nil, nil will be returned.

func (QueryRange) UpperBoundValue

func (qr QueryRange) UpperBoundValue() interface{}

UpperBoundValue returns the value for the upper bound. If the upper bound is nil, nil will be returned.

type QueryRanges

type QueryRanges map[string]QueryRange

QueryRanges defines a mapping between a composite event key and a QueryRange.

e.g.account.number => queryRange{lowerBound: 1, upperBound: 5}

func LookForRanges

func LookForRanges(conditions []syntax.Condition) (ranges QueryRanges, indexes []int)

LookForRanges returns a mapping of QueryRanges and the matching indexes in the provided query conditions.

type Service

type Service struct {
	service.BaseService
	// contains filtered or unexported fields
}

Service connects event bus, transaction and block indexers together in order to index transactions and blocks coming from the event bus.

func NewService

func NewService(args ServiceArgs) *Service

NewService constructs a new indexer service from the given arguments.

func (*Service) OnStart

func (is *Service) OnStart(ctx context.Context) error

OnStart implements part of service.Service. It registers an observer for the indexer if the underlying event sinks support indexing.

TODO(creachadair): Can we get rid of the "enabled" check?

func (*Service) OnStop

func (is *Service) OnStop()

OnStop implements service.Service by closing the event sinks.

type ServiceArgs

type ServiceArgs struct {
	Sinks    []EventSink
	EventBus *eventbus.EventBus
	Metrics  *Metrics
	Logger   log.Logger
}

ServiceArgs are arguments for constructing a new indexer service.

type TxIndexer

type TxIndexer interface {
	// Index analyzes, indexes and stores transactions. For indexing multiple
	// Transacions must guarantee the Index of the TxResult is in order.
	// See Batch struct.
	Index(results []*abci.TxResult) error

	// Get returns the transaction specified by hash or nil if the transaction is not indexed
	// or stored.
	Get(hash []byte) (*abci.TxResult, error)

	// Search allows you to query for transactions.
	Search(ctx context.Context, q *query.Query) ([]*abci.TxResult, error)
}

TxIndexer interface defines methods to index and search transactions.