Fix p2p/Peers metric (#975)

Do not use server.PeerCount() when getting peers right after receiving add/remove peer event as it does not contain updated info.

metrics/node/metrics.go

@@ -1,22 +1,44 @@
 package node
 
 import (
+	"errors"
+	"flag"
+
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
 )
 
 var (
-	nodePeersGauge    = metrics.NewRegisteredGauge("p2p/Peers", nil)
-	nodeMaxPeersGauge = metrics.NewRegisteredGauge("p2p/MaxPeers", nil)
+	nodePeersCounter  metrics.Counter
+	nodeMaxPeersGauge metrics.Gauge
 )
 
-func updateNodeMetrics(node *node.Node) {
-	server := node.Server()
-	if server == nil {
-		logger.Error("server not available")
-		return
+func init() {
+	// When running tests, we want metrics to be enabled.
+	// Having init() in metrics_test.go does not work because
+	// this init() is executed first.
+	if flag.Lookup("test.v") != nil {
+		metrics.Enabled = true
 	}
 
-	nodePeersGauge.Update(int64(server.PeerCount()))
-	nodeMaxPeersGauge.Update(int64(server.MaxPeers))
+	nodePeersCounter = metrics.NewRegisteredCounter("p2p/Peers", nil)
+	nodeMaxPeersGauge = metrics.NewRegisteredGauge("p2p/MaxPeers", nil)
+}
+
+func updateNodeMetrics(node *node.Node, evType p2p.PeerEventType) error {
+	server := node.Server()
+	if server == nil {
+		return errors.New("p2p server is unavailable")
+	}
+
+	if evType == p2p.PeerEventTypeAdd {
+		nodePeersCounter.Inc(1)
+	} else if evType == p2p.PeerEventTypeDrop {
+		nodePeersCounter.Dec(1)
+	}
+
+	nodeMaxPeersGauge.Update(int64(server.MaxPeers))
+
+	return nil
 }

metrics/node/metrics_test.go

@@ -0,0 +1,47 @@
+package node
+
+import (
+	"testing"
+
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUpdateNodeMetricsPeersCounter(t *testing.T) {
+	var err error
+
+	n, err := node.New(&node.Config{
+		P2P: p2p.Config{
+			MaxPeers: 10,
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, n.Start())
+	defer func() { require.NoError(t, n.Stop()) }()
+
+	err = updateNodeMetrics(n, p2p.PeerEventTypeAdd)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), nodePeersCounter.Count())
+	require.Equal(t, int64(10), nodeMaxPeersGauge.Value())
+
+	err = updateNodeMetrics(n, p2p.PeerEventTypeAdd)
+	require.NoError(t, err)
+	require.Equal(t, int64(2), nodePeersCounter.Count())
+
+	// skip other events
+	err = updateNodeMetrics(n, p2p.PeerEventTypeMsgRecv)
+	require.NoError(t, err)
+	err = updateNodeMetrics(n, p2p.PeerEventTypeMsgSend)
+	require.NoError(t, err)
+	require.Equal(t, int64(2), nodePeersCounter.Count())
+
+	err = updateNodeMetrics(n, p2p.PeerEventTypeDrop)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), nodePeersCounter.Count())
+
+	n.Server().MaxPeers = 20
+	err = updateNodeMetrics(n, p2p.PeerEventTypeDrop)
+	require.NoError(t, err)
+	require.Equal(t, int64(20), nodeMaxPeersGauge.Value())
+}

metrics/node/subscribe.go

@@ -30,11 +30,13 @@ func SubscribeServerEvents(ctx context.Context, node *node.Node) error {
 		select {
 		case event := <-ch:
 			if isAddDropPeerEvent(event.Type) {
-				updateNodeMetrics(node)
+				if err := updateNodeMetrics(node, event.Type); err != nil {
+					log.Error("failed to update node metrics", "err", err)
+				}
 			}
 		case err := <-subscription.Err():
 			if err != nil {
-				logger.Warn("Subscription failed", "err", err)
+				logger.Error("Subscription failed", "err", err)
 			}
 			subscription.Unsubscribe()
 			return err

timesource/timesource_test.go

@@ -12,7 +12,7 @@ import (
 
 const (
 	// clockCompareDelta declares time required between multiple calls to time.Now
-	clockCompareDelta = 30 * time.Microsecond
+	clockCompareDelta = 100 * time.Microsecond
 )
 
 // we don't user real servers for tests, but logic depends on