Add feature flag to toggle gossip sub in p2p (#2322)

* add feature flag to enable gossip sub in p2p

* invert the enable/disable logic

* add the flag in k8s and fix tests

* gazellle

* return empty config if nil

k8s/beacon-chain/beacon-chain.deploy.yaml

@@ -56,6 +56,8 @@ spec:
             - --trace-sample-fraction=1.0
             - --datadir=/data
             - --enable-committees-cache 
+            # Disabling gossip sub until a larger beacon chain deployment.
+            - --disable-gossip-sub
           resources:
             requests:
               memory: "100Mi"

shared/featureconfig/config.go

@@ -32,12 +32,16 @@ type FeatureFlagConfig struct {
 	EnableHistoricalStatePruning bool // EnableHistoricalStatePruning when updating finalized states.
 	EnableCommitteesCache        bool // EnableCommitteesCache for state transition.
 	EnableBlockAncestorCache     bool //EnableBlockAncestorCache for fork choice optimization.
+	DisableGossipSub             bool // DisableGossipSub in p2p messaging.
 }
 
 var featureConfig *FeatureFlagConfig
 
 // FeatureConfig retrieves feature config.
 func FeatureConfig() *FeatureFlagConfig {
+	if featureConfig == nil {
+		return &FeatureFlagConfig{}
+	}
 	return featureConfig
 }
 
@@ -78,6 +82,10 @@ func ConfigureBeaconFeatures(ctx *cli.Context) {
 		log.Info("Enabled block ancestor cache")
 		cfg.EnableBlockAncestorCache = true
 	}
+	if ctx.GlobalBool(DisableGossipSubFlag.Name) {
+		log.Info("Disabled gossipsub, using floodsub")
+		cfg.DisableGossipSub = true
+	}
 
 	InitFeatureConfig(cfg)
 }

shared/featureconfig/flags.go

@@ -46,6 +46,11 @@ var (
 		Name:  "enable-historical-state-pruning",
 		Usage: "Enable database pruning of historical states after finalized epochs",
 	}
+	// DisableGossipSubFlag uses floodsub in place of gossipsub.
+	DisableGossipSubFlag = cli.BoolFlag{
+		Name:  "disable-gossip-sub",
+		Usage: "Disable gossip sub messaging and use floodsub messaging",
+	}
 )
 
 // ValidatorFlags contains a list of all the feature flags that apply to the validator client.
@@ -59,4 +64,5 @@ var BeaconChainFlags = []cli.Flag{
 	EnableCheckBlockStateRootFlag,
 	EnableHistoricalStatePruningFlag,
 	EnableBlockAncestorCacheFlag,
+	DisableGossipSubFlag,
 }

shared/p2p/BUILD.bazel

@@ -19,6 +19,7 @@ go_library(
     deps = [
         "//proto/beacon/p2p/v1:go_default_library",
         "//shared/event:go_default_library",
+        "//shared/featureconfig:go_default_library",
         "//shared/iputils:go_default_library",
         "@com_github_gogo_protobuf//io:go_default_library",
         "@com_github_gogo_protobuf//proto:go_default_library",

shared/p2p/service.go

@@ -24,6 +24,7 @@ import (
 	rhost "github.com/libp2p/go-libp2p/p2p/host/routed"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	"github.com/prysmaticlabs/prysm/shared/event"
+	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"github.com/prysmaticlabs/prysm/shared/iputils"
 	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
@@ -85,7 +86,12 @@ func NewServer(cfg *ServerConfig) (*Server, error) {
 	// distributed hash table by their peer ID.
 	h = rhost.Wrap(h, dht)
 
-	gsub, err := pubsub.NewFloodSub(ctx, h)
+	var gsub *pubsub.PubSub
+	if featureconfig.FeatureConfig().DisableGossipSub {
+		gsub, err = pubsub.NewFloodSub(ctx, h)
+	} else {
+		gsub, err = pubsub.NewGossipSub(ctx, h)
+	}
 	if err != nil {
 		cancel()
 		return nil, err