feat: add teleportr driver

go/teleportr/cmd/teleportr/main.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/urfave/cli"
 
+	"github.com/ethereum-optimism/optimism/go/teleportr"
 	"github.com/ethereum-optimism/optimism/go/teleportr/flags"
 )
 
@@ -33,9 +34,7 @@ func main() {
 	app.Usage = "Teleportr"
 	app.Description = "Teleportr bridge from L1 to L2"
 
-	app.Action = func(_ *cli.Context) error {
-		return nil
-	}
+	app.Action = teleportr.Main(GitVersion)
 	err := app.Run(os.Args)
 	if err != nil {
 		log.Crit("Application failed", "message", err)

go/teleportr/drivers/disburser/metrics.go

+package disburser
+
+import (
+	"github.com/ethereum-optimism/optimism/go/bss-core/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const methodLabel = "method"
+
+var (
+	// DBMethodUpsertDeposits is a label for UpsertDeposits db method.
+	DBMethodUpsertDeposits = prometheus.Labels{methodLabel: "upsert_deposits"}
+
+	// DBMethodConfirmedDeposits is a label for ConfirmedDeposits db method.
+	DBMethodConfirmedDeposits = prometheus.Labels{methodLabel: "confirmed_deposits"}
+
+	// DBMethodLastProcessedBlock is a label for LastProcessedBlock db method.
+	DBMethodLastProcessedBlock = prometheus.Labels{methodLabel: "last_processed_block"}
+
+	// DBMethodUpsertPendingTx is a label for UpsertPendingTx db method.
+	DBMethodUpsertPendingTx = prometheus.Labels{methodLabel: "upsert_pending_tx"}
+
+	// DBMethodListPendingTxs is a label for ListPendingTxs db method.
+	DBMethodListPendingTxs = prometheus.Labels{methodLabel: "list_pending_txs"}
+
+	// DBMethodUpsertDisbursement is a label for UpsertDisbursement db method.
+	DBMethodUpsertDisbursement = prometheus.Labels{methodLabel: "upsert_disbursement"}
+
+	// DBMethodLatestDisbursementID is a label for LatestDisbursementID db method.
+	DBMethodLatestDisbursementID = prometheus.Labels{methodLabel: "latest_disbursement_id"}
+
+	// DBMethodDeletePendingTx is a label for DeletePendingTx db method.
+	DBMethodDeletePendingTx = prometheus.Labels{methodLabel: "delete_pending_tx"}
+)
+
+// Metrics extends the BSS core metrics with additional metrics tracked by the
+// sequencer driver.
+type Metrics struct {
+	*metrics.Base
+
+	// FailedDatabaseMethods tracks the number of database failures for each
+	// known database method.
+	FailedDatabaseMethods *prometheus.GaugeVec
+
+	// DepositIDMismatch tracks whether or not our database is in sync with the
+	// disrburser contract. 1 means in sync, 0 means out of sync.
+	DepositIDMismatch prometheus.Gauge
+
+	// MissingDisbursements tracks the number of deposits that are missing
+	// disbursement below our supposed next deposit id.
+	MissingDisbursements prometheus.Gauge
+
+	// SuccessfulDisbursements tracks the number of disbursements that emit a
+	// success event from a given tx.
+	SuccessfulDisbursements prometheus.Gauge
+
+	// FailedDisbursements tracks the number of disbursements that emit a failed
+	// event from a given tx.
+	FailedDisbursements prometheus.Gauge
+
+	// PostgresLastDisbursedID tracks the latest disbursement id in postgres.
+	PostgresLastDisbursedID prometheus.Gauge
+
+	// ContractNextDisbursementID tracks the next disbursement id expected by
+	// the disburser contract.
+	ContractNextDisbursementID prometheus.Gauge
+}
+
+// NewMetrics initializes a new, extended metrics object.
+func NewMetrics(subsystem string) *Metrics {
+	base := metrics.NewBase(subsystem, "")
+	return &Metrics{
+		Base: base,
+		FailedDatabaseMethods: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Name:      "failed_database_operations",
+			Help:      "Tracks the number of database failures",
+			Subsystem: base.SubsystemName(),
+		}, []string{methodLabel}),
+		DepositIDMismatch: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "deposit_id_mismatch",
+			Help: "Set to 1 when the postgres and the disrburser contract " +
+				"disagree on the next deposit id, and 0 otherwise",
+			Subsystem: base.SubsystemName(),
+		}),
+		MissingDisbursements: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "missing_disbursements",
+			Help: "Number of deposits that are missing disbursements in " +
+				"postgres below our supposed next deposit id",
+			Subsystem: base.SubsystemName(),
+		}),
+		SuccessfulDisbursements: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "successful_disbursements",
+			Help: "Number of disbursements that emit a success event " +
+				"from a given tx",
+			Subsystem: base.SubsystemName(),
+		}),
+		FailedDisbursements: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "failed_disbursements",
+			Help: "Number of disbursements that emit a failed event " +
+				"from a given tx",
+			Subsystem: base.SubsystemName(),
+		}),
+		PostgresLastDisbursedID: promauto.NewGauge(prometheus.GaugeOpts{
+			Name:      "postgres_last_disbursed_id",
+			Help:      "Latest recorded disbursement id in postgres",
+			Subsystem: base.SubsystemName(),
+		}),
+		ContractNextDisbursementID: promauto.NewGauge(prometheus.GaugeOpts{
+			Name:      "contract_next_disbursement_id",
+			Help:      "Next disbursement id expected by the disburser contract",
+			Subsystem: base.SubsystemName(),
+		}),
+	}
+}

