diff --git a/config/config.go b/config/config.go index 5162d2d4f..7dc58c030 100644 --- a/config/config.go +++ b/config/config.go @@ -1055,14 +1055,20 @@ type TxIndexConfig struct { // // Options: // 1) "null" - // 2) "kv" (default) - the simplest possible indexer, + // 2) "kv" (default) - The simplest possible indexer, // backed by key-value storage (defaults to levelDB; see DBBackend). - // 3) "psql" - the indexer services backed by PostgreSQL. + // 3) "psql" - The indexer services backed by PostgreSQL. + // 4) "pubsub" - The indexer services backed by Google Cloud PubSub. Indexer string `mapstructure:"indexer"` // The PostgreSQL connection configuration, the connection format: // postgresql://:@:/? PsqlConn string `mapstructure:"psql-conn"` + + // PubsubProjectID defines the Google Cloud Pubsub project ID. Note, operators + // must ensure the GOOGLE_APPLICATION_CREDENTIALS environment variable is set + // to the location of their creds file. + PubsubProjectID string `mapstructure:"pubsub-project-id"` } // DefaultTxIndexConfig returns a default configuration for the transaction indexer. diff --git a/config/toml.go b/config/toml.go index 0f119db4c..e4aca674e 100644 --- a/config/toml.go +++ b/config/toml.go @@ -471,9 +471,10 @@ peer_query_maj23_sleep_duration = "{{ .Consensus.PeerQueryMaj23SleepDuration }}" # # Options: # 1) "null" -# 2) "kv" (default) - the simplest possible indexer, backed by key-value storage (defaults to levelDB; see DBBackend). +# 2) "kv" (default) - The simplest possible indexer, backed by key-value storage (defaults to levelDB; see DBBackend). # - When "kv" is chosen "tx.height" and "tx.hash" will always be indexed. -# 3) "psql" - the indexer services backed by PostgreSQL. +# 3) "psql" - The indexer services backed by PostgreSQL. +# 4) "pubsub" - The indexer services backed by Google Cloud Pubsub. # When "kv" or "psql" is chosen "tx.height" and "tx.hash" will always be indexed. indexer = "{{ .TxIndex.Indexer }}" @@ -481,6 +482,11 @@ indexer = "{{ .TxIndex.Indexer }}" # postgresql://:@:/? psql-conn = "{{ .TxIndex.PsqlConn }}" +# The Google Cloud Pubsub project ID. Note, operators must ensure the +# GOOGLE_APPLICATION_CREDENTIALS environment variable is set to the location of +# their creds file. +pubsub-project-id = "{{ .TxIndex.PubsubProjectID }}" + ####################################################### ### Instrumentation Configuration Options ### ####################################################### diff --git a/state/indexer/sink/kafka/kafka.go b/state/indexer/sink/kafka/kafka.go deleted file mode 100644 index c87a89b05..000000000 --- a/state/indexer/sink/kafka/kafka.go +++ /dev/null @@ -1,17 +0,0 @@ -package kafka - -import ( - "cloud.google.com/go/pubsub" - - "github.com/tendermint/tendermint/state/indexer" - "github.com/tendermint/tendermint/state/txindex" -) - -var ( - _ indexer.BlockIndexer = (*EventSink)(nil) - _ txindex.TxIndexer = (*EventSink)(nil) -) - -type EventSink struct { - client *pubsub.Client -} diff --git a/state/indexer/sink/pubsub/pubsub.go b/state/indexer/sink/pubsub/pubsub.go new file mode 100644 index 000000000..68ff836cc --- /dev/null +++ b/state/indexer/sink/pubsub/pubsub.go @@ -0,0 +1,44 @@ +package pubsub + +import ( + "context" + "fmt" + "os" + "time" + + "cloud.google.com/go/pubsub" + + "github.com/tendermint/tendermint/state/indexer" + "github.com/tendermint/tendermint/state/txindex" +) + +const credsEnvVar = "GOOGLE_APPLICATION_CREDENTIALS" + +var ( + _ indexer.BlockIndexer = (*EventSink)(nil) + _ txindex.TxIndexer = (*EventSink)(nil) +) + +type EventSink struct { + client *pubsub.Client + chainID string +} + +func NewEventSink(connStr, chainID string) (*EventSink, error) { + if s := os.Getenv(credsEnvVar); len(s) == 0 { + return nil, fmt.Errorf("missing '%s' environment variable", credsEnvVar) + } + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + c, err := pubsub.NewClient(ctx, "project-id") + if err != nil { + return nil, fmt.Errorf("failed to create Google Cloud Pubsub client: %w", err) + } + + return &EventSink{ + client: c, + chainID: chainID, + }, nil +}