fixes and cli integration for new options

Author: i-norden

cmd/geth/config.go

@@ -18,6 +18,7 @@ package main
 
 import (
 	"bufio"
+	"context"
 	"errors"
 	"fmt"
 	"math/big"
@@ -26,10 +27,7 @@ import (
 	"time"
 	"unicode"
 
-	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
-
-	"github.com/ethereum/go-ethereum/eth/downloader"
-	"github.com/ethereum/go-ethereum/statediff"
+	"github.com/naoina/toml"
 	"gopkg.in/urfave/cli.v1"
 
 	"github.com/ethereum/go-ethereum/accounts/external"
@@ -38,13 +36,18 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/usbwallet"
 	"github.com/ethereum/go-ethereum/cmd/utils"
 	"github.com/ethereum/go-ethereum/eth/catalyst"
+	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
 	"github.com/ethereum/go-ethereum/internal/ethapi"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/naoina/toml"
+	"github.com/ethereum/go-ethereum/statediff"
+	dumpdb "github.com/ethereum/go-ethereum/statediff/indexer/database/dump"
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
+	"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
+	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
 )
 
 var (
@@ -185,48 +188,82 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
 	}
 
 	if ctx.GlobalBool(utils.StateDiffFlag.Name) {
-		var dbConfig *sql.Config
+		var indexerConfig interfaces.Config
+		var clientName, nodeID string
 		if ctx.GlobalIsSet(utils.StateDiffWritingFlag.Name) {
-			dbConfig = new(sql.Config)
-			dbConfig.Hostname = ctx.GlobalString(utils.StateDiffDBHostFlag.Name)
-			dbConfig.Port = ctx.GlobalInt(utils.StateDiffDBPortFlag.Name)
-			dbConfig.DatabaseName = ctx.GlobalString(utils.StateDiffDBNameFlag.Name)
-			dbConfig.Username = ctx.GlobalString(utils.StateDiffDBUserFlag.Name)
-			dbConfig.Password = ctx.GlobalString(utils.StateDiffDBPasswordFlag.Name)
-
+			clientName = ctx.GlobalString(utils.StateDiffDBClientNameFlag.Name)
 			if ctx.GlobalIsSet(utils.StateDiffDBNodeIDFlag.Name) {
-				dbConfig.ID = ctx.GlobalString(utils.StateDiffDBNodeIDFlag.Name)
+				nodeID = ctx.GlobalString(utils.StateDiffDBNodeIDFlag.Name)
 			} else {
 				utils.Fatalf("Must specify node ID for statediff DB output")
 			}
 
-			if ctx.GlobalIsSet(utils.StateDiffDBClientNameFlag.Name) {
-				dbConfig.ClientName = ctx.GlobalString(utils.StateDiffDBClientNameFlag.Name)
-			} else {
-				utils.Fatalf("Must specify client name for statediff DB output")
-			}
-
-			if ctx.GlobalIsSet(utils.StateDiffDBMinConns.Name) {
-				dbConfig.MinConns = ctx.GlobalInt(utils.StateDiffDBMinConns.Name)
-			}
-			if ctx.GlobalIsSet(utils.StateDiffDBMaxConns.Name) {
-				dbConfig.MaxConns = ctx.GlobalInt(utils.StateDiffDBMaxConns.Name)
-			}
-			if ctx.GlobalIsSet(utils.StateDiffDBMaxIdleConns.Name) {
-				dbConfig.MaxIdle = ctx.GlobalInt(utils.StateDiffDBMaxIdleConns.Name)
-			}
-			if ctx.GlobalIsSet(utils.StateDiffDBMaxConnLifetime.Name) {
-				dbConfig.MaxConnLifetime = ctx.GlobalDuration(utils.StateDiffDBMaxConnLifetime.Name) * time.Second
-			}
-			if ctx.GlobalIsSet(utils.StateDiffDBMaxConnIdleTime.Name) {
-				dbConfig.MaxConnIdleTime = ctx.GlobalDuration(utils.StateDiffDBMaxConnIdleTime.Name) * time.Second
+			dbTypeStr := ctx.GlobalString(utils.StateDiffDBTypeFlag.Name)
+			dbType, err := shared.ResolveDBType(dbTypeStr)
+			if err != nil {
+				utils.Fatalf("%v", err)
 			}
-			if ctx.GlobalIsSet(utils.StateDiffDBConnTimeout.Name) {
-				dbConfig.ConnTimeout = ctx.GlobalDuration(utils.StateDiffDBConnTimeout.Name) * time.Second
+			switch dbType {
+			case shared.POSTGRES:
+				driverTypeStr := ctx.GlobalString(utils.StateDiffDBDriverTypeFlag.Name)
+				driverType, err := postgres.ResolveDriverType(driverTypeStr)
+				if err != nil {
+					utils.Fatalf("%v", err)
+				}
+				pgConfig := postgres.Config{
+					Hostname:     ctx.GlobalString(utils.StateDiffDBHostFlag.Name),
+					Port:         ctx.GlobalInt(utils.StateDiffDBPortFlag.Name),
+					DatabaseName: ctx.GlobalString(utils.StateDiffDBNameFlag.Name),
+					Username:     ctx.GlobalString(utils.StateDiffDBUserFlag.Name),
+					Password:     ctx.GlobalString(utils.StateDiffDBPasswordFlag.Name),
+					ID:           nodeID,
+					ClientName:   clientName,
+					Driver:       driverType,
+				}
+				if ctx.GlobalIsSet(utils.StateDiffDBMinConns.Name) {
+					pgConfig.MinConns = ctx.GlobalInt(utils.StateDiffDBMinConns.Name)
+				}
+				if ctx.GlobalIsSet(utils.StateDiffDBMaxConns.Name) {
+					pgConfig.MaxConns = ctx.GlobalInt(utils.StateDiffDBMaxConns.Name)
+				}
+				if ctx.GlobalIsSet(utils.StateDiffDBMaxIdleConns.Name) {
+					pgConfig.MaxIdle = ctx.GlobalInt(utils.StateDiffDBMaxIdleConns.Name)
+				}
+				if ctx.GlobalIsSet(utils.StateDiffDBMaxConnLifetime.Name) {
+					pgConfig.MaxConnLifetime = ctx.GlobalDuration(utils.StateDiffDBMaxConnLifetime.Name) * time.Second
+				}
+				if ctx.GlobalIsSet(utils.StateDiffDBMaxConnIdleTime.Name) {
+					pgConfig.MaxConnIdleTime = ctx.GlobalDuration(utils.StateDiffDBMaxConnIdleTime.Name) * time.Second
+				}
+				if ctx.GlobalIsSet(utils.StateDiffDBConnTimeout.Name) {
+					pgConfig.ConnTimeout = ctx.GlobalDuration(utils.StateDiffDBConnTimeout.Name) * time.Second
+				}
+				indexerConfig = pgConfig
+			case shared.DUMP:
+				dumpTypeStr := ctx.GlobalString(utils.StateDiffDBDumpDst.Name)
+				dumpType, err := dumpdb.ResolveDumpType(dumpTypeStr)
+				if err != nil {
+					utils.Fatalf("%v", err)
+				}
+				switch dumpType {
+				case dumpdb.STDERR:
+					indexerConfig = dumpdb.Config{Dump: os.Stdout}
+				case dumpdb.STDOUT:
+					indexerConfig = dumpdb.Config{Dump: os.Stderr}
+				case dumpdb.DISCARD:
+					indexerConfig = dumpdb.Config{Dump: dumpdb.NewDiscardWriterCloser()}
+				default:
+					utils.Fatalf("unrecognized dump destination: %s", dumpType)
+				}
+			default:
+				utils.Fatalf("unrecognized database type: %s", dbType)
 			}
 		}
-		p := statediff.ServiceParams{
-			DBParams:        dbConfig,
+		p := statediff.Config{
+			IndexerConfig:   indexerConfig,
+			ID:              nodeID,
+			ClientName:      clientName,
+			Context:         context.Background(),
 			EnableWriteLoop: ctx.GlobalBool(utils.StateDiffWritingFlag.Name),
 			NumWorkers:      ctx.GlobalUint(utils.StateDiffWorkersFlag.Name),
 		}

cmd/geth/main.go

@@ -149,6 +149,9 @@ var (
 		utils.GpoIgnoreGasPriceFlag,
 		utils.MinerNotifyFullFlag,
 		utils.StateDiffFlag,
+		utils.StateDiffDBTypeFlag,
+		utils.StateDiffDBDriverTypeFlag,
+		utils.StateDiffDBDumpDst,
 		utils.StateDiffDBNameFlag,
 		utils.StateDiffDBPasswordFlag,
 		utils.StateDiffDBUserFlag,

cmd/geth/usage.go

@@ -225,6 +225,9 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 		Name: "STATE DIFF",
 		Flags: []cli.Flag{
 			utils.StateDiffFlag,
+			utils.StateDiffDBTypeFlag,
+			utils.StateDiffDBDriverTypeFlag,
+			utils.StateDiffDBDumpDst,
 			utils.StateDiffDBNameFlag,
 			utils.StateDiffDBPasswordFlag,
 			utils.StateDiffDBUserFlag,

cmd/utils/flags.go

@@ -786,6 +786,21 @@ var (
 		Name:  "statediff",
 		Usage: "Enables the processing of state diffs between each block",
 	}
+	StateDiffDBTypeFlag = cli.StringFlag{
+		Name:  "statediff.db.type",
+		Usage: "Statediff database type",
+		Value: "postgres",
+	}
+	StateDiffDBDriverTypeFlag = cli.StringFlag{
+		Name:  "statediff.db.driver",
+		Usage: "Statediff database driver type",
+		Value: "pgx",
+	}
+	StateDiffDBDumpDst = cli.StringFlag{
+		Name:  "statediff.dump.dst",
+		Usage: "Statediff database dump destination (default is stdout)",
+		Value: "stdout",
+	}
 	StateDiffDBHostFlag = cli.StringFlag{
 		Name:  "statediff.db.host",
 		Usage: "Statediff database hostname/ip",
@@ -840,14 +855,16 @@ var (
 	StateDiffDBClientNameFlag = cli.StringFlag{
 		Name:  "statediff.db.clientname",
 		Usage: "Client name to use when writing state diffs to database",
+		Value: "go-ethereum",
 	}
 	StateDiffWritingFlag = cli.BoolFlag{
 		Name:  "statediff.writing",
 		Usage: "Activates progressive writing of state diffs to database as new block are synced",
 	}
 	StateDiffWorkersFlag = cli.UintFlag{
 		Name:  "statediff.workers",
-		Usage: "Number of concurrent workers to use during statediff processing (0 = 1)",
+		Usage: "Number of concurrent workers to use during statediff processing (default 1)",
+		Value: 1,
 	}
 )
 
@@ -1804,7 +1821,7 @@ func RegisterGraphQLService(stack *node.Node, backend ethapi.Backend, cfg node.C
 }
 
 // RegisterStateDiffService configures and registers a service to stream state diff data over RPC
-func RegisterStateDiffService(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params statediff.ServiceParams) {
+func RegisterStateDiffService(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params statediff.Config) {
 	if err := statediff.New(stack, ethServ, cfg, params); err != nil {
 		Fatalf("Failed to register the Statediff service: %v", err)
 	}

statediff/README.md

@@ -79,6 +79,9 @@ This service introduces a CLI flag namespace `statediff`
 `--statediff` flag is used to turn on the service
 `--statediff.writing` is used to tell the service to write state diff objects it produces from synced ChainEvents directly to a configured Postgres database
 `--statediff.workers` is used to set the number of concurrent workers to process state diff objects and write them into the database
+`--statediff.db.type` is the type of database we write out to (current options: postgres and dump)
+`--statediff.dump.dst` is the destination to write to when operating in database dump mode (stdout, stderr, discard)
+`--statediff.db.driver` is the specific driver to use for the database (current options for postgres: pgx and sqlx)
 `--statediff.db.host` is the hostname/ip to dial to connect to the database
 `--statediff.db.port` is the port to dial to connect to the database
 `--statediff.db.name` is the name of the database to connect to

statediff/indexer/database/dump/batch_tx.go

@@ -30,21 +30,22 @@ import (
 
 // BatchTx wraps a void with the state necessary for building the tx concurrently during trie difference iteration
 type BatchTx struct {
-	dump      io.Writer
-	quit      chan struct{}
-	iplds     chan models.IPLDModel
-	ipldCache models.IPLDBatch
+	BlockNumber uint64
+	dump        io.Writer
+	quit        chan struct{}
+	iplds       chan models.IPLDModel
+	ipldCache   models.IPLDBatch
 
-	close func(blockTx *BatchTx, err error) error
+	submit func(blockTx *BatchTx, err error) error
 }
 
 // Submit satisfies indexer.AtomicTx
 func (tx *BatchTx) Submit(err error) error {
-	return tx.close(tx, err)
+	return tx.submit(tx, err)
 }
 
 func (tx *BatchTx) flush() error {
-	if _, err := fmt.Fprintf(tx.dump, "%+v", tx.ipldCache); err != nil {
+	if _, err := fmt.Fprintf(tx.dump, "%+v\r\n", tx.ipldCache); err != nil {
 		return err
 	}
 	tx.ipldCache = models.IPLDBatch{}

statediff/indexer/database/dump/config.go

@@ -17,15 +17,63 @@
 package dump
 
 import (
+	"fmt"
 	"io"
+	"strings"
 
 	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
 )
 
+// DumpType to explicitly type the dump destination
+type DumpType string
+
+const (
+	STDOUT  = "Stdout"
+	STDERR  = "Stderr"
+	DISCARD = "Discard"
+	UNKNOWN = "Unknown"
+)
+
+// ResolveDumpType resolves the dump type for the provided string
+func ResolveDumpType(str string) (DumpType, error) {
+	switch strings.ToLower(str) {
+	case "stdout", "out", "std out":
+		return STDOUT, nil
+	case "stderr", "err", "std err":
+		return STDERR, nil
+	case "discard", "void", "devnull", "dev null":
+		return DISCARD, nil
+	default:
+		return UNKNOWN, fmt.Errorf("unrecognized dump type: %s", str)
+	}
+}
+
+// Config for data dump
 type Config struct {
 	Dump io.WriteCloser
 }
 
+// Type satisfies interfaces.Config
 func (c Config) Type() shared.DBType {
 	return shared.DUMP
 }
+
+// NewDiscardWriterCloser returns a discardWrapper wrapping io.Discard
+func NewDiscardWriterCloser() io.WriteCloser {
+	return discardWrapper{blackhole: io.Discard}
+}
+
+// discardWrapper wraps io.Discard with io.Closer
+type discardWrapper struct {
+	blackhole io.Writer
+}
+
+// Write satisfies io.Writer
+func (dw discardWrapper) Write(b []byte) (int, error) {
+	return dw.blackhole.Write(b)
+}
+
+// Close satisfies io.Closer
+func (dw discardWrapper) Close() error {
+	return nil
+}