JayT106
3 years ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
28 changed files with 1760 additions and 277 deletions
Split View
Diff Options
-
+1 -0CHANGELOG_PENDING.md
-
+10 -4config/config.go
-
+8 -2config/toml.go
-
+75 -47docs/architecture/adr-065-custom-event-indexing.md
-
+4 -0go.mod
-
+39 -0go.sum
-
+9 -11node/node.go
-
+95 -0node/node_test.go
-
+60 -32node/setup.go
-
+12 -7rpc/core/blocks.go
-
+1 -2rpc/core/env.go
-
+84 -76rpc/core/tx.go
-
+54 -0state/indexer/eventsink.go
-
+4 -5state/indexer/indexer.go
-
+52 -18state/indexer/indexer_service.go
-
+129 -13state/indexer/indexer_service_test.go
-
+61 -0state/indexer/sink/kv/kv.go
-
+351 -0state/indexer/sink/kv/kv_test.go
-
+51 -0state/indexer/sink/null/null.go
-
+39 -0state/indexer/sink/null/null_test.go
-
+197 -0state/indexer/sink/psql/psql.go
-
+365 -0state/indexer/sink/psql/psql_test.go
-
+31 -0state/indexer/sink/psql/schema.sql
-
+11 -46state/indexer/tx/kv/kv.go
-
+1 -1state/indexer/tx/kv/kv_bench_test.go
-
+12 -12state/indexer/tx/kv/kv_test.go
-
+1 -1state/indexer/tx/null/null.go
-
+3 -0types/events.go
+ 1
- 0
CHANGELOG_PENDING.md
View File
+ 10
- 4
config/config.go
View File
+ 8
- 2
config/toml.go
View File
+ 75
- 47
docs/architecture/adr-065-custom-event-indexing.md
View File
+ 4
- 0
go.mod
View File
+ 39
- 0
go.sum
View File
+ 9
- 11
node/node.go
View File
+ 95
- 0
node/node_test.go
View File
+ 60
- 32
node/setup.go
View File
+ 12
- 7
rpc/core/blocks.go
View File
+ 1
- 2
rpc/core/env.go
View File
+ 84
- 76
rpc/core/tx.go
View File
+ 54
- 0
state/indexer/eventsink.go
View File
| @ -0,0 +1,54 @@ | |||
| package indexer | |||
| import ( | |||
| "context" | |||
| abci "github.com/tendermint/tendermint/abci/types" | |||
| "github.com/tendermint/tendermint/libs/pubsub/query" | |||
| "github.com/tendermint/tendermint/types" | |||
| ) | |||
| type EventSinkType string | |||
| const ( | |||
| NULL EventSinkType = "null" | |||
| KV EventSinkType = "kv" | |||
| PSQL EventSinkType = "psql" | |||
| ) | |||
| // 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 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 | |||
| } | |||
+ 4
- 5
state/indexer/indexer.go
View File
+ 52
- 18
state/indexer/indexer_service.go
View File
+ 129
- 13
state/indexer/indexer_service_test.go
View File
+ 61
- 0
state/indexer/sink/kv/kv.go
View File
| @ -0,0 +1,61 @@ | |||
| package kv | |||
| import ( | |||
| "context" | |||
| abci "github.com/tendermint/tendermint/abci/types" | |||
| "github.com/tendermint/tendermint/libs/pubsub/query" | |||
| "github.com/tendermint/tendermint/state/indexer" | |||
| kvb "github.com/tendermint/tendermint/state/indexer/block/kv" | |||
| kvt "github.com/tendermint/tendermint/state/indexer/tx/kv" | |||
| "github.com/tendermint/tendermint/types" | |||
| dbm "github.com/tendermint/tm-db" | |||
| ) | |||
| var _ indexer.EventSink = (*EventSink)(nil) | |||
| // The EventSink is an aggregator for redirecting the call path of the tx/block kvIndexer. | |||
| // For the implementation details please see the kv.go in the indexer/block and indexer/tx folder. | |||
| type EventSink struct { | |||
| txi *kvt.TxIndex | |||
| bi *kvb.BlockerIndexer | |||
| } | |||
| func NewEventSink(store dbm.DB) indexer.EventSink { | |||
| return &EventSink{ | |||
| txi: kvt.NewTxIndex(store), | |||
| bi: kvb.New(store), | |||
| } | |||
| } | |||
| func (kves *EventSink) Type() indexer.EventSinkType { | |||
| return indexer.KV | |||
| } | |||
| func (kves *EventSink) IndexBlockEvents(bh types.EventDataNewBlockHeader) error { | |||
| return kves.bi.Index(bh) | |||
| } | |||
| func (kves *EventSink) IndexTxEvents(results []*abci.TxResult) error { | |||
| return kves.txi.Index(results) | |||
| } | |||
| func (kves *EventSink) SearchBlockEvents(ctx context.Context, q *query.Query) ([]int64, error) { | |||
| return kves.bi.Search(ctx, q) | |||
| } | |||
| func (kves *EventSink) SearchTxEvents(ctx context.Context, q *query.Query) ([]*abci.TxResult, error) { | |||
| return kves.txi.Search(ctx, q) | |||
| } | |||
| func (kves *EventSink) GetTxByHash(hash []byte) (*abci.TxResult, error) { | |||
| return kves.txi.Get(hash) | |||
| } | |||
| func (kves *EventSink) HasBlock(h int64) (bool, error) { | |||
| return kves.bi.Has(h) | |||
| } | |||
| func (kves *EventSink) Stop() error { | |||
| return nil | |||
| } | |||
+ 351
- 0
state/indexer/sink/kv/kv_test.go
View File
| @ -0,0 +1,351 @@ | |||
| package kv | |||
| import ( | |||
| "context" | |||
| "fmt" | |||
| "testing" | |||
| "github.com/gogo/protobuf/proto" | |||
| "github.com/stretchr/testify/assert" | |||
| "github.com/stretchr/testify/require" | |||
| abci "github.com/tendermint/tendermint/abci/types" | |||
| "github.com/tendermint/tendermint/libs/pubsub/query" | |||
| "github.com/tendermint/tendermint/state/indexer" | |||
| kvtx "github.com/tendermint/tendermint/state/indexer/tx/kv" | |||
| "github.com/tendermint/tendermint/types" | |||
| db "github.com/tendermint/tm-db" | |||
| ) | |||
| func TestType(t *testing.T) { | |||
| kvSink := NewEventSink(db.NewMemDB()) | |||
| assert.Equal(t, indexer.KV, kvSink.Type()) | |||
| } | |||
| func TestStop(t *testing.T) { | |||
| kvSink := NewEventSink(db.NewMemDB()) | |||
| assert.Nil(t, kvSink.Stop()) | |||
| } | |||
| func TestBlockFuncs(t *testing.T) { | |||
| store := db.NewPrefixDB(db.NewMemDB(), []byte("block_events")) | |||
| indexer := NewEventSink(store) | |||
| require.NoError(t, indexer.IndexBlockEvents(types.EventDataNewBlockHeader{ | |||
| Header: types.Header{Height: 1}, | |||
| ResultBeginBlock: abci.ResponseBeginBlock{ | |||
| Events: []abci.Event{ | |||
| { | |||
| Type: "begin_event", | |||
| Attributes: []abci.EventAttribute{ | |||
| { | |||
| Key: "proposer", | |||
| Value: "FCAA001", | |||
| Index: true, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| ResultEndBlock: abci.ResponseEndBlock{ | |||
| Events: []abci.Event{ | |||
| { | |||
| Type: "end_event", | |||
| Attributes: []abci.EventAttribute{ | |||
| { | |||
| Key: "foo", | |||
| Value: "100", | |||
| Index: true, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| })) | |||
| b, e := indexer.HasBlock(1) | |||
| assert.Nil(t, e) | |||
| assert.True(t, b) | |||
| for i := 2; i < 12; i++ { | |||
| var index bool | |||
| if i%2 == 0 { | |||
| index = true | |||
| } | |||
| require.NoError(t, indexer.IndexBlockEvents(types.EventDataNewBlockHeader{ | |||
| Header: types.Header{Height: int64(i)}, | |||
| ResultBeginBlock: abci.ResponseBeginBlock{ | |||
| Events: []abci.Event{ | |||
| { | |||
| Type: "begin_event", | |||
| Attributes: []abci.EventAttribute{ | |||
| { | |||
| Key: "proposer", | |||
| Value: "FCAA001", | |||
| Index: true, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| ResultEndBlock: abci.ResponseEndBlock{ | |||
| Events: []abci.Event{ | |||
| { | |||
| Type: "end_event", | |||
| Attributes: []abci.EventAttribute{ | |||
| { | |||
| Key: "foo", | |||
| Value: fmt.Sprintf("%d", i), | |||
| Index: index, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| })) | |||
| } | |||
| testCases := map[string]struct { | |||
| q *query.Query | |||
| results []int64 | |||
| }{ | |||
| "block.height = 100": { | |||
| q: query.MustParse("block.height = 100"), | |||
| results: []int64{}, | |||
| }, | |||
| "block.height = 5": { | |||
| q: query.MustParse("block.height = 5"), | |||
| results: []int64{5}, | |||
| }, | |||
| "begin_event.key1 = 'value1'": { | |||
| q: query.MustParse("begin_event.key1 = 'value1'"), | |||
| results: []int64{}, | |||
| }, | |||
| "begin_event.proposer = 'FCAA001'": { | |||
| q: query.MustParse("begin_event.proposer = 'FCAA001'"), | |||
| results: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11}, | |||
| }, | |||
| "end_event.foo <= 5": { | |||
| q: query.MustParse("end_event.foo <= 5"), | |||
| results: []int64{2, 4}, | |||
| }, | |||
| "end_event.foo >= 100": { | |||
| q: query.MustParse("end_event.foo >= 100"), | |||
| results: []int64{1}, | |||
| }, | |||
| "block.height > 2 AND end_event.foo <= 8": { | |||
| q: query.MustParse("block.height > 2 AND end_event.foo <= 8"), | |||
| results: []int64{4, 6, 8}, | |||
| }, | |||
| "begin_event.proposer CONTAINS 'FFFFFFF'": { | |||
| q: query.MustParse("begin_event.proposer CONTAINS 'FFFFFFF'"), | |||
| results: []int64{}, | |||
| }, | |||
| "begin_event.proposer CONTAINS 'FCAA001'": { | |||
| q: query.MustParse("begin_event.proposer CONTAINS 'FCAA001'"), | |||
| results: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11}, | |||
| }, | |||
| } | |||
| for name, tc := range testCases { | |||
| tc := tc | |||
| t.Run(name, func(t *testing.T) { | |||
| results, err := indexer.SearchBlockEvents(context.Background(), tc.q) | |||
| require.NoError(t, err) | |||
| require.Equal(t, tc.results, results) | |||
| }) | |||
| } | |||
| } | |||
| func TestTxSearchWithCancelation(t *testing.T) { | |||
| indexer := NewEventSink(db.NewMemDB()) | |||
| txResult := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "1", Index: true}}}, | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "owner", Value: "Ivan", Index: true}}}, | |||
| {Type: "", Attributes: []abci.EventAttribute{{Key: "not_allowed", Value: "Vlad", Index: true}}}, | |||
| }) | |||
| err := indexer.IndexTxEvents([]*abci.TxResult{txResult}) | |||
| require.NoError(t, err) | |||
| r, e := indexer.GetTxByHash(types.Tx("HELLO WORLD").Hash()) | |||
| assert.Nil(t, e) | |||
| assert.Equal(t, r, txResult) | |||
| ctx, cancel := context.WithCancel(context.Background()) | |||
| cancel() | |||
| results, err := indexer.SearchTxEvents(ctx, query.MustParse("account.number = 1")) | |||
| assert.NoError(t, err) | |||
| assert.Empty(t, results) | |||
| } | |||
| func TestTxSearchDeprecatedIndexing(t *testing.T) { | |||
| esdb := db.NewMemDB() | |||
| indexer := NewEventSink(esdb) | |||
| // index tx using events indexing (composite key) | |||
| txResult1 := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "1", Index: true}}}, | |||
| }) | |||
| hash1 := types.Tx(txResult1.Tx).Hash() | |||
| err := indexer.IndexTxEvents([]*abci.TxResult{txResult1}) | |||
| require.NoError(t, err) | |||
| // index tx also using deprecated indexing (event as key) | |||
| txResult2 := txResultWithEvents(nil) | |||
| txResult2.Tx = types.Tx("HELLO WORLD 2") | |||
| hash2 := types.Tx(txResult2.Tx).Hash() | |||
| b := esdb.NewBatch() | |||
| rawBytes, err := proto.Marshal(txResult2) | |||
| require.NoError(t, err) | |||
| depKey := []byte(fmt.Sprintf("%s/%s/%d/%d", | |||
| "sender", | |||
| "addr1", | |||
| txResult2.Height, | |||
| txResult2.Index, | |||
| )) | |||
| err = b.Set(depKey, hash2) | |||
| require.NoError(t, err) | |||
| err = b.Set(kvtx.KeyFromHeight(txResult2), hash2) | |||
| require.NoError(t, err) | |||
| err = b.Set(hash2, rawBytes) | |||
| require.NoError(t, err) | |||
| err = b.Write() | |||
| require.NoError(t, err) | |||
| testCases := []struct { | |||
| q string | |||
| results []*abci.TxResult | |||
| }{ | |||
| // search by hash | |||
| {fmt.Sprintf("tx.hash = '%X'", hash1), []*abci.TxResult{txResult1}}, | |||
| // search by hash | |||
| {fmt.Sprintf("tx.hash = '%X'", hash2), []*abci.TxResult{txResult2}}, | |||
| // search by exact match (one key) | |||
| {"account.number = 1", []*abci.TxResult{txResult1}}, | |||
| {"account.number >= 1 AND account.number <= 5", []*abci.TxResult{txResult1}}, | |||
| // search by range (lower bound) | |||
| {"account.number >= 1", []*abci.TxResult{txResult1}}, | |||
| // search by range (upper bound) | |||
| {"account.number <= 5", []*abci.TxResult{txResult1}}, | |||
| // search using not allowed key | |||
| {"not_allowed = 'boom'", []*abci.TxResult{}}, | |||
| // search for not existing tx result | |||
| {"account.number >= 2 AND account.number <= 5", []*abci.TxResult{}}, | |||
| // search using not existing key | |||
| {"account.date >= TIME 2013-05-03T14:45:00Z", []*abci.TxResult{}}, | |||
| // search by deprecated key | |||
| {"sender = 'addr1'", []*abci.TxResult{txResult2}}, | |||
| } | |||
| ctx := context.Background() | |||
| for _, tc := range testCases { | |||
| tc := tc | |||
| t.Run(tc.q, func(t *testing.T) { | |||
| results, err := indexer.SearchTxEvents(ctx, query.MustParse(tc.q)) | |||
| require.NoError(t, err) | |||
| for _, txr := range results { | |||
| for _, tr := range tc.results { | |||
| assert.True(t, proto.Equal(tr, txr)) | |||
| } | |||
| } | |||
| }) | |||
| } | |||
| } | |||
| func TestTxSearchOneTxWithMultipleSameTagsButDifferentValues(t *testing.T) { | |||
| indexer := NewEventSink(db.NewMemDB()) | |||
| txResult := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "1", Index: true}}}, | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "2", Index: true}}}, | |||
| }) | |||
| err := indexer.IndexTxEvents([]*abci.TxResult{txResult}) | |||
| require.NoError(t, err) | |||
| ctx := context.Background() | |||
| results, err := indexer.SearchTxEvents(ctx, query.MustParse("account.number >= 1")) | |||
| assert.NoError(t, err) | |||
| assert.Len(t, results, 1) | |||
| for _, txr := range results { | |||
| assert.True(t, proto.Equal(txResult, txr)) | |||
| } | |||
| } | |||
| func TestTxSearchMultipleTxs(t *testing.T) { | |||
| indexer := NewEventSink(db.NewMemDB()) | |||
| // indexed first, but bigger height (to test the order of transactions) | |||
| txResult := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "1", Index: true}}}, | |||
| }) | |||
| txResult.Tx = types.Tx("Bob's account") | |||
| txResult.Height = 2 | |||
| txResult.Index = 1 | |||
| err := indexer.IndexTxEvents([]*abci.TxResult{txResult}) | |||
| require.NoError(t, err) | |||
| // indexed second, but smaller height (to test the order of transactions) | |||
| txResult2 := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "2", Index: true}}}, | |||
| }) | |||
| txResult2.Tx = types.Tx("Alice's account") | |||
| txResult2.Height = 1 | |||
| txResult2.Index = 2 | |||
| err = indexer.IndexTxEvents([]*abci.TxResult{txResult2}) | |||
| require.NoError(t, err) | |||
| // indexed third (to test the order of transactions) | |||
| txResult3 := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "3", Index: true}}}, | |||
| }) | |||
| txResult3.Tx = types.Tx("Jack's account") | |||
| txResult3.Height = 1 | |||
| txResult3.Index = 1 | |||
| err = indexer.IndexTxEvents([]*abci.TxResult{txResult3}) | |||
| require.NoError(t, err) | |||
| // indexed fourth (to test we don't include txs with similar events) | |||
| // https://github.com/tendermint/tendermint/issues/2908 | |||
| txResult4 := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number.id", Value: "1", Index: true}}}, | |||
| }) | |||
| txResult4.Tx = types.Tx("Mike's account") | |||
| txResult4.Height = 2 | |||
| txResult4.Index = 2 | |||
| err = indexer.IndexTxEvents([]*abci.TxResult{txResult4}) | |||
| require.NoError(t, err) | |||
| ctx := context.Background() | |||
| results, err := indexer.SearchTxEvents(ctx, query.MustParse("account.number >= 1")) | |||
| assert.NoError(t, err) | |||
| require.Len(t, results, 3) | |||
| } | |||
| func txResultWithEvents(events []abci.Event) *abci.TxResult { | |||
| tx := types.Tx("HELLO WORLD") | |||
| return &abci.TxResult{ | |||
| Height: 1, | |||
| Index: 0, | |||
| Tx: tx, | |||
| Result: abci.ResponseDeliverTx{ | |||
| Data: []byte{0}, | |||
| Code: abci.CodeTypeOK, | |||
| Log: "", | |||
| Events: events, | |||
| }, | |||
| } | |||
| } | |||
+ 51
- 0
state/indexer/sink/null/null.go
View File
| @ -0,0 +1,51 @@ | |||
| package null | |||
| import ( | |||
| "context" | |||
| abci "github.com/tendermint/tendermint/abci/types" | |||
| "github.com/tendermint/tendermint/libs/pubsub/query" | |||
| "github.com/tendermint/tendermint/state/indexer" | |||
| "github.com/tendermint/tendermint/types" | |||
| ) | |||
| var _ indexer.EventSink = (*EventSink)(nil) | |||
| // EventSink implements a no-op indexer. | |||
| type EventSink struct{} | |||
| func NewEventSink() indexer.EventSink { | |||
| return &EventSink{} | |||
| } | |||
| func (nes *EventSink) Type() indexer.EventSinkType { | |||
| return indexer.NULL | |||
| } | |||
| func (nes *EventSink) IndexBlockEvents(bh types.EventDataNewBlockHeader) error { | |||
| return nil | |||
| } | |||
| func (nes *EventSink) IndexTxEvents(results []*abci.TxResult) error { | |||
| return nil | |||
| } | |||
| func (nes *EventSink) SearchBlockEvents(ctx context.Context, q *query.Query) ([]int64, error) { | |||
| return nil, nil | |||
| } | |||
| func (nes *EventSink) SearchTxEvents(ctx context.Context, q *query.Query) ([]*abci.TxResult, error) { | |||
| return nil, nil | |||
| } | |||
| func (nes *EventSink) GetTxByHash(hash []byte) (*abci.TxResult, error) { | |||
| return nil, nil | |||
| } | |||
| func (nes *EventSink) HasBlock(h int64) (bool, error) { | |||
| return false, nil | |||
| } | |||
| func (nes *EventSink) Stop() error { | |||
| return nil | |||
| } | |||
+ 39
- 0
state/indexer/sink/null/null_test.go
View File
| @ -0,0 +1,39 @@ | |||
| package null | |||
| import ( | |||
| "context" | |||
| "testing" | |||
| "github.com/stretchr/testify/assert" | |||
| "github.com/tendermint/tendermint/state/indexer" | |||
| "github.com/tendermint/tendermint/types" | |||
| ) | |||
| func TestNullEventSink(t *testing.T) { | |||
| nullIndexer := NewEventSink() | |||
| assert.Nil(t, nullIndexer.IndexTxEvents(nil)) | |||
| assert.Nil(t, nullIndexer.IndexBlockEvents(types.EventDataNewBlockHeader{})) | |||
| val1, err1 := nullIndexer.SearchBlockEvents(context.TODO(), nil) | |||
| assert.Nil(t, val1) | |||
| assert.Nil(t, err1) | |||
| val2, err2 := nullIndexer.SearchTxEvents(context.TODO(), nil) | |||
| assert.Nil(t, val2) | |||
| assert.Nil(t, err2) | |||
| val3, err3 := nullIndexer.GetTxByHash(nil) | |||
| assert.Nil(t, val3) | |||
| assert.Nil(t, err3) | |||
| val4, err4 := nullIndexer.HasBlock(0) | |||
| assert.False(t, val4) | |||
| assert.Nil(t, err4) | |||
| } | |||
| func TestType(t *testing.T) { | |||
| nullIndexer := NewEventSink() | |||
| assert.Equal(t, indexer.NULL, nullIndexer.Type()) | |||
| } | |||
| func TestStop(t *testing.T) { | |||
| nullIndexer := NewEventSink() | |||
| assert.Nil(t, nullIndexer.Stop()) | |||
| } | |||
+ 197
- 0
state/indexer/sink/psql/psql.go
View File
| @ -0,0 +1,197 @@ | |||
| package psql | |||
| import ( | |||
| "context" | |||
| "database/sql" | |||
| "errors" | |||
| "fmt" | |||
| "time" | |||
| sq "github.com/Masterminds/squirrel" | |||
| proto "github.com/gogo/protobuf/proto" | |||
| abci "github.com/tendermint/tendermint/abci/types" | |||
| "github.com/tendermint/tendermint/libs/pubsub/query" | |||
| "github.com/tendermint/tendermint/state/indexer" | |||
| "github.com/tendermint/tendermint/types" | |||
| ) | |||
| var _ indexer.EventSink = (*EventSink)(nil) | |||
| const ( | |||
| TableEventBlock = "block_events" | |||
| TableEventTx = "tx_events" | |||
| TableResultTx = "tx_results" | |||
| DriverName = "postgres" | |||
| ) | |||
| // EventSink is an indexer backend providing the tx/block index services. | |||
| type EventSink struct { | |||
| store *sql.DB | |||
| chainID string | |||
| } | |||
| func NewEventSink(connStr string, chainID string) (indexer.EventSink, *sql.DB, error) { | |||
| db, err := sql.Open(DriverName, connStr) | |||
| if err != nil { | |||
| return nil, nil, err | |||
| } | |||
| return &EventSink{ | |||
| store: db, | |||
| chainID: chainID, | |||
| }, db, nil | |||
| } | |||
| func (es *EventSink) Type() indexer.EventSinkType { | |||
| return indexer.PSQL | |||
| } | |||
| func (es *EventSink) IndexBlockEvents(h types.EventDataNewBlockHeader) error { | |||
| sqlStmt := sq. | |||
| Insert(TableEventBlock). | |||
| Columns("key", "value", "height", "type", "created_at", "chain_id"). | |||
| PlaceholderFormat(sq.Dollar) | |||
| ts := time.Now() | |||
| // index the reserved block height index | |||
| sqlStmt = sqlStmt. | |||
| Values(types.BlockHeightKey, fmt.Sprint(h.Header.Height), h.Header.Height, "", ts, es.chainID) | |||
| // index begin_block events | |||
| sqlStmt, err := indexBlockEvents( | |||
| sqlStmt, h.ResultBeginBlock.Events, types.EventTypeBeginBlock, h.Header.Height, ts, es.chainID) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| // index end_block events | |||
| sqlStmt, err = indexBlockEvents( | |||
| sqlStmt, h.ResultEndBlock.Events, types.EventTypeEndBlock, h.Header.Height, ts, es.chainID) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| _, err = sqlStmt.RunWith(es.store).Exec() | |||
| return err | |||
| } | |||
| func (es *EventSink) IndexTxEvents(txr []*abci.TxResult) error { | |||
| // index the tx result | |||
| var txid uint32 | |||
| sqlStmtTxResult := sq. | |||
| Insert(TableResultTx). | |||
| Columns("tx_result", "created_at"). | |||
| PlaceholderFormat(sq.Dollar). | |||
| RunWith(es.store). | |||
| Suffix("RETURNING \"id\"") | |||
| sqlStmtEvents := sq. | |||
| Insert(TableEventTx). | |||
| Columns("key", "value", "height", "hash", "tx_result_id", "created_at", "chain_id"). | |||
| PlaceholderFormat(sq.Dollar) | |||
| ts := time.Now() | |||
| for _, tx := range txr { | |||
| txBz, err := proto.Marshal(tx) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| sqlStmtTxResult = sqlStmtTxResult.Values(txBz, ts) | |||
| // execute sqlStmtTxResult db query and retrieve the txid | |||
| err = sqlStmtTxResult.QueryRow().Scan(&txid) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| // index the reserved height and hash indices | |||
| hash := fmt.Sprintf("%X", types.Tx(tx.Tx).Hash()) | |||
| sqlStmtEvents = sqlStmtEvents.Values(types.TxHashKey, hash, tx.Height, hash, txid, ts, es.chainID) | |||
| sqlStmtEvents = sqlStmtEvents.Values(types.TxHeightKey, fmt.Sprint(tx.Height), tx.Height, hash, txid, ts, es.chainID) | |||
| for _, event := range tx.Result.Events { | |||
| // only index events with a non-empty type | |||
| if len(event.Type) == 0 { | |||
| continue | |||
| } | |||
| for _, attr := range event.Attributes { | |||
| if len(attr.Key) == 0 { | |||
| continue | |||
| } | |||
| // index if `index: true` is set | |||
| compositeTag := fmt.Sprintf("%s.%s", event.Type, attr.Key) | |||
| // ensure event does not conflict with a reserved prefix key | |||
| if compositeTag == types.TxHashKey || compositeTag == types.TxHeightKey { | |||
| return fmt.Errorf("event type and attribute key \"%s\" is reserved; please use a different key", compositeTag) | |||
| } | |||
| if attr.GetIndex() { | |||
| sqlStmtEvents = sqlStmtEvents.Values(compositeTag, attr.Value, tx.Height, hash, txid, ts, es.chainID) | |||
| } | |||
| } | |||
| } | |||
| } | |||
| // execute sqlStmtEvents db query... | |||
| _, err := sqlStmtEvents.RunWith(es.store).Exec() | |||
| return err | |||
| } | |||
| func (es *EventSink) SearchBlockEvents(ctx context.Context, q *query.Query) ([]int64, error) { | |||
| return nil, errors.New("block search is not supported via the postgres event sink") | |||
| } | |||
| func (es *EventSink) SearchTxEvents(ctx context.Context, q *query.Query) ([]*abci.TxResult, error) { | |||
| return nil, errors.New("tx search is not supported via the postgres event sink") | |||
| } | |||
| func (es *EventSink) GetTxByHash(hash []byte) (*abci.TxResult, error) { | |||
| return nil, errors.New("getTxByHash is not supported via the postgres event sink") | |||
| } | |||
| func (es *EventSink) HasBlock(h int64) (bool, error) { | |||
| return false, errors.New("hasBlock is not supported via the postgres event sink") | |||
| } | |||
| func indexBlockEvents( | |||
| sqlStmt sq.InsertBuilder, | |||
| events []abci.Event, | |||
| ty string, | |||
| height int64, | |||
| ts time.Time, | |||
| chainID string, | |||
| ) (sq.InsertBuilder, error) { | |||
| for _, event := range events { | |||
| // only index events with a non-empty type | |||
| if len(event.Type) == 0 { | |||
| continue | |||
| } | |||
| for _, attr := range event.Attributes { | |||
| if len(attr.Key) == 0 { | |||
| continue | |||
| } | |||
| // index iff the event specified index:true and it's not a reserved event | |||
| compositeKey := fmt.Sprintf("%s.%s", event.Type, attr.Key) | |||
| if compositeKey == types.BlockHeightKey { | |||
| return sqlStmt, fmt.Errorf( | |||
| "event type and attribute key \"%s\" is reserved; please use a different key", compositeKey) | |||
| } | |||
| if attr.GetIndex() { | |||
| sqlStmt = sqlStmt.Values(compositeKey, attr.Value, height, ty, ts, chainID) | |||
| } | |||
| } | |||
| } | |||
| return sqlStmt, nil | |||
| } | |||
| func (es *EventSink) Stop() error { | |||
| return es.store.Close() | |||
| } | |||
+ 365
- 0
state/indexer/sink/psql/psql_test.go
View File
| @ -0,0 +1,365 @@ | |||
| package psql | |||
| import ( | |||
| "context" | |||
| "database/sql" | |||
| "errors" | |||
| "fmt" | |||
| "io/ioutil" | |||
| "os" | |||
| "testing" | |||
| "time" | |||
| sq "github.com/Masterminds/squirrel" | |||
| schema "github.com/adlio/schema" | |||
| proto "github.com/gogo/protobuf/proto" | |||
| _ "github.com/lib/pq" | |||
| dockertest "github.com/ory/dockertest" | |||
| "github.com/ory/dockertest/docker" | |||
| "github.com/stretchr/testify/assert" | |||
| "github.com/stretchr/testify/require" | |||
| abci "github.com/tendermint/tendermint/abci/types" | |||
| "github.com/tendermint/tendermint/state/indexer" | |||
| "github.com/tendermint/tendermint/types" | |||
| ) | |||
| var db *sql.DB | |||
| var resource *dockertest.Resource | |||
| var chainID = "test-chainID" | |||
| var ( | |||
| user = "postgres" | |||
| password = "secret" | |||
| port = "5432" | |||
| dsn = "postgres://%s:%s@localhost:%s/%s?sslmode=disable" | |||
| dbName = "postgres" | |||
| ) | |||
| func TestType(t *testing.T) { | |||
| pool, err := setupDB(t) | |||
| require.NoError(t, err) | |||
| psqlSink := &EventSink{store: db, chainID: chainID} | |||
| assert.Equal(t, indexer.PSQL, psqlSink.Type()) | |||
| require.NoError(t, teardown(t, pool)) | |||
| } | |||
| func TestBlockFuncs(t *testing.T) { | |||
| pool, err := setupDB(t) | |||
| require.NoError(t, err) | |||
| indexer := &EventSink{store: db, chainID: chainID} | |||
| require.NoError(t, indexer.IndexBlockEvents(getTestBlockHeader())) | |||
| r, err := verifyBlock(1) | |||
| assert.True(t, r) | |||
| require.NoError(t, err) | |||
| r, err = verifyBlock(2) | |||
| assert.False(t, r) | |||
| require.NoError(t, err) | |||
| r, err = indexer.HasBlock(1) | |||
| assert.False(t, r) | |||
| assert.Equal(t, errors.New("hasBlock is not supported via the postgres event sink"), err) | |||
| r, err = indexer.HasBlock(2) | |||
| assert.False(t, r) | |||
| assert.Equal(t, errors.New("hasBlock is not supported via the postgres event sink"), err) | |||
| r2, err := indexer.SearchBlockEvents(context.TODO(), nil) | |||
| assert.Nil(t, r2) | |||
| assert.Equal(t, errors.New("block search is not supported via the postgres event sink"), err) | |||
| require.NoError(t, verifyTimeStamp(TableEventBlock)) | |||
| require.NoError(t, teardown(t, pool)) | |||
| } | |||
| func TestTxFuncs(t *testing.T) { | |||
| pool, err := setupDB(t) | |||
| assert.Nil(t, err) | |||
| indexer := &EventSink{store: db, chainID: chainID} | |||
| txResult := txResultWithEvents([]abci.Event{ | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "number", Value: "1", Index: true}}}, | |||
| {Type: "account", Attributes: []abci.EventAttribute{{Key: "owner", Value: "Ivan", Index: true}}}, | |||
| {Type: "", Attributes: []abci.EventAttribute{{Key: "not_allowed", Value: "Vlad", Index: true}}}, | |||
| }) | |||
| err = indexer.IndexTxEvents([]*abci.TxResult{txResult}) | |||
| require.NoError(t, err) | |||
| tx, err := verifyTx(types.Tx(txResult.Tx).Hash()) | |||
| require.NoError(t, err) | |||
| assert.Equal(t, txResult, tx) | |||
| require.NoError(t, verifyTimeStamp(TableEventTx)) | |||
| require.NoError(t, verifyTimeStamp(TableResultTx)) | |||
| tx, err = indexer.GetTxByHash(types.Tx(txResult.Tx).Hash()) | |||
| assert.Nil(t, tx) | |||
| assert.Equal(t, errors.New("getTxByHash is not supported via the postgres event sink"), err) | |||
| r2, err := indexer.SearchTxEvents(context.TODO(), nil) | |||
| assert.Nil(t, r2) | |||
| assert.Equal(t, errors.New("tx search is not supported via the postgres event sink"), err) | |||
| assert.Nil(t, teardown(t, pool)) | |||
| } | |||
| func TestStop(t *testing.T) { | |||
| pool, err := setupDB(t) | |||
| require.NoError(t, err) | |||
| indexer := &EventSink{store: db} | |||
| require.NoError(t, indexer.Stop()) | |||
| defer db.Close() | |||
| require.NoError(t, pool.Purge(resource)) | |||
| } | |||
| func getTestBlockHeader() types.EventDataNewBlockHeader { | |||
| return types.EventDataNewBlockHeader{ | |||
| Header: types.Header{Height: 1}, | |||
| ResultBeginBlock: abci.ResponseBeginBlock{ | |||
| Events: []abci.Event{ | |||
| { | |||
| Type: "begin_event", | |||
| Attributes: []abci.EventAttribute{ | |||
| { | |||
| Key: "proposer", | |||
| Value: "FCAA001", | |||
| Index: true, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| ResultEndBlock: abci.ResponseEndBlock{ | |||
| Events: []abci.Event{ | |||
| { | |||
| Type: "end_event", | |||
| Attributes: []abci.EventAttribute{ | |||
| { | |||
| Key: "foo", | |||
| Value: "100", | |||
| Index: true, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| }, | |||
| } | |||
| } | |||
| func readSchema() ([]*schema.Migration, error) { | |||
| filename := "schema.sql" | |||
| contents, err := ioutil.ReadFile(filename) | |||
| if err != nil { | |||
| return nil, fmt.Errorf("failed to read sql file from '%s': %w", filename, err) | |||
| } | |||
| mg := &schema.Migration{} | |||
| mg.ID = time.Now().Local().String() + " db schema" | |||
| mg.Script = string(contents) | |||
| return append([]*schema.Migration{}, mg), nil | |||
| } | |||
| func resetDB(t *testing.T) { | |||
| q := "DROP TABLE IF EXISTS block_events,tx_events,tx_results" | |||
| _, err := db.Exec(q) | |||
| require.NoError(t, err) | |||
| q = "DROP TYPE IF EXISTS block_event_type" | |||
| _, err = db.Exec(q) | |||
| require.NoError(t, err) | |||
| } | |||
| func txResultWithEvents(events []abci.Event) *abci.TxResult { | |||
| tx := types.Tx("HELLO WORLD") | |||
| return &abci.TxResult{ | |||
| Height: 1, | |||
| Index: 0, | |||
| Tx: tx, | |||
| Result: abci.ResponseDeliverTx{ | |||
| Data: []byte{0}, | |||
| Code: abci.CodeTypeOK, | |||
| Log: "", | |||
| Events: events, | |||
| }, | |||
| } | |||
| } | |||
| func verifyTx(hash []byte) (*abci.TxResult, error) { | |||
| join := fmt.Sprintf("%s ON %s.id = tx_result_id", TableEventTx, TableResultTx) | |||
| sqlStmt := sq. | |||
| Select("tx_result", fmt.Sprintf("%s.id", TableResultTx), "tx_result_id", "hash", "chain_id"). | |||
| Distinct().From(TableResultTx). | |||
| InnerJoin(join). | |||
| Where(fmt.Sprintf("hash = $1 AND chain_id = '%s'", chainID), fmt.Sprintf("%X", hash)) | |||
| rows, err := sqlStmt.RunWith(db).Query() | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| defer rows.Close() | |||
| if rows.Next() { | |||
| var txResult []byte | |||
| var txResultID, txid int | |||
| var h, cid string | |||
| err = rows.Scan(&txResult, &txResultID, &txid, &h, &cid) | |||
| if err != nil { | |||
| return nil, nil | |||
| } | |||
| msg := new(abci.TxResult) | |||
| err = proto.Unmarshal(txResult, msg) | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| return msg, err | |||
| } | |||
| // No result | |||
| return nil, nil | |||
| } | |||
| func verifyTimeStamp(tb string) error { | |||
| // We assume the tx indexing time would not exceed 2 second from now | |||
| sqlStmt := sq. | |||
| Select(fmt.Sprintf("%s.created_at", tb)). | |||
| Distinct().From(tb). | |||
| Where(fmt.Sprintf("%s.created_at >= $1", tb), time.Now().Add(-2*time.Second)) | |||
| rows, err := sqlStmt.RunWith(db).Query() | |||
| if err != nil { | |||
| return err | |||
| } | |||
| defer rows.Close() | |||
| if rows.Next() { | |||
| var ts string | |||
| err = rows.Scan(&ts) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| return nil | |||
| } | |||
| return errors.New("no result") | |||
| } | |||
| func verifyBlock(h int64) (bool, error) { | |||
| sqlStmt := sq. | |||
| Select("height"). | |||
| Distinct(). | |||
| From(TableEventBlock). | |||
| Where(fmt.Sprintf("height = %d", h)) | |||
| rows, err := sqlStmt.RunWith(db).Query() | |||
| if err != nil { | |||
| return false, err | |||
| } | |||
| defer rows.Close() | |||
| if !rows.Next() { | |||
| return false, nil | |||
| } | |||
| sqlStmt = sq. | |||
| Select("type, height", "chain_id"). | |||
| Distinct(). | |||
| From(TableEventBlock). | |||
| Where(fmt.Sprintf("height = %d AND type = '%s' AND chain_id = '%s'", h, types.EventTypeBeginBlock, chainID)) | |||
| rows, err = sqlStmt.RunWith(db).Query() | |||
| if err != nil { | |||
| return false, err | |||
| } | |||
| if !rows.Next() { | |||
| return false, nil | |||
| } | |||
| sqlStmt = sq. | |||
| Select("type, height"). | |||
| Distinct(). | |||
| From(TableEventBlock). | |||
| Where(fmt.Sprintf("height = %d AND type = '%s'", h, types.EventTypeEndBlock)) | |||
| rows, err = sqlStmt.RunWith(db).Query() | |||
| if err != nil { | |||
| return false, err | |||
| } | |||
| return rows.Next(), nil | |||
| } | |||
| func setupDB(t *testing.T) (*dockertest.Pool, error) { | |||
| t.Helper() | |||
| pool, err := dockertest.NewPool(os.Getenv("DOCKER_URL")) | |||
| require.NoError(t, err) | |||
| resource, err = pool.RunWithOptions(&dockertest.RunOptions{ | |||
| Repository: DriverName, | |||
| Tag: "13", | |||
| Env: []string{ | |||
| "POSTGRES_USER=" + user, | |||
| "POSTGRES_PASSWORD=" + password, | |||
| "POSTGRES_DB=" + dbName, | |||
| "listen_addresses = '*'", | |||
| }, | |||
| ExposedPorts: []string{port}, | |||
| }, func(config *docker.HostConfig) { | |||
| // set AutoRemove to true so that stopped container goes away by itself | |||
| config.AutoRemove = true | |||
| config.RestartPolicy = docker.RestartPolicy{ | |||
| Name: "no", | |||
| } | |||
| }) | |||
| require.NoError(t, err) | |||
| // Set the container to expire in a minute to avoid orphaned containers | |||
| // hanging around | |||
| _ = resource.Expire(60) | |||
| conn := fmt.Sprintf(dsn, user, password, resource.GetPort(port+"/tcp"), dbName) | |||
| if err = pool.Retry(func() error { | |||
| var err error | |||
| _, db, err = NewEventSink(conn, chainID) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| return db.Ping() | |||
| }); err != nil { | |||
| require.NoError(t, err) | |||
| } | |||
| resetDB(t) | |||
| sm, err := readSchema() | |||
| assert.Nil(t, err) | |||
| assert.Nil(t, schema.NewMigrator().Apply(db, sm)) | |||
| return pool, nil | |||
| } | |||
| func teardown(t *testing.T, pool *dockertest.Pool) error { | |||
| t.Helper() | |||
| // When you're done, kill and remove the container | |||
| assert.Nil(t, pool.Purge(resource)) | |||
| return db.Close() | |||
| } | |||
+ 31
- 0
state/indexer/sink/psql/schema.sql
View File
| @ -0,0 +1,31 @@ | |||
| CREATE TYPE block_event_type AS ENUM ('begin_block', 'end_block', ''); | |||
| CREATE TABLE block_events ( | |||
| id SERIAL PRIMARY KEY, | |||
| key VARCHAR NOT NULL, | |||
| value VARCHAR NOT NULL, | |||
| height INTEGER NOT NULL, | |||
| type block_event_type, | |||
| created_at TIMESTAMPTZ NOT NULL, | |||
| chain_id VARCHAR NOT NULL | |||
| ); | |||
| CREATE TABLE tx_results ( | |||
| id SERIAL PRIMARY KEY, | |||
| tx_result BYTEA NOT NULL, | |||
| created_at TIMESTAMPTZ NOT NULL | |||
| ); | |||
| CREATE TABLE tx_events ( | |||
| id SERIAL PRIMARY KEY, | |||
| key VARCHAR NOT NULL, | |||
| value VARCHAR NOT NULL, | |||
| height INTEGER NOT NULL, | |||
| hash VARCHAR NOT NULL, | |||
| tx_result_id SERIAL, | |||
| created_at TIMESTAMPTZ NOT NULL, | |||
| chain_id VARCHAR NOT NULL, | |||
| FOREIGN KEY (tx_result_id) | |||
| REFERENCES tx_results(id) | |||
| ON DELETE CASCADE | |||
| ); | |||
| CREATE INDEX idx_block_events_key_value ON block_events(key, value); | |||
| CREATE INDEX idx_tx_events_key_value ON tx_events(key, value); | |||
| CREATE INDEX idx_tx_events_hash ON tx_events(hash); | |||