Observability - instrument P2P component

cli/operator/node.go

@@ -268,7 +268,6 @@ var StartNodeCmd = &cobra.Command{
 			}
 		}
 
-		cfg.P2pNetworkConfig.Metrics = metricsReporter
 		cfg.P2pNetworkConfig.MessageValidator = messageValidator
 		cfg.SSVOptions.ValidatorOptions.MessageValidator = messageValidator
 
@@ -670,7 +669,7 @@ func setupP2P(logger *zap.Logger, db basedb.Database, mr metricsreporter.Metrics
 	}
 	cfg.P2pNetworkConfig.NetworkPrivateKey = netPrivKey
 
-	n, err := p2pv1.New(logger, &cfg.P2pNetworkConfig, mr)
+	n, err := p2pv1.New(logger, &cfg.P2pNetworkConfig)
 	if err != nil {
 		logger.Fatal("failed to setup p2p network", zap.Error(err))
 	}

network/p2p/config.go

@@ -17,7 +17,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/ssvlabs/ssv/message/validation"
-	"github.com/ssvlabs/ssv/monitoring/metricsreporter"
 	"github.com/ssvlabs/ssv/network"
 	"github.com/ssvlabs/ssv/network/commons"
 	"github.com/ssvlabs/ssv/networkconfig"
@@ -75,8 +74,6 @@ type Config struct {
 	Network networkconfig.NetworkConfig
 	// MessageValidator validates incoming messages.
 	MessageValidator validation.MessageValidator
-	// Metrics report metrics.
-	Metrics metricsreporter.MetricsReporter
 
 	PubsubMsgCacheTTL         time.Duration `yaml:"PubsubMsgCacheTTL" env:"PUBSUB_MSG_CACHE_TTL" env-description:"How long a message ID will be remembered as seen"`
 	PubsubOutQueueSize        int           `yaml:"PubsubOutQueueSize" env:"PUBSUB_OUT_Q_SIZE" env-description:"The size that we assign to the outbound pubsub message queue"`

network/p2p/observability.go

@@ -0,0 +1,84 @@
+package p2pv1
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/ssvlabs/ssv/network/peers"
+	"github.com/ssvlabs/ssv/network/topics"
+	"github.com/ssvlabs/ssv/observability"
+)
+
+const (
+	observabilityName      = "github.com/ssvlabs/ssv/network/p2p"
+	observabilityNamespace = "ssv.p2p"
+)
+
+var (
+	meter = otel.Meter(observabilityName)
+
+	peersConnectedGauge = observability.NewMetric(
+		meter.Int64Gauge(
+			metricName("peers.connected"),
+			metric.WithUnit("{peer}"),
+			metric.WithDescription("number of connected peers")))
+
+	peersByTopicCounter = observability.NewMetric(
+		meter.Int64Gauge(
+			metricName("peers.per_topic"),
+			metric.WithUnit("{peer}"),
+			metric.WithDescription("number of connected peers per topic")))
+
+	peerIdentityGauge = observability.NewMetric(
+		meter.Int64Gauge(
+			metricName("peers.per_version"),
+			metric.WithUnit("{peer}"),
+			metric.WithDescription("number of connected peers per node version")))
+)
+
+func metricName(name string) string {
+	return fmt.Sprintf("%s.%s", observabilityNamespace, name)
+}
+
+func recordPeerCount(ctx context.Context, host host.Host) func() {
+	return func() {
+		peers := host.Network().Peers()
+		peersConnectedGauge.Record(ctx, int64(len(peers)))
+	}
+}
+
+func recordPeerCountPerTopic(ctx context.Context, ctrl topics.Controller) func() {
+	return func() {
+		for _, topicName := range ctrl.Topics() {
+			peers, err := ctrl.Peers(topicName)
+			if err != nil {
+				return
+			}
+			peersByTopicCounter.Record(ctx, int64(len(peers)), metric.WithAttributes(attribute.String("ssv.p2p.topic.name", topicName)))
+		}
+	}
+}
+
+func recordPeerIdentities(ctx context.Context, host host.Host, index peers.Index) func() {
+	return func() {
+		peersByVersion := make(map[string]int64)
+		for _, pid := range host.Network().Peers() {
+			nodeVersion := "unknown"
+			ni := index.NodeInfo(pid)
+			if ni != nil {
+				if ni.Metadata != nil {
+					nodeVersion = ni.Metadata.NodeVersion
+				}
+			}
+			peersByVersion[nodeVersion]++
+		}
+		for version, peerCount := range peersByVersion {
+			peerIdentityGauge.Record(ctx, peerCount, metric.WithAttributes(attribute.String("ssv.node.version", version)))
+		}
+	}
+}

network/p2p/p2p.go

