Add some metrics improvements (#15922)

* Define TCP and QUIC as `InternetProtocol` (no functional change).

* Group types. (No functional changes)

* Rename variables and use range syntax.

* Add `p2pMaxPeers` and `p2pPeerCountDirectionType` metrics

* `p2p_subscribed_topic_peer_total`: Reset to avoid dangling values.

* `validateConfig`:
- Use `Warning` with fields instead of `Warnf`.
- Avoid to both modify in place the input value and return it.

* Add `p2p_minimum_peers_per_subnet` metric.

* `beaconConfig` => `cfg`.

#15880 (comment)

* Add changelog

---------

Co-authored-by: james-prysm <[email protected]>

Author: nalepae

beacon-chain/blockchain/service.go

@@ -472,8 +472,8 @@ func (s *Service) removeStartupState() {
 func (s *Service) updateCustodyInfoInDB(slot primitives.Slot) (primitives.Slot, uint64, error) {
 	isSubscribedToAllDataSubnets := flags.Get().SubscribeAllDataSubnets
 
-	beaconConfig := params.BeaconConfig()
-	custodyRequirement := beaconConfig.CustodyRequirement
+	cfg := params.BeaconConfig()
+	custodyRequirement := cfg.CustodyRequirement
 
 	// Check if the node was previously subscribed to all data subnets, and if so,
 	// store the new status accordingly.
@@ -493,7 +493,7 @@ func (s *Service) updateCustodyInfoInDB(slot primitives.Slot) (primitives.Slot,
 	// Compute the custody group count.
 	custodyGroupCount := custodyRequirement
 	if isSubscribedToAllDataSubnets {
-		custodyGroupCount = beaconConfig.NumberOfCustodyGroups
+		custodyGroupCount = cfg.NumberOfCustodyGroups
 	}
 
 	// Safely compute the fulu fork slot.
@@ -536,11 +536,11 @@ func spawnCountdownIfPreGenesis(ctx context.Context, genesisTime time.Time, db d
 }
 
 func fuluForkSlot() (primitives.Slot, error) {
-	beaconConfig := params.BeaconConfig()
+	cfg := params.BeaconConfig()
 
-	fuluForkEpoch := beaconConfig.FuluForkEpoch
-	if fuluForkEpoch == beaconConfig.FarFutureEpoch {
-		return beaconConfig.FarFutureSlot, nil
+	fuluForkEpoch := cfg.FuluForkEpoch
+	if fuluForkEpoch == cfg.FarFutureEpoch {
+		return cfg.FarFutureSlot, nil
 	}
 
 	forkFuluSlot, err := slots.EpochStart(fuluForkEpoch)

beacon-chain/core/helpers/validators.go

@@ -401,7 +401,7 @@ func ComputeProposerIndex(bState state.ReadOnlyBeaconState, activeIndices []prim
 		return 0, errors.New("empty active indices list")
 	}
 	hashFunc := hash.CustomSHA256Hasher()
-	beaconConfig := params.BeaconConfig()
+	cfg := params.BeaconConfig()
 	seedBuffer := make([]byte, len(seed)+8)
 	copy(seedBuffer, seed[:])
 
@@ -426,14 +426,14 @@ func ComputeProposerIndex(bState state.ReadOnlyBeaconState, activeIndices []prim
 			offset := (i % 16) * 2
 			randomValue := uint64(randomBytes[offset]) | uint64(randomBytes[offset+1])<<8
 
-			if effectiveBal*fieldparams.MaxRandomValueElectra >= beaconConfig.MaxEffectiveBalanceElectra*randomValue {
+			if effectiveBal*fieldparams.MaxRandomValueElectra >= cfg.MaxEffectiveBalanceElectra*randomValue {
 				return candidateIndex, nil
 			}
 		} else {
 			binary.LittleEndian.PutUint64(seedBuffer[len(seed):], i/32)
 			randomByte := hashFunc(seedBuffer)[i%32]
 
-			if effectiveBal*fieldparams.MaxRandomByte >= beaconConfig.MaxEffectiveBalance*uint64(randomByte) {
+			if effectiveBal*fieldparams.MaxRandomByte >= cfg.MaxEffectiveBalance*uint64(randomByte) {
 				return candidateIndex, nil
 			}
 		}

beacon-chain/core/peerdas/das_core.go

@@ -89,14 +89,14 @@ func CustodyGroups(nodeId enode.ID, custodyGroupCount uint64) ([]uint64, error)
 // ComputeColumnsForCustodyGroup computes the columns for a given custody group.
 // https://github.com/ethereum/consensus-specs/blob/master/specs/fulu/das-core.md#compute_columns_for_custody_group
 func ComputeColumnsForCustodyGroup(custodyGroup uint64) ([]uint64, error) {
-	beaconConfig := params.BeaconConfig()
-	numberOfCustodyGroups := beaconConfig.NumberOfCustodyGroups
+	cfg := params.BeaconConfig()
+	numberOfCustodyGroups := cfg.NumberOfCustodyGroups
 
 	if custodyGroup >= numberOfCustodyGroups {
 		return nil, ErrCustodyGroupTooLarge
 	}
 
-	numberOfColumns := beaconConfig.NumberOfColumns
+	numberOfColumns := cfg.NumberOfColumns
 
 	columnsPerGroup := numberOfColumns / numberOfCustodyGroups
 
@@ -112,9 +112,9 @@ func ComputeColumnsForCustodyGroup(custodyGroup uint64) ([]uint64, error) {
 // ComputeCustodyGroupForColumn computes the custody group for a given column.
 // It is the reciprocal function of ComputeColumnsForCustodyGroup.
 func ComputeCustodyGroupForColumn(columnIndex uint64) (uint64, error) {
-	beaconConfig := params.BeaconConfig()
-	numberOfColumns := beaconConfig.NumberOfColumns
-	numberOfCustodyGroups := beaconConfig.NumberOfCustodyGroups
+	cfg := params.BeaconConfig()
+	numberOfColumns := cfg.NumberOfColumns
+	numberOfCustodyGroups := cfg.NumberOfCustodyGroups
 
 	if columnIndex >= numberOfColumns {
 		return 0, ErrIndexTooLarge

beacon-chain/core/peerdas/validator.go

@@ -84,10 +84,10 @@ func ValidatorsCustodyRequirement(state beaconState.ReadOnlyBeaconState, validat
 		totalNodeBalance += validator.EffectiveBalance()
 	}
 
-	beaconConfig := params.BeaconConfig()
-	numberOfCustodyGroups := beaconConfig.NumberOfCustodyGroups
-	validatorCustodyRequirement := beaconConfig.ValidatorCustodyRequirement
-	balancePerAdditionalCustodyGroup := beaconConfig.BalancePerAdditionalCustodyGroup
+	cfg := params.BeaconConfig()
+	numberOfCustodyGroups := cfg.NumberOfCustodyGroups
+	validatorCustodyRequirement := cfg.ValidatorCustodyRequirement
+	balancePerAdditionalCustodyGroup := cfg.BalancePerAdditionalCustodyGroup
 
 	count := totalNodeBalance / balancePerAdditionalCustodyGroup
 	return min(max(count, validatorCustodyRequirement), numberOfCustodyGroups), nil

beacon-chain/core/time/slot_epoch_test.go

@@ -196,7 +196,7 @@ func TestAltairCompatible(t *testing.T) {
 }
 
 func TestCanUpgradeTo(t *testing.T) {
-	beaconConfig := params.BeaconConfig()
+	cfg := params.BeaconConfig()
 
 	outerTestCases := []struct {
 		name        string
@@ -205,40 +205,40 @@ func TestCanUpgradeTo(t *testing.T) {
 	}{
 		{
 			name:        "Altair",
-			forkEpoch:   &beaconConfig.AltairForkEpoch,
+			forkEpoch:   &cfg.AltairForkEpoch,
 			upgradeFunc: time.CanUpgradeToAltair,
 		},
 		{
 			name:        "Bellatrix",
-			forkEpoch:   &beaconConfig.BellatrixForkEpoch,
+			forkEpoch:   &cfg.BellatrixForkEpoch,
 			upgradeFunc: time.CanUpgradeToBellatrix,
 		},
 		{
 			name:        "Capella",
-			forkEpoch:   &beaconConfig.CapellaForkEpoch,
+			forkEpoch:   &cfg.CapellaForkEpoch,
 			upgradeFunc: time.CanUpgradeToCapella,
 		},
 		{
 			name:        "Deneb",
-			forkEpoch:   &beaconConfig.DenebForkEpoch,
+			forkEpoch:   &cfg.DenebForkEpoch,
 			upgradeFunc: time.CanUpgradeToDeneb,
 		},
 		{
 			name:        "Electra",
-			forkEpoch:   &beaconConfig.ElectraForkEpoch,
+			forkEpoch:   &cfg.ElectraForkEpoch,
 			upgradeFunc: time.CanUpgradeToElectra,
 		},
 		{
 			name:        "Fulu",
-			forkEpoch:   &beaconConfig.FuluForkEpoch,
+			forkEpoch:   &cfg.FuluForkEpoch,
 			upgradeFunc: time.CanUpgradeToFulu,
 		},
 	}
 
 	for _, otc := range outerTestCases {
 		params.SetupTestConfigCleanup(t)
 		*otc.forkEpoch = 5
-		params.OverrideBeaconConfig(beaconConfig)
+		params.OverrideBeaconConfig(cfg)
 
 		innerTestCases := []struct {
 			name string

beacon-chain/p2p/config.go

@@ -7,6 +7,7 @@ import (
 	statefeed "github.com/OffchainLabs/prysm/v6/beacon-chain/core/feed/state"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/db"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/startup"
+	"github.com/sirupsen/logrus"
 )
 
 // This is the default queue size used if we have specified an invalid one.
@@ -63,12 +64,17 @@ func (cfg *Config) connManagerLowHigh() (int, int) {
 	return low, high
 }
 
-// validateConfig validates whether the values provided are accurate and will set
-// the appropriate values for those that are invalid.
-func validateConfig(cfg *Config) *Config {
-	if cfg.QueueSize == 0 {
-		log.Warnf("Invalid pubsub queue size of %d initialized, setting the quese size as %d instead", cfg.QueueSize, defaultPubsubQueueSize)
-		cfg.QueueSize = defaultPubsubQueueSize
+// validateConfig validates whether the provided config has valid values and sets
+// the invalid ones to default.
+func validateConfig(cfg *Config) {
+	if cfg.QueueSize > 0 {
+		return
 	}
-	return cfg
+
+	log.WithFields(logrus.Fields{
+		"queueSize": cfg.QueueSize,
+		"default":   defaultPubsubQueueSize,
+	}).Warning("Invalid pubsub queue size, setting the queue size to the default value")
+
+	cfg.QueueSize = defaultPubsubQueueSize
 }

beacon-chain/p2p/custody.go

@@ -259,11 +259,11 @@ func (s *Service) custodyGroupCountFromPeerENR(pid peer.ID) uint64 {
 }
 
 func fuluForkSlot() (primitives.Slot, error) {
-	beaconConfig := params.BeaconConfig()
+	cfg := params.BeaconConfig()
 
-	fuluForkEpoch := beaconConfig.FuluForkEpoch
-	if fuluForkEpoch == beaconConfig.FarFutureEpoch {
-		return beaconConfig.FarFutureSlot, nil
+	fuluForkEpoch := cfg.FuluForkEpoch
+	if fuluForkEpoch == cfg.FarFutureEpoch {
+		return cfg.FarFutureSlot, nil
 	}
 
 	forkFuluSlot, err := slots.EpochStart(fuluForkEpoch)

beacon-chain/p2p/monitoring.go

@@ -3,6 +3,7 @@ package p2p
 import (
 	"strings"
 
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/peers"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/prometheus/client_golang/prometheus"
@@ -26,12 +27,25 @@ var (
 		Help: "The number of peers in a given state.",
 	},
 		[]string{"state"})
+	p2pMaxPeers = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "p2p_max_peers",
+		Help: "The target maximum number of peers.",
+	})
+	p2pPeerCountDirectionType = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "p2p_peer_count_direction_type",
+		Help: "The number of peers in a given direction and type.",
+	},
+		[]string{"direction", "type"})
 	connectedPeersCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "connected_libp2p_peers",
 		Help: "Tracks the total number of connected libp2p peers by agent string",
 	},
 		[]string{"agent"},
 	)
+	minimumPeersPerSubnet = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "p2p_minimum_peers_per_subnet",
+		Help: "The minimum number of peers to connect to per subnet",
+	})
 	avgScoreConnectedClients = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "connected_libp2p_peers_average_scores",
 		Help: "Tracks the overall p2p scores of connected libp2p peers by agent string",
@@ -174,35 +188,45 @@ var (
 )
 
 func (s *Service) updateMetrics() {
+	store := s.Host().Peerstore()
 	connectedPeers := s.peers.Connected()
+
 	p2pPeerCount.WithLabelValues("Connected").Set(float64(len(connectedPeers)))
 	p2pPeerCount.WithLabelValues("Disconnected").Set(float64(len(s.peers.Disconnected())))
 	p2pPeerCount.WithLabelValues("Connecting").Set(float64(len(s.peers.Connecting())))
 	p2pPeerCount.WithLabelValues("Disconnecting").Set(float64(len(s.peers.Disconnecting())))
 	p2pPeerCount.WithLabelValues("Bad").Set(float64(len(s.peers.Bad())))
 
-	store := s.Host().Peerstore()
-	numConnectedPeersByClient := make(map[string]float64)
+	upperTCP := strings.ToUpper(string(peers.TCP))
+	upperQUIC := strings.ToUpper(string(peers.QUIC))
+
+	p2pPeerCountDirectionType.WithLabelValues("inbound", upperTCP).Set(float64(len(s.peers.InboundConnectedWithProtocol(peers.TCP))))
+	p2pPeerCountDirectionType.WithLabelValues("inbound", upperQUIC).Set(float64(len(s.peers.InboundConnectedWithProtocol(peers.QUIC))))
+	p2pPeerCountDirectionType.WithLabelValues("outbound", upperTCP).Set(float64(len(s.peers.OutboundConnectedWithProtocol(peers.TCP))))
+	p2pPeerCountDirectionType.WithLabelValues("outbound", upperQUIC).Set(float64(len(s.peers.OutboundConnectedWithProtocol(peers.QUIC))))
+
+	connectedPeersCountByClient := make(map[string]float64)
 	peerScoresByClient := make(map[string][]float64)
-	for i := 0; i < len(connectedPeers); i++ {
-		p := connectedPeers[i]
+	for _, p := range connectedPeers {
 		pid, err := peer.Decode(p.String())
 		if err != nil {
 			log.WithError(err).Debug("Could not decode peer string")
 			continue
 		}
 
 		foundName := agentFromPid(pid, store)
-		numConnectedPeersByClient[foundName] += 1
+		connectedPeersCountByClient[foundName] += 1
 
 		// Get peer scoring data.
 		overallScore := s.peers.Scorers().Score(pid)
 		peerScoresByClient[foundName] = append(peerScoresByClient[foundName], overallScore)
 	}
+
 	connectedPeersCount.Reset() // Clear out previous results.
-	for agent, total := range numConnectedPeersByClient {
+	for agent, total := range connectedPeersCountByClient {
 		connectedPeersCount.WithLabelValues(agent).Set(total)
 	}
+
 	avgScoreConnectedClients.Reset() // Clear out previous results.
 	for agent, scoringData := range peerScoresByClient {
 		avgScore := average(scoringData)

beacon-chain/p2p/peers/status.go

@@ -81,29 +81,31 @@ const (
 type InternetProtocol string
 
 const (
-	TCP  = "tcp"
-	QUIC = "quic"
+	TCP  = InternetProtocol("tcp")
+	QUIC = InternetProtocol("quic")
 )
 
-// Status is the structure holding the peer status information.
-type Status struct {
-	ctx                   context.Context
-	scorers               *scorers.Service
-	store                 *peerdata.Store
-	ipTracker             map[string]uint64
-	rand                  *rand.Rand
-	ipColocationWhitelist []*net.IPNet
-}
-
-// StatusConfig represents peer status service params.
-type StatusConfig struct {
-	// PeerLimit specifies maximum amount of concurrent peers that are expected to be connect to the node.
-	PeerLimit int
-	// ScorerParams holds peer scorer configuration params.
-	ScorerParams *scorers.Config
-	// IPColocationWhitelist contains CIDR ranges that are exempt from IP colocation limits.
-	IPColocationWhitelist []*net.IPNet
-}
+type (
+	// Status is the structure holding the peer status information.
+	Status struct {
+		ctx                   context.Context
+		scorers               *scorers.Service
+		store                 *peerdata.Store
+		ipTracker             map[string]uint64
+		rand                  *rand.Rand
+		ipColocationWhitelist []*net.IPNet
+	}
+
+	// StatusConfig represents peer status service params.
+	StatusConfig struct {
+		// PeerLimit specifies maximum amount of concurrent peers that are expected to be connect to the node.
+		PeerLimit int
+		// ScorerParams holds peer scorer configuration params.
+		ScorerParams *scorers.Config
+		// IPColocationWhitelist contains CIDR ranges that are exempt from IP colocation limits.
+		IPColocationWhitelist []*net.IPNet
+	}
+)
 
 // NewStatus creates a new status entity.
 func NewStatus(ctx context.Context, config *StatusConfig) *Status {

beacon-chain/p2p/rpc_topic_mappings.go

@@ -345,17 +345,17 @@ func TopicFromMessage(msg string, epoch primitives.Epoch) (string, error) {
 		return "", errors.Errorf("%s: %s", invalidRPCMessageType, msg)
 	}
 
-	beaconConfig := params.BeaconConfig()
+	cfg := params.BeaconConfig()
 
 	// Check if the message is to be updated in fulu.
-	if epoch >= beaconConfig.FuluForkEpoch {
+	if epoch >= cfg.FuluForkEpoch {
 		if version, ok := fuluMapping[msg]; ok {
 			return protocolPrefix + msg + version, nil
 		}
 	}
 
 	// Check if the message is to be updated in altair.
-	if epoch >= beaconConfig.AltairForkEpoch {
+	if epoch >= cfg.AltairForkEpoch {
 		if version, ok := altairMapping[msg]; ok {
 			return protocolPrefix + msg + version, nil
 		}