Merge pull request #322 from bloxapp/stage (v20)

.k8/yamls/ssv-exporter.yml

@@ -139,6 +139,8 @@ spec:
           name: port-15000
           hostPort: 15000
         env:
+          - name: SHARE_CONFIG
+            value: "./data/share.yaml"
           - name: ETH_1_ADDR
             valueFrom:
               secretKeyRef:

beacon/beacon.go

@@ -27,8 +27,8 @@ type Beacon interface {
 	// GetDuties returns duties for the passed validators indices
 	GetDuties(epoch spec.Epoch, validatorIndices []spec.ValidatorIndex) ([]*Duty, error)
 
-	// GetIndices returns indices for each pubkey from the node
-	GetIndices(validatorPubKeys []spec.BLSPubKey) (map[spec.ValidatorIndex]*api.Validator, error)
+	// GetValidatorData returns metadata (balance, index, status, more) for each pubkey from the node
+	GetValidatorData(validatorPubKeys []spec.BLSPubKey) (map[spec.ValidatorIndex]*api.Validator, error)
 
 	// GetAttestationData returns attestation data by the given slot and committee index
 	GetAttestationData(slot spec.Slot, committeeIndex spec.CommitteeIndex) (*spec.AttestationData, error)

beacon/duty.go

@@ -21,4 +21,3 @@ type Duty struct {
 	// ValidatorCommitteeIndex is the index of the validator in the list of validators in the committee.
 	ValidatorCommitteeIndex uint64
 }
-

beacon/duty_data.go

@@ -56,8 +56,9 @@ type IsInputValueSignedData interface {
 type InputValueAttestation struct {
 	Attestation *phase0.Attestation
 }
+
 // isInputValueSignedData implementation
- func (*InputValueAttestation) isInputValueSignedData() {}
+func (*InputValueAttestation) isInputValueSignedData() {}
 
 // GetSignedData returns input data
 func (m *DutyData) GetSignedData() IsInputValueSignedData {
@@ -73,4 +74,4 @@ func (m *DutyData) GetAttestation() *phase0.Attestation {
 		return x.Attestation
 	}
 	return nil
-}
+}

beacon/goclient/goclient.go

@@ -142,7 +142,8 @@ func (gc *goClient) GetDuties(epoch spec.Epoch, validatorIndices []spec.Validato
 	return nil, errors.New("client does not support AttesterDutiesProvider")
 }
 
-func (gc *goClient) GetIndices(validatorPubKeys []spec.BLSPubKey) (map[spec.ValidatorIndex]*api.Validator, error) {
+// GetValidatorData returns metadata (balance, index, status, more) for each pubkey from the node
+func (gc *goClient) GetValidatorData(validatorPubKeys []spec.BLSPubKey) (map[spec.ValidatorIndex]*api.Validator, error) {
 	if provider, isProvider := gc.client.(eth2client.ValidatorsProvider); isProvider {
 		validatorsMap, err := provider.ValidatorsByPubKey(gc.ctx, "head", validatorPubKeys) // TODO maybe need to get the chainId (head) as var
 		if err != nil {

beacon/test_utils.go

@@ -0,0 +1,95 @@
+package beacon
+
+import (
+	v1 "github.com/attestantio/go-eth2-client/api/v1"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/herumi/bls-eth-go-binary/bls"
+	"sync"
+)
+
+func init() {
+	_ = bls.Init(bls.BLS12_381)
+}
+
+// NewMockBeacon creates a new mock implementation of beacon client
+func NewMockBeacon(dutiesResults map[uint64][]*Duty, validatorsData map[spec.BLSPubKey]*v1.Validator) Beacon {
+	return &mockBeacon{
+		indicesMap:     map[spec.BLSPubKey]spec.ValidatorIndex{},
+		indicesLock:    sync.Mutex{},
+		dutiesResults:  dutiesResults,
+		validatorsData: validatorsData,
+	}
+}
+
+type mockBeacon struct {
+	indicesMap     map[spec.BLSPubKey]spec.ValidatorIndex
+	indicesLock    sync.Mutex
+	dutiesResults  map[uint64][]*Duty
+	validatorsData map[spec.BLSPubKey]*v1.Validator
+}
+
+func (m *mockBeacon) ExtendIndexMap(index spec.ValidatorIndex, pubKey spec.BLSPubKey) {
+	m.indicesLock.Lock()
+	defer m.indicesLock.Unlock()
+
+	m.indicesMap[pubKey] = index
+}
+
+func (m *mockBeacon) GetDuties(epoch spec.Epoch, validatorIndices []spec.ValidatorIndex) ([]*Duty, error) {
+	return m.dutiesResults[uint64(epoch)], nil
+}
+
+func (m *mockBeacon) GetValidatorData(validatorPubKeys []spec.BLSPubKey) (map[spec.ValidatorIndex]*v1.Validator, error) {
+	results := map[spec.ValidatorIndex]*v1.Validator{}
+	for _, pk := range validatorPubKeys {
+		if data, ok := m.validatorsData[pk]; ok {
+			results[data.Index] = data
+		}
+	}
+	return results, nil
+}
+
+func (m *mockBeacon) GetAttestationData(slot spec.Slot, committeeIndex spec.CommitteeIndex) (*spec.AttestationData, error) {
+	return nil, nil
+}
+
+func (m *mockBeacon) SignAttestation(data *spec.AttestationData, duty *Duty, shareKey *bls.SecretKey) (*spec.Attestation, []byte, error) {
+	return nil, nil, nil
+}
+
+func (m *mockBeacon) SubmitAttestation(attestation *spec.Attestation) error {
+	return nil
+}
+
+func (m *mockBeacon) SubscribeToCommitteeSubnet(subscription []*v1.BeaconCommitteeSubscription) error {
+	return nil
+}
+
+// NewMockValidatorMetadataStorage creates a new mock implementation of ValidatorMetadataStorage
+func NewMockValidatorMetadataStorage() ValidatorMetadataStorage {
+	return &mockValidatorMetadataStorage{
+		map[string]*ValidatorMetadata{},
+		sync.Mutex{},
+	}
+}
+
+type mockValidatorMetadataStorage struct {
+	data map[string]*ValidatorMetadata
+	lock sync.Mutex
+}
+
+func (m *mockValidatorMetadataStorage) UpdateValidatorMetadata(pk string, metadata *ValidatorMetadata) error {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	m.data[pk] = metadata
+
+	return nil
+}
+
+func (m *mockValidatorMetadataStorage) Size() int {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	return len(m.data)
+}

beacon/types.go

@@ -25,4 +25,4 @@ const (
 	RoleTypeAttester
 	RoleTypeAggregator
 	RoleTypeProposer
-)
+)

beacon/validator_metadata_test.go

@@ -0,0 +1,91 @@
+package beacon
+
+import (
+	"encoding/hex"
+	v1 "github.com/attestantio/go-eth2-client/api/v1"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/bloxapp/ssv/utils/logex"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"strings"
+	"sync/atomic"
+	"testing"
+)
+
+func init() {
+	logex.Build("test", zap.InfoLevel, nil)
+}
+
+func TestValidatorMetadata_Status(t *testing.T) {
+	t.Run("ready", func(t *testing.T) {
+		meta := &ValidatorMetadata{
+			Status: v1.ValidatorStateActiveOngoing,
+		}
+		require.True(t, meta.Activated())
+		require.False(t, meta.Exiting())
+		require.False(t, meta.Slashed())
+	})
+
+	t.Run("exited", func(t *testing.T) {
+		meta := &ValidatorMetadata{
+			Status: v1.ValidatorStateWithdrawalPossible,
+		}
+		require.True(t, meta.Exiting())
+		require.True(t, meta.Activated())
+		require.False(t, meta.Slashed())
+	})
+
+	t.Run("exitedSlashed", func(t *testing.T) {
+		meta := &ValidatorMetadata{
+			Status: v1.ValidatorStateExitedSlashed,
+		}
+		require.True(t, meta.Slashed())
+		require.True(t, meta.Exiting())
+		require.True(t, meta.Activated())
+	})
+}
+
+func TestUpdateValidatorsMetadata(t *testing.T) {
+	var updateCount uint64
+	pks := []string{
+		"a17bb48a3f8f558e29d08ede97d6b7b73823d8dc2e0530fe8b747c93d7d6c2755957b7ffb94a7cec830456fd5492ba19",
+		"a0cf5642ed5aa82178a5f79e00292c5b700b67fbf59630ce4f542c392495d9835a99c826aa2459a67bc80867245386c6",
+		"8bafb7165f42e1179f83b7fcd8fe940e60ed5933fac176fdf75a60838c688eaa3b57717be637bde1d5cebdadb8e39865",
+	}
+
+	decodeds := make([][]byte, len(pks))
+	blsPubKeys := make([]spec.BLSPubKey, len(pks))
+	for i, pk := range pks {
+		blsPubKey := spec.BLSPubKey{}
+		decoded, _ := hex.DecodeString(pk)
+		copy(blsPubKey[:], decoded[:])
+		decodeds[i] = decoded
+		blsPubKeys[i] = blsPubKey
+	}
+
+	data := map[spec.BLSPubKey]*v1.Validator{}
+	data[blsPubKeys[0]] = &v1.Validator{
+		Index:     spec.ValidatorIndex(210961),
+		Status:    v1.ValidatorStateWithdrawalPossible,
+		Validator: &spec.Validator{PublicKey: blsPubKeys[0]},
+	}
+	data[blsPubKeys[1]] = &v1.Validator{
+		Index:     spec.ValidatorIndex(213820),
+		Status:    v1.ValidatorStateActiveOngoing,
+		Validator: &spec.Validator{PublicKey: blsPubKeys[1]},
+	}
+	bc := NewMockBeacon(map[uint64][]*Duty{}, data)
+
+	storage := NewMockValidatorMetadataStorage()
+
+	onUpdated := func(pk string, meta *ValidatorMetadata) {
+		joined := strings.Join(pks, ":")
+		require.True(t, strings.Contains(joined, pk))
+		require.True(t, meta.Index == spec.ValidatorIndex(210961) || meta.Index == spec.ValidatorIndex(213820))
+		atomic.AddUint64(&updateCount, 1)
+	}
+	err := UpdateValidatorsMetadata([][]byte{blsPubKeys[0][:], blsPubKeys[1][:], blsPubKeys[2][:]}, storage, bc, onUpdated)
+	require.Nil(t, err)
+	require.Equal(t, uint64(2), updateCount)
+	require.Equal(t, 2, storage.(*mockValidatorMetadataStorage).Size())
+}

beacon/validator_metdata.go

@@ -0,0 +1,107 @@
+package beacon
+
+import (
+	"encoding/hex"
+	v1 "github.com/attestantio/go-eth2-client/api/v1"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/bloxapp/ssv/utils/logex"
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+)
+
+// ValidatorMetadataStorage interface for validator metadata
+type ValidatorMetadataStorage interface {
+	UpdateValidatorMetadata(pk string, metadata *ValidatorMetadata) error
+}
+
+// ValidatorMetadata represents validator metdata from beacon
+type ValidatorMetadata struct {
+	Balance spec.Gwei           `json:"balance"`
+	Status  v1.ValidatorState   `json:"status"`
+	Index   spec.ValidatorIndex `json:"index"` // pointer in order to support nil
+}
+
+// Activated returns true if the validator is not unknown. It might be pending activation or active
+func (m *ValidatorMetadata) Activated() bool {
+	return m.Status.HasActivated()
+}
+
+// Exiting returns true if the validator is existing or exited
+func (m *ValidatorMetadata) Exiting() bool {
+	return m.Status.IsExited() || m.Status.HasExited()
+}
+
+// Slashed returns true if the validator is existing or exited due to slashing
+func (m *ValidatorMetadata) Slashed() bool {
+	return m.Status == v1.ValidatorStateExitedSlashed || m.Status == v1.ValidatorStateActiveSlashed
+}
+
+// OnUpdated represents a function to be called once validator's metadata was updated
+type OnUpdated func(pk string, meta *ValidatorMetadata)
+
+// UpdateValidatorsMetadata updates validator information for the given public keys
+func UpdateValidatorsMetadata(pubKeys [][]byte, collection ValidatorMetadataStorage, bc Beacon, onUpdated OnUpdated) error {
+	logger := logex.GetLogger(zap.String("who", "UpdateValidatorsMetadata"))
+
+	results, err := FetchValidatorsMetadata(bc, pubKeys)
+	if err != nil {
+		return errors.Wrap(err, "failed to get validator data from Beacon")
+	}
+	logger.Debug("got validators metadata", zap.Int("pks count", len(pubKeys)),
+		zap.Int("results count", len(results)))
+
+	var errs []error
+	for pk, meta := range results {
+		if err := collection.UpdateValidatorMetadata(pk, meta); err != nil {
+			logger.Error("failed to update validator metadata",
+				zap.String("pk", pk), zap.Error(err))
+			errs = append(errs, err)
+		}
+		if onUpdated != nil {
+			onUpdated(pk, meta)
+		}
+		logger.Debug("managed to update validator metadata",
+			zap.String("pk", pk), zap.Any("metadata", *meta))
+	}
+	if len(errs) > 0 {
+		logger.Error("could not process validators returned from beacon",
+			zap.Int("count", len(errs)), zap.Any("errs", errs))
+		return errors.Errorf("could not process %d validators returned from beacon", len(errs))
+	}
+
+	return nil
+}
+
+// FetchValidatorsMetadata is fetching validators data from beacon
+func FetchValidatorsMetadata(bc Beacon, pubKeys [][]byte) (map[string]*ValidatorMetadata, error) {
+	logger := logex.GetLogger(zap.String("who", "FetchValidatorsMetadata"))
+	if len(pubKeys) == 0 {
+		return nil, nil
+	}
+	var pubkeys []spec.BLSPubKey
+	for _, pk := range pubKeys {
+		blsPubKey := spec.BLSPubKey{}
+		copy(blsPubKey[:], pk)
+		pubkeys = append(pubkeys, blsPubKey)
+	}
+	logger.Debug("fetching metadata for public keys", zap.Int("total", len(pubkeys)))
+	validatorsIndexMap, err := bc.GetValidatorData(pubkeys)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get validators data from beacon")
+	}
+	logger.Debug("got validators metadata", zap.Int("pks count", len(pubKeys)),
+		zap.Int("results count", len(validatorsIndexMap)))
+	ret := make(map[string]*ValidatorMetadata)
+	for index, v := range validatorsIndexMap {
+		pk := hex.EncodeToString(v.Validator.PublicKey[:])
+		meta := &ValidatorMetadata{
+			Balance: v.Balance,
+			Status:  v.Status,
+			Index:   v.Index,
+		}
+		ret[pk] = meta
+		// once fetched, the internal map in go-client should be updated
+		bc.ExtendIndexMap(index, v.Validator.PublicKey)
+	}
+	return ret, nil
+}