Store parsigdb metrics

Author: KaloyanTanev

app/app.go

@@ -567,7 +567,27 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		return err
 	}
 
-	parSigDB := parsigdb.NewMemDB(lock.Threshold, deadlinerFunc("parsigdb"))
+	var slotDuration uint64
+	var genesisTime time.Time
+	if conf.TestnetConfig.IsNonZero() {
+		slotDuration = 12
+		genesisTime = time.Unix(conf.TestnetConfig.GenesisTimestamp, 0)
+	} else {
+		network, err := eth2util.ForkVersionToNetwork(lock.ForkVersion)
+		if err != nil {
+			network = "mainnet"
+		}
+		slotDuration, err = eth2util.NetworkToSlotDuration(network)
+		if err != nil {
+			return err
+		}
+		genesisTime, err = eth2util.NetworkToGenesisTime(network)
+		if err != nil {
+			return err
+		}
+
+	}
+	parSigDB := parsigdb.NewMemDB(ctx, lock.Threshold, deadlinerFunc("parsigdb"), parsigdb.NewMemDBMetadata(slotDuration, genesisTime))
 
 	var parSigEx core.ParSigEx
 	if conf.TestConfig.ParSigExFunc != nil {

core/dutydb/memory.go

@@ -34,7 +34,6 @@ func NewMemDB(deadliner core.Deadliner) *MemDB {
 }
 
 // MemDB is an in-memory dutyDB implementation.
-// It is a placeholder for the badgerDB implementation.
 type MemDB struct {
 	mu sync.Mutex
 

core/parsigdb/memory.go

@@ -6,26 +6,41 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"strconv"
 	"sync"
+	"time"
 
 	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/app/log"
 	"github.com/obolnetwork/charon/app/z"
 	"github.com/obolnetwork/charon/core"
 )
 
+// NewMemDBMetadata returns a new in-memory partial signature database instance.
+func NewMemDBMetadata(slotDuration uint64, genesisTime time.Time) MemDBMetadata {
+	return MemDBMetadata{
+		slotDuration: slotDuration,
+		genesisTime:  genesisTime,
+	}
+}
+
+type MemDBMetadata struct {
+	slotDuration uint64
+	genesisTime  time.Time
+}
+
 // NewMemDB returns a new in-memory partial signature database instance.
-func NewMemDB(threshold int, deadliner core.Deadliner) *MemDB {
+func NewMemDB(ctx context.Context, threshold int, deadliner core.Deadliner, metadata MemDBMetadata) *MemDB {
 	return &MemDB{
 		entries:    make(map[key][]core.ParSignedData),
 		keysByDuty: make(map[core.Duty][]key),
 		threshold:  threshold,
 		deadliner:  deadliner,
+		metadata:   metadata,
 	}
 }
 
-// MemDB is a placeholder in-memory partial signature database.
-// It will be replaced with a BadgerDB implementation.
+// MemDB is an in-memory partial signature database.
 type MemDB struct {
 	mu           sync.Mutex
 	internalSubs []func(context.Context, core.Duty, core.ParSignedDataSet) error
@@ -35,6 +50,8 @@ type MemDB struct {
 	keysByDuty map[core.Duty][]key
 	threshold  int
 	deadliner  core.Deadliner
+
+	metadata MemDBMetadata
 }
 
 // SubscribeInternal registers a callback when an internal
@@ -146,6 +163,18 @@ func (db *MemDB) store(k key, value core.ParSignedData) ([]core.ParSignedData, b
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
+	now := time.Now().UnixMilli()
+
+	slotStart := (uint64(db.metadata.genesisTime.Unix()) + k.Duty.Slot*db.metadata.slotDuration) * 1000 // in ms
+	timeSinceSlotStart := float64(now-int64(slotStart)) / 1000                                          // in seconds
+	switch k.Duty.Type {
+	case core.DutyAttester:
+		timeSinceSlotStart -= 4.0
+	case core.DutyAggregator, core.DutySyncContribution:
+		timeSinceSlotStart -= 8.0
+	}
+	parsigStored.WithLabelValues(k.Duty.Type.String(), strconv.FormatInt(int64(value.ShareIdx), 10)).Observe(timeSinceSlotStart)
+
 	for _, s := range db.entries[k] {
 		if s.ShareIdx == value.ShareIdx {
 			equal, err := parSignedDataEqual(s, value)

core/parsigdb/memory_internal_test.go

@@ -5,6 +5,7 @@ package parsigdb
 import (
 	"context"
 	"testing"
+	"time"
 
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
 	"github.com/attestantio/go-eth2-client/spec/altair"
@@ -13,6 +14,7 @@ import (
 
 	"github.com/obolnetwork/charon/cluster"
 	"github.com/obolnetwork/charon/core"
+	"github.com/obolnetwork/charon/eth2util"
 	"github.com/obolnetwork/charon/testutil"
 )
 
@@ -115,7 +117,7 @@ func TestMemDBThreshold(t *testing.T) {
 	)
 
 	deadliner := newTestDeadliner()
-	db := NewMemDB(th, deadliner)
+	db := NewMemDB(t.Context(), th, deadliner, NewMemDBMetadata(eth2util.Mainnet.SlotDuration, time.Unix(eth2util.Mainnet.GenesisTimestamp, 0)))
 
 	ctx := t.Context()
 

core/parsigdb/metrics.go

@@ -14,3 +14,11 @@ var exitCounter = promauto.NewCounterVec(prometheus.CounterOpts{
 	Name:      "exit_total",
 	Help:      "Total number of partially signed voluntary exits per public key",
 }, []string{"pubkey"}) // Ok to use pubkey (high cardinality) here since these are very rare
+
+var parsigStored = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Namespace: "core",
+	Subsystem: "parsigdb",
+	Name:      "store",
+	Help:      "Latency of partial signatures received since earliest expected time, per duty, per peer index",
+	Buckets:   []float64{.001, 0.01, 0.05, .1, .25, .5, .75, 1, 1.25, 1.5, 1.75, 2.0, 2.25, 2.5, 2.75, 3, 5},
+}, []string{"duty", "peer_idx"})

dkg/dkg.go

@@ -242,7 +242,7 @@ func Run(ctx context.Context, conf Config) (err error) {
 		return errors.Wrap(err, "get peer IDs")
 	}
 
-	ex := newExchanger(p2pNode, nodeIdx.PeerIdx, peerIDs, []sigType{
+	ex := newExchanger(ctx, p2pNode, nodeIdx.PeerIdx, peerIDs, []sigType{
 		sigLock,
 		sigDepositData,
 		sigValidatorRegistration,

dkg/exchanger.go

@@ -56,7 +56,7 @@ type exchanger struct {
 	sigDatasChan  chan map[core.PubKey][]core.ParSignedData
 }
 
-func newExchanger(p2pNode host.Host, peerIdx int, peers []peer.ID, sigTypes []sigType, timeout time.Duration) *exchanger {
+func newExchanger(ctx context.Context, p2pNode host.Host, peerIdx int, peers []peer.ID, sigTypes []sigType, timeout time.Duration) *exchanger {
 	// Partial signature roots not known yet, so skip verification in parsigex, rather verify before we aggregate.
 	noopVerifier := func(context.Context, core.Duty, core.PubKey, core.ParSignedData) error {
 		return nil
@@ -82,7 +82,7 @@ func newExchanger(p2pNode host.Host, peerIdx int, peers []peer.ID, sigTypes []si
 
 	ex := &exchanger{
 		// threshold is len(peers) to wait until we get all the partial sigs from all the peers per DV
-		sigdb:    parsigdb.NewMemDB(len(peers), noopDeadliner{}),
+		sigdb:    parsigdb.NewMemDB(ctx, len(peers), noopDeadliner{}, parsigdb.NewMemDBMetadata(0, time.Now())), // metadata timestamps are used for metrics, irrelevant for DKG
 		sigex:    parsigex.NewParSigEx(p2pNode, p2p.Send, peerIdx, peers, noopVerifier, dutyGaterFunc, p2p.WithSendTimeout(timeout), p2p.WithReceiveTimeout(timeout)),
 		sigTypes: st,
 		sigData: dataByPubkey{

dkg/exchanger_internal_test.go

@@ -94,7 +94,7 @@ func TestExchanger(t *testing.T) {
 	}
 
 	for i := range nodes {
-		ex := newExchanger(hosts[i], i, peers, expectedSigTypes, 8*time.Second)
+		ex := newExchanger(t.Context(), hosts[i], i, peers, expectedSigTypes, 8*time.Second)
 		exchangers = append(exchangers, ex)
 	}
 

dkg/protocol_addoperators.go

@@ -85,7 +85,7 @@ func (p *addOperatorsProtocol) PostInit(ctx context.Context, pctx *ProtocolConte
 
 	p.allENRs = append(p.allENRs, p.newENRs...)
 
-	pctx.SigExchanger = newExchanger(pctx.ThisNode, pctx.ThisNodeIdx.PeerIdx, pctx.PeerIDs, []sigType{sigLock}, pctx.Config.Timeout)
+	pctx.SigExchanger = newExchanger(ctx, pctx.ThisNode, pctx.ThisNodeIdx.PeerIdx, pctx.PeerIDs, []sigType{sigLock}, pctx.Config.Timeout)
 	pctx.Caster = bcast.New(pctx.ThisNode, pctx.PeerIDs, pctx.ENRPrivateKey)
 	pctx.NodeSigCaster = newNodeSigBcast(pctx.Peers, pctx.ThisNodeIdx, pctx.Caster)
 

dkg/protocol_removeoperators.go

@@ -151,7 +151,7 @@ func (p *removeOperatorsProtocol) PostInit(ctx context.Context, pctx *ProtocolCo
 			PeerIdx:  nodeIdx,
 			ShareIdx: nodeIdx + 1,
 		}
-		pctx.SigExchanger = newExchanger(pctx.ThisNode, nodeIdx, newPeerIDs, []sigType{sigLock}, pctx.Config.Timeout)
+		pctx.SigExchanger = newExchanger(ctx, pctx.ThisNode, nodeIdx, newPeerIDs, []sigType{sigLock}, pctx.Config.Timeout)
 	}
 
 	reshareConfig := pedersen.NewReshareConfig(len(pctx.Lock.Validators), p.newThreshold, nil, oldPeerIDs)