Merge pull request #519 from shutter-network/fix/peer-metrics-above-t…

…arget

Fix: Always update libp2p peer metrics

rolling-shutter/go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/AdamSLevy/jsonrpc2/v14 v14.1.0
 	github.com/benbjohnson/clock v1.3.5
 	github.com/bitwurx/jrpc2 v0.0.0-20220302204700-52c6dbbeb536
+	github.com/deckarep/golang-set/v2 v2.1.0
 	github.com/deepmap/oapi-codegen v1.9.1
 	github.com/ethereum/go-ethereum v1.13.11
 	github.com/ferranbt/fastssz v0.1.3
@@ -85,7 +86,6 @@ require (
 	github.com/creachadair/taskgroup v0.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
-	github.com/deckarep/golang-set/v2 v2.1.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
 	github.com/dgraph-io/badger/v2 v2.2007.4 // indirect
 	github.com/dgraph-io/ristretto v0.0.3 // indirect

rolling-shutter/p2p/dht.go

@@ -5,6 +5,7 @@ import (
 	"math/rand"
 	"time"
 
+	mapset "github.com/deckarep/golang-set/v2"
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/discovery"
@@ -78,9 +79,15 @@ func findPeers(ctx context.Context, h host.Host, d discovery.Discoverer, ns stri
 		case <-ctx.Done():
 			return ctx.Err()
 		case <-ticker.C:
+			allPeerIDs := mapset.NewSet[peer.ID]()
+			for _, peerID := range h.Network().Peers() {
+				allPeerIDs.Add(peerID)
+			}
+
 			peersBefore := len(h.Network().Peers())
 			if peersBefore >= peerTarget {
 				log.Debug().Int("peers-before", peersBefore).Int("peer-target", peerTarget).Msg("have enough peers")
+				updatePeersMetrics(h, allPeerIDs)
 				continue
 			}
 
@@ -91,22 +98,19 @@ func findPeers(ctx context.Context, h host.Host, d discovery.Discoverer, ns stri
 
 			newConnections := 0
 			failedDials := 0
-			ourId := h.ID().String()
 
-			randomizedPeers := randomizePeers(peers)
+			for _, p := range peers {
+				allPeerIDs.Add(p.ID)
+			}
+			updatePeersMetrics(h, allPeerIDs)
 
+			randomizedPeers := randomizePeers(peers)
 			for _, p := range randomizedPeers {
 				collectPeerAddresses(p)
 				if p.ID == h.ID() {
 					continue
 				}
-				connectedness := h.Network().Connectedness(p.ID)
-				metricsP2PPeerConnectedness.WithLabelValues(ourId, p.ID.String()).Set(float64(connectedness))
-				peerPing := h.Peerstore().LatencyEWMA(p.ID)
-				if peerPing != 0 {
-					metricsP2PPeerPing.WithLabelValues(ourId, p.ID.String()).Set(peerPing.Seconds())
-				}
-				if connectedness != network.Connected {
+				if h.Network().Connectedness(p.ID) != network.Connected {
 					_, err = h.Network().DialPeer(ctx, p.ID)
 					if err != nil {
 						log.Debug().

rolling-shutter/p2p/metrics.go

@@ -1,6 +1,8 @@
 package p2p
 
 import (
+	mapset "github.com/deckarep/golang-set/v2"
+	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -68,3 +70,15 @@ func init() {
 	prometheus.MustRegister(metricsP2PPeerConnectedness)
 	prometheus.MustRegister(metricsP2PPeerPing)
 }
+
+func updatePeersMetrics(h host.Host, peerIds mapset.Set[peer.ID]) {
+	ourID := h.ID().String()
+	for p := range peerIds.Iterator().C {
+		connectedness := h.Network().Connectedness(p)
+		metricsP2PPeerConnectedness.WithLabelValues(ourID, p.String()).Set(float64(connectedness))
+		peerPing := h.Peerstore().LatencyEWMA(p)
+		if peerPing != 0 {
+			metricsP2PPeerPing.WithLabelValues(ourID, p.String()).Set(peerPing.Seconds())
+		}
+	}
+}