Merge pull request #1452 from hyperledger/eth-consumer

Add continuity test to vent

Author: Casey Kuhlman

Makefile

@@ -169,7 +169,7 @@ build_burrow_debug:
 .PHONY: install
 install: build_burrow
 	mkdir -p ${BIN_PATH}
-	install -T ${REPO}/bin/burrow ${BIN_PATH}/burrow
+	install ${REPO}/bin/burrow ${BIN_PATH}/burrow
 
 # build burrow with checks for race conditions
 .PHONY: build_race_db
@@ -262,9 +262,10 @@ start_ganache: $(PID_DIR)/ganache.pid
 stop_ganache: $(PID_DIR)/ganache.pid
 	@kill $(shell cat $<) && echo "Ganache process stopped." && rm $< || rm $<
 
+# For local debug
 .PHONY: postgres
 postgres:
-	docker-compose up
+	docker run -e POSTGRES_HOST_AUTH_METHOD=trust -p 5432:5432 postgres:11-alpine
 
 .PHONY: test_restore
 test_restore:

cmd/burrow/commands/dump.go

@@ -71,7 +71,7 @@ func Dump(output Output) func(cmd *cli.Cmd) {
 				}
 
 				// Include all logging by default
-				logger, err := logconfig.New().NewLogger()
+				logger, err := logconfig.New().Logger()
 				if err != nil {
 					output.Fatalf("could not make logger: %v", err)
 				}

cmd/burrow/commands/vent.go

@@ -21,6 +21,26 @@ import (
 	cli "github.com/jawher/mow.cli"
 )
 