@@ -86,7 +86,6 @@ type p2pNetwork struct {
 	connHandler  connections.ConnHandler
 	connGater    connmgr.ConnectionGater
 	trustedPeers []*peer.AddrInfo
-	metrics      Metrics
 
 	state int32
 
@@ -107,7 +106,7 @@ type p2pNetwork struct {
 }
 
 // New creates a new p2p network
-func New(logger *zap.Logger, cfg *Config, mr Metrics) (*p2pNetwork, error) {
+func New(logger *zap.Logger, cfg *Config) (*p2pNetwork, error) {
 	ctx, cancel := context.WithCancel(cfg.Ctx)
 
 	logger = logger.Named(logging.NameP2PNetwork)
@@ -127,7 +126,6 @@ func New(logger *zap.Logger, cfg *Config, mr Metrics) (*p2pNetwork, error) {
 		operatorPKHashToPKCache: hashmap.New[string, []byte](),
 		operatorSigner:          cfg.OperatorSigner,
 		operatorDataStore:       cfg.OperatorDataStore,
-		metrics:                 mr,
 	}
 	if err := n.parseTrustedPeers(); err != nil {
 		return nil, err
@@ -259,14 +257,12 @@ func (n *p2pNetwork) Start(logger *zap.Logger) error {
 	go n.startDiscovery(logger, connector)
 
 	async.Interval(n.ctx, connManagerBalancingInterval, n.peersBalancing(logger))
-	// don't report metrics in tests
-	if n.cfg.Metrics != nil {
-		async.Interval(n.ctx, peersReportingInterval, n.reportAllPeers(logger))
 
-		async.Interval(n.ctx, peerIdentitiesReportingInterval, n.reportPeerIdentities(logger))
+	async.Interval(n.ctx, peersReportingInterval, recordPeerCount(n.ctx, n.host))
 
-		async.Interval(n.ctx, topicsReportingInterval, n.reportTopics(logger))
-	}
+	async.Interval(n.ctx, peerIdentitiesReportingInterval, recordPeerIdentities(n.ctx, n.host, n.idx))
+
+	async.Interval(n.ctx, topicsReportingInterval, recordPeerCountPerTopic(n.ctx, n.topicsCtrl))
 
 	if err := n.subscribeToSubnets(logger); err != nil {
 		return err
@@ -291,7 +287,6 @@ func (n *p2pNetwork) peersBalancing(logger *zap.Logger) func() {
 		// Check if it has the maximum number of connections
 		currentCount := len(allPeers)
 		if currentCount < n.cfg.MaxPeers {
-			_ = n.idx.GetSubnetsStats() // trigger metrics update
 			return
 		}
 

network/p2p/p2p_pubsub.go

@@ -16,9 +16,7 @@ import (
 	"github.com/ssvlabs/ssv/network"
 	"github.com/ssvlabs/ssv/network/commons"
 	"github.com/ssvlabs/ssv/network/records"
-	genesismessage "github.com/ssvlabs/ssv/protocol/genesis/message"
 	"github.com/ssvlabs/ssv/protocol/genesis/ssv/genesisqueue"
-	"github.com/ssvlabs/ssv/protocol/v2/message"
 	p2pprotocol "github.com/ssvlabs/ssv/protocol/v2/p2p"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/queue"
 )
@@ -231,10 +229,8 @@ func (n *p2pNetwork) handlePubsubMessages(logger *zap.Logger) func(ctx context.C
 		switch m := msg.ValidatorData.(type) {
 		case *queue.SSVMessage:
 			decodedMsg = m
-			metricsRouterIncoming.WithLabelValues(message.MsgTypeToString(m.MsgType)).Inc()
 		case *genesisqueue.GenesisSSVMessage:
 			decodedMsg = m
-			metricsRouterIncoming.WithLabelValues(genesismessage.MsgTypeToString(m.MsgType)).Inc()
 		case nil:
 			return errors.New("message was not decoded")
 		default:

network/p2p/test_utils.go

@@ -14,7 +14,6 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/ssvlabs/ssv/message/validation"
-	"github.com/ssvlabs/ssv/monitoring/metricsreporter"
 	"github.com/ssvlabs/ssv/network"
 	"github.com/ssvlabs/ssv/network/commons"
 	p2pcommons "github.com/ssvlabs/ssv/network/commons"
@@ -180,7 +179,6 @@ func (ln *LocalNet) NewTestP2pNetwork(ctx context.Context, nodeIndex uint64, key
 	cfg.Ctx = ctx
 	cfg.Subnets = "00000000000000000100000400000400" // calculated for topics 64, 90, 114; PAY ATTENTION for future test scenarios which use more than one eth-validator we need to make this field dynamically changing
 	cfg.NodeStorage = nodeStorage
-	cfg.Metrics = nil
 	cfg.MessageValidator = validation.New(
 		networkconfig.TestNetwork,
 		nodeStorage.ValidatorStore(),
@@ -224,8 +222,7 @@ func (ln *LocalNet) NewTestP2pNetwork(ctx context.Context, nodeIndex uint64, key
 
 	cfg.OperatorDataStore = operatordatastore.New(&registrystorage.OperatorData{ID: nodeIndex + 1})
 
-	mr := metricsreporter.New()
-	p, err := New(logger, cfg, mr)
+	p, err := New(logger, cfg)
 	if err != nil {
 		return nil, err
 	}