node: Removed support for writing attestation events to bigtable
The Spy can be modified to listen to attestation events on the p2p network and write them to a database instead.
This commit is contained in:
Kevin Peters
kev1n-peters
parent
406a43d03e
commit
957f3307de
|
|
@ -15,8 +15,6 @@ venv
|
|||
.env.0x
|
||||
devnet-consts.json
|
||||
!/scripts/devnet-consts.json
|
||||
bigtable-admin.json
|
||||
bigtable-writer.json
|
||||
**/cert.pem
|
||||
**/payer-mainnet.json
|
||||
**/payer-testnet.json
|
||||
|
|
|
|||
|
|
@ -172,13 +172,6 @@ Set the include path:
|
|||
|
||||
|
||||
|
||||
### BigTable event persistence
|
||||
|
||||
Guardian events can be persisted to a cloud BigTable instance by passing a GCP project and service account key to Tilt.
|
||||
Launch the devnet with flags supplying your database info to forward events to your cloud BigTable, rather than the local devnet BigTable emulator:
|
||||
|
||||
tilt up -- --num=1 --gcpProject=your-project-id --bigTableKeyPath=./your-service-account-key.json
|
||||
|
||||
### Algorand
|
||||
|
||||
Node logs:
|
||||
|
|
|
|||
30
Tiltfile
30
Tiltfile
|
|
@ -45,11 +45,6 @@ config.define_string("num", False, "Number of guardian nodes to run")
|
|||
#
|
||||
config.define_string("namespace", False, "Kubernetes namespace to use")
|
||||
|
||||
# These arguments will enable writing Guardian events to a cloud BigTable instance.
|
||||
# Writing to a cloud BigTable is optional. These arguments are not required to run the devnet.
|
||||
config.define_string("gcpProject", False, "GCP project ID for BigTable persistence")
|
||||
config.define_string("bigTableKeyPath", False, "Path to BigTable json key file")
|
||||
|
||||
# When running Tilt on a server, this can be used to set the public hostname Tilt runs on
|
||||
# for service links in the UI to work.
|
||||
config.define_string("webHost", False, "Public hostname for port forwards")
|
||||
|
|
@ -83,8 +78,6 @@ config.define_bool("generic_relayer", False, "Enable the generic relayer off-cha
|
|||
cfg = config.parse()
|
||||
num_guardians = int(cfg.get("num", "1"))
|
||||
namespace = cfg.get("namespace", "wormhole")
|
||||
gcpProject = cfg.get("gcpProject", "")
|
||||
bigTableKeyPath = cfg.get("bigTableKeyPath", "")
|
||||
webHost = cfg.get("webHost", "localhost")
|
||||
ci = cfg.get("ci", False)
|
||||
algorand = cfg.get("algorand", ci)
|
||||
|
|
@ -141,14 +134,6 @@ docker_build(
|
|||
|
||||
# node
|
||||
|
||||
if bigTableKeyPath != "":
|
||||
k8s_yaml_with_ns(
|
||||
secret_yaml_generic(
|
||||
"node-bigtable-key",
|
||||
from_file = "bigtable-key.json=" + bigTableKeyPath,
|
||||
),
|
||||
)
|
||||
|
||||
docker_build(
|
||||
ref = "guardiand-image",
|
||||
context = ".",
|
||||
|
|
@ -187,21 +172,6 @@ def build_node_yaml():
|
|||
container["command"] = command_with_dlv(container["command"])
|
||||
print(container["command"])
|
||||
|
||||
if gcpProject != "":
|
||||
container["command"] += [
|
||||
"--bigTablePersistenceEnabled",
|
||||
"--bigTableInstanceName",
|
||||
"wormhole",
|
||||
"--bigTableTableName",
|
||||
"v2Events",
|
||||
"--bigTableTopicName",
|
||||
"new-vaa-devnet",
|
||||
"--bigTableKeyPath",
|
||||
"/tmp/mounted-keys/bigtable-key.json",
|
||||
"--bigTableGCPProject",
|
||||
gcpProject,
|
||||
]
|
||||
|
||||
if aptos:
|
||||
container["command"] += [
|
||||
"--aptosRPC",
|
||||
|
|
|
|||
|
|
@ -44,13 +44,6 @@ spec:
|
|||
# mount shared between containers for runtime state
|
||||
- name: node-rundir
|
||||
emptyDir: {}
|
||||
- name: node-bigtable-key
|
||||
secret:
|
||||
secretName: node-bigtable-key
|
||||
optional: true
|
||||
items:
|
||||
- key: bigtable-key.json
|
||||
path: bigtable-key.json
|
||||
- name: node-wormchain-key
|
||||
secret:
|
||||
secretName: node-wormchain-key
|
||||
|
|
@ -66,8 +59,6 @@ spec:
|
|||
volumeMounts:
|
||||
- mountPath: /run/node
|
||||
name: node-rundir
|
||||
- mountPath: /tmp/mounted-keys/bigtable
|
||||
name: node-bigtable-key
|
||||
- mountPath: /tmp/mounted-keys/wormchain
|
||||
name: node-wormchain-key
|
||||
command:
|
||||
|
|
|
|||
|
|
@ -39,7 +39,6 @@ import (
|
|||
"github.com/certusone/wormhole/node/pkg/devnet"
|
||||
"github.com/certusone/wormhole/node/pkg/node"
|
||||
"github.com/certusone/wormhole/node/pkg/p2p"
|
||||
"github.com/certusone/wormhole/node/pkg/reporter"
|
||||
"github.com/certusone/wormhole/node/pkg/supervisor"
|
||||
libp2p_crypto "github.com/libp2p/go-libp2p/core/crypto"
|
||||
"github.com/libp2p/go-libp2p/core/peer"
|
||||
|
|
@ -204,13 +203,6 @@ var (
|
|||
// Loki cloud logging parameters
|
||||
telemetryLokiURL *string
|
||||
|
||||
bigTablePersistenceEnabled *bool
|
||||
bigTableGCPProject *string
|
||||
bigTableInstanceName *string
|
||||
bigTableTableName *string
|
||||
bigTableTopicName *string
|
||||
bigTableKeyPath *string
|
||||
|
||||
chainGovernorEnabled *bool
|
||||
|
||||
gatewayRelayerContract *string
|
||||
|
|
@ -376,13 +368,6 @@ func init() {
|
|||
|
||||
telemetryLokiURL = NodeCmd.Flags().String("telemetryLokiURL", "", "Loki cloud logging URL")
|
||||
|
||||
bigTablePersistenceEnabled = NodeCmd.Flags().Bool("bigTablePersistenceEnabled", false, "Turn on forwarding events to BigTable")
|
||||
bigTableGCPProject = NodeCmd.Flags().String("bigTableGCPProject", "", "Google Cloud project ID for storing events")
|
||||
bigTableInstanceName = NodeCmd.Flags().String("bigTableInstanceName", "", "BigTable instance name for storing events")
|
||||
bigTableTableName = NodeCmd.Flags().String("bigTableTableName", "", "BigTable table name to store events in")
|
||||
bigTableTopicName = NodeCmd.Flags().String("bigTableTopicName", "", "GCP topic name to publish to")
|
||||
bigTableKeyPath = NodeCmd.Flags().String("bigTableKeyPath", "", "Path to json Service Account key")
|
||||
|
||||
chainGovernorEnabled = NodeCmd.Flags().Bool("chainGovernorEnabled", false, "Run the chain governor")
|
||||
|
||||
gatewayRelayerContract = NodeCmd.Flags().String("gatewayRelayerContract", "", "Address of the smart contract on wormchain to receive relayed VAAs")
|
||||
|
|
@ -752,24 +737,6 @@ func runNode(cmd *cobra.Command, args []string) {
|
|||
}
|
||||
}
|
||||
|
||||
if *bigTablePersistenceEnabled {
|
||||
if *bigTableGCPProject == "" {
|
||||
logger.Fatal("Please specify --bigTableGCPProject")
|
||||
}
|
||||
if *bigTableInstanceName == "" {
|
||||
logger.Fatal("Please specify --bigTableInstanceName")
|
||||
}
|
||||
if *bigTableTableName == "" {
|
||||
logger.Fatal("Please specify --bigTableTableName")
|
||||
}
|
||||
if *bigTableTopicName == "" {
|
||||
logger.Fatal("Please specify --bigTableTopicName")
|
||||
}
|
||||
if *bigTableKeyPath == "" {
|
||||
logger.Fatal("Please specify --bigTableKeyPath")
|
||||
}
|
||||
}
|
||||
|
||||
if *telemetryKey != "" && *telemetryServiceAccountFile != "" {
|
||||
logger.Fatal("Please do not specify both --telemetryKey and --telemetryServiceAccountFile")
|
||||
}
|
||||
|
|
@ -1480,18 +1447,6 @@ func runNode(cmd *cobra.Command, args []string) {
|
|||
}
|
||||
}
|
||||
|
||||
if *bigTablePersistenceEnabled {
|
||||
bigTableConnectionConfig := &reporter.BigTableConnectionConfig{
|
||||
GcpProjectID: *bigTableGCPProject,
|
||||
GcpInstanceName: *bigTableInstanceName,
|
||||
TableName: *bigTableTableName,
|
||||
TopicName: *bigTableTopicName,
|
||||
GcpKeyFilePath: *bigTableKeyPath,
|
||||
}
|
||||
|
||||
guardianOptions = append(guardianOptions, node.GuardianOptionBigTablePersistence(bigTableConnectionConfig))
|
||||
}
|
||||
|
||||
// Run supervisor with Guardian Node as root.
|
||||
supervisor.New(rootCtx, logger, guardianNode.Run(rootCtxCancel, guardianOptions...),
|
||||
// It's safer to crash and restart the process in case we encounter a panic,
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ module github.com/certusone/wormhole/node
|
|||
go 1.20
|
||||
|
||||
require (
|
||||
cloud.google.com/go/bigtable v1.18.1
|
||||
github.com/celo-org/celo-blockchain v1.5.5
|
||||
github.com/cenkalti/backoff/v4 v4.2.0
|
||||
github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf
|
||||
|
|
@ -46,7 +45,6 @@ require (
|
|||
|
||||
require (
|
||||
cloud.google.com/go/logging v1.7.0
|
||||
cloud.google.com/go/pubsub v1.33.0
|
||||
github.com/CosmWasm/wasmd v0.30.0
|
||||
github.com/algorand/go-algorand-sdk v1.23.0
|
||||
github.com/benbjohnson/clock v1.3.5
|
||||
|
|
@ -72,7 +70,6 @@ require (
|
|||
cloud.google.com/go v0.110.6 // indirect
|
||||
cloud.google.com/go/compute v1.23.0 // indirect
|
||||
cloud.google.com/go/compute/metadata v0.2.3 // indirect
|
||||
cloud.google.com/go/iam v1.1.1 // indirect
|
||||
cloud.google.com/go/longrunning v0.5.1 // indirect
|
||||
contrib.go.opencensus.io/exporter/stackdriver v0.13.11 // indirect
|
||||
filippo.io/edwards25519 v1.0.0 // indirect
|
||||
|
|
@ -97,11 +94,8 @@ require (
|
|||
github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 // indirect
|
||||
github.com/c2h5oh/datasize v0.0.0-20200112174442-28bbd4740fee // indirect
|
||||
github.com/celo-org/celo-bls-go v0.2.4 // indirect
|
||||
github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
|
||||
github.com/cespare/xxhash v1.1.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.2.0 // indirect
|
||||
github.com/cncf/udpa/go v0.0.0-20220112060539-c52dc94e7fbe // indirect
|
||||
github.com/cncf/xds/go v0.0.0-20230607035331-e9ce68804cb4 // indirect
|
||||
github.com/coinbase/rosetta-sdk-go v0.7.0 // indirect
|
||||
github.com/confio/ics23/go v0.9.0 // indirect
|
||||
github.com/containerd/cgroups v1.1.0 // indirect
|
||||
|
|
@ -132,8 +126,6 @@ require (
|
|||
github.com/dvsekhvalnov/jose2go v1.5.0 // indirect
|
||||
github.com/edsrzf/mmap-go v1.1.0 // indirect
|
||||
github.com/elastic/gosigar v0.14.2 // indirect
|
||||
github.com/envoyproxy/go-control-plane v0.11.1-0.20230524094728-9239064ad72f // indirect
|
||||
github.com/envoyproxy/protoc-gen-validate v0.10.1 // indirect
|
||||
github.com/fatih/color v1.14.1 // indirect
|
||||
github.com/felixge/httpsnoop v1.0.3 // indirect
|
||||
github.com/flynn/noise v1.0.0 // indirect
|
||||
|
|
@ -354,7 +346,6 @@ require (
|
|||
golang.org/x/term v0.11.0 // indirect
|
||||
golang.org/x/text v0.12.0 // indirect
|
||||
golang.org/x/tools v0.12.0 // indirect
|
||||
golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
|
||||
gonum.org/v1/gonum v0.12.0 // indirect
|
||||
google.golang.org/appengine v1.6.7 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20230807174057-1744710a1577 // indirect
|
||||
|
|
|
|||
13
node/go.sum
13
node/go.sum
|
|
@ -102,8 +102,6 @@ cloud.google.com/go/bigquery v1.42.0/go.mod h1:8dRTJxhtG+vwBKzE5OseQn/hiydoQN3Ee
|
|||
cloud.google.com/go/bigquery v1.43.0/go.mod h1:ZMQcXHsl+xmU1z36G2jNGZmKp9zNY5BUua5wDgmNCfw=
|
||||
cloud.google.com/go/bigquery v1.44.0/go.mod h1:0Y33VqXTEsbamHJvJHdFmtqHvMIY28aK1+dFsvaChGc=
|
||||
cloud.google.com/go/bigtable v1.2.0/go.mod h1:JcVAOl45lrTmQfLj7T6TxyMzIN/3FGGcFm+2xVAli2o=
|
||||
cloud.google.com/go/bigtable v1.18.1 h1:SxQk9Bj6OKxeiuvevG/KBjqGn/7X8heZbWfK0tYkFd8=
|
||||
cloud.google.com/go/bigtable v1.18.1/go.mod h1:NAVyfJot9jlo+KmgWLUJ5DJGwNDoChzAcrecLpmuAmY=
|
||||
cloud.google.com/go/billing v1.4.0/go.mod h1:g9IdKBEFlItS8bTtlrZdVLWSSdSyFUZKXNS02zKMOZY=
|
||||
cloud.google.com/go/billing v1.5.0/go.mod h1:mztb1tBc3QekhjSgmpf/CV4LzWXLzCArwpLmP2Gm88s=
|
||||
cloud.google.com/go/billing v1.6.0/go.mod h1:WoXzguj+BeHXPbKfNWkqVtDdzORazmCjraY+vrxcyvI=
|
||||
|
|
@ -229,7 +227,6 @@ cloud.google.com/go/iam v0.6.0/go.mod h1:+1AH33ueBne5MzYccyMHtEKqLE4/kJOibtffMHD
|
|||
cloud.google.com/go/iam v0.7.0/go.mod h1:H5Br8wRaDGNc8XP3keLc4unfUUZeyH3Sfl9XpQEYOeg=
|
||||
cloud.google.com/go/iam v0.8.0/go.mod h1:lga0/y3iH6CX7sYqypWJ33hf7kkfXJag67naqGESjkE=
|
||||
cloud.google.com/go/iam v1.1.1 h1:lW7fzj15aVIXYHREOqjRBV9PsH0Z6u8Y46a1YGvQP4Y=
|
||||
cloud.google.com/go/iam v1.1.1/go.mod h1:A5avdyVL2tCppe4unb0951eI9jreack+RJ0/d+KUZOU=
|
||||
cloud.google.com/go/iap v1.4.0/go.mod h1:RGFwRJdihTINIe4wZ2iCP0zF/qu18ZwyKxrhMhygBEc=
|
||||
cloud.google.com/go/iap v1.5.0/go.mod h1:UH/CGgKd4KyohZL5Pt0jSKE4m3FR51qg6FKQ/z/Ix9A=
|
||||
cloud.google.com/go/ids v1.1.0/go.mod h1:WIuwCaYVOzHIj2OhN9HAwvW+DBdmUAdcWlFxRl+KubM=
|
||||
|
|
@ -239,7 +236,6 @@ cloud.google.com/go/iot v1.4.0/go.mod h1:dIDxPOn0UvNDUMD8Ger7FIaTuvMkj+aGk94RPP0
|
|||
cloud.google.com/go/kms v1.4.0/go.mod h1:fajBHndQ+6ubNw6Ss2sSd+SWvjL26RNo/dr7uxsnnOA=
|
||||
cloud.google.com/go/kms v1.5.0/go.mod h1:QJS2YY0eJGBg3mnDfuaCyLauWwBJiHRboYxJ++1xJNg=
|
||||
cloud.google.com/go/kms v1.6.0/go.mod h1:Jjy850yySiasBUDi6KFUwUv2n1+o7QZFyuUJg6OgjA0=
|
||||
cloud.google.com/go/kms v1.15.0 h1:xYl5WEaSekKYN5gGRyhjvZKM22GVBBCzegGNVPy+aIs=
|
||||
cloud.google.com/go/language v1.4.0/go.mod h1:F9dRpNFQmJbkaop6g0JhSBXCNlO90e1KWx5iDdxbWic=
|
||||
cloud.google.com/go/language v1.6.0/go.mod h1:6dJ8t3B+lUYfStgls25GusK04NLh3eDLQnWM3mdEbhI=
|
||||
cloud.google.com/go/language v1.7.0/go.mod h1:DJ6dYN/W+SQOjF8e1hLQXMF21AkH2w9wiPzPCJa2MIE=
|
||||
|
|
@ -308,8 +304,6 @@ cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjp
|
|||
cloud.google.com/go/pubsub v1.5.0/go.mod h1:ZEwJccE3z93Z2HWvstpri00jOg7oO4UZDtKhwDwqF0w=
|
||||
cloud.google.com/go/pubsub v1.26.0/go.mod h1:QgBH3U/jdJy/ftjPhTkyXNj543Tin1pRYcdcPRnFIRI=
|
||||
cloud.google.com/go/pubsub v1.27.1/go.mod h1:hQN39ymbV9geqBnfQq6Xf63yNhUAhv9CZhzp5O6qsW0=
|
||||
cloud.google.com/go/pubsub v1.33.0 h1:6SPCPvWav64tj0sVX/+npCBKhUi/UjJehy9op/V3p2g=
|
||||
cloud.google.com/go/pubsub v1.33.0/go.mod h1:f+w71I33OMyxf9VpMVcZbnG5KSUkCOUHYpFd5U1GdRc=
|
||||
cloud.google.com/go/pubsublite v1.5.0/go.mod h1:xapqNQ1CuLfGi23Yda/9l4bBCKz/wC3KIJ5gKcxveZg=
|
||||
cloud.google.com/go/recaptchaenterprise v1.3.1/go.mod h1:OdD+q+y4XGeAlxRaMn1Y7/GveP6zmq76byL6tjPE7d4=
|
||||
cloud.google.com/go/recaptchaenterprise/v2 v2.1.0/go.mod h1:w9yVqajwroDNTfGuhmOjPDN//rZGySaf6PtFVcSCa7o=
|
||||
|
|
@ -810,7 +804,6 @@ github.com/cenkalti/backoff/v4 v4.2.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyY
|
|||
github.com/census-instrumentation/opencensus-proto v0.2.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/census-instrumentation/opencensus-proto v0.3.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/census-instrumentation/opencensus-proto v0.4.1 h1:iKLQ0xPNFxR/2hzXZMrBo8f1j86j5WHzznCCQxV/b8g=
|
||||
github.com/census-instrumentation/opencensus-proto v0.4.1/go.mod h1:4T9NM4+4Vw91VeyqjLS6ao50K5bOcLKN6Q42XnYaRYw=
|
||||
github.com/certifi/gocertifi v0.0.0-20191021191039-0944d244cd40/go.mod h1:sGbDF6GwGcLpkNXPUTkMRoywsNa/ol15pxFe6ERfguA=
|
||||
github.com/certifi/gocertifi v0.0.0-20200922220541-2c3bb06c6054/go.mod h1:sGbDF6GwGcLpkNXPUTkMRoywsNa/ol15pxFe6ERfguA=
|
||||
|
|
@ -847,7 +840,6 @@ github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGX
|
|||
github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
|
||||
github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
|
||||
github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
|
||||
github.com/cncf/udpa/go v0.0.0-20220112060539-c52dc94e7fbe h1:QQ3GSy+MqSHxm/d8nCtnAiZdYFd45cYZPs8vOOIYKfk=
|
||||
github.com/cncf/udpa/go v0.0.0-20220112060539-c52dc94e7fbe/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
|
||||
github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
|
|
@ -858,7 +850,6 @@ github.com/cncf/xds/go v0.0.0-20211130200136-a8f946100490/go.mod h1:eXthEFrGJvWH
|
|||
github.com/cncf/xds/go v0.0.0-20220314180256-7f1daf1720fc/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20230105202645-06c439db220b/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20230607035331-e9ce68804cb4 h1:/inchEIKaYC1Akx+H+gqO04wryn5h75LSazbRlnya1k=
|
||||
github.com/cncf/xds/go v0.0.0-20230607035331-e9ce68804cb4/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa/go.mod h1:zn76sxSg3SzpJ0PPJaLDCu+Bu0Lg3sKTORVIj19EIF8=
|
||||
github.com/cockroachdb/datadriven v0.0.0-20200714090401-bf6692d28da5/go.mod h1:h6jFvWxBdQXxjopDMZyH2UVceIRfR84bdzbkoKrsWNo=
|
||||
github.com/cockroachdb/errors v1.2.4/go.mod h1:rQD95gz6FARkaKkQXUksEje/d9a6wBJoCr5oaCLELYA=
|
||||
|
|
@ -1200,14 +1191,12 @@ github.com/envoyproxy/go-control-plane v0.10.1/go.mod h1:AY7fTTXNdv/aJ2O5jwpxAPO
|
|||
github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE=
|
||||
github.com/envoyproxy/go-control-plane v0.10.3/go.mod h1:fJJn/j26vwOu972OllsvAgJJM//w9BV6Fxbg2LuVd34=
|
||||
github.com/envoyproxy/go-control-plane v0.11.1-0.20230524094728-9239064ad72f h1:7T++XKzy4xg7PKy+bM+Sa9/oe1OC88yz2hXQUISoXfA=
|
||||
github.com/envoyproxy/go-control-plane v0.11.1-0.20230524094728-9239064ad72f/go.mod h1:sfYdkwUW4BA3PbKjySwjJy+O4Pu0h62rlqCMHNk+K+Q=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.0.14/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.6.2/go.mod h1:2t7qjJNvHPx8IjnBOzl9E9/baC+qXE/TeeyBRzgJDws=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.6.7/go.mod h1:dyJXwwfPK2VSqiB9Klm1J6romD608Ba7Hij42vrOBCo=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.9.1/go.mod h1:OKNgG7TCp5pF4d6XftA0++PMirau2/yoOwVac3AbF2w=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.10.1 h1:c0g45+xCJhdgFGw7a5QAfdS4byAbud7miNWJ1WwEVf8=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.10.1/go.mod h1:DRjgyB0I43LtJapqN6NiRwroiAU2PaFuvk/vjgh61ss=
|
||||
github.com/esimonov/ifshort v1.0.4/go.mod h1:Pe8zjlRrJ80+q2CxHLfEOfTwxCZ4O+MuhcHcfgNWTk0=
|
||||
github.com/ethereum/go-ethereum v1.9.25/go.mod h1:vMkFiYLHI4tgPw4k2j4MHKoovchFE8plZ0M9VMk4/oM=
|
||||
github.com/ethereum/go-ethereum v1.10.4/go.mod h1:nEE0TP5MtxGzOMd7egIrbPJMQBnhVU3ELNxhBglIzhg=
|
||||
|
|
@ -3927,7 +3916,6 @@ golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8T
|
|||
golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
|
||||
golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
|
||||
golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 h1:H2TDz8ibqkAF6YGhCdN3jS9O0/s90v0rJh3X/OLHEUk=
|
||||
golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
|
||||
gonum.org/v1/gonum v0.0.0-20180816165407-929014505bf4/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
|
||||
gonum.org/v1/gonum v0.0.0-20181121035319-3f7ecaa7e8ca/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
|
||||
|
|
@ -4430,7 +4418,6 @@ nhooyr.io/websocket v1.8.6/go.mod h1:B70DZP8IakI65RVQ51MsWP/8jndNma26DVA/nFSCgW0
|
|||
nhooyr.io/websocket v1.8.7 h1:usjR2uOr/zjjkVMy0lW+PPohFok7PCow5sDjLgX4P4g=
|
||||
nhooyr.io/websocket v1.8.7/go.mod h1:B70DZP8IakI65RVQ51MsWP/8jndNma26DVA/nFSCgW0=
|
||||
pack.ag/amqp v0.11.2/go.mod h1:4/cbmt4EJXSKlG6LCfWHoqmN0uFdy5i/+YFz+fTfhV4=
|
||||
rsc.io/binaryregexp v0.2.0 h1:HfqmD5MEmC0zvwBuF187nq9mdnXjXsSivRiXN7SmRkE=
|
||||
rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
|
||||
rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=
|
||||
rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
|
||||
|
|
|
|||
|
|
@ -720,7 +720,7 @@ func (s *nodePrivilegedService) fetchMissing(
|
|||
|
||||
// Inject into the gossip signed VAA receive path.
|
||||
// This has the same effect as if the VAA was received from the network
|
||||
// (verifying signature, publishing to BigTable, storing in local DB...).
|
||||
// (verifying signature, storing in local DB...).
|
||||
s.signedInC <- &gossipv1.SignedVAAWithQuorum{
|
||||
Vaa: vaaBytes,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -11,7 +11,6 @@ import (
|
|||
"github.com/certusone/wormhole/node/pkg/governor"
|
||||
"github.com/certusone/wormhole/node/pkg/gwrelayer"
|
||||
gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
|
||||
"github.com/certusone/wormhole/node/pkg/reporter"
|
||||
"github.com/certusone/wormhole/node/pkg/supervisor"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
|
@ -55,13 +54,12 @@ type G struct {
|
|||
gk *ecdsa.PrivateKey
|
||||
|
||||
// components
|
||||
db *db.Database
|
||||
gst *common.GuardianSetState
|
||||
acct *accountant.Accountant
|
||||
gov *governor.ChainGovernor
|
||||
gatewayRelayer *gwrelayer.GatewayRelayer
|
||||
attestationEvents *reporter.AttestationEventReporter
|
||||
publicrpcServer *grpc.Server
|
||||
db *db.Database
|
||||
gst *common.GuardianSetState
|
||||
acct *accountant.Accountant
|
||||
gov *governor.ChainGovernor
|
||||
gatewayRelayer *gwrelayer.GatewayRelayer
|
||||
publicrpcServer *grpc.Server
|
||||
|
||||
// runnables
|
||||
runnablesWithScissors map[string]supervisor.Runnable
|
||||
|
|
@ -98,7 +96,7 @@ func NewGuardianNode(
|
|||
}
|
||||
|
||||
// initializeBasic sets up everything that every GuardianNode needs before any options can be applied.
|
||||
func (g *G) initializeBasic(logger *zap.Logger, rootCtxCancel context.CancelFunc) {
|
||||
func (g *G) initializeBasic(rootCtxCancel context.CancelFunc) {
|
||||
g.rootCtxCancel = rootCtxCancel
|
||||
|
||||
// Setup various channels...
|
||||
|
|
@ -114,9 +112,6 @@ func (g *G) initializeBasic(logger *zap.Logger, rootCtxCancel context.CancelFunc
|
|||
// Guardian set state managed by processor
|
||||
g.gst = common.NewGuardianSetState(nil)
|
||||
|
||||
// provides methods for reporting progress toward message attestation, and channels for receiving attestation lifecycle events.
|
||||
g.attestationEvents = reporter.EventListener(logger)
|
||||
|
||||
// allocate maps
|
||||
g.runnablesWithScissors = make(map[string]supervisor.Runnable)
|
||||
g.runnables = make(map[string]supervisor.Runnable)
|
||||
|
|
@ -158,7 +153,7 @@ func (g *G) Run(rootCtxCancel context.CancelFunc, options ...*GuardianOption) su
|
|||
return func(ctx context.Context) error {
|
||||
logger := supervisor.Logger(ctx)
|
||||
|
||||
g.initializeBasic(logger, rootCtxCancel)
|
||||
g.initializeBasic(rootCtxCancel)
|
||||
if err := g.applyOptions(ctx, logger, options); err != nil {
|
||||
logger.Fatal("failed to initialize GuardianNode", zap.Error(err))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -913,10 +913,10 @@ func TestGuardianConfigs(t *testing.T) {
|
|||
{
|
||||
name: "double-configuration",
|
||||
opts: []*GuardianOption{
|
||||
GuardianOptionBigTablePersistence(nil),
|
||||
GuardianOptionBigTablePersistence(nil),
|
||||
GuardianOptionDatabase(nil),
|
||||
GuardianOptionDatabase(nil),
|
||||
},
|
||||
err: "Component bigtable is already configured and cannot be configured a second time",
|
||||
err: "Component db is already configured and cannot be configured a second time",
|
||||
},
|
||||
}
|
||||
runGuardianConfigTests(t, tc)
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@ import (
|
|||
"github.com/certusone/wormhole/node/pkg/processor"
|
||||
gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
|
||||
"github.com/certusone/wormhole/node/pkg/readiness"
|
||||
"github.com/certusone/wormhole/node/pkg/reporter"
|
||||
"github.com/certusone/wormhole/node/pkg/supervisor"
|
||||
"github.com/certusone/wormhole/node/pkg/watchers"
|
||||
"github.com/certusone/wormhole/node/pkg/watchers/ibc"
|
||||
|
|
@ -453,15 +452,6 @@ func GuardianOptionPublicWeb(listenAddr string, publicGRPCSocketPath string, tls
|
|||
}}
|
||||
}
|
||||
|
||||
func GuardianOptionBigTablePersistence(config *reporter.BigTableConnectionConfig) *GuardianOption {
|
||||
return &GuardianOption{
|
||||
name: "bigtable",
|
||||
f: func(ctx context.Context, logger *zap.Logger, g *G) error {
|
||||
g.runnables["bigtable"] = reporter.BigTableWriter(g.attestationEvents, config)
|
||||
return nil
|
||||
}}
|
||||
}
|
||||
|
||||
// GuardianOptionDatabase configures the main database to be used for this guardian node.
|
||||
// Dependencies: none
|
||||
func GuardianOptionDatabase(db *db.Database) *GuardianOption {
|
||||
|
|
@ -493,7 +483,6 @@ func GuardianOptionProcessor() *GuardianOption {
|
|||
g.signedInC.readC,
|
||||
g.gk,
|
||||
g.gst,
|
||||
g.attestationEvents,
|
||||
g.gov,
|
||||
g.acct,
|
||||
g.acctC.readC,
|
||||
|
|
|
|||
|
|
@ -12,7 +12,6 @@ import (
|
|||
"go.uber.org/zap"
|
||||
|
||||
"github.com/certusone/wormhole/node/pkg/common"
|
||||
"github.com/certusone/wormhole/node/pkg/reporter"
|
||||
"github.com/wormhole-foundation/wormhole/sdk/vaa"
|
||||
)
|
||||
|
||||
|
|
@ -106,7 +105,5 @@ func (p *Processor) handleMessage(k *common.MessagePublication) {
|
|||
messagesSignedTotal.With(prometheus.Labels{
|
||||
"emitter_chain": k.EmitterChain.String()}).Add(1)
|
||||
|
||||
p.attestationEvents.ReportMessagePublication(&reporter.MessagePublication{VAA: v.VAA, InitiatingTxID: k.TxHash})
|
||||
|
||||
p.broadcastSignature(v, s, k.TxHash.Bytes())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -316,5 +316,4 @@ func (p *Processor) handleInboundSignedVAAWithQuorum(ctx context.Context, m *gos
|
|||
p.logger.Error("failed to store signed VAA", zap.Error(err))
|
||||
return
|
||||
}
|
||||
p.attestationEvents.ReportVAAQuorum(v)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@ import (
|
|||
"github.com/certusone/wormhole/node/pkg/common"
|
||||
"github.com/certusone/wormhole/node/pkg/gwrelayer"
|
||||
gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
|
||||
"github.com/certusone/wormhole/node/pkg/reporter"
|
||||
"github.com/certusone/wormhole/node/pkg/supervisor"
|
||||
"github.com/wormhole-foundation/wormhole/sdk/vaa"
|
||||
|
||||
|
|
@ -105,8 +104,6 @@ type Processor struct {
|
|||
// gk is the node's guardian private key
|
||||
gk *ecdsa.PrivateKey
|
||||
|
||||
attestationEvents *reporter.AttestationEventReporter
|
||||
|
||||
logger *zap.Logger
|
||||
|
||||
db *db.Database
|
||||
|
|
@ -158,7 +155,6 @@ func NewProcessor(
|
|||
signedInC <-chan *gossipv1.SignedVAAWithQuorum,
|
||||
gk *ecdsa.PrivateKey,
|
||||
gst *common.GuardianSetState,
|
||||
attestationEvents *reporter.AttestationEventReporter,
|
||||
g *governor.ChainGovernor,
|
||||
acct *accountant.Accountant,
|
||||
acctReadC <-chan *common.MessagePublication,
|
||||
|
|
@ -176,8 +172,6 @@ func NewProcessor(
|
|||
gst: gst,
|
||||
db: db,
|
||||
|
||||
attestationEvents: attestationEvents,
|
||||
|
||||
logger: supervisor.Logger(ctx),
|
||||
state: &aggregationState{observationMap{}},
|
||||
ourAddr: crypto.PubkeyToAddress(gk.PublicKey),
|
||||
|
|
|
|||
|
|
@ -35,7 +35,6 @@ func (v *VAA) HandleQuorum(sigs []*vaa.Signature, hash string, p *Processor) {
|
|||
}
|
||||
|
||||
p.broadcastSignedVAA(signed)
|
||||
p.attestationEvents.ReportVAAQuorum(signed)
|
||||
p.state.signatures[hash].submitted = true
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,114 +0,0 @@
|
|||
package reporter
|
||||
|
||||
import (
|
||||
"math/rand"
|
||||
"sync"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/wormhole-foundation/wormhole/sdk/vaa"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
)
|
||||
|
||||
const maxClientId = 1e6
|
||||
|
||||
type (
|
||||
// MessagePublication is a VAA along with a transaction identifer from the EmiterChain
|
||||
MessagePublication struct {
|
||||
VAA vaa.VAA
|
||||
// The native transaction identifier from the EmitterAddress interaction.
|
||||
InitiatingTxID common.Hash
|
||||
}
|
||||
)
|
||||
|
||||
type lifecycleEventChannels struct {
|
||||
// channel for each event
|
||||
MessagePublicationC chan *MessagePublication
|
||||
VAAQuorumC chan *vaa.VAA
|
||||
}
|
||||
|
||||
type AttestationEventReporter struct {
|
||||
mu sync.RWMutex
|
||||
logger *zap.Logger
|
||||
|
||||
subs map[int]*lifecycleEventChannels
|
||||
}
|
||||
type activeSubscription struct {
|
||||
ClientId int
|
||||
Channels *lifecycleEventChannels
|
||||
}
|
||||
|
||||
func EventListener(logger *zap.Logger) *AttestationEventReporter {
|
||||
events := &AttestationEventReporter{
|
||||
logger: logger.Named("eventlistener"),
|
||||
subs: map[int]*lifecycleEventChannels{},
|
||||
}
|
||||
return events
|
||||
}
|
||||
|
||||
// getUniqueClientId loops to generate & test integers for existence as key of map. returns an int that is not a key in map.
|
||||
func (re *AttestationEventReporter) getUniqueClientId() int {
|
||||
clientId := 0
|
||||
found := true
|
||||
for found {
|
||||
clientId = rand.Intn(maxClientId) //#nosec G404 The clientIds don't need to be unpredictable. They just need to be unique.
|
||||
_, found = re.subs[clientId]
|
||||
}
|
||||
return clientId
|
||||
}
|
||||
|
||||
func (re *AttestationEventReporter) Subscribe() *activeSubscription {
|
||||
re.mu.Lock()
|
||||
defer re.mu.Unlock()
|
||||
|
||||
clientId := re.getUniqueClientId()
|
||||
re.logger.Debug("Subscribe for client", zap.Int("clientId", clientId))
|
||||
channels := &lifecycleEventChannels{
|
||||
MessagePublicationC: make(chan *MessagePublication, 500),
|
||||
// TODO: This channel only needs to be this big due to Pythnet traffic. Once the explorer no longer reads these from bigtable, we can stop writing Pyth messages to this channel.
|
||||
VAAQuorumC: make(chan *vaa.VAA, 1000),
|
||||
}
|
||||
re.subs[clientId] = channels
|
||||
sub := &activeSubscription{ClientId: clientId, Channels: channels}
|
||||
return sub
|
||||
}
|
||||
|
||||
func (re *AttestationEventReporter) Unsubscribe(clientId int) {
|
||||
re.mu.Lock()
|
||||
defer re.mu.Unlock()
|
||||
|
||||
re.logger.Debug("Unsubscribe for client", zap.Int("clientId", clientId))
|
||||
delete(re.subs, clientId)
|
||||
}
|
||||
|
||||
// ReportMessagePublication is invoked when an on-chain message is observed.
|
||||
func (re *AttestationEventReporter) ReportMessagePublication(msg *MessagePublication) {
|
||||
re.mu.RLock()
|
||||
defer re.mu.RUnlock()
|
||||
|
||||
for client, sub := range re.subs {
|
||||
select {
|
||||
case sub.MessagePublicationC <- msg:
|
||||
re.logger.Debug("published MessagePublication to client", zap.Int("client", client))
|
||||
default:
|
||||
re.logger.Error("channel overflow when attempting to publish MessagePublication to client", zap.Int("client", client))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ReportVAAQuorum is invoked when quorum is reached.
|
||||
func (re *AttestationEventReporter) ReportVAAQuorum(msg *vaa.VAA) {
|
||||
re.mu.RLock()
|
||||
defer re.mu.RUnlock()
|
||||
|
||||
for client, sub := range re.subs {
|
||||
select {
|
||||
case sub.VAAQuorumC <- msg:
|
||||
re.logger.Debug("published VAAQuorum to client", zap.Int("client", client))
|
||||
default:
|
||||
re.logger.Error("channel overflow when attempting to publish VAAQuorum to client", zap.Int("client", client))
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,36 +0,0 @@
|
|||
package reporter
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestGetUniqueClientId(t *testing.T) {
|
||||
/*
|
||||
Rationale:
|
||||
Pro: This test does not have false positives. It is guaranteed to fail if the magic value for the maximum client ID changes.
|
||||
Con: It takes ca. 0.463s to run this test
|
||||
*/
|
||||
|
||||
var almostFullMap = make(map[int]*lifecycleEventChannels, maxClientId)
|
||||
|
||||
firstExpectedValue := 0
|
||||
secondExpectedValue := 1
|
||||
|
||||
// build a full map
|
||||
for i := 0; i < maxClientId; i++ {
|
||||
almostFullMap[i] = nil
|
||||
}
|
||||
|
||||
// Test that we can find the empty slot in the map
|
||||
delete(almostFullMap, firstExpectedValue)
|
||||
re := AttestationEventReporter{sync.RWMutex{}, nil, almostFullMap}
|
||||
assert.Equal(t, re.getUniqueClientId(), firstExpectedValue)
|
||||
|
||||
// Test that we can find a different empty slot in the map
|
||||
almostFullMap[firstExpectedValue] = nil
|
||||
delete(almostFullMap, secondExpectedValue)
|
||||
assert.Equal(t, re.getUniqueClientId(), secondExpectedValue)
|
||||
}
|
||||
|
|
@ -1,173 +0,0 @@
|
|||
package reporter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"github.com/certusone/wormhole/node/pkg/supervisor"
|
||||
"github.com/wormhole-foundation/wormhole/sdk/vaa"
|
||||
"go.uber.org/zap"
|
||||
"google.golang.org/api/option"
|
||||
|
||||
"cloud.google.com/go/bigtable"
|
||||
"cloud.google.com/go/pubsub"
|
||||
)
|
||||
|
||||
type BigTableConnectionConfig struct {
|
||||
GcpProjectID string
|
||||
GcpInstanceName string
|
||||
GcpKeyFilePath string
|
||||
TableName string
|
||||
TopicName string
|
||||
}
|
||||
|
||||
type bigTableWriter struct {
|
||||
connectionConfig *BigTableConnectionConfig
|
||||
events *AttestationEventReporter
|
||||
}
|
||||
|
||||
// rowKey returns a string with the input vales delimited by colons.
|
||||
func MakeRowKey(emitterChain vaa.ChainID, emitterAddress vaa.Address, sequence uint64) string {
|
||||
// left-pad the sequence with zeros to 16 characters, because bigtable keys are stored lexicographically
|
||||
return fmt.Sprintf("%d:%s:%016d", emitterChain, emitterAddress, sequence)
|
||||
}
|
||||
|
||||
func BigTableWriter(events *AttestationEventReporter, connectionConfig *BigTableConnectionConfig) func(ctx context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
|
||||
e := &bigTableWriter{events: events, connectionConfig: connectionConfig}
|
||||
|
||||
hostname, err := os.Hostname()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
errC := make(chan error)
|
||||
logger := supervisor.Logger(ctx)
|
||||
|
||||
client, err := bigtable.NewClient(ctx,
|
||||
e.connectionConfig.GcpProjectID,
|
||||
e.connectionConfig.GcpInstanceName,
|
||||
option.WithCredentialsFile(e.connectionConfig.GcpKeyFilePath))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create BigTable client: %w", err)
|
||||
}
|
||||
tbl := client.Open(e.connectionConfig.TableName)
|
||||
|
||||
pubsubClient, err := pubsub.NewClient(ctx,
|
||||
e.connectionConfig.GcpProjectID,
|
||||
option.WithCredentialsFile(e.connectionConfig.GcpKeyFilePath))
|
||||
if err != nil {
|
||||
logger.Error("failed to create GCP PubSub client", zap.Error(err))
|
||||
return fmt.Errorf("failed to create GCP PubSub client: %w", err)
|
||||
}
|
||||
logger.Info("GCP PubSub.NewClient initialized")
|
||||
|
||||
pubsubTopic := pubsubClient.Topic(e.connectionConfig.TopicName)
|
||||
logger.Info("GCP PubSub.Topic initialized",
|
||||
zap.String("Topic", e.connectionConfig.TopicName))
|
||||
// call to subscribe to event channels
|
||||
sub := e.events.Subscribe()
|
||||
logger.Info("subscribed to AttestationEvents")
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case msg := <-sub.Channels.MessagePublicationC:
|
||||
colFam := "MessagePublication"
|
||||
mutation := bigtable.NewMutation()
|
||||
ts := bigtable.Now()
|
||||
|
||||
mutation.Set(colFam, "Version", ts, []byte(fmt.Sprint(msg.VAA.Version)))
|
||||
mutation.Set(colFam, "GuardianSetIndex", ts, []byte(fmt.Sprint(msg.VAA.GuardianSetIndex)))
|
||||
mutation.Set(colFam, "Timestamp", ts, []byte(ts.Time().String()))
|
||||
mutation.Set(colFam, "Nonce", ts, []byte(fmt.Sprint(msg.VAA.Nonce)))
|
||||
mutation.Set(colFam, "EmitterChain", ts, []byte(msg.VAA.EmitterChain.String()))
|
||||
mutation.Set(colFam, "EmitterAddress", ts, []byte(msg.VAA.EmitterAddress.String()))
|
||||
mutation.Set(colFam, "Sequence", ts, []byte(fmt.Sprint(msg.VAA.Sequence)))
|
||||
mutation.Set(colFam, "InitiatingTxID", ts, []byte(msg.InitiatingTxID.Hex()))
|
||||
mutation.Set(colFam, "Payload", ts, msg.VAA.Payload)
|
||||
|
||||
mutation.Set(colFam, "ReporterHostname", ts, []byte(hostname))
|
||||
|
||||
// filter to see if there is a row with this rowKey, and has a value for EmitterAddress
|
||||
filter := bigtable.ChainFilters(
|
||||
bigtable.FamilyFilter(colFam),
|
||||
bigtable.ColumnFilter("EmitterAddress"))
|
||||
conditionalMutation := bigtable.NewCondMutation(filter, nil, mutation)
|
||||
|
||||
rowKey := MakeRowKey(msg.VAA.EmitterChain, msg.VAA.EmitterAddress, msg.VAA.Sequence)
|
||||
err := tbl.Apply(ctx, rowKey, conditionalMutation)
|
||||
if err != nil {
|
||||
logger.Error("Failed to write message publication to BigTable",
|
||||
zap.String("rowKey", rowKey),
|
||||
zap.String("columnFamily", colFam),
|
||||
zap.Error(err))
|
||||
errC <- err
|
||||
}
|
||||
case msg := <-sub.Channels.VAAQuorumC:
|
||||
colFam := "QuorumState"
|
||||
mutation := bigtable.NewMutation()
|
||||
ts := bigtable.Now()
|
||||
|
||||
b, marshalErr := msg.Marshal()
|
||||
if marshalErr != nil {
|
||||
logger.Error("failed to marshal VAAQuorum VAA.")
|
||||
}
|
||||
mutation.Set(colFam, "SignedVAA", ts, b)
|
||||
|
||||
// filter to see if this row already has the signature.
|
||||
filter := bigtable.ChainFilters(
|
||||
bigtable.FamilyFilter(colFam),
|
||||
bigtable.ColumnFilter("SignedVAA"))
|
||||
conditionalMutation := bigtable.NewCondMutation(filter, nil, mutation)
|
||||
|
||||
rowKey := MakeRowKey(msg.EmitterChain, msg.EmitterAddress, msg.Sequence)
|
||||
err := tbl.Apply(ctx, rowKey, conditionalMutation)
|
||||
if err != nil {
|
||||
logger.Error("Failed to write persistence info to BigTable",
|
||||
zap.String("rowKey", rowKey),
|
||||
zap.String("columnFamily", colFam),
|
||||
zap.Error(err))
|
||||
errC <- err
|
||||
}
|
||||
publishResult := pubsubTopic.Publish(ctx, &pubsub.Message{Data: b})
|
||||
if _, err = publishResult.Get(ctx); err != nil {
|
||||
logger.Error("Failed getting GCP PubSub publish reciept",
|
||||
zap.String("rowKey", rowKey),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
e.events.Unsubscribe(sub.ClientId)
|
||||
if err = client.Close(); err != nil {
|
||||
logger.Error("Could not close BigTable client", zap.Error(err))
|
||||
}
|
||||
if pubsubErr := pubsubClient.Close(); pubsubErr != nil {
|
||||
logger.Error("Could not close GCP PubSub client", zap.Error(pubsubErr))
|
||||
}
|
||||
return ctx.Err()
|
||||
case err := <-errC:
|
||||
logger.Error("bigtablewriter encountered an error", zap.Error(err))
|
||||
|
||||
e.events.Unsubscribe(sub.ClientId)
|
||||
|
||||
// try to close the connection before returning
|
||||
if closeErr := client.Close(); closeErr != nil {
|
||||
logger.Error("Could not close BigTable client", zap.Error(closeErr))
|
||||
}
|
||||
if pubsubErr := pubsubClient.Close(); pubsubErr != nil {
|
||||
logger.Error("Could not close GCP PubSub client", zap.Error(pubsubErr))
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue