op-node: Add a histogram to report current peer scores

op-node/metrics/histogram_snapshot.go

@@ -0,0 +1,44 @@
+package metrics
+
+import (
+	"sync/atomic"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type ReplaceableHistogramVec struct {
+	current        *atomic.Value
+	opts           prometheus.HistogramOpts
+	variableLabels []string
+}
+
+func NewReplaceableHistogramVec(registry *prometheus.Registry, opts prometheus.HistogramOpts, variableLabels []string) *ReplaceableHistogramVec {
+	metric := &ReplaceableHistogramVec{
+		current:        &atomic.Value{},
+		opts:           opts,
+		variableLabels: variableLabels,
+	}
+	metric.current.Store(prometheus.NewHistogramVec(opts, variableLabels))
+	registry.MustRegister(metric)
+	return metric
+}
+
+func (c *ReplaceableHistogramVec) Replace(updater func(h *prometheus.HistogramVec)) {
+	h := prometheus.NewHistogramVec(c.opts, c.variableLabels)
+	updater(h)
+	c.current.Store(h)
+}
+
+func (c *ReplaceableHistogramVec) Describe(ch chan<- *prometheus.Desc) {
+	collector, ok := c.current.Load().(prometheus.Collector)
+	if ok {
+		collector.Describe(ch)
+	}
+}
+
+func (c *ReplaceableHistogramVec) Collect(ch chan<- prometheus.Metric) {
+	collector, ok := c.current.Load().(prometheus.Collector)
+	if ok {
+		collector.Collect(ch)
+	}
+}

op-node/metrics/metrics.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	ophttp "github.com/ethereum-optimism/optimism/op-node/http"
+	"github.com/ethereum-optimism/optimism/op-node/p2p/store"
 	"github.com/ethereum-optimism/optimism/op-service/metrics"
 
 	pb "github.com/libp2p/go-libp2p-pubsub/pb"
@@ -66,7 +67,7 @@ type Metricer interface {
 	Document() []metrics.DocumentedMetric
 	RecordChannelInputBytes(num int)
 	// P2P Metrics
-	SetPeerScores(scores map[string]float64)
+	SetPeerScores(scores map[string]float64, allScores []store.PeerScores)
 	ClientPayloadByNumberEvent(num uint64, resultCode byte, duration time.Duration)
 	ServerPayloadByNumberEvent(num uint64, resultCode byte, duration time.Duration)
 	PayloadsQuarantineSize(n int)
@@ -132,15 +133,16 @@ type Metrics struct {
 	TransactionsSequencedTotal prometheus.Counter
 
 	// P2P Metrics
-	PeerCount         prometheus.Gauge
-	StreamCount       prometheus.Gauge
-	PeerScores        *prometheus.GaugeVec
-	GossipEventsTotal *prometheus.CounterVec
-	BandwidthTotal    *prometheus.GaugeVec
-	PeerUnbans        prometheus.Counter
-	IPUnbans          prometheus.Counter
-	Dials             *prometheus.CounterVec
-	Accepts           *prometheus.CounterVec
+	PeerCount           prometheus.Gauge
+	StreamCount         prometheus.Gauge
+	PeerScores          *prometheus.GaugeVec
+	GossipEventsTotal   *prometheus.CounterVec
+	BandwidthTotal      *prometheus.GaugeVec
+	PeerUnbans          prometheus.Counter
+	IPUnbans            prometheus.Counter
+	Dials               *prometheus.CounterVec
+	Accepts             *prometheus.CounterVec
+	PeerScoresHistogram *ReplaceableHistogramVec
 
 	ChannelInputBytes prometheus.Counter
 
@@ -161,6 +163,7 @@ func NewMetrics(procName string) *Metrics {
 	registry.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
 	registry.MustRegister(collectors.NewGoCollector())
 	factory := metrics.With(registry)
+
 	return &Metrics{
 		Info: factory.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: ns,
@@ -321,6 +324,12 @@ func NewMetrics(procName string) *Metrics {
 		}, []string{
 			"band",
 		}),
+		PeerScoresHistogram: NewReplaceableHistogramVec(registry, prometheus.HistogramOpts{
+			Namespace: ns,
+			Name:      "peer_scores_histogram",
+			Help:      "Histogram of currrently connected peer scores",
+			Buckets:   []float64{-100, -40, -20, -10, -5, -2, -1, -0.5, -0.05, 0, 0.05, 0.5, 1, 2, 5, 10, 20, 40},
+		}, []string{"type"}),
 		StreamCount: factory.NewGauge(prometheus.GaugeOpts{
 			Namespace: ns,
 			Subsystem: "p2p",
@@ -452,7 +461,18 @@ func NewMetrics(procName string) *Metrics {
 
 // SetPeerScores updates the peer score [prometheus.GaugeVec].
 // This takes a map of labels to scores.
-func (m *Metrics) SetPeerScores(scores map[string]float64) {
+func (m *Metrics) SetPeerScores(scores map[string]float64, allScores []store.PeerScores) {
+	m.PeerScoresHistogram.Replace(func(h *prometheus.HistogramVec) {
+		for _, scores := range allScores {
+			h.WithLabelValues("total").Observe(scores.Gossip.Total)
+			h.WithLabelValues("ipColocation").Observe(scores.Gossip.IPColocationFactor)
+			h.WithLabelValues("behavioralPenalty").Observe(scores.Gossip.BehavioralPenalty)
+			h.WithLabelValues("blocksFirstMessage").Observe(scores.Gossip.Blocks.FirstMessageDeliveries)
+			h.WithLabelValues("blocksTimeInMesh").Observe(scores.Gossip.Blocks.TimeInMesh)
+			h.WithLabelValues("blocksMessageDeliveries").Observe(scores.Gossip.Blocks.MeshMessageDeliveries)
+			h.WithLabelValues("blocksInvalidMessageDeliveries").Observe(scores.Gossip.Blocks.InvalidMessageDeliveries)
+		}
+	})
 	for label, score := range scores {
 		m.PeerScores.WithLabelValues(label).Set(score)
 	}
@@ -785,7 +805,7 @@ func (n *noopMetricer) RecordSequencerReset() {
 func (n *noopMetricer) RecordGossipEvent(evType int32) {
 }
 
-func (n *noopMetricer) SetPeerScores(scores map[string]float64) {
+func (n *noopMetricer) SetPeerScores(scores map[string]float64, allScores []store.PeerScores) {
 }
 
 func (n *noopMetricer) IncPeerCount() {

op-node/p2p/peer_scorer.go

@@ -96,7 +96,7 @@ type Peerstore interface {
 	// Peers returns all of the peer IDs stored across all inner stores.
 	Peers() peer.IDSlice
 
-	SetScore(id peer.ID, diff store.ScoreDiff) error
+	SetScore(id peer.ID, diff store.ScoreDiff) (store.PeerScores, error)
 }
 
 // Scorer is a peer scorer that scores peers based on application-specific metrics.
@@ -106,8 +106,9 @@ type Scorer interface {
 	SnapshotHook() pubsub.ExtendedPeerScoreInspectFn
 }
 
+//go:generate mockery --name ScoreMetrics --output mocks/
 type ScoreMetrics interface {
-	SetPeerScores(map[string]float64)
+	SetPeerScores(map[string]float64, []store.PeerScores)
 }
 
 // NewScorer returns a new peer scorer.
@@ -129,6 +130,7 @@ func (s *scorer) SnapshotHook() pubsub.ExtendedPeerScoreInspectFn {
 	blocksTopicName := blocksTopicV1(s.cfg)
 	return func(m map[peer.ID]*pubsub.PeerScoreSnapshot) {
 		scoreMap := make(map[string]float64)
+		allScores := make([]store.PeerScores, len(m))
 		// Zero out all bands.
 		for _, b := range s.bandScoreThresholds.bands {
 			scoreMap[b.band] = 0
@@ -147,15 +149,16 @@ func (s *scorer) SnapshotHook() pubsub.ExtendedPeerScoreInspectFn {
 				diff.Blocks.FirstMessageDeliveries = topSnap.FirstMessageDeliveries
 				diff.Blocks.InvalidMessageDeliveries = topSnap.InvalidMessageDeliveries
 			}
-			if err := s.peerStore.SetScore(id, &diff); err != nil {
+			if peerScores, err := s.peerStore.SetScore(id, &diff); err != nil {
 				s.log.Warn("Unable to update peer gossip score", "err", err)
+				allScores = append(allScores, peerScores)
 			}
 		}
 		for _, snap := range m {
 			band := s.bandScoreThresholds.Bucket(snap.Score)
 			scoreMap[band] += 1
 		}
-		s.metricer.SetPeerScores(scoreMap)
+		s.metricer.SetPeerScores(scoreMap, allScores)
 	}
 }
 

op-node/p2p/peer_scorer_test.go

@@ -22,15 +22,15 @@ type PeerScorerTestSuite struct {
 	suite.Suite
 
 	mockStore    *p2pMocks.Peerstore
-	mockMetricer *p2pMocks.GossipMetricer
+	mockMetricer *p2pMocks.ScoreMetrics
 	bandScorer   *p2p.BandScoreThresholds
 	logger       log.Logger
 }
 
 // SetupTest sets up the test suite.
 func (testSuite *PeerScorerTestSuite) SetupTest() {
 	testSuite.mockStore = &p2pMocks.Peerstore{}
-	testSuite.mockMetricer = &p2pMocks.GossipMetricer{}
+	testSuite.mockMetricer = &p2pMocks.ScoreMetrics{}
 	bandScorer, err := p2p.NewBandScorer("-40:graylist;0:friend;")
 	testSuite.NoError(err)
 	testSuite.bandScorer = bandScorer
@@ -77,14 +77,15 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 	)
 	inspectFn := scorer.SnapshotHook()
 
+	scores := store.PeerScores{}
 	// Expect updating the peer store
-	testSuite.mockStore.On("SetScore", peer.ID("peer1"), &store.GossipScores{Total: float64(-100)}).Return(nil).Once()
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), &store.GossipScores{Total: float64(-100)}).Return(scores, nil).Once()
 
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
 		"friend":   0,
 		"graylist": 1,
-	}).Return(nil).Once()
+	}, []store.PeerScores{scores}).Return(nil).Once()
 
 	// Apply the snapshot
 	snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{
@@ -95,13 +96,13 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 	inspectFn(snapshotMap)
 
 	// Expect updating the peer store
-	testSuite.mockStore.On("SetScore", peer.ID("peer1"), &store.GossipScores{Total: 0}).Return(nil).Once()
+	testSuite.mockStore.On("SetScore", peer.ID("peer1"), &store.GossipScores{Total: 0}).Return(scores, nil).Once()
 
 	// The metricer should then be called with the peer score band map
 	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
 		"friend":   1,
 		"graylist": 0,
-	}).Return(nil).Once()
+	}, []store.PeerScores{scores}).Return(nil).Once()
 
 	// Apply the snapshot
 	snapshotMap = map[peer.ID]*pubsub.PeerScoreSnapshot{
@@ -111,33 +112,3 @@ func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHook() {
 	}
 	inspectFn(snapshotMap)
 }
-
-// TestScorer_SnapshotHookBlocksPeer tests running the snapshot hook on the peer scorer with a peer score below the threshold.
-// This implies that the peer should be blocked.
-func (testSuite *PeerScorerTestSuite) TestScorer_SnapshotHookBlocksPeer() {
-	scorer := p2p.NewScorer(
-		&rollup.Config{L2ChainID: big.NewInt(123)},
-		testSuite.mockStore,
-		testSuite.mockMetricer,
-		testSuite.bandScorer,
-		testSuite.logger,
-	)
-	inspectFn := scorer.SnapshotHook()
-
-	// Expect updating the peer store
-	testSuite.mockStore.On("SetScore", peer.ID("peer1"), &store.GossipScores{Total: float64(-101)}).Return(nil).Once()
-
-	// The metricer should then be called with the peer score band map
-	testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
-		"friend":   0,
-		"graylist": 1,
-	}).Return(nil)
-
-	// Apply the snapshot
-	snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{
-		peer.ID("peer1"): {
-			Score: -101,
-		},
-	}
-	inspectFn(snapshotMap)
-}

op-node/p2p/peer_scores_test.go

@@ -36,15 +36,15 @@ type PeerScoresTestSuite struct {
 	suite.Suite
 
 	mockStore    *p2pMocks.Peerstore
-	mockMetricer *p2pMocks.GossipMetricer
+	mockMetricer *p2pMocks.ScoreMetrics
 	bandScorer   BandScoreThresholds
 	logger       log.Logger
 }
 
 // SetupTest sets up the test suite.
 func (testSuite *PeerScoresTestSuite) SetupTest() {
 	testSuite.mockStore = &p2pMocks.Peerstore{}
-	testSuite.mockMetricer = &p2pMocks.GossipMetricer{}
+	testSuite.mockMetricer = &p2pMocks.ScoreMetrics{}
 	bandScorer, err := NewBandScorer("0:graylist;")
 	testSuite.NoError(err)
 	testSuite.bandScorer = *bandScorer
@@ -157,7 +157,7 @@ func (testSuite *PeerScoresTestSuite) TestNegativeScores() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	testSuite.mockMetricer.On("SetPeerScores", mock.Anything).Return(nil)
+	testSuite.mockMetricer.On("SetPeerScores", mock.Anything, mock.Anything).Return(nil)
 
 	// Construct 20 hosts using the [getNetHosts] function.
 	hosts := getNetHosts(testSuite, ctx, 20)

op-node/p2p/store/iface.go

@@ -38,7 +38,7 @@ type ScoreDatastore interface {
 	GetPeerScores(id peer.ID) (PeerScores, error)
 
 	// SetScore applies the given store diff to the specified peer
-	SetScore(id peer.ID, diff ScoreDiff) error
+	SetScore(id peer.ID, diff ScoreDiff) (PeerScores, error)
 }
 
 // ScoreDiff defines a type-safe batch of changes to apply to the peer-scoring record of the peer.