Exporter (#113)

* add CLI and design doc

* refactor observers

* ibft sync basis (read)

* add storage for operator information

* align with enr config

* refcator eth1

* remove timeout for eth1 sync

* eth1 and p2p config alignment

* update config files

* refactor ssv node > operator node

* refactor operator storage

* add stopper

* add count for calculating indices

* add config for eth1 SyncOffset

* add tests

* use normalized value of syncOffset

* update make and docker

* update docker compose

* use interfaces to abstract ws library

* support get validators

* add gorilla adapter

* close connections for /query endpoint, add port config

* review fixes

* go fmt

Author: amirylm

.air.toml

@@ -14,16 +14,16 @@ bin = "${BUILD_PATH}/cmd/ssvnode"
 
 # Customize binary.
 # This is how you start to run your application. Since my application will works like CLI, so to run it, like to make a CLI call.
-full_bin="make BUILD_PATH=${BUILD_PATH} start-node"
+full_bin="make BUILD_PATH=${BUILD_PATH} ${RUN_CMD}"
 
 # This log file places in your tmp_dir.
 log = "air_errors.log"
 # Watch these filename extensions.
 include_ext = ["go", "yaml"]
 # Ignore these filename extensions or directories.
-exclude_dir = ["/bin"]
+exclude_dir = ["/bin", "/data"]
 # It's not necessary to trigger build each time file changes if it's too frequent.
-delay = 1000 # ms
+delay = 10000 # ms
 
 [log]
 # Show log time

.dockerignore

@@ -1,2 +1,3 @@
 bin
 data
+docs

Makefile

@@ -16,7 +16,6 @@ endif
 # node command builder
 NODE_COMMAND=--config=${CONFIG_PATH}
 
-
 ifneq ($(SHARE_CONFIG),)
   NODE_COMMAND+= --share-config=${SHARE_CONFIG}
 endif
@@ -62,7 +61,6 @@ ifdef DEBUG_PORT
 	@echo "Running node-${NODE_ID} in debug mode"
 	@dlv  --continue --accept-multiclient --headless --listen=:${DEBUG_PORT} --api-version=2 exec \
 	 ${BUILD_PATH} start-node -- ${NODE_COMMAND}
-
 else
 	@echo "Running node on address: ${HOST_ADDRESS})"
 	@${BUILD_PATH} start-node ${NODE_COMMAND}
@@ -79,7 +77,6 @@ docker-image:
 	@echo "node ${NODES_ID}"
 	@sudo docker rm -f ssv_node && docker run -d --env-file .env --restart unless-stopped --name=ssv_node -p 13000:13000 -p 12000:12000 'bloxstaking/ssv-node:latest' make BUILD_PATH=/go/bin/ssvnode start-node
 
-
 NODES=ssv-node-1 ssv-node-2 ssv-node-3 ssv-node-4
 .PHONY: docker-all
 docker-all:
@@ -96,8 +93,12 @@ docker-debug:
 stop:
 	@docker-compose  down
 
-
 .PHONY: start-boot-node
 start-boot-node:
 	@echo "Running start-boot-node"
 	${BUILD_PATH} start-boot-node
+
+.PHONY: start-exporter
+start-exporter:
+	@echo "Running start-exporter"
+	${BUILD_PATH} start-exporter ${NODE_COMMAND}

cli/bootnode/boot_node.go

@@ -13,7 +13,7 @@ import (
 
 type config struct {
 	global_config.GlobalConfig `yaml:"global"`
-	Options bootnode.Options `yaml:"bootnode"`
+	Options                    bootnode.Options `yaml:"bootnode"`
 }
 
 var cfg config
@@ -33,7 +33,7 @@ var StartBootNodeCmd = &cobra.Command{
 		Logger := logex.Build(cmd.Parent().Short, loggerLevel)
 
 		if err != nil {
-			Logger.Warn(fmt.Sprintf("Default log level set to %s", loggerLevel),zap.Error(err))
+			Logger.Warn(fmt.Sprintf("Default log level set to %s", loggerLevel), zap.Error(err))
 		}
 
 		cfg.Options.Logger = Logger

cli/cli.go

@@ -2,7 +2,8 @@ package cli
 
 import (
 	"github.com/bloxapp/ssv/cli/bootnode"
-	"github.com/bloxapp/ssv/cli/ssvnode"
+	"github.com/bloxapp/ssv/cli/exporter"
+	"github.com/bloxapp/ssv/cli/operator"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 	"log"
@@ -13,10 +14,10 @@ var Logger *zap.Logger
 
 // RootCmd represents the root command of SSV CLI
 var RootCmd = &cobra.Command{
-	Use:  "ssvnode",
-	Short:  "ssv-node",
-	Long: `SSV node is a CLI for running SSV-related operations.`,
-	PersistentPreRun: func(cmd *cobra.Command, args []string){
+	Use:   "ssvnode",
+	Short: "ssv-node",
+	Long:  `SSV node is a CLI for running SSV-related operations.`,
+	PersistentPreRun: func(cmd *cobra.Command, args []string) {
 	},
 }
 
@@ -29,7 +30,9 @@ func Execute(appName, version string) {
 		log.Fatal("failed to execute root command", zap.Error(err))
 	}
 }
-func init()  {
+
+func init() {
 	RootCmd.AddCommand(bootnode.StartBootNodeCmd)
-	RootCmd.AddCommand(ssvnode.StartNodeCmd)
-}
+	RootCmd.AddCommand(exporter.StartExporterNodeCmd)
+	RootCmd.AddCommand(operator.StartNodeCmd)
+}

cli/exporter/node.go

@@ -0,0 +1,95 @@
+package exporter
+
+import (
+	"crypto/rsa"
+	"fmt"
+	global_config "github.com/bloxapp/ssv/cli/config"
+	"github.com/bloxapp/ssv/eth1"
+	"github.com/bloxapp/ssv/eth1/goeth"
+	"github.com/bloxapp/ssv/exporter"
+	"github.com/bloxapp/ssv/exporter/api"
+	"github.com/bloxapp/ssv/exporter/api/adapters/gorilla"
+	"github.com/bloxapp/ssv/network/p2p"
+	"github.com/bloxapp/ssv/storage"
+	"github.com/bloxapp/ssv/storage/basedb"
+	"github.com/bloxapp/ssv/utils/logex"
+	"github.com/ilyakaznacheev/cleanenv"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+	"log"
+	"net/http"
+)
+
+type config struct {
+	global_config.GlobalConfig `yaml:"global"`
+	DBOptions                  basedb.Options `yaml:"db"`
+	P2pNetworkConfig           p2p.Config     `yaml:"p2p"`
+
+	ExporterConfig exporter.Options `yaml:"exporter"`
+
+	ETH1Addr       string `yaml:"ETH1Addr" env-required:"true"`
+	ETH1SyncOffset string `yaml:"ETH1SyncOffset" env:"ETH_1_SYNC_OFFSET"`
+	Network        string `yaml:"Network" env-default:"prater"`
+}
+
+var cfg config
+
+var globalArgs global_config.Args
+
+// StartExporterNodeCmd is the command to start SSV boot node
+var StartExporterNodeCmd = &cobra.Command{
+	Use:   "start-exporter",
+	Short: "Starts exporter node",
+	Run: func(cmd *cobra.Command, args []string) {
+		if err := cleanenv.ReadConfig(globalArgs.ConfigPath, &cfg); err != nil {
+			log.Fatal(err)
+		}
+		// configure logger and db
+		loggerLevel, err := logex.GetLoggerLevelValue(cfg.LogLevel)
+		Logger := logex.Build(cmd.Parent().Short, loggerLevel)
+		if err != nil {
+			Logger.Warn(fmt.Sprintf("Default log level set to %s", loggerLevel), zap.Error(err))
+		}
+		cfg.DBOptions.Logger = Logger
+		db, err := storage.GetStorageFactory(cfg.DBOptions)
+		if err != nil {
+			Logger.Fatal("failed to create db!", zap.Error(err))
+		}
+
+		network, err := p2p.New(cmd.Context(), Logger, &cfg.P2pNetworkConfig)
+		if err != nil {
+			Logger.Fatal("failed to create network", zap.Error(err))
+		}
+		eth1Client, err := goeth.NewEth1Client(goeth.ClientOptions{
+			Ctx: cmd.Context(), Logger: Logger, NodeAddr: cfg.ETH1Addr,
+			// using an empty private key provider
+			// because the exporter doesn't run in the context of an operator
+			PrivKeyProvider: func() (*rsa.PrivateKey, error) {
+				return nil, nil
+			},
+		})
+		if err != nil {
+			Logger.Fatal("failed to create eth1 client", zap.Error(err))
+		}
+
+		cfg.ExporterConfig.Eth1Client = eth1Client
+		cfg.ExporterConfig.Logger = Logger
+		cfg.ExporterConfig.Network = network
+		cfg.ExporterConfig.DB = db
+		cfg.ExporterConfig.Ctx = cmd.Context()
+		cfg.ExporterConfig.WS = api.NewWsServer(Logger, gorilla.NewGorillaAdapter(Logger), http.NewServeMux())
+
+		exporterNode := exporter.New(cfg.ExporterConfig)
+
+		if err := exporterNode.StartEth1(eth1.HexStringToSyncOffset(cfg.ETH1SyncOffset)); err != nil {
+			Logger.Fatal("failed to start eth1", zap.Error(err))
+		}
+		if err := exporterNode.Start(); err != nil {
+			Logger.Fatal("failed to start exporter", zap.Error(err))
+		}
+	},
+}
+
+func init() {
+	global_config.ProcessArgs(&cfg, &globalArgs, StartExporterNodeCmd)
+}

cli/generate_operator_keys.go

@@ -17,7 +17,7 @@ var generateOperatorKeysCmd = &cobra.Command{
 		logger := logex.Build(RootCmd.Short, zapcore.DebugLevel)
 
 		pk, sk, err := rsaencryption.GenerateKeys()
-		if err != nil{
+		if err != nil {
 			logger.Fatal("Failed to generate operator keys", zap.Error(err))
 		}
 		logger.Info("generated public key (base64)", zap.Any("pk", pk))

cli/operator/node.go

@@ -0,0 +1,124 @@
+package operator
+
+import (
+	"fmt"
+	"github.com/bloxapp/eth2-key-manager/core"
+	"github.com/bloxapp/ssv/beacon/prysmgrpc"
+	global_config "github.com/bloxapp/ssv/cli/config"
+	"github.com/bloxapp/ssv/eth1"
+	"github.com/bloxapp/ssv/eth1/goeth"
+	"github.com/bloxapp/ssv/network/p2p"
+	"github.com/bloxapp/ssv/operator"
+	"github.com/bloxapp/ssv/shared/params"
+	"github.com/bloxapp/ssv/storage"
+	"github.com/bloxapp/ssv/storage/basedb"
+	"github.com/bloxapp/ssv/utils/logex"
+	"github.com/ilyakaznacheev/cleanenv"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+	"log"
+)
+
+type config struct {
+	global_config.GlobalConfig `yaml:"global"`
+	DBOptions                  basedb.Options   `yaml:"db"`
+	SSVOptions                 operator.Options `yaml:"ssv"`
+
+	Network           string `yaml:"Network" env-default:"prater"`
+	BeaconNodeAddr    string `yaml:"BeaconNodeAddr" env:"BEACON_NODE_ADDR" env-required:"true"`
+	OperatorKey       string `yaml:"OperatorKey" env:"OPERATOR_KEY" env-description:"Operator private key, used to decrypt contract events"`
+	ETH1Addr          string `yaml:"ETH1Addr" env:"ETH_1_ADDR" env-required:"true"`
+	ETH1SyncOffset    string `yaml:"ETH1SyncOffset" env:"ETH_1_SYNC_OFFSET"`
+	SmartContractAddr string `yaml:"SmartContractAddr" env:"SMART_CONTRACT_ADDR_KEY" env-description:"smart contract addr listen to event from" env-default:""`
+
+	P2pNetworkConfig p2p.Config `yaml:"p2p"`
+}
+
+var cfg config
+
+var globalArgs global_config.Args
+
+// StartNodeCmd is the command to start SSV node
+var StartNodeCmd = &cobra.Command{
+	Use:   "start-node",
+	Short: "Starts an instance of SSV node",
+	Run: func(cmd *cobra.Command, args []string) {
+		if err := cleanenv.ReadConfig(globalArgs.ConfigPath, &cfg); err != nil {
+			log.Fatal(err)
+		}
+		if globalArgs.ShareConfigPath != "" {
+			if err := cleanenv.ReadConfig(globalArgs.ShareConfigPath, &cfg); err != nil {
+				log.Fatal(err)
+			}
+		}
+		loggerLevel, err := logex.GetLoggerLevelValue(cfg.LogLevel)
+		Logger := logex.Build(cmd.Parent().Short, loggerLevel)
+
+		if err != nil {
+			Logger.Warn(fmt.Sprintf("Default log level set to %s", loggerLevel), zap.Error(err))
+		}
+		cfg.DBOptions.Logger = Logger
+		db, err := storage.GetStorageFactory(cfg.DBOptions)
+		if err != nil {
+			Logger.Fatal("failed to create db!", zap.Error(err))
+		}
+
+		// TODO Not refactored yet Start:
+		eth2Network := core.NetworkFromString(cfg.Network)
+		beaconClient, err := prysmgrpc.New(cmd.Context(), Logger, eth2Network, []byte("BloxStaking"), cfg.BeaconNodeAddr)
+		if err != nil {
+			Logger.Fatal("failed to create beacon client", zap.Error(err))
+		}
+
+		network, err := p2p.New(cmd.Context(), Logger, &cfg.P2pNetworkConfig)
+		if err != nil {
+			Logger.Fatal("failed to create network", zap.Error(err))
+		}
+		// end Non refactored
+
+		ctx := cmd.Context()
+		cfg.SSVOptions.Context = ctx
+		cfg.SSVOptions.Logger = Logger
+		cfg.SSVOptions.DB = db
+		cfg.SSVOptions.Beacon = &beaconClient
+		cfg.SSVOptions.ETHNetwork = &eth2Network
+		cfg.SSVOptions.ValidatorOptions.ETHNetwork = &eth2Network
+		cfg.SSVOptions.ValidatorOptions.Logger = Logger
+		cfg.SSVOptions.ValidatorOptions.Context = ctx
+		cfg.SSVOptions.ValidatorOptions.DB = db
+		cfg.SSVOptions.ValidatorOptions.Network = network
+		cfg.SSVOptions.ValidatorOptions.Beacon = beaconClient
+
+		operatorStorage := operator.NewOperatorNodeStorage(db, Logger)
+		if err := operatorStorage.SetupPrivateKey(cfg.OperatorKey); err != nil {
+			Logger.Fatal("failed to setup operator private key", zap.Error(err))
+		}
+		// create new eth1 client
+		if cfg.SmartContractAddr != "" {
+			Logger.Info("using smart contract addr from cfg", zap.String("addr", cfg.SmartContractAddr))
+			params.SsvConfig().OperatorContractAddress = cfg.SmartContractAddr // TODO need to remove config and use in eth2 option cfg
+		}
+		cfg.SSVOptions.Eth1Client, err = goeth.NewEth1Client(goeth.ClientOptions{
+			Ctx:             cmd.Context(),
+			Logger:          Logger,
+			NodeAddr:        cfg.ETH1Addr,
+			PrivKeyProvider: operatorStorage.GetPrivateKey,
+		})
+		if err != nil {
+			Logger.Fatal("failed to create eth1 client", zap.Error(err))
+		}
+
+		operatorNode := operator.New(cfg.SSVOptions)
+
+		if err := operatorNode.StartEth1(eth1.HexStringToSyncOffset(cfg.ETH1SyncOffset)); err != nil {
+			Logger.Fatal("failed to start eth1", zap.Error(err))
+		}
+		if err := operatorNode.Start(); err != nil {
+			Logger.Fatal("failed to start SSV node", zap.Error(err))
+		}
+	},
+}
+
+func init() {
+	global_config.ProcessArgs(&cfg, &globalArgs, StartNodeCmd)
+}