Commit 218828d1 authored by Andreas Bigger's avatar Andreas Bigger

fix peer scoring and related metrics

parent b49833fe
...@@ -34,6 +34,15 @@ var ( ...@@ -34,6 +34,15 @@ var (
Value: "none", Value: "none",
EnvVar: p2pEnv("PEER_SCORING"), EnvVar: p2pEnv("PEER_SCORING"),
} }
PeerScoreBands = cli.StringFlag{
Name: "p2p.score.bands",
Usage: "Sets the peer score bands used primarily for peer score metrics. " +
"Should be provided in following format: <threshold>:<label>;<threshold>:<label>;..." +
"For example: -40:graylist;-20:restricted;0:nopx;20:friend;",
Required: false,
Value: "-40:graylist;-20:restricted;0:nopx;20:friend;",
EnvVar: p2pEnv("SCORE_BANDS"),
}
// Banning Flag - whether or not we want to act on the scoring // Banning Flag - whether or not we want to act on the scoring
Banning = cli.BoolFlag{ Banning = cli.BoolFlag{
...@@ -276,6 +285,9 @@ var p2pFlags = []cli.Flag{ ...@@ -276,6 +285,9 @@ var p2pFlags = []cli.Flag{
NoDiscovery, NoDiscovery,
P2PPrivPath, P2PPrivPath,
P2PPrivRaw, P2PPrivRaw,
PeerScoring,
Banning,
TopicScoring,
ListenIP, ListenIP,
ListenTCPPort, ListenTCPPort,
ListenUDPPort, ListenUDPPort,
......
...@@ -15,7 +15,6 @@ import ( ...@@ -15,7 +15,6 @@ import (
pb "github.com/libp2p/go-libp2p-pubsub/pb" pb "github.com/libp2p/go-libp2p-pubsub/pb"
libp2pmetrics "github.com/libp2p/go-libp2p/core/metrics" libp2pmetrics "github.com/libp2p/go-libp2p/core/metrics"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/collectors" "github.com/prometheus/client_golang/prometheus/collectors"
"github.com/prometheus/client_golang/prometheus/promhttp" "github.com/prometheus/client_golang/prometheus/promhttp"
...@@ -66,7 +65,7 @@ type Metricer interface { ...@@ -66,7 +65,7 @@ type Metricer interface {
RecordSequencerSealingTime(duration time.Duration) RecordSequencerSealingTime(duration time.Duration)
Document() []metrics.DocumentedMetric Document() []metrics.DocumentedMetric
// P2P Metrics // P2P Metrics
RecordPeerScoring(peerID peer.ID, score float64) SetPeerScores(scores map[string]float64)
} }
// Metrics tracks all the metrics for the op-node. // Metrics tracks all the metrics for the op-node.
...@@ -287,21 +286,24 @@ func NewMetrics(procName string) *Metrics { ...@@ -287,21 +286,24 @@ func NewMetrics(procName string) *Metrics {
Name: "peer_count", Name: "peer_count",
Help: "Count of currently connected p2p peers", Help: "Count of currently connected p2p peers",
}), }),
StreamCount: factory.NewGauge(prometheus.GaugeOpts{ // Notice: We cannot use peer ids as [Labels] in the GaugeVec
Namespace: ns, // since peer ids would open a service attack vector.
Subsystem: "p2p", // Each peer id would be a separate metric, flooding prometheus.
Name: "stream_count", //
Help: "Count of currently connected p2p streams", // [Labels]: https://prometheus.io/docs/practices/naming/#labels
}),
PeerScores: factory.NewGaugeVec(prometheus.GaugeOpts{ PeerScores: factory.NewGaugeVec(prometheus.GaugeOpts{
Namespace: ns, Namespace: ns,
Subsystem: "p2p", Subsystem: "p2p",
Name: "peer_scores", Name: "peer_scores",
Help: "Peer scoring", Help: "Count of peer scores grouped by score",
}, []string{ }, []string{
// No label names here since peer ids would open a service attack vector. "band",
// Each peer id would be a separate metric, flooding prometheus. }),
// See: https://prometheus.io/docs/practices/naming/#labels StreamCount: factory.NewGauge(prometheus.GaugeOpts{
Namespace: ns,
Subsystem: "p2p",
Name: "stream_count",
Help: "Count of currently connected p2p streams",
}), }),
GossipEventsTotal: factory.NewCounterVec(prometheus.CounterOpts{ GossipEventsTotal: factory.NewCounterVec(prometheus.CounterOpts{
Namespace: ns, Namespace: ns,
...@@ -350,6 +352,14 @@ func NewMetrics(procName string) *Metrics { ...@@ -350,6 +352,14 @@ 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) {
for label, score := range scores {
m.PeerScores.WithLabelValues(label).Set(score)
}
}
// RecordInfo sets a pseudo-metric that contains versioning and // RecordInfo sets a pseudo-metric that contains versioning and
// config info for the opnode. // config info for the opnode.
func (m *Metrics) RecordInfo(version string) { func (m *Metrics) RecordInfo(version string) {
...@@ -491,10 +501,6 @@ func (m *Metrics) RecordGossipEvent(evType int32) { ...@@ -491,10 +501,6 @@ func (m *Metrics) RecordGossipEvent(evType int32) {
m.GossipEventsTotal.WithLabelValues(pb.TraceEvent_Type_name[evType]).Inc() m.GossipEventsTotal.WithLabelValues(pb.TraceEvent_Type_name[evType]).Inc()
} }
func (m *Metrics) RecordPeerScoring(peerID peer.ID, score float64) {
m.PeerScores.WithLabelValues(peerID.String()).Set(score)
}
func (m *Metrics) IncPeerCount() { func (m *Metrics) IncPeerCount() {
m.PeerCount.Inc() m.PeerCount.Inc()
} }
...@@ -627,7 +633,7 @@ func (n *noopMetricer) RecordSequencerReset() { ...@@ -627,7 +633,7 @@ func (n *noopMetricer) RecordSequencerReset() {
func (n *noopMetricer) RecordGossipEvent(evType int32) { func (n *noopMetricer) RecordGossipEvent(evType int32) {
} }
func (n *noopMetricer) RecordPeerScoring(peerID peer.ID, score float64) { func (n *noopMetricer) SetPeerScores(scores map[string]float64) {
} }
func (n *noopMetricer) IncPeerCount() { func (n *noopMetricer) IncPeerCount() {
......
package p2p
import (
"testing"
"github.com/stretchr/testify/require"
)
// TestBandScorer_ParseDefault tests the [BandScorer.Parse] function
// on the default band scores cli flag value.
func TestBandScorer_ParseDefault(t *testing.T) {
defaultScoringBands := "-40:graylist;-20:restricted;0:nopx;20:friend;"
// Create a new band scorer.
bandScorer := NewBandScorer()
require.NoError(t, bandScorer.Parse(defaultScoringBands))
// Validate the [BandScorer] internals.
require.Len(t, bandScorer.(*bandScoreThresholds).bands, 4)
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["graylist"], float64(-40))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["restricted"], float64(-20))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["nopx"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["friend"], float64(20))
require.Equal(t, bandScorer.(*bandScoreThresholds).lowestBand, "graylist")
}
// TestBandScorer_ParseEmpty tests the [BandScorer.Parse] function
// on an empty string.
func TestBandScorer_ParseEmpty(t *testing.T) {
// Create a band scorer on an empty string.
bandScorer := NewBandScorer()
require.NoError(t, bandScorer.Parse(""))
// Validate the [BandScorer] internals.
require.Len(t, bandScorer.(*bandScoreThresholds).bands, 0)
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["graylist"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["restricted"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["nopx"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["friend"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).lowestBand, "")
}
// TestBandScorer_ParseWhitespace tests the [BandScorer.Parse] function
// on a variety of whitespaced strings.
func TestBandScorer_ParseWhitespace(t *testing.T) {
// Create a band scorer on an empty string.
bandScorer := NewBandScorer()
require.NoError(t, bandScorer.Parse(" ; ; ; "))
// Validate the [BandScorer] internals.
require.Len(t, bandScorer.(*bandScoreThresholds).bands, 0)
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["graylist"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["restricted"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["nopx"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).bands["friend"], float64(0))
require.Equal(t, bandScorer.(*bandScoreThresholds).lowestBand, "")
}
...@@ -58,6 +58,10 @@ func NewConfig(ctx *cli.Context, blockTime uint64) (*p2p.Config, error) { ...@@ -58,6 +58,10 @@ func NewConfig(ctx *cli.Context, blockTime uint64) (*p2p.Config, error) {
return nil, fmt.Errorf("failed to load p2p peer scoring options: %w", err) return nil, fmt.Errorf("failed to load p2p peer scoring options: %w", err)
} }
if err := loadPeerScoreBands(conf, ctx); err != nil {
return nil, fmt.Errorf("failed to load p2p peer score bands: %w", err)
}
if err := loadBanningOption(conf, ctx); err != nil { if err := loadBanningOption(conf, ctx); err != nil {
return nil, fmt.Errorf("failed to load banning option: %w", err) return nil, fmt.Errorf("failed to load banning option: %w", err)
} }
...@@ -121,6 +125,17 @@ func loadPeerScoringParams(conf *p2p.Config, ctx *cli.Context, blockTime uint64) ...@@ -121,6 +125,17 @@ func loadPeerScoringParams(conf *p2p.Config, ctx *cli.Context, blockTime uint64)
return nil return nil
} }
// loadPeerScoreBands loads [p2p.BandScorer] from the CLI context.
func loadPeerScoreBands(conf *p2p.Config, ctx *cli.Context) error {
scoreBands := ctx.GlobalString(flags.PeerScoreBands.Name)
bandScorer := p2p.NewBandScorer()
if err := bandScorer.Parse(scoreBands); err != nil {
return err
}
conf.BandScoreThresholds = bandScorer
return nil
}
// loadBanningOption loads whether or not to ban peers from the CLI context. // loadBanningOption loads whether or not to ban peers from the CLI context.
func loadBanningOption(conf *p2p.Config, ctx *cli.Context) error { func loadBanningOption(conf *p2p.Config, ctx *cli.Context) error {
ban := ctx.GlobalBool(flags.Banning.Name) ban := ctx.GlobalBool(flags.Banning.Name)
......
...@@ -54,6 +54,9 @@ type Config struct { ...@@ -54,6 +54,9 @@ type Config struct {
PeerScoring pubsub.PeerScoreParams PeerScoring pubsub.PeerScoreParams
TopicScoring pubsub.TopicScoreParams TopicScoring pubsub.TopicScoreParams
// Peer Score Band Thresholds
BandScoreThresholds BandScorer
// Whether to ban peers based on their [PeerScoring] score. // Whether to ban peers based on their [PeerScoring] score.
BanningEnabled bool BanningEnabled bool
...@@ -151,6 +154,10 @@ func (conf *Config) PeerScoringParams() *pubsub.PeerScoreParams { ...@@ -151,6 +154,10 @@ func (conf *Config) PeerScoringParams() *pubsub.PeerScoreParams {
return &conf.PeerScoring return &conf.PeerScoring
} }
func (conf *Config) PeerBandScorer() BandScorer {
return conf.BandScoreThresholds
}
func (conf *Config) BanPeers() bool { func (conf *Config) BanPeers() bool {
return conf.BanningEnabled return conf.BanningEnabled
} }
......
...@@ -55,6 +55,7 @@ type GossipSetupConfigurables interface { ...@@ -55,6 +55,7 @@ type GossipSetupConfigurables interface {
TopicScoringParams() *pubsub.TopicScoreParams TopicScoringParams() *pubsub.TopicScoreParams
BanPeers() bool BanPeers() bool
ConfigureGossip(params *pubsub.GossipSubParams) []pubsub.Option ConfigureGossip(params *pubsub.GossipSubParams) []pubsub.Option
PeerBandScorer() BandScorer
} }
type GossipRuntimeConfig interface { type GossipRuntimeConfig interface {
...@@ -64,7 +65,8 @@ type GossipRuntimeConfig interface { ...@@ -64,7 +65,8 @@ type GossipRuntimeConfig interface {
//go:generate mockery --name GossipMetricer //go:generate mockery --name GossipMetricer
type GossipMetricer interface { type GossipMetricer interface {
RecordGossipEvent(evType int32) RecordGossipEvent(evType int32)
RecordPeerScoring(peerID peer.ID, score float64) // Peer Scoring Metric Funcs
SetPeerScores(map[string]float64)
} }
func blocksTopicV1(cfg *rollup.Config) string { func blocksTopicV1(cfg *rollup.Config) string {
......
// Code generated by mockery v2.22.1. DO NOT EDIT.
package mocks
import mock "github.com/stretchr/testify/mock"
// BandScorer is an autogenerated mock type for the BandScorer type
type BandScorer struct {
mock.Mock
}
// Bucket provides a mock function with given fields: score
func (_m *BandScorer) Bucket(score float64) string {
ret := _m.Called(score)
var r0 string
if rf, ok := ret.Get(0).(func(float64) string); ok {
r0 = rf(score)
} else {
r0 = ret.Get(0).(string)
}
return r0
}
// Parse provides a mock function with given fields: str
func (_m *BandScorer) Parse(str string) error {
ret := _m.Called(str)
var r0 error
if rf, ok := ret.Get(0).(func(string) error); ok {
r0 = rf(str)
} else {
r0 = ret.Error(0)
}
return r0
}
// Reset provides a mock function with given fields:
func (_m *BandScorer) Reset() {
_m.Called()
}
type mockConstructorTestingTNewBandScorer interface {
mock.TestingT
Cleanup(func())
}
// NewBandScorer creates a new instance of BandScorer. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
func NewBandScorer(t mockConstructorTestingTNewBandScorer) *BandScorer {
mock := &BandScorer{}
mock.Mock.Test(t)
t.Cleanup(func() { mock.AssertExpectations(t) })
return mock
}
// Code generated by mockery v2.14.0. DO NOT EDIT. // Code generated by mockery v2.22.1. DO NOT EDIT.
package mocks package mocks
...@@ -123,13 +123,16 @@ func (_m *ConnectionGater) InterceptUpgraded(_a0 network.Conn) (bool, control.Di ...@@ -123,13 +123,16 @@ func (_m *ConnectionGater) InterceptUpgraded(_a0 network.Conn) (bool, control.Di
ret := _m.Called(_a0) ret := _m.Called(_a0)
var r0 bool var r0 bool
var r1 control.DisconnectReason
if rf, ok := ret.Get(0).(func(network.Conn) (bool, control.DisconnectReason)); ok {
return rf(_a0)
}
if rf, ok := ret.Get(0).(func(network.Conn) bool); ok { if rf, ok := ret.Get(0).(func(network.Conn) bool); ok {
r0 = rf(_a0) r0 = rf(_a0)
} else { } else {
r0 = ret.Get(0).(bool) r0 = ret.Get(0).(bool)
} }
var r1 control.DisconnectReason
if rf, ok := ret.Get(1).(func(network.Conn) control.DisconnectReason); ok { if rf, ok := ret.Get(1).(func(network.Conn) control.DisconnectReason); ok {
r1 = rf(_a0) r1 = rf(_a0)
} else { } else {
......
// Code generated by mockery v2.14.0. DO NOT EDIT. // Code generated by mockery v2.22.1. DO NOT EDIT.
package mocks package mocks
import ( import mock "github.com/stretchr/testify/mock"
mock "github.com/stretchr/testify/mock"
peer "github.com/libp2p/go-libp2p/core/peer"
)
// GossipMetricer is an autogenerated mock type for the GossipMetricer type // GossipMetricer is an autogenerated mock type for the GossipMetricer type
type GossipMetricer struct { type GossipMetricer struct {
...@@ -18,9 +14,9 @@ func (_m *GossipMetricer) RecordGossipEvent(evType int32) { ...@@ -18,9 +14,9 @@ func (_m *GossipMetricer) RecordGossipEvent(evType int32) {
_m.Called(evType) _m.Called(evType)
} }
// RecordPeerScoring provides a mock function with given fields: peerID, score // SetPeerScores provides a mock function with given fields: _a0
func (_m *GossipMetricer) RecordPeerScoring(peerID peer.ID, score float64) { func (_m *GossipMetricer) SetPeerScores(_a0 map[string]float64) {
_m.Called(peerID, score) _m.Called(_a0)
} }
type mockConstructorTestingTNewGossipMetricer interface { type mockConstructorTestingTNewGossipMetricer interface {
......
// Code generated by mockery v2.14.0. DO NOT EDIT. // Code generated by mockery v2.22.1. DO NOT EDIT.
package mocks package mocks
......
// Code generated by mockery v2.14.0. DO NOT EDIT. // Code generated by mockery v2.22.1. DO NOT EDIT.
package mocks package mocks
......
package p2p package p2p
import ( import (
"fmt"
"strconv"
"strings"
log "github.com/ethereum/go-ethereum/log" log "github.com/ethereum/go-ethereum/log"
pubsub "github.com/libp2p/go-libp2p-pubsub" pubsub "github.com/libp2p/go-libp2p-pubsub"
peer "github.com/libp2p/go-libp2p/core/peer" peer "github.com/libp2p/go-libp2p/core/peer"
) )
type scorer struct { type scorer struct {
peerStore Peerstore peerStore Peerstore
metricer GossipMetricer metricer GossipMetricer
log log.Logger log log.Logger
gater PeerGater gater PeerGater
bandScoreThresholds BandScorer
}
// bandScoreThresholds holds the thresholds for classifying peers
// into different score bands.
type bandScoreThresholds struct {
bands map[string]float64
lowestBand string
}
// BandScorer is an interface for placing peer scores
// into various bands.
//
// Implementations are expected to construct internals using the
// [Parse] function and then expose the [Bucket] function for
// downstream [BandScorer] consumers.
//
//go:generate mockery --name BandScorer --output mocks/
type BandScorer interface {
Parse(str string) error
Bucket(score float64) string
Reset()
}
// NewBandScorer constructs a new [BandScorer] instance.
func NewBandScorer() BandScorer {
return &bandScoreThresholds{
bands: make(map[string]float64),
}
}
// Reset wipes the internal state of the [BandScorer].
func (s *bandScoreThresholds) Reset() {
s.bands = make(map[string]float64)
}
// Parse creates a [BandScorer] from a given string.
func (s *bandScoreThresholds) Parse(str string) error {
var lowestThreshold float64
for i, band := range strings.Split(str, ";") {
// Skip empty band strings.
band := strings.TrimSpace(band)
if band == "" {
continue
}
split := strings.Split(band, ":")
if len(split) != 2 {
return fmt.Errorf("invalid score band: %s", band)
}
threshold, err := strconv.ParseFloat(split[0], 64)
if err != nil {
return err
}
s.bands[split[1]] = threshold
if threshold < lowestThreshold || i == 0 {
s.lowestBand = split[1]
lowestThreshold = threshold
}
}
return nil
}
// Bucket returns the appropriate band for a given score.
func (s *bandScoreThresholds) Bucket(score float64) string {
for band, threshold := range s.bands {
if score >= threshold {
return band
}
}
// If there is no band threshold lower than the score,
// the peer must be placed in the lowest bucket.
return s.lowestBand
} }
// Peerstore is a subset of the libp2p peerstore.Peerstore interface. // Peerstore is a subset of the libp2p peerstore.Peerstore interface.
...@@ -34,12 +110,13 @@ type Scorer interface { ...@@ -34,12 +110,13 @@ type Scorer interface {
} }
// NewScorer returns a new peer scorer. // NewScorer returns a new peer scorer.
func NewScorer(peerGater PeerGater, peerStore Peerstore, metricer GossipMetricer, log log.Logger) Scorer { func NewScorer(peerGater PeerGater, peerStore Peerstore, metricer GossipMetricer, bandScoreThresholds BandScorer, log log.Logger) Scorer {
return &scorer{ return &scorer{
peerStore: peerStore, peerStore: peerStore,
metricer: metricer, metricer: metricer,
log: log, log: log,
gater: peerGater, gater: peerGater,
bandScoreThresholds: bandScoreThresholds,
} }
} }
...@@ -48,13 +125,20 @@ func NewScorer(peerGater PeerGater, peerStore Peerstore, metricer GossipMetricer ...@@ -48,13 +125,20 @@ func NewScorer(peerGater PeerGater, peerStore Peerstore, metricer GossipMetricer
// The returned [pubsub.ExtendedPeerScoreInspectFn] is called with a mapping of peer IDs to peer score snapshots. // The returned [pubsub.ExtendedPeerScoreInspectFn] is called with a mapping of peer IDs to peer score snapshots.
func (s *scorer) SnapshotHook() pubsub.ExtendedPeerScoreInspectFn { func (s *scorer) SnapshotHook() pubsub.ExtendedPeerScoreInspectFn {
return func(m map[peer.ID]*pubsub.PeerScoreSnapshot) { return func(m map[peer.ID]*pubsub.PeerScoreSnapshot) {
// Reset the score bands
s.bandScoreThresholds.Reset()
// First clear the peer score bands
scoreMap := make(map[string]float64)
for id, snap := range m { for id, snap := range m {
// Record peer score in the metricer // Increment the bucket for the peer's score
s.metricer.RecordPeerScoring(id, snap.Score) band := s.bandScoreThresholds.Bucket(snap.Score)
scoreMap[band] += 1
// Update with the peer gater // Update with the peer gater
s.gater.Update(id, snap.Score) s.gater.Update(id, snap.Score)
} }
s.metricer.SetPeerScores(scoreMap)
} }
} }
......
...@@ -17,18 +17,19 @@ type PeerScorerTestSuite struct { ...@@ -17,18 +17,19 @@ type PeerScorerTestSuite struct {
suite.Suite suite.Suite
// mockConnGater *p2pMocks.ConnectionGater // mockConnGater *p2pMocks.ConnectionGater
mockGater *p2pMocks.PeerGater mockGater *p2pMocks.PeerGater
mockStore *p2pMocks.Peerstore mockStore *p2pMocks.Peerstore
mockMetricer *p2pMocks.GossipMetricer mockMetricer *p2pMocks.GossipMetricer
logger log.Logger mockBandScorer *p2pMocks.BandScorer
logger log.Logger
} }
// SetupTest sets up the test suite. // SetupTest sets up the test suite.
func (testSuite *PeerScorerTestSuite) SetupTest() { func (testSuite *PeerScorerTestSuite) SetupTest() {
testSuite.mockGater = &p2pMocks.PeerGater{} testSuite.mockGater = &p2pMocks.PeerGater{}
// testSuite.mockConnGater = &p2pMocks.ConnectionGater{}
testSuite.mockStore = &p2pMocks.Peerstore{} testSuite.mockStore = &p2pMocks.Peerstore{}
testSuite.mockMetricer = &p2pMocks.GossipMetricer{} testSuite.mockMetricer = &p2pMocks.GossipMetricer{}
testSuite.mockBandScorer = &p2pMocks.BandScorer{}
testSuite.logger = testlog.Logger(testSuite.T(), log.LvlError) testSuite.logger = testlog.Logger(testSuite.T(), log.LvlError)
} }
...@@ -43,6 +44,7 @@ func (testSuite *PeerScorerTestSuite) TestPeerScorerOnConnect() { ...@@ -43,6 +44,7 @@ func (testSuite *PeerScorerTestSuite) TestPeerScorerOnConnect() {
testSuite.mockGater, testSuite.mockGater,
testSuite.mockStore, testSuite.mockStore,
testSuite.mockMetricer, testSuite.mockMetricer,
testSuite.mockBandScorer,
testSuite.logger, testSuite.logger,
) )
scorer.OnConnect() scorer.OnConnect()
...@@ -54,6 +56,7 @@ func (testSuite *PeerScorerTestSuite) TestPeerScorerOnDisconnect() { ...@@ -54,6 +56,7 @@ func (testSuite *PeerScorerTestSuite) TestPeerScorerOnDisconnect() {
testSuite.mockGater, testSuite.mockGater,
testSuite.mockStore, testSuite.mockStore,
testSuite.mockMetricer, testSuite.mockMetricer,
testSuite.mockBandScorer,
testSuite.logger, testSuite.logger,
) )
scorer.OnDisconnect() scorer.OnDisconnect()
...@@ -65,17 +68,23 @@ func (testSuite *PeerScorerTestSuite) TestSnapshotHook() { ...@@ -65,17 +68,23 @@ func (testSuite *PeerScorerTestSuite) TestSnapshotHook() {
testSuite.mockGater, testSuite.mockGater,
testSuite.mockStore, testSuite.mockStore,
testSuite.mockMetricer, testSuite.mockMetricer,
testSuite.mockBandScorer,
testSuite.logger, testSuite.logger,
) )
inspectFn := scorer.SnapshotHook() inspectFn := scorer.SnapshotHook()
// Mock the snapshot updates // Mock the band scorer calls
// This doesn't return anything testSuite.mockBandScorer.On("Reset").Return(nil)
testSuite.mockMetricer.On("RecordPeerScoring", peer.ID("peer1"), float64(-100)).Return(nil) testSuite.mockBandScorer.On("Bucket", float64(-100)).Return("graylist")
// Mock the peer gater call // Mock the peer gater call
testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-100)).Return(nil) testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-100)).Return(nil)
// The metricer should then be called with the peer score band map
testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
"graylist": 1,
}).Return(nil)
// Apply the snapshot // Apply the snapshot
snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{ snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{
peer.ID("peer1"): { peer.ID("peer1"): {
...@@ -92,17 +101,23 @@ func (testSuite *PeerScorerTestSuite) TestSnapshotHookBlockPeer() { ...@@ -92,17 +101,23 @@ func (testSuite *PeerScorerTestSuite) TestSnapshotHookBlockPeer() {
testSuite.mockGater, testSuite.mockGater,
testSuite.mockStore, testSuite.mockStore,
testSuite.mockMetricer, testSuite.mockMetricer,
testSuite.mockBandScorer,
testSuite.logger, testSuite.logger,
) )
inspectFn := scorer.SnapshotHook() inspectFn := scorer.SnapshotHook()
// Mock the snapshot updates // Mock the band scorer calls
// This doesn't return anything testSuite.mockBandScorer.On("Reset").Return(nil)
testSuite.mockMetricer.On("RecordPeerScoring", peer.ID("peer1"), float64(-101)).Return(nil) testSuite.mockBandScorer.On("Bucket", float64(-101)).Return("graylist")
// Mock the peer gater call // Mock the peer gater call
testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-101)).Return(nil) testSuite.mockGater.On("Update", peer.ID("peer1"), float64(-101)).Return(nil)
// The metricer should then be called with the peer score band map
testSuite.mockMetricer.On("SetPeerScores", map[string]float64{
"graylist": 1,
}).Return(nil)
// Apply the snapshot // Apply the snapshot
snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{ snapshotMap := map[peer.ID]*pubsub.PeerScoreSnapshot{
peer.ID("peer1"): { peer.ID("peer1"): {
......
...@@ -14,7 +14,7 @@ func ConfigurePeerScoring(h host.Host, g ConnectionGater, gossipConf GossipSetup ...@@ -14,7 +14,7 @@ func ConfigurePeerScoring(h host.Host, g ConnectionGater, gossipConf GossipSetup
peerScoreThresholds := NewPeerScoreThresholds() peerScoreThresholds := NewPeerScoreThresholds()
banEnabled := gossipConf.BanPeers() banEnabled := gossipConf.BanPeers()
peerGater := NewPeerGater(g, log, banEnabled) peerGater := NewPeerGater(g, log, banEnabled)
scorer := NewScorer(peerGater, h.Peerstore(), m, log) scorer := NewScorer(peerGater, h.Peerstore(), m, gossipConf.PeerBandScorer(), log)
opts := []pubsub.Option{} opts := []pubsub.Option{}
// Check the app specific score since libp2p doesn't export it's [validate] function :/ // Check the app specific score since libp2p doesn't export it's [validate] function :/
if peerScoreParams != nil && peerScoreParams.AppSpecificScore != nil { if peerScoreParams != nil && peerScoreParams.AppSpecificScore != nil {
......
...@@ -11,7 +11,7 @@ import ( ...@@ -11,7 +11,7 @@ import (
p2pMocks "github.com/ethereum-optimism/optimism/op-node/p2p/mocks" p2pMocks "github.com/ethereum-optimism/optimism/op-node/p2p/mocks"
testlog "github.com/ethereum-optimism/optimism/op-node/testlog" testlog "github.com/ethereum-optimism/optimism/op-node/testlog"
mock "github.com/stretchr/testify/mock" "github.com/stretchr/testify/mock"
suite "github.com/stretchr/testify/suite" suite "github.com/stretchr/testify/suite"
log "github.com/ethereum/go-ethereum/log" log "github.com/ethereum/go-ethereum/log"
...@@ -27,10 +27,11 @@ import ( ...@@ -27,10 +27,11 @@ import (
type PeerScoresTestSuite struct { type PeerScoresTestSuite struct {
suite.Suite suite.Suite
mockGater *p2pMocks.ConnectionGater mockGater *p2pMocks.ConnectionGater
mockStore *p2pMocks.Peerstore mockStore *p2pMocks.Peerstore
mockMetricer *p2pMocks.GossipMetricer mockMetricer *p2pMocks.GossipMetricer
logger log.Logger mockBandScorer *p2pMocks.BandScorer
logger log.Logger
} }
// SetupTest sets up the test suite. // SetupTest sets up the test suite.
...@@ -38,6 +39,7 @@ func (testSuite *PeerScoresTestSuite) SetupTest() { ...@@ -38,6 +39,7 @@ func (testSuite *PeerScoresTestSuite) SetupTest() {
testSuite.mockGater = &p2pMocks.ConnectionGater{} testSuite.mockGater = &p2pMocks.ConnectionGater{}
testSuite.mockStore = &p2pMocks.Peerstore{} testSuite.mockStore = &p2pMocks.Peerstore{}
testSuite.mockMetricer = &p2pMocks.GossipMetricer{} testSuite.mockMetricer = &p2pMocks.GossipMetricer{}
testSuite.mockBandScorer = &p2pMocks.BandScorer{}
testSuite.logger = testlog.Logger(testSuite.T(), log.LvlError) testSuite.logger = testlog.Logger(testSuite.T(), log.LvlError)
} }
...@@ -68,6 +70,7 @@ func newGossipSubs(testSuite *PeerScoresTestSuite, ctx context.Context, hosts [] ...@@ -68,6 +70,7 @@ func newGossipSubs(testSuite *PeerScoresTestSuite, ctx context.Context, hosts []
rt := pubsub.DefaultGossipSubRouter(h) rt := pubsub.DefaultGossipSubRouter(h)
opts := []pubsub.Option{} opts := []pubsub.Option{}
opts = append(opts, p2p.ConfigurePeerScoring(h, testSuite.mockGater, &p2p.Config{ opts = append(opts, p2p.ConfigurePeerScoring(h, testSuite.mockGater, &p2p.Config{
BandScoreThresholds: testSuite.mockBandScorer,
PeerScoring: pubsub.PeerScoreParams{ PeerScoring: pubsub.PeerScoreParams{
AppSpecificScore: func(p peer.ID) float64 { AppSpecificScore: func(p peer.ID) float64 {
if p == hosts[0].ID() { if p == hosts[0].ID() {
...@@ -118,8 +121,10 @@ func (testSuite *PeerScoresTestSuite) TestNegativeScores() { ...@@ -118,8 +121,10 @@ func (testSuite *PeerScoresTestSuite) TestNegativeScores() {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()
testSuite.mockMetricer.On("RecordPeerScoring", mock.Anything, float64(0)).Return(nil) testSuite.mockBandScorer.On("Reset").Return(nil)
testSuite.mockMetricer.On("RecordPeerScoring", mock.Anything, float64(-1000)).Return(nil) testSuite.mockBandScorer.On("Bucket", float64(0)).Return("nopx")
testSuite.mockBandScorer.On("Bucket", float64(-1000)).Return("graylist")
testSuite.mockMetricer.On("SetPeerScores", mock.Anything).Return(nil)
testSuite.mockGater.On("ListBlockedPeers").Return([]peer.ID{}) testSuite.mockGater.On("ListBlockedPeers").Return([]peer.ID{})
......
...@@ -68,6 +68,10 @@ func (p *Prepared) PeerScoringParams() *pubsub.PeerScoreParams { ...@@ -68,6 +68,10 @@ func (p *Prepared) PeerScoringParams() *pubsub.PeerScoreParams {
return nil return nil
} }
func (p *Prepared) PeerBandScorer() BandScorer {
return nil
}
func (p *Prepared) BanPeers() bool { func (p *Prepared) BanPeers() bool {
return false return false
} }
......
...@@ -61,6 +61,8 @@ services: ...@@ -61,6 +61,8 @@ services:
--p2p.listen.ip=0.0.0.0 --p2p.listen.ip=0.0.0.0
--p2p.listen.tcp=9003 --p2p.listen.tcp=9003
--p2p.listen.udp=9003 --p2p.listen.udp=9003
--p2p.scoring.peers=light
--p2p.ban.peers=true
--snapshotlog.file=/op_log/snapshot.log --snapshotlog.file=/op_log/snapshot.log
--p2p.priv.path=/config/p2p-node-key.txt --p2p.priv.path=/config/p2p-node-key.txt
--metrics.enabled --metrics.enabled
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment