Merge pull request #7 from tdahar/dev

Dev: First stable version of this software

.env-example

@@ -0,0 +1,5 @@
+LIVE_METRICS_CMD="eth-cl-live-metrics"
+LIVE_METRICS_LOG_LEVEL="debug" 
+LIVE_METRICS_BN_ENDPOINTS="prysm/localhost:3500,lh/localhost:5052"
+LIVE_METRICS_DB_ENDPOINT="postgresql://db_user:db_password@db_ip:db_port/db_name"
+LIVE_METRICS_DB_WORKERS="5"

.gitignore

@@ -13,3 +13,6 @@
 
 # Dependency directories (remove the comment below to include it)
 # vendor/
+
+.vscode/**
+block-scorer

.gitmodules

@@ -0,0 +1,4 @@
+[submodule "go-eth2-client"]
+	path = go-eth2-client
+	url = https://github.com/tdahar/go-eth2-client
+	branch = feature/skip-randao

Makefile

@@ -0,0 +1,32 @@
+#!make
+GOCC=go
+MKDIR_P=mkdir -p
+GIT_SUBM=git submodule
+
+include .env
+
+BIN_PATH=./build
+BIN="./build/eth_cl_live_metrics"
+LIVE_METRICS_CMD="live-metrics"
+
+.PHONY: check dependencies build install clean
+
+build:
+	$(GOCC) build -o $(BIN)
+
+install:
+	$(GOCC) install
+
+dependencies:
+	$(GIT_SUBM) update --init
+
+clean:
+	rm -r $(BIN_PATH)
+
+run: 
+	$(BIN) $(LIVE_METRICS_CMD) \
+    	--log-level=${LIVE_METRICS_LOG_LEVEL} \
+        --bn-endpoints=${LIVE_METRICS_BN_ENDPOINTS} \
+        --db-endpoint=${LIVE_METRICS_DB_ENDPOINT} \
+        --db-workers=${LIVE_METRICS_DB_WORKERS} \
+	--metrics=${LIVE_METRICS_METRICS}

README.md

@@ -1,2 +1,2 @@
 # block-scorer
-Tool to ask for blocks from the Ethereum CL clients and evaluate block score
+Tool to ask for blocks from the Ethereum CL clients and evaluate block score.

go.mod

@@ -0,0 +1,59 @@
+module github.com/tdahar/eth-cl-live-metrics
+
+go 1.17
+
+require (
+	github.com/attestantio/go-eth2-client v0.11.7
+	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.14.0
+	github.com/rs/zerolog v1.28.0
+	github.com/sirupsen/logrus v1.9.0
+	github.com/urfave/cli/v2 v2.16.3
+)
+
+require github.com/google/go-cmp v0.5.8 // indirect
+
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.37.0 // indirect
+	github.com/prometheus/procfs v0.8.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+)
+
+require (
+	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
+	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.13.0 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgproto3/v2 v2.3.1 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
+	github.com/jackc/pgtype v1.12.0 // indirect
+	github.com/jackc/pgx/v4 v4.17.2
+	github.com/jackc/puddle v1.3.0 // indirect
+	github.com/klauspost/cpuid/v2 v2.0.11 // 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
+	github.com/r3labs/sse/v2 v2.7.4 // indirect
+	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
+	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa // indirect
+	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
+	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // 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
+)
+
+replace github.com/attestantio/go-eth2-client => ./go-eth2-client

main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/sirupsen/logrus"
+	"github.com/tdahar/eth-cl-live-metrics/pkg/cmd"
+	cli "github.com/urfave/cli/v2"
+)
+
+var (
+	Version = "v1.0.0"
+	CliName = "Eth CL Live Metrics"
+	log     = logrus.WithField(
+		"cli", "CliName",
+	)
+)
+
+func main() {
+	fmt.Println(CliName, Version)
+
+	// Set the general log configurations for the entire tool
+	logrus.SetLevel(logrus.InfoLevel)
+
+	app := &cli.App{
+		Name:      CliName,
+		Usage:     "Tinny client that requests and processes the Beacon Block proposals for each client.",
+		UsageText: "live-metrics [commands] [arguments...]",
+		Authors: []*cli.Author{
+			{
+				Name:  "Tarun",
+				Email: "[email protected]",
+			},
+		},
+		EnableBashCompletion: true,
+		Commands: []*cli.Command{
+			cmd.AnalyzerCommand,
+		},
+	}
+	// generate the block analyzer
+	if err := app.RunContext(context.Background(), os.Args); err != nil {
+		log.Errorf("error: %v\n", err)
+		os.Exit(1)
+	}
+}

pkg/analysis/additional_structs/epoch.go

@@ -0,0 +1,79 @@
+package additional_structs
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"sync"
+
+	api "github.com/attestantio/go-eth2-client/api/v1"
+	"github.com/attestantio/go-eth2-client/http"
+	"github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/sirupsen/logrus"
+)
+
+var (
+	log = logrus.WithField(
+		"module", "Epoch Data")
+)
+
+type EpochStructs struct {
+	mu                       sync.Mutex
+	Api                      *http.Service
+	CurrentBeaconCommittees  []*api.BeaconCommittee
+	CurrentEpoch             uint64
+	PreviousBeaconCommittees []*api.BeaconCommittee
+	PreviousEpoch            uint64
+}
+
+func NewEpochData(iApi *http.Service) EpochStructs {
+
+	return EpochStructs{
+		Api:                      iApi,
+		CurrentBeaconCommittees:  make([]*api.BeaconCommittee, 0),
+		CurrentEpoch:             0,
+		PreviousBeaconCommittees: make([]*api.BeaconCommittee, 0),
+		PreviousEpoch:            0,
+	}
+}
+
+func (e *EpochStructs) RequestNewBeaconCommittee(slot uint64) error {
+	epochCommittees, err := e.Api.BeaconCommittees(context.Background(), strconv.Itoa(int(slot)))
+
+	if err != nil {
+		return fmt.Errorf("could not request beacon committees for epoch %d: %s", int(slot/32), err)
+	}
+
+	// keep in mind we can only receive attestations to 32 blocks before
+	e.PreviousBeaconCommittees = e.CurrentBeaconCommittees
+	e.PreviousEpoch = e.CurrentEpoch
+	e.CurrentBeaconCommittees = epochCommittees
+	e.CurrentEpoch = uint64(slot / 32)
+
+	return nil
+
+}
+
+func (e *EpochStructs) GetBeaconCommittee(slot uint64, index uint64) []phase0.ValidatorIndex {
+	log := log.WithField("routine", "epoch-structs")
+	e.mu.Lock()
+	// if the epoch requested is newer than the data we have
+	if slot/32 > e.CurrentEpoch {
+		log.Debugf("Requesting new beacon committee for %d", slot/32)
+		e.RequestNewBeaconCommittee(slot)
+	}
+	e.mu.Unlock()
+
+	committeeList := e.PreviousBeaconCommittees
+	if slot/32 == e.CurrentEpoch {
+		committeeList = e.CurrentBeaconCommittees
+	}
+
+	for _, item := range committeeList {
+		if item.Slot == phase0.Slot(slot) && item.Index == phase0.CommitteeIndex(index) {
+			return item.Validators
+		}
+	}
+
+	return nil
+}

pkg/analysis/analyzer.go

@@ -0,0 +1,133 @@
+package analysis
+
+import (
+	"context"
+	"time"
+
+	"github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/prysmaticlabs/go-bitfield"
+	"github.com/sirupsen/logrus"
+	"github.com/tdahar/eth-cl-live-metrics/pkg/analysis/additional_structs"
+	"github.com/tdahar/eth-cl-live-metrics/pkg/client_api"
+	"github.com/tdahar/eth-cl-live-metrics/pkg/postgresql"
+)
+
+var (
+	moduleName = "Analysis"
+	log        = logrus.WithField(
+		"module", moduleName)
+)
+
+type ClientLiveData struct {
+	ctx              context.Context
+	Eth2Provider     client_api.APIClient                                       // connection to the beacon node
+	AttHistory       map[phase0.Slot]map[phase0.CommitteeIndex]bitfield.Bitlist // 32 slots of attestation per slot and committeeIndex
+	BlockRootHistory map[phase0.Slot]phase0.Root                                // 64 slots of roots
+	log              *logrus.Entry                                              // each analyzer has its own logger
+	ProcessNewHead   chan struct{}
+	DBClient         *postgresql.PostgresDBService
+	EpochData        additional_structs.EpochStructs
+	CurrentHeadSlot  uint64
+	Monitoring       MonitoringMetrics
+}
+
+func NewBlockAnalyzer(ctx context.Context, label string, cliEndpoint string, timeout time.Duration, dbClient *postgresql.PostgresDBService) (*ClientLiveData, error) {
+	client, err := client_api.NewAPIClient(ctx, label, cliEndpoint, timeout)
+	if err != nil {
+		log.Errorf("could not create eth2 client: %s", err)
+		return &ClientLiveData{}, err
+	}
+	return &ClientLiveData{
+		ctx:              ctx,
+		Eth2Provider:     *client,
+		DBClient:         dbClient,
+		AttHistory:       make(map[phase0.Slot]map[phase0.CommitteeIndex]bitfield.Bitlist),
+		BlockRootHistory: make(map[phase0.Slot]phase0.Root),
+		log:              log.WithField("label", label),
+		EpochData:        additional_structs.NewEpochData(client.Api),
+		CurrentHeadSlot:  0,
+		ProcessNewHead:   make(chan struct{}),
+		Monitoring:       MonitoringMetrics{},
+	}, nil
+}
+
+// Asks for a block proposal to the client and stores score in the database
+func (b *ClientLiveData) ProposeNewBlock(slot phase0.Slot) {
+	log := b.log.WithField("task", "generate-block")
+	log.Debugf("processing new block: %d\n", slot)
+
+	randaoReveal := phase0.BLSSignature{}
+	graffiti := []byte("")
+	snapshot := time.Now()
+	block, err := b.Eth2Provider.Api.BeaconBlockProposal(b.ctx, slot, randaoReveal, graffiti) // ask for block proposal
+	blockTime := time.Since(snapshot).Seconds()                                               // time to generate block
+
+	for i := range b.AttHistory {
+		if i+32 < slot { // attestations can only reference 32 slots back
+			delete(b.AttHistory, i) // remove old entries from the map
+		}
+	}
+
+	for i := range b.BlockRootHistory {
+		if i+64 < slot { // attestations can only reference 32 slots back
+			delete(b.BlockRootHistory, i) // remove old entries from the map
+		}
+	}
+
+	metrics := postgresql.BlockMetricsModel{
+		Slot:  int(slot),
+		Label: b.Eth2Provider.Label,
+		Score: -1,
+	}
+	if err != nil {
+		log.Errorf("error requesting block from %s: %s", b.Eth2Provider.Label, err)
+		b.Monitoring.ProposalStatus = 0
+
+	} else {
+		// for now we just have Bellatrix
+		newMetrics, err := b.BellatrixBlockMetrics(block.Bellatrix, blockTime)
+		if err != nil {
+			log.Errorf("error analyzing block from %s: %s", b.Eth2Provider.Label, err)
+			b.Monitoring.ProposalStatus = 0
+		} else {
+			b.Monitoring.ProposalStatus = 1
+			metrics = newMetrics
+			log.Infof("Block Generation Time: %f", blockTime)
+			log.Infof("Metrics: %+v", metrics)
+		}
+
+	}
+
+	// Store in DB
+	params := make([]interface{}, 0)
+	params = append(params, metrics.Slot)
+	params = append(params, metrics.Label)
+	params = append(params, metrics.Score)
+	params = append(params, metrics.Duration)
+	params = append(params, metrics.CorrectSource)
+	params = append(params, metrics.CorrectTarget)
+	params = append(params, metrics.CorrectHead)
+	params = append(params, metrics.Sync1Bits)
+	params = append(params, metrics.AttNum)
+	params = append(params, metrics.NewVotes)
+	params = append(params, metrics.AttesterSlashings)
+	params = append(params, metrics.ProposerSlashings)
+	params = append(params, metrics.ProposerSlashingScore)
+	params = append(params, metrics.AttesterSlashingScore)
+	params = append(params, metrics.SyncScore)
+
+	writeTask := postgresql.WriteTask{
+		QueryString: postgresql.InsertNewScore,
+		Params:      params,
+	}
+	b.Monitoring.ProposalStatus = 1
+
+	b.DBClient.WriteChan <- writeTask
+
+	// We block the update attestations as new head could impact attestations of the proposed block
+	// b.ProcessNewHead <- struct{}{} // Allow the new head to update attestations
+}
+
+type MonitoringMetrics struct {
+	ProposalStatus int
+}