p2p --next: Register p2p peer count metrics (#3301)

beacon-chain/p2p/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "handshake.go",
         "interfaces.go",
         "log.go",
+        "monitoring.go",
         "options.go",
         "rpc_topic_mappings.go",
         "sender.go",
@@ -42,6 +43,8 @@ go_library(
         "@com_github_libp2p_go_libp2p_pubsub//:go_default_library",
         "@com_github_multiformats_go_multiaddr//:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
     ],
 )

beacon-chain/p2p/discovery_test.go

@@ -22,7 +22,7 @@ func createAddrAndPrivKey(t *testing.T) (net.IP, *ecdsa.PrivateKey) {
 		t.Fatalf("Could not get ip: %v", err)
 	}
 	ipAddr := net.ParseIP(ip)
-	pkey, err := privKey(&Config{})
+	pkey, err := privKey(&Config{Encoding: "ssz"})
 	if err != nil {
 		t.Fatalf("Could not get private key: %v", err)
 	}
@@ -64,6 +64,7 @@ func TestStartDiscV5_DiscoverAllPeers(t *testing.T) {
 
 	cfg := &Config{
 		BootstrapNodeAddr: bootNode.String(),
+		Encoding: "ssz",
 	}
 
 	var listeners []*discv5.Network
@@ -121,7 +122,7 @@ func TestMultiAddrConversion_OK(t *testing.T) {
 }
 
 func TestStaticPeering_PeersAreAdded(t *testing.T) {
-	cfg := &Config{}
+	cfg := &Config{Encoding: "ssz"}
 	port := 3000
 	var staticPeers []string
 	var hosts []host.Host

beacon-chain/p2p/monitoring.go

@@ -0,0 +1,52 @@
+package p2p
+
+import (
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	p2pTopicPeerCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "p2p_topic_peer_count",
+		Help: "The number of peers subscribed to a topic",
+	},
+	[]string{"topic"})
+)
+
+func registerMetrics(s *Service) {
+
+	// Metrics with a single value can use GaugeFunc, CounterFunc, etc.
+	if err := prometheus.DefaultRegisterer.Register(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "p2p_peer_count",
+		Help: "The number of currently connected peers",
+	}, func() float64 {
+		return float64(peerCount(s.host))
+	})); err != nil {
+		// This should only happen in tests.
+		log.WithError(err).Error("Failed to register metric")
+	}
+
+
+	// Metrics with labels, polled every 10s.
+	go func() {
+		for {
+			updateP2PTopicPeerCount(s)
+
+			time.Sleep(10 * time.Second)
+		}
+	}()
+}
+
+func peerCount(h host.Host) int {
+	return len(h.Network().Peers())
+}
+
+func updateP2PTopicPeerCount(s *Service) {
+	for topic := range GossipTopicMappings {
+		topic += s.Encoding().ProtocolSuffix()
+		p2pTopicPeerCount.WithLabelValues(topic).Set(float64(len(s.pubsub.ListPeers(topic))))
+	}
+}

beacon-chain/p2p/options_test.go

@@ -30,6 +30,7 @@ func TestPrivateKeyLoading(t *testing.T) {
 	log.WithField("file", file.Name()).WithField("key", keyStr).Info("Wrote key to file")
 	cfg := &Config{
 		PrivateKey: file.Name(),
+		Encoding: "ssz",
 	}
 	pKey, err := privKey(cfg)
 	if err != nil {

beacon-chain/p2p/service.go

@@ -102,6 +102,8 @@ func (s *Service) Start() {
 
 	s.started = true
 
+	registerMetrics(s)
+
 	multiAddrs := s.host.Network().ListenAddresses()
 	log.Infof("Node currently listening at %s", multiAddrs[1].String())
 }

beacon-chain/p2p/service_test.go

@@ -93,6 +93,7 @@ func TestService_Start_OnlyStartsOnce(t *testing.T) {
 	cfg := &Config{
 		Port:    2000,
 		UDPPort: 2000,
+		Encoding: "ssz",
 	}
 	s, _ := NewService(cfg)
 	s.dv5Listener = &mockListener{}
@@ -125,6 +126,7 @@ func TestListenForNewNodes(t *testing.T) {
 
 	cfg := &Config{
 		BootstrapNodeAddr: bootNode.String(),
+		Encoding: "ssz",
 	}
 	var listeners []*discv5.Network
 	var hosts []host.Host

shared/deprecated-p2p/monitoring.go

@@ -10,10 +10,6 @@ import (
 )
 
 var (
-	peerCountMetric = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "p2p_peer_count",
-		Help: "The number of currently connected peers",
-	})
 	propagationTimeMetric = promauto.NewHistogram(prometheus.HistogramOpts{
 		Name:    "p2p_propagation_time_sec",
 		Help:    "The time between message sent/received from peer",
@@ -24,6 +20,10 @@ var (
 // starPeerWatcher updates the peer count metric and calls to reconnect any VIP
 // peers such as the bootnode peer, the relay node peer or the static peers.
 func startPeerWatcher(ctx context.Context, h host.Host, reconnectPeers ...string) {
+	peerCountMetric := promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "p2p_peer_count",
+		Help: "The number of currently connected peers",
+	})
 
 	go (func() {
 		for {