go/teleportr/go.mod

@@ -3,16 +3,60 @@ module github.com/ethereum-optimism/optimism/go/teleportr
 go 1.17
 
 require (
+	github.com/ethereum-optimism/optimism/go/bss-core v0.0.0-20220218171106-67a0414d7606
 	github.com/ethereum/go-ethereum v1.10.15
 	github.com/google/uuid v1.3.0
 	github.com/lib/pq v1.10.4
 	github.com/stretchr/testify v1.7.0
+	github.com/urfave/cli v1.22.5
 )
 
 require (
+	github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 // indirect
+	github.com/VictoriaMetrics/fastcache v1.6.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/btcsuite/btcd v0.22.0-beta // indirect
+	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/deckarep/golang-set v0.0.0-20180603214616-504e848d77ea // indirect
+	github.com/decred/base58 v1.0.3 // indirect
+	github.com/decred/dcrd/crypto/blake256 v1.0.0 // indirect
+	github.com/decred/dcrd/crypto/ripemd160 v1.0.1 // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v3 v3.0.0 // indirect
+	github.com/decred/dcrd/hdkeychain/v3 v3.0.0 // indirect
+	github.com/getsentry/sentry-go v0.11.0 // indirect
+	github.com/go-ole/go-ole v1.2.1 // indirect
+	github.com/go-stack/stack v1.8.0 // indirect
+	github.com/golang/protobuf v1.4.3 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/gorilla/websocket v1.4.2 // indirect
+	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
+	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
+	github.com/holiman/uint256 v1.2.0 // indirect
+	github.com/mattn/go-runewidth v0.0.9 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/olekukonko/tablewriter v0.0.5 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v1.11.0 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.26.0 // indirect
+	github.com/prometheus/procfs v0.6.0 // indirect
+	github.com/prometheus/tsdb v0.7.1 // indirect
+	github.com/rjeczalik/notify v0.9.1 // indirect
+	github.com/russross/blackfriday/v2 v2.0.1 // indirect
+	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect
+	github.com/shurcooL/sanitized_anchor_name v1.0.0 // indirect
+	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect
+	github.com/tklauser/go-sysconf v0.3.5 // indirect
+	github.com/tklauser/numcpus v0.2.2 // indirect
+	github.com/tyler-smith/go-bip39 v1.0.1-0.20181017060643-dbb3b84ba2ef // indirect
 	golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2 // indirect
 	golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912 // indirect
+	google.golang.org/protobuf v1.26.0-rc.1 // indirect
+	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )
+
+replace github.com/ethereum-optimism/optimism/go/bss-core => ../bss-core

go/teleportr/teleportr.go