+type LogLevel string
+
+const (
+	LogLevelNone  LogLevel = "none"
+	LogLevelInfo  LogLevel = "info"
+	LogLevelTrace LogLevel = "trace"
+)
+
+func logConfig(level LogLevel) *logconfig.LoggingConfig {
+	logConf := logconfig.New()
+	switch level {
+	case LogLevelNone:
+		return logConf.None()
+	case LogLevelTrace:
+		return logConf.WithTrace()
+	default:
+		return logConf
+	}
+}
+
 // Vent consumes EVM events and commits to a DB
 func Vent(output Output) func(cmd *cli.Cmd) {
 	return func(cmd *cli.Cmd) {
@@ -32,9 +52,14 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 				dbOpts := sqlDBOpts(cmd, cfg)
 				grpcAddrOpt := cmd.StringOpt("chain-addr", cfg.ChainAddress, "Address to connect to the Hyperledger Burrow gRPC server")
 				httpAddrOpt := cmd.StringOpt("http-addr", cfg.HTTPListenAddress, "Address to bind the HTTP server")
-				logLevelOpt := cmd.StringOpt("log-level", cfg.LogLevel, "Logging level (error, warn, info, debug)")
+				logLevelOpt := cmd.StringOpt("log-level", string(LogLevelInfo), "Logging level (none, info, trace)")
 				watchAddressesOpt := cmd.StringsOpt("watch", nil, "Add contract address to global watch filter")
 				minimumHeightOpt := cmd.IntOpt("minimum-height", 0, "Only process block greater than or equal to height passed")
+				maxRetriesOpt := cmd.IntOpt("max-retries", int(cfg.BlockConsumerConfig.MaxRetries), "Maximum number of retries when consuming blocks")
+				backoffDurationOpt := cmd.StringOpt("backoff", "",
+					"The minimum duration to wait before asking for new blocks - increases exponentially when errors occur. Values like 200ms, 1s, 2m")
+				batchSizeOpt := cmd.IntOpt("batch-size", int(cfg.BlockConsumerConfig.MaxBlockBatchSize),
+					"The maximum number of blocks from which to request events in a single call - will reduce logarithmically to 1 when errors occur.")
 				abiFileOpt := cmd.StringsOpt("abi", cfg.AbiFileOrDirs, "EVM Contract ABI file or folder")
 				specFileOrDirOpt := cmd.StringsOpt("spec", cfg.SpecFileOrDirs, "SQLSol specification file or folder")
 				dbBlockOpt := cmd.BoolOpt("blocks", false, "Create block tables and persist related data")
@@ -43,16 +68,21 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 				announceEveryOpt := cmd.StringOpt("announce-every", "5s", "Announce vent status every period as a Go duration, e.g. 1ms, 3s, 1h")
 
 				cmd.Before = func() {
+					var err error
 					// Rather annoying boilerplate here... but there is no way to pass mow.cli a pointer for it to fill you value
 					cfg.DBAdapter = *dbOpts.adapter
 					cfg.DBURL = *dbOpts.url
 					cfg.DBSchema = *dbOpts.schema
 					cfg.ChainAddress = *grpcAddrOpt
 					cfg.HTTPListenAddress = *httpAddrOpt
-					cfg.LogLevel = *logLevelOpt
 					cfg.WatchAddresses = make([]crypto.Address, len(*watchAddressesOpt))
 					cfg.MinimumHeight = uint64(*minimumHeightOpt)
-					var err error
+					cfg.BlockConsumerConfig.MaxRetries = uint64(*maxRetriesOpt)
+					cfg.BlockConsumerConfig.BaseBackoffDuration, err = parseDuration(*backoffDurationOpt)
+					if err != nil {
+						output.Fatalf("could not parse backoff duration: %w", err)
+					}
+					cfg.BlockConsumerConfig.MaxBlockBatchSize = uint64(*batchSizeOpt)
 					for i, wa := range *watchAddressesOpt {
 						cfg.WatchAddresses[i], err = crypto.AddressFromHexString(wa)
 						if err != nil {
@@ -68,29 +98,31 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 						cfg.SpecOpt |= sqlsol.Tx
 					}
 
-					if *announceEveryOpt != "" {
-						var err error
-						cfg.AnnounceEvery, err = time.ParseDuration(*announceEveryOpt)
-						if err != nil {
-							output.Fatalf("could not parse announce-every duration %s: %v", *announceEveryOpt, err)
-						}
+					cfg.AnnounceEvery, err = parseDuration(*announceEveryOpt)
+					if err != nil {
+						output.Fatalf("could not parse announce-every duration %s: %v", *announceEveryOpt, err)
 					}
 				}
 
 				cmd.Spec = "--spec=<spec file or dir>... [--abi=<abi file or dir>...] " +
 					"[--watch=<contract address>...] [--minimum-height=<lowest height from which to read>] " +
+					"[--max-retries=<max block request retries>] [--backoff=<minimum backoff duration>] " +
+					"[--batch-size=<minimum block batch size>] " +
 					"[--db-adapter] [--db-url] [--db-schema] [--blocks] [--txs] [--chain-addr] [--http-addr] " +
 					"[--log-level] [--announce-every=<duration>]"
 
 				cmd.Action = func() {
-					log, err := logconfig.New().NewLogger()
+					logger, err := logConfig(LogLevel(*logLevelOpt)).Logger()
 					if err != nil {
 						output.Fatalf("failed to load logger: %v", err)
 					}
 
-					log = log.With("service", "vent")
-					consumer := service.NewConsumer(cfg, log, make(chan types.EventData))
-					server := service.NewServer(cfg, log, consumer)
+					logger = logger.With("service", "vent")
+					consumer := service.NewConsumer(cfg, logger, make(chan types.EventData))
+					if err != nil {
+						output.Fatalf("Could not create Vent Consumer: %v", err)
+					}
+					server := service.NewServer(cfg, logger, consumer)
 
 					projection, err := sqlsol.SpecLoader(cfg.SpecFileOrDirs, cfg.SpecOpt)
 					if err != nil {
@@ -194,7 +226,7 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 				}
 
 				cmd.Action = func() {
-					log, err := logconfig.New().NewLogger()
+					log, err := logconfig.New().Logger()
 					if err != nil {
 						output.Fatalf("failed to load logger: %v", err)
 					}
@@ -230,6 +262,13 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 	}
 }
 
+func parseDuration(duration string) (time.Duration, error) {
+	if duration == "" {
+		return 0, nil
+	}
+	return time.ParseDuration(duration)
+}
+
 type dbOpts struct {
 	adapter *string
 	url     *string

core/config.go

@@ -34,7 +34,7 @@ func (kern *Kernel) LoadKeysFromConfig(conf *keys.KeysConfig) (err error) {
 
 // LoadLoggerFromConfig adds a logging configuration to the kernel
 func (kern *Kernel) LoadLoggerFromConfig(conf *logconfig.LoggingConfig) error {
-	logger, err := conf.NewLogger()
+	logger, err := conf.Logger()
 	kern.SetLogger(logger)
 	return err
 }

crypto/signature_test.go

@@ -0,0 +1,16 @@
+package crypto
+
+import (
+	"fmt"
+	"math/big"
+	"testing"
+)
+
+func TestGetEthChainID(t *testing.T) {
+	chainIDString := "BurrowChain_FAB3C1-AB0FD1"
+	chainID := GetEthChainID(chainIDString)
+	b := new(big.Int).SetBytes([]byte(chainIDString))
+	fmt.Println(b)
+	fmt.Println(chainID)
+	fmt.Printf("%X", chainID.Bytes())
+}

logging/logconfig/config.go

@@ -43,6 +43,16 @@ func New() *LoggingConfig {
 	}
 }
 
+func (lc *LoggingConfig) WithTrace() *LoggingConfig {
+	lc.Trace = true
+	return lc
+}
+
+func (lc *LoggingConfig) None() *LoggingConfig {
+	lc.RootSink = nil
+	return lc
+}
+
 func (lc *LoggingConfig) Root(configure func(sink *SinkConfig) *SinkConfig) *LoggingConfig {
 	lc.RootSink = configure(Sink())
 	return lc
@@ -65,8 +75,16 @@ func (lc *LoggingConfig) JSONString() string {
 	return JSONString(lc)
 }
 
+func (lc *LoggingConfig) MustLogger() *logging.Logger {
+	logger, err := lc.Logger()
+	if err != nil {
+		panic(err)
+	}
+	return logger
+}
+
 // Obtain a logger from this LoggingConfig
-func (lc *LoggingConfig) NewLogger() (*logging.Logger, error) {
+func (lc *LoggingConfig) Logger() (*logging.Logger, error) {
 	outputLogger, errCh, err := newLogger(lc)
 	if err != nil {
 		return nil, err

logging/logconfig/sinks.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/eapache/channels"
 	"github.com/go-kit/kit/log"
+	"github.com/hyperledger/burrow/logging"
 	"github.com/hyperledger/burrow/logging/loggers"
 	"github.com/hyperledger/burrow/logging/structure"
 )
@@ -159,19 +160,42 @@ func Sink() *SinkConfig {
 	return &SinkConfig{}
 }
 
-func (sinkConfig *SinkConfig) AddSinks(sinks ...*SinkConfig) *SinkConfig {
-	sinkConfig.Sinks = append(sinkConfig.Sinks, sinks...)
-	return sinkConfig
+func (sc *SinkConfig) MustLogger() *logging.Logger {
+	return sc.LoggingConfig().MustLogger()
 }
 
-func (sinkConfig *SinkConfig) SetTransform(transform *TransformConfig) *SinkConfig {
-	sinkConfig.Transform = transform
-	return sinkConfig
+func (sc *SinkConfig) Logger() (*logging.Logger, error) {
+	return sc.LoggingConfig().Logger()
 }
 
-func (sinkConfig *SinkConfig) SetOutput(output *OutputConfig) *SinkConfig {
-	sinkConfig.Output = output
-	return sinkConfig
+// Wrap this sink as RootSink of LoggingCOnfig
+func (sc *SinkConfig) LoggingConfig() *LoggingConfig {
+	lc := New()
+	lc.RootSink = sc
+	return lc
+}
+
+func (sc *SinkConfig) AddSinks(sinks ...*SinkConfig) *SinkConfig {
+	sc.Sinks = append(sc.Sinks, sinks...)
+	return sc
+}
+
+func (sc *SinkConfig) SetTransform(transform *TransformConfig) *SinkConfig {
+	sc.Transform = transform
+	return sc
+}
+
+func (sc *SinkConfig) FilterScope(scope string) *SinkConfig {
+	return sc.SetTransform(FilterTransform(IncludeWhenAllMatch, structure.ScopeKey, scope))
+}
+
+func (sc *SinkConfig) Terminal() *SinkConfig {
+	return sc.SetOutput(StderrOutput().SetFormat(TerminalFormat))
+}
+
+func (sc *SinkConfig) SetOutput(output *OutputConfig) *SinkConfig {
+	sc.Output = output
+	return sc
 }
 
 func (outputConfig *OutputConfig) SetFormat(format string) *OutputConfig {
@@ -298,8 +322,11 @@ func VectoriseTransform() *TransformConfig {
 }
 
 // Logger formation
-func (sinkConfig *SinkConfig) BuildLogger() (log.Logger, map[string]*loggers.CaptureLogger, error) {
-	return BuildLoggerFromSinkConfig(sinkConfig, make(map[string]*loggers.CaptureLogger))
+func (sc *SinkConfig) BuildLogger() (log.Logger, map[string]*loggers.CaptureLogger, error) {
+	if sc == nil {
+		return log.NewNopLogger(), nil, nil
+	}
+	return BuildLoggerFromSinkConfig(sc, make(map[string]*loggers.CaptureLogger))
 }
 
 func BuildLoggerFromSinkConfig(sinkConfig *SinkConfig, captures map[string]*loggers.CaptureLogger) (log.Logger,

rpc/lib/rpc_test.go

@@ -78,7 +78,7 @@ func TestMain(m *testing.M) {
 
 // launch unix and tcp servers
 func setup() {
-	logger, err := logconfig.New().NewLogger()
+	logger, err := logconfig.New().Logger()
 	if err != nil {
 		panic(err)
 	}