Merge PR #7121: Support Event Indexing

CHANGELOG.md

@@ -158,9 +158,9 @@ be used to retrieve the actual proposal `Content`. Also the `NewMsgSubmitProposa
 * (genesis) [\#7000](https://github.com/cosmos/cosmos-sdk/pull/7000) The root `GenesisState` is now decoded using `encoding/json` instead of amino so `int64` and `uint64` types are now encoded as integers as opposed to strings.
 * (types) [\#7032](https://github.com/cosmos/cosmos-sdk/pull/7032) All types ending with `ID` (e.g. `ProposalID`) now end with `Id` (e.g. `ProposalId`), to match default Protobuf generated format. Also see [\#7033](https://github.com/cosmos/cosmos-sdk/pull/7033) for more details.
 
-
 ### Features
 
+* (events) [\#7121](https://github.com/cosmos/cosmos-sdk/pull/7121) The application now drives what events are indexed by Tendermint via the `index-events` configuration in `app.toml`, which is a list of events taking the form `{eventType}.{attributeKey}`.
 * [\#6089](https://github.com/cosmos/cosmos-sdk/pull/6089) Transactions can now have a `TimeoutHeight` set which allows the transaction to be rejected if it's committed at a height greater than the timeout.
 * (tests) [\#6489](https://github.com/cosmos/cosmos-sdk/pull/6489) Introduce package `testutil`, new in-process testing network framework for use in integration and unit tests.
 * (crypto/multisig) [\#6241](https://github.com/cosmos/cosmos-sdk/pull/6241) Add Multisig type directly to the repo. Previously this was in tendermint.

baseapp/abci.go

@@ -145,6 +145,7 @@ func (app *BaseApp) BeginBlock(req abci.RequestBeginBlock) (res abci.ResponseBeg
 
 	if app.beginBlocker != nil {
 		res = app.beginBlocker(app.deliverState.ctx, req)
+		res.Events = sdk.MarkEventsToIndex(res.Events, app.indexEvents)
 	}
 	// set the signed validators for addition to context in deliverTx
 	app.voteInfos = req.LastCommitInfo.GetVotes()
@@ -161,6 +162,7 @@ func (app *BaseApp) EndBlock(req abci.RequestEndBlock) (res abci.ResponseEndBloc
 
 	if app.endBlocker != nil {
 		res = app.endBlocker(app.deliverState.ctx, req)
+		res.Events = sdk.MarkEventsToIndex(res.Events, app.indexEvents)
 	}
 
 	if cp := app.GetConsensusParams(app.deliverState.ctx); cp != nil {
@@ -207,7 +209,7 @@ func (app *BaseApp) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
 		GasUsed:   int64(gInfo.GasUsed),   // TODO: Should type accept unsigned ints?
 		Log:       result.Log,
 		Data:      result.Data,
-		Events:    result.Events,
+		Events:    sdk.MarkEventsToIndex(result.Events, app.indexEvents),
 	}
 }
 
@@ -245,7 +247,7 @@ func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx
 		GasUsed:   int64(gInfo.GasUsed),   // TODO: Should type accept unsigned ints?
 		Log:       result.Log,
 		Data:      result.Data,
-		Events:    result.Events,
+		Events:    sdk.MarkEventsToIndex(result.Events, app.indexEvents),
 	}
 }
 

baseapp/baseapp.go

@@ -98,6 +98,10 @@ type BaseApp struct { // nolint: maligned
 
 	// trace set will return full stack traces for errors in ABCI Log field
 	trace bool
+
+	// indexEvents defines the set of events in the form {eventType}.{attributeKey},
+	// which informs Tendermint what to index. If empty, all events will be indexed.
+	indexEvents map[string]struct{}
 }
 
 // NewBaseApp returns a reference to an initialized BaseApp. It accepts a
@@ -283,6 +287,14 @@ func (app *BaseApp) setTrace(trace bool) {
 	app.trace = trace
 }
 
+func (app *BaseApp) setIndexEvents(ie []string) {
+	app.indexEvents = make(map[string]struct{})
+
+	for _, e := range ie {
+		app.indexEvents[e] = struct{}{}
+	}
+}
+
 // Router returns the router of the BaseApp.
 func (app *BaseApp) Router() sdk.Router {
 	if app.sealed {

baseapp/options.go

@@ -43,6 +43,11 @@ func SetTrace(trace bool) func(*BaseApp) {
 	return func(app *BaseApp) { app.setTrace(trace) }
 }
 
+// SetIndexEvents provides a BaseApp option function that sets the events to index.
+func SetIndexEvents(ie []string) func(*BaseApp) {
+	return func(app *BaseApp) { app.setIndexEvents(ie) }
+}
+
 // SetInterBlockCache provides a BaseApp option function that sets the
 // inter-block cache.
 func SetInterBlockCache(cache sdk.MultiStorePersistentCache) func(*BaseApp) {

server/config/config.go

@@ -45,6 +45,10 @@ type BaseConfig struct {
 
 	// InterBlockCache enables inter-block caching.
 	InterBlockCache bool `mapstructure:"inter-block-cache"`
+
+	// IndexEvents defines the set of events in the form {eventType}.{attributeKey},
+	// which informs Tendermint what to index. If empty, all events will be indexed.
+	IndexEvents []string `mapstructure:"index-events"`
 }
 
 // APIConfig defines the API listener configuration.
@@ -134,6 +138,7 @@ func DefaultConfig() *Config {
 			PruningKeepRecent: "0",
 			PruningKeepEvery:  "0",
 			PruningInterval:   "0",
+			IndexEvents:       make([]string, 0),
 		},
 		Telemetry: telemetry.Config{
 			Enabled:      false,
@@ -175,6 +180,7 @@ func GetConfig(v *viper.Viper) Config {
 			PruningInterval:   v.GetString("pruning-interval"),
 			HaltHeight:        v.GetUint64("halt-height"),
 			HaltTime:          v.GetUint64("halt-time"),
+			IndexEvents:       v.GetStringSlice("index-events"),
 		},
 		Telemetry: telemetry.Config{
 			ServiceName:             v.GetString("telemetry.service-name"),

server/config/toml.go

@@ -47,6 +47,13 @@ halt-time = {{ .BaseConfig.HaltTime }}
 # InterBlockCache enables inter-block caching.
 inter-block-cache = {{ .BaseConfig.InterBlockCache }}
 
+# IndexEvents defines the set of events in the form {eventType}.{attributeKey},
+# which informs Tendermint what to index. If empty, all events will be indexed.
+#
+# Example:
+# ["message.sender", "message.recipient"]
+index-events = {{ .BaseConfig.IndexEvents }}
+
 ###############################################################################
 ###                         Telemetry Configuration                         ###
 ###############################################################################

server/start.go

@@ -48,6 +48,7 @@ const (
 	FlagPruningKeepRecent = "pruning-keep-recent"
 	FlagPruningKeepEvery  = "pruning-keep-every"
 	FlagPruningInterval   = "pruning-interval"
+	FlagIndexEvents       = "index-events"
 )
 
 // GRPC-related flags.

simapp/simd/cmd/root.go

@@ -187,6 +187,7 @@ func newApp(logger log.Logger, db dbm.DB, traceStore io.Writer, appOpts serverty
 		baseapp.SetHaltTime(cast.ToUint64(appOpts.Get(server.FlagHaltTime))),
 		baseapp.SetInterBlockCache(cache),
 		baseapp.SetTrace(cast.ToBool(appOpts.Get(server.FlagTrace))),
+		baseapp.SetIndexEvents(cast.ToStringSlice(appOpts.Get(server.FlagIndexEvents))),
 	)
 }
 

types/events.go

@@ -199,3 +199,30 @@ func StringifyEvents(events []abci.Event) StringEvents {
 
 	return res.Flatten()
 }
+
+// MarkEventsToIndex returns the set of ABCI events, where each event's attribute
+// has it's index value marked based on the provided set of events to index.
+func MarkEventsToIndex(events []abci.Event, indexSet map[string]struct{}) []abci.Event {
+	updatedEvents := make([]abci.Event, len(events))
+	for i, e := range events {
+		updatedEvent := abci.Event{
+			Type:       e.Type,
+			Attributes: make([]abci.EventAttribute, len(e.Attributes)),
+		}
+
+		for j, attr := range e.Attributes {
+			_, index := indexSet[fmt.Sprintf("%s.%s", e.Type, attr.Key)]
+			updatedAttr := abci.EventAttribute{
+				Key:   attr.Key,
+				Value: attr.Value,
+				Index: index,
+			}
+
+			updatedEvent.Attributes[j] = updatedAttr
+		}
+
+		updatedEvents[i] = updatedEvent
+	}
+
+	return updatedEvents
+}

types/events_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	abci "github.com/tendermint/tendermint/abci/types"
 )
 
 func TestAppendEvents(t *testing.T) {
@@ -69,3 +70,89 @@ func TestStringifyEvents(t *testing.T) {
 	expectedJSONStr := "[{\"type\":\"message\",\"attributes\":[{\"key\":\"sender\",\"value\":\"foo\"},{\"key\":\"module\",\"value\":\"bank\"}]}]"
 	require.Equal(t, expectedJSONStr, string(bz))
 }
+
+func TestMarkEventsToIndex(t *testing.T) {
+	events := []abci.Event{
+		{
+			Type: "message",
+			Attributes: []abci.EventAttribute{
+				{Key: []byte("sender"), Value: []byte("foo")},
+				{Key: []byte("recipient"), Value: []byte("bar")},
+			},
+		},
+		{
+			Type: "staking",
+			Attributes: []abci.EventAttribute{
+				{Key: []byte("deposit"), Value: []byte("5")},
+				{Key: []byte("unbond"), Value: []byte("10")},
+			},
+		},
+	}
+
+	testCases := map[string]struct {
+		events   []abci.Event
+		indexSet map[string]struct{}
+		expected []abci.Event
+	}{
+		"empty index set": {
+			events:   events,
+			expected: events,
+			indexSet: map[string]struct{}{},
+		},
+		"index some events": {
+			events: events,
+			expected: []abci.Event{
+				{
+					Type: "message",
+					Attributes: []abci.EventAttribute{
+						{Key: []byte("sender"), Value: []byte("foo"), Index: true},
+						{Key: []byte("recipient"), Value: []byte("bar")},
+					},
+				},
+				{
+					Type: "staking",
+					Attributes: []abci.EventAttribute{
+						{Key: []byte("deposit"), Value: []byte("5"), Index: true},
+						{Key: []byte("unbond"), Value: []byte("10")},
+					},
+				},
+			},
+			indexSet: map[string]struct{}{
+				"message.sender":  {},
+				"staking.deposit": {},
+			},
+		},
+		"index all events": {
+			events: events,
+			expected: []abci.Event{
+				{
+					Type: "message",
+					Attributes: []abci.EventAttribute{
+						{Key: []byte("sender"), Value: []byte("foo"), Index: true},
+						{Key: []byte("recipient"), Value: []byte("bar"), Index: true},
+					},
+				},
+				{
+					Type: "staking",
+					Attributes: []abci.EventAttribute{
+						{Key: []byte("deposit"), Value: []byte("5"), Index: true},
+						{Key: []byte("unbond"), Value: []byte("10"), Index: true},
+					},
+				},
+			},
+			indexSet: map[string]struct{}{
+				"message.sender":    {},
+				"message.recipient": {},
+				"staking.deposit":   {},
+				"staking.unbond":    {},
+			},
+		},
+	}
+
+	for name, tc := range testCases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			require.Equal(t, tc.expected, MarkEventsToIndex(tc.events, tc.indexSet))
+		})
+	}
+}