+package teleportr
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	bsscore "github.com/ethereum-optimism/optimism/go/bss-core"
+	"github.com/ethereum-optimism/optimism/go/bss-core/dial"
+	"github.com/ethereum-optimism/optimism/go/bss-core/metrics"
+	"github.com/ethereum-optimism/optimism/go/bss-core/txmgr"
+	"github.com/ethereum-optimism/optimism/go/teleportr/db"
+	"github.com/ethereum-optimism/optimism/go/teleportr/drivers/disburser"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/urfave/cli"
+)
+
+func Main(gitVersion string) func(ctx *cli.Context) error {
+	return func(cliCtx *cli.Context) error {
+		cfg, err := NewConfig(cliCtx)
+		if err != nil {
+			return err
+		}
+
+		log.Info("Initializing teleportr")
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		var logHandler log.Handler
+		if cfg.LogTerminal {
+			logHandler = log.StreamHandler(os.Stdout, log.TerminalFormat(true))
+		} else {
+			logHandler = log.StreamHandler(os.Stdout, log.JSONFormat())
+		}
+
+		logLevel, err := log.LvlFromString(cfg.LogLevel)
+		if err != nil {
+			return err
+		}
+
+		log.Root().SetHandler(log.LvlFilterHandler(logLevel, logHandler))
+
+		disburserPrivKey, disburserAddr, err := bsscore.ParseWalletPrivKeyAndContractAddr(
+			"Teleportr", cfg.Mnemonic, cfg.DisburserHDPath,
+			cfg.DisburserPrivKey, cfg.DisburserAddress,
+		)
+		if err != nil {
+			return err
+		}
+
+		depositAddr, err := bsscore.ParseAddress(cfg.DepositAddress)
+		if err != nil {
+			return err
+		}
+
+		l1Client, err := dial.L1EthClientWithTimeout(ctx, cfg.L1EthRpc, cfg.DisableHTTP2)
+		if err != nil {
+			return err
+		}
+		defer l1Client.Close()
+
+		l2Client, err := dial.L1EthClientWithTimeout(ctx, cfg.L2EthRpc, cfg.DisableHTTP2)
+		if err != nil {
+			return err
+		}
+		defer l2Client.Close()
+
+		database, err := db.Open(db.Config{
+			Host:      cfg.PostgresHost,
+			Port:      uint16(cfg.PostgresPort),
+			User:      cfg.PostgresUser,
+			Password:  cfg.PostgresPassword,
+			DBName:    cfg.PostgresDBName,
+			EnableSSL: cfg.PostgresEnableSSL,
+		})
+		if err != nil {
+			return err
+		}
+		defer database.Close()
+
+		if cfg.MetricsServerEnable {
+			go metrics.RunServer(cfg.MetricsHostname, cfg.MetricsPort)
+		}
+
+		chainID, err := l1Client.ChainID(ctx)
+		if err != nil {
+			return err
+		}
+
+		txManagerConfig := txmgr.Config{
+			ResubmissionTimeout:       cfg.ResubmissionTimeout,
+			ReceiptQueryInterval:      time.Second,
+			NumConfirmations:          1, // L2 insta confs
+			SafeAbortNonceTooLowCount: cfg.SafeAbortNonceTooLowCount,
+		}
+
+		teleportrDriver, err := disburser.NewDriver(disburser.Config{
+			Name:                 "Teleportr",
+			L1Client:             l1Client,
+			L2Client:             l2Client,
+			Database:             database,
+			MaxTxSize:            cfg.MaxL2TxSize,
+			NumConfirmations:     cfg.NumDepositConfirmations,
+			DeployBlockNumber:    cfg.DepositDeployBlockNumber,
+			FilterQueryMaxBlocks: cfg.FilterQueryMaxBlocks,
+			DepositAddr:          depositAddr,
+			DisburserAddr:        disburserAddr,
+			ChainID:              chainID,
+			PrivKey:              disburserPrivKey,
+		})
+		if err != nil {
+			return err
+		}
+
+		teleportrService := bsscore.NewService(bsscore.ServiceConfig{
+			Context:         ctx,
+			Driver:          teleportrDriver,
+			PollInterval:    cfg.PollInterval,
+			ClearPendingTx:  false,
+			L1Client:        l1Client,
+			TxManagerConfig: txManagerConfig,
+		})
+
+		services := []*bsscore.Service{teleportrService}
+		teleportr, err := bsscore.NewBatchSubmitter(ctx, cancel, services)
+		if err != nil {
+			return err
+		}
+
+		log.Info("Starting teleportr")
+
+		err = teleportr.Start()
+		if err != nil {
+			return err
+		}
+		defer teleportr.Stop()
+
+		log.Info("Teleportr started")
+
+		interruptChannel := make(chan os.Signal, 1)
+		signal.Notify(interruptChannel, []os.Signal{
+			os.Interrupt,
+			os.Kill,
+			syscall.SIGTERM,
+			syscall.SIGQUIT,
+		}...)
+		<-interruptChannel
+
+		return nil
+	}
+}