*: store parsigdb metrics (#4214)

Add histogram metric to track when charon received partial sig from peer compared to the earliest expected time.

Mind you that there might be differences when validator clients send their sync duties.

category: feature
ticket: none

Author: KaloyanTanev

app/app.go

@@ -567,7 +567,32 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		return err
 	}
 
-	parSigDB := parsigdb.NewMemDB(lock.Threshold, deadlinerFunc("parsigdb"))
+	var (
+		slotDuration uint64
+		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(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(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,21 @@ 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
+	default:
+	}
+
+	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(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/exchanger.go

@@ -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(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/pedersen/reshare.go

@@ -124,6 +124,7 @@ func RunReshareDKG(ctx context.Context, config *Config, board *Board, shares []s
 		for _, removedPeerID := range config.Reshare.RemovedPeers {
 			if idx, ok := config.PeerMap[removedPeerID]; ok && idx.PeerIdx == int(node.Index) {
 				isRemoving = true
+
 				if idx.PeerIdx == thisNodeIndex {
 					thisIsRemovedNode = true
 				}
@@ -136,6 +137,7 @@ func RunReshareDKG(ctx context.Context, config *Config, board *Board, shares []s
 		for _, addedPeerID := range config.Reshare.AddedPeers {
 			if idx, ok := config.PeerMap[addedPeerID]; ok && idx.PeerIdx == int(node.Index) {
 				isNewlyAdded = true
+
 				if idx.PeerIdx == thisNodeIndex {
 					thisIsAddedNode = true
 				}

docs/metrics.md

@@ -57,6 +57,7 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance.
 | `core_consensus_timeout_total` | Counter | Total count of consensus timeouts by protocol, duty, and timer | `protocol, duty, timer` |
 | `core_fetcher_proposal_blinded` | Gauge | Whether the fetched proposal was blinded (1) or local (2) |  |
 | `core_parsigdb_exit_total` | Counter | Total number of partially signed voluntary exits per public key | `pubkey` |
+| `core_parsigdb_store` | Histogram | Latency of partial signatures received since earliest expected time, per duty, per peer index | `duty, peer_idx` |
 | `core_scheduler_current_epoch` | Gauge | The current epoch |  |
 | `core_scheduler_current_slot` | Gauge | The current slot |  |
 | `core_scheduler_duty_total` | Counter | The total count of duties scheduled by type | `duty` |

eth2util/network.go

@@ -27,6 +27,8 @@ type Network struct {
 	GenesisTimestamp int64
 	// CapellaHardFork represents capella fork version, used for computing domains for signatures
 	CapellaHardFork string
+	// SlotDuration represents slot duration in seconds
+	SlotDuration uint64
 }
 
 // IsNonZero checks if each field in this struct is not equal to its zero value.
@@ -42,34 +44,39 @@ var (
 		GenesisForkVersionHex: "0x00000000",
 		GenesisTimestamp:      1606824023,
 		CapellaHardFork:       "0x03000000",
+		SlotDuration:          12,
 	}
 	Goerli = Network{
 		ChainID:               5,
 		Name:                  "goerli",
 		GenesisForkVersionHex: "0x00001020",
 		GenesisTimestamp:      1616508000,
 		CapellaHardFork:       "0x03001020",
+		SlotDuration:          12,
 	}
 	Gnosis = Network{
 		ChainID:               100,
 		Name:                  "gnosis",
 		GenesisForkVersionHex: "0x00000064",
 		GenesisTimestamp:      1638993340,
 		CapellaHardFork:       "0x03000064",
+		SlotDuration:          5,
 	}
 	Chiado = Network{
 		ChainID:               10200,
 		Name:                  "chiado",
 		GenesisForkVersionHex: "0x0000006f",
 		GenesisTimestamp:      1665396300,
 		CapellaHardFork:       "0x0300006f",
+		SlotDuration:          5,
 	}
 	Sepolia = Network{
 		ChainID:               11155111,
 		Name:                  "sepolia",
 		GenesisForkVersionHex: "0x90000069",
 		GenesisTimestamp:      1655733600,
 		CapellaHardFork:       "0x90000072",
+		SlotDuration:          12,
 	}
 	// Holesky metadata taken from https://github.com/eth-clients/holesky#metadata.
 	Holesky = Network{
@@ -78,6 +85,7 @@ var (
 		GenesisForkVersionHex: "0x01017000",
 		GenesisTimestamp:      1696000704,
 		CapellaHardFork:       "0x04017000",
+		SlotDuration:          12,
 	}
 	// Hoodi metadata taken from https://github.com/eth-clients/hoodi/#metadata.
 	Hoodi = Network{
@@ -86,6 +94,7 @@ var (
 		GenesisForkVersionHex: "0x10000910",
 		GenesisTimestamp:      1742213400,
 		CapellaHardFork:       "0x40000910",
+		SlotDuration:          12,
 	}
 )
 
@@ -192,6 +201,15 @@ func NetworkToGenesisTime(name string) (time.Time, error) {
 	return time.Unix(network.GenesisTimestamp, 0), nil
 }
 
+func NetworkToSlotDuration(name string) (uint64, error) {
+	network, err := networkFromName(name)
+	if err != nil {
+		return 0, err
+	}
+
+	return network.SlotDuration, nil
+}
+
 func ForkVersionToGenesisTime(forkVersion []byte) (time.Time, error) {
 	network, err := networkFromForkVersion(fmt.Sprintf("%#x", forkVersion))
 	if err != nil {