Merge pull request #648 from filecoin-project/feat/pubsub-tracer

pubsub tracing

node/builder.go

@@ -312,6 +312,9 @@ func ConfigFullNode(c interface{}) Option {
 	return Options(
 		ConfigCommon(&cfg.Common),
 		Override(HeadMetricsKey, metrics.SendHeadNotifs(cfg.Metrics.Nickname)),
+		If(cfg.Metrics.PubsubTracing,
+			Override(new(*pubsub.PubSub), lp2p.GossipSub(lp2p.PubsubTracer())),
+		),
 	)
 }
 

node/config/def.go

@@ -41,7 +41,8 @@ type Libp2p struct {
 // // Full Node
 
 type Metrics struct {
-	Nickname string
+	Nickname      string
+	PubsubTracing bool
 }
 
 // // Storage Miner

node/modules/lp2p/pubsub.go

@@ -1,21 +1,45 @@
 package lp2p
 
 import (
+	"context"
+
 	host "github.com/libp2p/go-libp2p-core/host"
+	peer "github.com/libp2p/go-libp2p-core/peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	ma "github.com/multiformats/go-multiaddr"
 	"go.uber.org/fx"
 
 	"github.com/filecoin-project/lotus/node/modules/helpers"
 )
 
-func FloodSub(pubsubOptions ...pubsub.Option) interface{} {
-	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host) (service *pubsub.PubSub, err error) {
-		return pubsub.NewFloodSub(helpers.LifecycleCtx(mctx, lc), host, pubsubOptions...)
+type PubsubOpt func(host.Host) pubsub.Option
+
+func PubsubTracer() PubsubOpt {
+	return func(host host.Host) pubsub.Option {
+		pi, err := peer.AddrInfoFromP2pAddr(ma.StringCast("/ip4/147.75.67.199/tcp/4001/p2p/QmTd6UvR47vUidRNZ1ZKXHrAFhqTJAD27rKL9XYghEKgKX"))
+		if err != nil {
+			panic(err)
+		}
+
+		tr, err := pubsub.NewRemoteTracer(context.TODO(), host, *pi)
+		if err != nil {
+			panic(err)
+		}
+
+		return pubsub.WithEventTracer(tr)
 	}
 }
 
-func GossipSub(pubsubOptions ...pubsub.Option) interface{} {
+func GossipSub(pubsubOptions ...PubsubOpt) interface{} {
 	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host) (service *pubsub.PubSub, err error) {
-		return pubsub.NewGossipSub(helpers.LifecycleCtx(mctx, lc), host, pubsubOptions...)
+		return pubsub.NewGossipSub(helpers.LifecycleCtx(mctx, lc), host, paresOpts(host, pubsubOptions)...)
 	}
 }
+
+func paresOpts(host host.Host, in []PubsubOpt) []pubsub.Option {
+	out := make([]pubsub.Option, len(in))
+	for k, v := range in {
+		out[k] = v(host)
+	}
+	return out
+}

node/options.go

@@ -40,6 +40,12 @@ func ApplyIf(check func(s *Settings) bool, opts ...Option) Option {
 	}
 }
 
+func If(b bool, opts ...Option) Option {
+	return ApplyIf(func(s *Settings) bool {
+		return b
+	}, opts...)
+}
+
 // Override option changes constructor for a given type
 func Override(typ, constructor interface{}) Option {
 	return func(s *Settings) error {