Merge pull request #12 from cortze/dev

Stable version optimizing DB interaction, DB total size and State processing times

.env

@@ -0,0 +1,10 @@
+STATE_ANALYZER_CMD="rewards"
+STATE_ANALYZER_LOG_LEVEL="debug" 
+STATE_ANALYZER_BN_ENDPOINT="http://beacon_node_ip:api-port"
+STATE_ANALYZER_OUTFOLDER="results" 
+STATE_ANALYZER_INIT_SLOT="10000" 
+STATE_ANALYZER_FINAL_SLOT="10050"
+STATE_ANALYZER_VALIDATOR_INDEXES="test_validators.json"
+STATE_ANALYZER_DB_URL="postgresql://db_user:db_password@db_ip:db_port/db_name"
+STATE_ANALYZER_WORKERS_NUM="50"
+STATE_ANALYZER_DB_WORKERS_NUM="4"

.gitignore

@@ -1 +1,7 @@
 results/
+eth2-state-analyzer
+.vscode/**
+test*.json
+.ipynb_checkpoints/**
+metrika/**
+.config

Makefile

@@ -0,0 +1,33 @@
+#!make
+
+GOCC=go
+MKDIR_P=mkdir -p
+
+BIN_PATH=./build
+BIN="./build/eth2-state-analyzer"
+
+include .env
+
+.PHONY: check build install run clean
+
+build: 
+	$(GOCC) build -o $(BIN)
+
+install:
+	$(GOCC) install
+
+run: 
+	$(BIN) $(STATE_ANALYZER_CMD) \
+        --log-level=${STATE_ANALYZER_LOG_LEVEL} \
+        --bn-endpoint=${STATE_ANALYZER_BN_ENDPOINT} \
+        --outfolder=${STATE_ANALYZER_OUTFOLDER} \
+        --init-slot=${STATE_ANALYZER_INIT_SLOT} \
+        --final-slot=${STATE_ANALYZER_FINAL_SLOT} \
+        --validator-indexes=${STATE_ANALYZER_VALIDATOR_INDEXES} \
+        --db-url=${STATE_ANALYZER_DB_URL} \
+        --workers-num=${STATE_ANALYZER_WORKERS_NUM} \
+        --db-workers-num=${STATE_ANALYZER_DB_WORKERS_NUM}
+
+clean:
+	rm -r $(BIN_PATH)
+

README.md

@@ -1,2 +1 @@
-# api-benchmark
-Little client that measures API's response times
+# Eth2 State Analyzer

cmd/reward_cmd.go

@@ -1,6 +1,9 @@
 package cmd
 
 import (
+	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	"github.com/pkg/errors"
@@ -40,17 +43,31 @@ var RewardsCommand = &cli.Command{
 		&cli.StringFlag{
 			Name:  "log-level",
 			Usage: "log level: debug, warn, info, error",
+		},
+		&cli.StringFlag{
+			Name:  "db-url",
+			Usage: "example: postgresql://beaconchain:beaconchain@localhost:5432/beacon_states_kiln",
+		},
+		&cli.IntFlag{
+			Name:  "workers-num",
+			Usage: "example: 50",
+		},
+		&cli.IntFlag{
+			Name:  "db-workers-num",
+			Usage: "example: 50",
 		}},
 }
 
 var logRewardsRewards = logrus.WithField(
 	"module", "RewardsCommand",
 )
 
-var QueryTimeout = 30 * time.Second
+var QueryTimeout = 90 * time.Second
 
 // CrawlAction is the function that is called when running `eth2`.
 func LaunchRewardsCalculator(c *cli.Context) error {
+	coworkers := 1
+	dbWorkers := 1
 	logRewardsRewards.Info("parsing flags")
 	// check if a config file is set
 	if !c.IsSet("bn-endpoint") {
@@ -71,10 +88,23 @@ func LaunchRewardsCalculator(c *cli.Context) error {
 	if c.IsSet("log-level") {
 		logrus.SetLevel(utils.ParseLogLevel(c.String("log-level")))
 	}
+	if !c.IsSet("db-url") {
+		return errors.New("db-url not provided")
+	}
+	if !c.IsSet("workers-num") {
+		logRewardsRewards.Infof("workers-num flag not provided, default: 1")
+	} else {
+		coworkers = c.Int("workers-num")
+	}
+	if !c.IsSet("db-workers-num") {
+		logRewardsRewards.Infof("db-workers-num flag not provided, default: 1")
+	} else {
+		dbWorkers = c.Int("db-workers-num")
+	}
 	bnEndpoint := c.String("bn-endpoint")
-	outputFile := c.String("outfolder")
 	initSlot := uint64(c.Int("init-slot"))
 	finalSlot := uint64(c.Int("final-slot"))
+	dbUrl := c.String("db-url")
 
 	validatorIndexes, err := utils.GetValIndexesFromJson(c.String("validator-indexes"))
 	if err != nil {
@@ -86,16 +116,33 @@ func LaunchRewardsCalculator(c *cli.Context) error {
 	if err != nil {
 		return err
 	}
+
 	// generate the state analyzer
-	stateAnalyzer, err := analyzer.NewStateAnalyzer(c.Context, cli, initSlot, finalSlot, validatorIndexes)
+	stateAnalyzer, err := analyzer.NewStateAnalyzer(c.Context, cli, initSlot, finalSlot, validatorIndexes, dbUrl, coworkers, dbWorkers)
 	if err != nil {
 		return err
 	}
 
-	stateAnalyzer.Run()
+	procDoneC := make(chan struct{})
+	sigtermC := make(chan os.Signal)
+
+	signal.Notify(sigtermC, syscall.SIGINT, syscall.SIGTERM, os.Interrupt, syscall.SIGTERM)
 
-	// TODO: Compose Results
-	stateAnalyzer.ExportToCsv(outputFile)
+	go func() {
+		stateAnalyzer.Run()
+		procDoneC <- struct{}{}
+	}()
+
+	select {
+	case <-sigtermC:
+		logRewardsRewards.Info("Sudden shutdown detected, controlled shutdown of the cli triggered")
+		stateAnalyzer.Close()
+
+	case <-procDoneC:
+		logRewardsRewards.Info("Process successfully finish!")
+	}
+	close(sigtermC)
+	close(procDoneC)
 
 	return nil
 }

go.mod

@@ -4,6 +4,7 @@ go 1.17
 
 require (
 	github.com/attestantio/go-eth2-client v0.11.3
+	github.com/jackc/pgx/v4 v4.16.1
 	github.com/pkg/errors v0.9.1
 	github.com/rs/zerolog v1.26.1
 	github.com/sirupsen/logrus v1.6.0
@@ -15,17 +16,28 @@ require (
 	github.com/fatih/color v1.13.0 // indirect
 	github.com/ferranbt/fastssz v0.0.0-20220103083642-bc5fefefa28b // indirect
 	github.com/goccy/go-yaml v1.9.5 // indirect
+	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
+	github.com/jackc/pgconn v1.12.1 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgproto3/v2 v2.3.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
+	github.com/jackc/pgtype v1.11.0 // indirect
+	github.com/jackc/puddle v1.2.1 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.11 // indirect
 	github.com/konsorten/go-windows-terminal-sequences v1.0.3 // indirect
+	github.com/magiconair/properties v1.8.6 // indirect
 	github.com/mattn/go-colorable v0.1.12 // indirect
 	github.com/mattn/go-isatty v0.0.14 // indirect
 	github.com/minio/sha256-simd v1.0.0 // indirect
 	github.com/mitchellh/mapstructure v1.4.3 // indirect
 	github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 // indirect
 	github.com/r3labs/sse/v2 v2.7.4 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	golang.org/x/crypto v0.0.0-20211215165025-cf75a172585e // indirect
 	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
 	golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9 // indirect
+	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	gopkg.in/cenkalti/backoff.v1 v1.1.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect