update metrics with fork digest for p2p (#5251)

* update metrics with fork digest for p2p

* update p2p metrics

* update metrics using att values

* wrapped up

* fix bug

Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

beacon-chain/p2p/monitoring.go

@@ -6,11 +6,6 @@ import (
 )
 
 var (
-	p2pTopicPeerCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "p2p_topic_peer_count",
-		Help: "The number of peers subscribed to a given topic.",
-	},
-		[]string{"topic"})
 	p2pPeerCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "p2p_peer_count",
 		Help: "The number of peers in a given state.",
@@ -19,10 +14,6 @@ var (
 )
 
 func (s *Service) updateMetrics() {
-	for topic := range GossipTopicMappings {
-		topic += s.Encoding().ProtocolSuffix()
-		p2pTopicPeerCount.WithLabelValues(topic).Set(float64(len(s.pubsub.ListPeers(topic))))
-	}
 	p2pPeerCount.WithLabelValues("Connected").Set(float64(len(s.peers.Connected())))
 	p2pPeerCount.WithLabelValues("Disconnected").Set(float64(len(s.peers.Disconnected())))
 	p2pPeerCount.WithLabelValues("Connecting").Set(float64(len(s.peers.Connecting())))

beacon-chain/sync/metrics.go

@@ -1,11 +1,23 @@
 package sync
 
 import (
+	"fmt"
+	"reflect"
+	"strings"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	pb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
+	"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
 )
 
 var (
+	topicPeerCount = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "p2p_topic_peer_count",
+			Help: "The number of peers subscribed to a given topic.",
+		}, []string{"topic"},
+	)
 	messageReceivedCounter = promauto.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "p2p_message_received_total",
@@ -58,3 +70,32 @@ var (
 		},
 	)
 )
+
+func (r *Service) updateMetrics() {
+	// We update the dynamic subnet topics.
+	digest, err := r.p2p.ForkDigest()
+	if err != nil {
+		log.WithError(err).Errorf("Could not compute fork digest")
+	}
+	indices := r.committeeIndices()
+	attTopic := p2p.GossipTypeMapping[reflect.TypeOf(&pb.Attestation{})]
+	attTopic += r.p2p.Encoding().ProtocolSuffix()
+	for _, committeeIdx := range indices {
+		formattedTopic := fmt.Sprintf(attTopic, digest, committeeIdx)
+		topicPeerCount.WithLabelValues(formattedTopic).Set(float64(len(r.p2p.PubSub().ListPeers(formattedTopic))))
+	}
+	// We update all other gossip topics.
+	for topic := range p2p.GossipTopicMappings {
+		// We already updated attestation subnet topics.
+		if strings.Contains(topic, "committee_index") {
+			continue
+		}
+		topic += r.p2p.Encoding().ProtocolSuffix()
+		if !strings.Contains(topic, "%x") {
+			topicPeerCount.WithLabelValues(topic).Set(float64(len(r.p2p.PubSub().ListPeers(topic))))
+			continue
+		}
+		formattedTopic := fmt.Sprintf(topic, digest)
+		topicPeerCount.WithLabelValues(formattedTopic).Set(float64(len(r.p2p.PubSub().ListPeers(formattedTopic))))
+	}
+}

beacon-chain/sync/service.go

@@ -137,6 +137,9 @@ func (r *Service) Start() {
 	r.maintainPeerStatuses()
 	r.resyncIfBehind()
 	r.refreshENR()
+
+	// Update sync metrics.
+	runutil.RunEvery(r.ctx, time.Second*10, r.updateMetrics)
 }
 
 // Stop the regular sync service.