diff --git a/cmd/nebula/cmd_crawl.go b/cmd/nebula/cmd_crawl.go index 7e6c8822..516a0aa6 100644 --- a/cmd/nebula/cmd_crawl.go +++ b/cmd/nebula/cmd_crawl.go @@ -36,7 +36,6 @@ var crawlConfig = &config.Crawl{ Network: string(config.NetworkIPFS), BootstrapPeers: cli.NewStringSlice(), Protocols: cli.NewStringSlice(string(kaddht.ProtocolDHT)), - AddrTrackTypeStr: "public", AddrDialTypeStr: "public", KeepENR: false, CheckExposed: false, @@ -74,17 +73,6 @@ var CrawlCommand = &cli.Command{ fmt.Println(crawlConfig.String()) } - switch config.AddrType(strings.ToLower(crawlConfig.AddrTrackTypeStr)) { - case config.AddrTypePrivate: - crawlConfig.AddrTrackTypeStr = string(config.AddrTypePrivate) - case config.AddrTypePublic: - crawlConfig.AddrTrackTypeStr = string(config.AddrTypePublic) - case config.AddrTypeAny: - crawlConfig.AddrTrackTypeStr = string(config.AddrTypeAny) - default: - return fmt.Errorf("unknown type of addresses to track: %s (supported values are private, public, any)", crawlConfig.AddrTrackTypeStr) - } - switch config.AddrType(strings.ToLower(crawlConfig.AddrDialTypeStr)) { case config.AddrTypePrivate: crawlConfig.AddrDialTypeStr = string(config.AddrTypePrivate) @@ -153,25 +141,19 @@ var CrawlCommand = &cli.Command{ Destination: &crawlConfig.PersistNeighbors, }, &cli.StringFlag{ - Name: "addr-track-type", - Usage: "Which type addresses should be stored to the database (private, public, any)", - EnvVars: []string{"NEBULA_CRAWL_ADDR_TRACK_TYPE"}, - Value: crawlConfig.AddrTrackTypeStr, - Destination: &crawlConfig.AddrTrackTypeStr, + Name: "network", + Usage: "Which network should be crawled. Presets default bootstrap peers and protocol. Run: `nebula networks` for more information.", + EnvVars: []string{"NEBULA_CRAWL_NETWORK"}, + Value: crawlConfig.Network, + Destination: &crawlConfig.Network, }, &cli.StringFlag{ Name: "addr-dial-type", - Usage: "Which type of addresses should Nebula try to dial (private, public, any)", + Usage: "LIBP2P/NON-ETHEREUM: Which type of addresses should Nebula try to dial (private, public, any)", EnvVars: []string{"NEBULA_CRAWL_ADDR_DIAL_TYPE"}, Value: crawlConfig.AddrDialTypeStr, Destination: &crawlConfig.AddrDialTypeStr, - }, - &cli.StringFlag{ - Name: "network", - Usage: "Which network should be crawled. Presets default bootstrap peers and protocol. Run: `nebula networks` for more information.", - EnvVars: []string{"NEBULA_CRAWL_NETWORK"}, - Value: crawlConfig.Network, - Destination: &crawlConfig.Network, + Category: flagCategoryNetwork, }, &cli.BoolFlag{ Name: "gossipsub-px", @@ -318,7 +300,6 @@ func CrawlAction(c *cli.Context) error { CrawlWorkerCount: cfg.CrawlWorkerCount, BootstrapPeers: bpEnodes, AddrDialType: cfg.AddrDialType(), - AddrTrackType: cfg.AddrTrackType(), TracerProvider: cfg.Root.TracerProvider, MeterProvider: cfg.Root.MeterProvider, LogErrors: cfg.Root.LogErrors, @@ -424,7 +405,6 @@ func CrawlAction(c *cli.Context) error { BootstrapPeers: bpEnodes, CrawlWorkerCount: cfg.CrawlWorkerCount, AddrDialType: cfg.AddrDialType(), - AddrTrackType: cfg.AddrTrackType(), KeepENR: crawlConfig.KeepENR, TracerProvider: cfg.Root.TracerProvider, MeterProvider: cfg.Root.MeterProvider, @@ -473,7 +453,6 @@ func CrawlAction(c *cli.Context) error { CheckExposed: cfg.CheckExposed, BootstrapPeers: bpAddrInfos, AddrDialType: cfg.AddrDialType(), - AddrTrackType: cfg.AddrTrackType(), TracerProvider: cfg.Root.TracerProvider, MeterProvider: cfg.Root.MeterProvider, GossipSubPX: cfg.EnableGossipSubPX, diff --git a/config/config.go b/config/config.go index f0897171..0ab65322 100644 --- a/config/config.go +++ b/config/config.go @@ -402,9 +402,6 @@ type Crawl struct { // The network to crawl Network string - // Which type addresses should be stored to the database (private, public, both) - AddrTrackTypeStr string - // Which type of addresses should Nebula try to dial (private, public, both) AddrDialTypeStr string @@ -427,10 +424,6 @@ type Crawl struct { EnableGossipSubPX bool } -func (c *Crawl) AddrTrackType() AddrType { - return AddrType(c.AddrTrackTypeStr) -} - func (c *Crawl) AddrDialType() AddrType { return AddrType(c.AddrDialTypeStr) } diff --git a/core/engine_test.go b/core/engine_test.go index 91c2b05a..c22f00a5 100644 --- a/core/engine_test.go +++ b/core/engine_test.go @@ -12,7 +12,6 @@ import ( "github.com/stretchr/testify/require" "go.uber.org/goleak" - "github.com/dennis-tra/nebula-crawler/config" "github.com/dennis-tra/nebula-crawler/db" "github.com/dennis-tra/nebula-crawler/utils" ) @@ -246,9 +245,7 @@ func TestNewEngine_Run_parking_peers(t *testing.T) { } crawler := newTestCrawler() - writerCfg := &CrawlWriterConfig{ - AddrTrackType: config.AddrTypeAny, - } + writerCfg := &CrawlWriterConfig{} writer := NewCrawlWriter[*testPeerInfo]("1", db.NewNoopClient(), writerCfg) crawler.On("Work", mock.IsType(ctx), bootstrapPeer). diff --git a/core/handler_crawl.go b/core/handler_crawl.go index 65dd2038..101c5e8d 100644 --- a/core/handler_crawl.go +++ b/core/handler_crawl.go @@ -53,6 +53,9 @@ type CrawlResult[I PeerInfo[I]] struct { // discovery protocol. ExtraMaddrs []ma.Multiaddr + // All addresses that the remote peer claims to listen on. + ListenMaddrs []ma.Multiaddr + // The multi address of the connection that we have established to the peer ConnectMaddr ma.Multiaddr @@ -99,11 +102,15 @@ func (r CrawlResult[I]) PeerInfo() I { } func (r CrawlResult[I]) LogEntry() *log.Entry { + rtSize := -1 + if r.RoutingTable != nil { + rtSize = len(r.RoutingTable.Neighbors) + } logEntry := log.WithFields(log.Fields{ "remoteID": r.Info.ID().ShortString(), "isDialable": r.ConnectError == nil && r.CrawlError == nil, "duration": r.CrawlDuration(), - "rtSize": len(r.RoutingTable.Neighbors), + "rtSize": rtSize, }) if r.ConnectError != nil { diff --git a/core/writer_crawl.go b/core/writer_crawl.go index 4858a5db..0a09bba9 100644 --- a/core/writer_crawl.go +++ b/core/writer_crawl.go @@ -8,14 +8,10 @@ import ( "github.com/libp2p/go-libp2p/core/peer" log "github.com/sirupsen/logrus" - "github.com/dennis-tra/nebula-crawler/config" "github.com/dennis-tra/nebula-crawler/db" - "github.com/dennis-tra/nebula-crawler/utils" ) -type CrawlWriterConfig struct { - AddrTrackType config.AddrType -} +type CrawlWriterConfig struct{} // CrawlWriter handles the insert/upsert/update operations for a particular crawl result. type CrawlWriter[I PeerInfo[I]] struct { @@ -47,16 +43,6 @@ func (w *CrawlWriter[I]) Work(ctx context.Context, task CrawlResult[I]) (WriteRe logEntry.Debugln("Storing peer") defer logEntry.Debugln("Stored peer") - maddrs := task.Info.Addrs() - switch w.cfg.AddrTrackType { - case config.AddrTypePrivate: - maddrs = utils.FilterPublicMaddrs(maddrs) - case config.AddrTypePublic: - maddrs = utils.FilterPrivateMaddrs(maddrs) - default: - // noop - } - var ( errorBits uint16 neighbors []peer.ID @@ -80,6 +66,7 @@ func (w *CrawlWriter[I]) Work(ctx context.Context, task CrawlResult[I]) (WriteRe DialMaddrs: task.DialMaddrs, FilteredMaddrs: task.FilteredMaddrs, ExtraMaddrs: task.ExtraMaddrs, + ListenMaddrs: task.ListenMaddrs, ConnectMaddr: task.ConnectMaddr, DialErrors: task.DialErrors, ConnectDuration: task.ConnectDuration(), diff --git a/db/ch.go b/db/ch.go index d486a28e..b46ec073 100644 --- a/db/ch.go +++ b/db/ch.go @@ -385,6 +385,7 @@ type ClickHouseVisit struct { DialMaddrs []string `ch:"dial_maddrs"` FilteredMaddrs []string `ch:"filtered_maddrs"` ExtraMaddrs []string `ch:"extra_maddrs"` + ListenMaddrs []string `ch:"listen_maddrs"` DialErrors []string `ch:"dial_errors"` ConnectMaddr *string `ch:"connect_maddr"` CrawlError *string `ch:"crawl_error"` @@ -605,6 +606,7 @@ func (c *ClickHouseClient) InsertVisit(ctx context.Context, args *VisitArgs) err DialMaddrs: utils.MaddrsToAddrs(args.DialMaddrs), FilteredMaddrs: utils.MaddrsToAddrs(args.FilteredMaddrs), ExtraMaddrs: utils.MaddrsToAddrs(args.ExtraMaddrs), + ListenMaddrs: utils.MaddrsToAddrs(args.ListenMaddrs), ConnectMaddr: connMaddrStr, DialErrors: args.DialErrors, CrawlError: crawlErrStr, diff --git a/db/client.go b/db/client.go index 2ef56399..a1045e0a 100644 --- a/db/client.go +++ b/db/client.go @@ -46,6 +46,7 @@ type VisitArgs struct { DialMaddrs []ma.Multiaddr FilteredMaddrs []ma.Multiaddr ExtraMaddrs []ma.Multiaddr + ListenMaddrs []ma.Multiaddr DialErrors []string ConnectMaddr ma.Multiaddr DialDuration time.Duration diff --git a/db/json.go b/db/json.go index 9a85b248..061fec91 100644 --- a/db/json.go +++ b/db/json.go @@ -161,6 +161,9 @@ func (c *JSONClient) InsertCrawlProperties(ctx context.Context, properties map[s type JSONVisit struct { PeerID peer.ID Maddrs []ma.Multiaddr + FilteredMaddrs []ma.Multiaddr + ListenMaddrs []ma.Multiaddr + ConnectMaddr ma.Multiaddr Protocols []string AgentVersion string ConnectDuration string @@ -175,7 +178,10 @@ type JSONVisit struct { func (c *JSONClient) InsertVisit(ctx context.Context, args *VisitArgs) error { data := JSONVisit{ PeerID: args.PeerID, - Maddrs: slices.Concat(args.DialMaddrs, args.FilteredMaddrs, args.ExtraMaddrs), + Maddrs: slices.Concat(args.DialMaddrs), + FilteredMaddrs: args.FilteredMaddrs, + ListenMaddrs: args.ListenMaddrs, + ConnectMaddr: args.ConnectMaddr, Protocols: args.Protocols, AgentVersion: args.AgentVersion, ConnectDuration: args.ConnectDuration.String(), diff --git a/db/migrations/chcluster/000006_add_missing_maddrs_column_to_visits_table.down.sql b/db/migrations/chcluster/000006_add_missing_maddrs_column_to_visits_table.down.sql new file mode 100644 index 00000000..87c73305 --- /dev/null +++ b/db/migrations/chcluster/000006_add_missing_maddrs_column_to_visits_table.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE visits + DROP COLUMN listen_maddrs; \ No newline at end of file diff --git a/db/migrations/chcluster/000006_add_missing_maddrs_column_to_visits_table.up.sql b/db/migrations/chcluster/000006_add_missing_maddrs_column_to_visits_table.up.sql new file mode 100644 index 00000000..3735d175 --- /dev/null +++ b/db/migrations/chcluster/000006_add_missing_maddrs_column_to_visits_table.up.sql @@ -0,0 +1,2 @@ +ALTER TABLE visits + ADD COLUMN listen_maddrs Array(String) DEFAULT [] AFTER extra_maddrs; \ No newline at end of file diff --git a/db/migrations/chlocal/000003_create_visits_table.up.sql b/db/migrations/chlocal/000003_create_visits_table.up.sql index d97f3c83..0614e198 100644 --- a/db/migrations/chlocal/000003_create_visits_table.up.sql +++ b/db/migrations/chlocal/000003_create_visits_table.up.sql @@ -1,3 +1,5 @@ +-- DO NOT EDIT: This file was generated with: just generate-local-clickhouse-migrations + -- Captures the information when crawling a peer CREATE TABLE visits ( diff --git a/db/migrations/chlocal/000004_create_neighbors_table.up.sql b/db/migrations/chlocal/000004_create_neighbors_table.up.sql index 4cd2fa2e..80b3303e 100644 --- a/db/migrations/chlocal/000004_create_neighbors_table.up.sql +++ b/db/migrations/chlocal/000004_create_neighbors_table.up.sql @@ -1,3 +1,5 @@ +-- DO NOT EDIT: This file was generated with: just generate-local-clickhouse-migrations + CREATE TABLE neighbors ( -- identifies the crawl that produced this neighbor mapping diff --git a/db/migrations/chlocal/000006_add_missing_maddrs_column_to_visits_table.down.sql b/db/migrations/chlocal/000006_add_missing_maddrs_column_to_visits_table.down.sql new file mode 100644 index 00000000..97d84c45 --- /dev/null +++ b/db/migrations/chlocal/000006_add_missing_maddrs_column_to_visits_table.down.sql @@ -0,0 +1,4 @@ +-- DO NOT EDIT: This file was generated with: just generate-local-clickhouse-migrations + +ALTER TABLE visits + DROP COLUMN listen_maddrs; \ No newline at end of file diff --git a/db/migrations/chlocal/000006_add_missing_maddrs_column_to_visits_table.up.sql b/db/migrations/chlocal/000006_add_missing_maddrs_column_to_visits_table.up.sql new file mode 100644 index 00000000..77039b97 --- /dev/null +++ b/db/migrations/chlocal/000006_add_missing_maddrs_column_to_visits_table.up.sql @@ -0,0 +1,4 @@ +-- DO NOT EDIT: This file was generated with: just generate-local-clickhouse-migrations + +ALTER TABLE visits + ADD COLUMN listen_maddrs Array(String) DEFAULT [] AFTER extra_maddrs; \ No newline at end of file diff --git a/discv4/crawler.go b/discv4/crawler.go index bd398673..731bda02 100644 --- a/discv4/crawler.go +++ b/discv4/crawler.go @@ -157,6 +157,7 @@ func (c *Crawler) Work(ctx context.Context, task PeerInfo) (core.CrawlResult[Pee DialMaddrs: devp2pResult.DialMaddrs, FilteredMaddrs: filteredMaddrs, ExtraMaddrs: []ma.Multiaddr{}, + ListenMaddrs: []ma.Multiaddr{}, ConnectMaddr: connectMaddr, DialErrors: db.MaddrErrors(devp2pResult.DialMaddrs, devp2pResult.ConnectError), ConnectError: devp2pResult.ConnectError, diff --git a/discv4/driver_crawler.go b/discv4/driver_crawler.go index 1632085a..3ffd4bd3 100644 --- a/discv4/driver_crawler.go +++ b/discv4/driver_crawler.go @@ -155,7 +155,6 @@ type CrawlDriverConfig struct { DialTimeout time.Duration BootstrapPeers []*enode.Node AddrDialType config.AddrType - AddrTrackType config.AddrType MeterProvider metric.MeterProvider TracerProvider trace.TracerProvider LogErrors bool @@ -175,9 +174,7 @@ func (cfg *CrawlDriverConfig) CrawlerConfig() *CrawlerConfig { } func (cfg *CrawlDriverConfig) WriterConfig() *core.CrawlWriterConfig { - return &core.CrawlWriterConfig{ - AddrTrackType: cfg.AddrTrackType, - } + return &core.CrawlWriterConfig{} } type CrawlDriver struct { diff --git a/discv5/crawler.go b/discv5/crawler.go index 83208d03..6da65735 100644 --- a/discv5/crawler.go +++ b/discv5/crawler.go @@ -16,7 +16,6 @@ import ( "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/peerstore" - basichost "github.com/libp2p/go-libp2p/p2p/host/basic" "github.com/libp2p/go-msgio/pbio" ma "github.com/multiformats/go-multiaddr" log "github.com/sirupsen/logrus" @@ -27,6 +26,7 @@ import ( "github.com/dennis-tra/nebula-crawler/core" "github.com/dennis-tra/nebula-crawler/db" pgmodels "github.com/dennis-tra/nebula-crawler/db/models/pg" + nebp2p "github.com/dennis-tra/nebula-crawler/libp2p" ) const MaxCrawlRetriesAfterTimeout = 2 // magic @@ -45,7 +45,7 @@ type CrawlerConfig struct { type Crawler struct { id string cfg *CrawlerConfig - host *basichost.BasicHost + host *nebp2p.Host listener *discover.UDPv5 crawledPeers int done chan struct{} @@ -146,7 +146,7 @@ func (c *Crawler) Work(ctx context.Context, task PeerInfo) (core.CrawlResult[Pee // construct a slice with multi addresses that the peer replied with which // we didn't know before. var extraMaddrs []ma.Multiaddr - for _, maddr := range libp2pResult.ListenAddrs { + for _, maddr := range libp2pResult.ListenMaddrs { if _, ok := knownMaddrsMap[string(maddr.Bytes())]; ok { continue } @@ -164,6 +164,7 @@ func (c *Crawler) Work(ctx context.Context, task PeerInfo) (core.CrawlResult[Pee DialMaddrs: libp2pResult.DialMaddrs, FilteredMaddrs: filteredMaddrs, ExtraMaddrs: extraMaddrs, + ListenMaddrs: libp2pResult.ListenMaddrs, ConnectMaddr: connectMaddr, DialErrors: db.MaddrErrors(libp2pResult.DialMaddrs, libp2pResult.ConnectError), ConnectError: libp2pResult.ConnectError, @@ -299,7 +300,7 @@ type Libp2pResult struct { ConnectMaddr ma.Multiaddr Agent string Protocols []string - ListenAddrs []ma.Multiaddr + ListenMaddrs []ma.Multiaddr ConnClosedImmediately bool // whether conn was no error but still unconnected GenTCPAddr bool // whether a TCP address was generated WakuClusterID uint32 @@ -322,6 +323,10 @@ func (c *Crawler) crawlLibp2p(ctx context.Context, pi PeerInfo) chan Libp2pResul Addrs: result.DialMaddrs, } + // register the given peer (before connecting) to receive + // the identify result on the returned channel + identifyChan := c.host.RegisterIdentify(pi.ID()) + var conn network.Conn result.ConnectStartTime = time.Now() conn, result.ConnectError = c.connect(ctx, addrInfo) // use filtered addr list @@ -350,33 +355,28 @@ func (c *Crawler) crawlLibp2p(ctx context.Context, pi PeerInfo) chan Libp2pResul select { case <-timeoutCtx.Done(): // identification timed out. - case <-c.host.IDService().IdentifyWait(conn): + case identify, more := <-identifyChan: // identification may have succeeded. - } - - // Extract information from peer store - ps := c.host.Peerstore() - - // Extract agent - if agent, err := ps.Get(pi.ID(), "AgentVersion"); err == nil { - result.Agent = agent.(string) - } + if !more { + break + } - // Extract protocols - if protocols, err := ps.GetProtocols(pi.ID()); err == nil { - result.Protocols = make([]string, len(protocols)) - for i := range protocols { - result.Protocols[i] = string(protocols[i]) + result.Agent = identify.AgentVersion + result.ListenMaddrs = identify.ListenAddrs + result.Protocols = make([]string, len(identify.Protocols)) + for i := range identify.Protocols { + result.Protocols[i] = string(identify.Protocols[i]) } } - // Extract listen addresses - result.ListenAddrs = ps.Addrs(pi.ID()) } else { // if there was a connection error, parse it to a known one result.ConnectErrorStr = db.NetError(result.ConnectError) } + // deregister peer from identify messages + c.host.DeregisterIdentify(pi.ID()) + // Free connection resources if err := c.host.Network().ClosePeer(pi.ID()); err != nil { log.WithError(err).WithField("remoteID", pi.ID().ShortString()).Warnln("Could not close connection to peer") diff --git a/discv5/driver_crawler.go b/discv5/driver_crawler.go index 1d036b96..db348716 100644 --- a/discv5/driver_crawler.go +++ b/discv5/driver_crawler.go @@ -19,10 +19,8 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/libp2p/go-libp2p" mplex "github.com/libp2p/go-libp2p-mplex" - libp2pconfig "github.com/libp2p/go-libp2p/config" "github.com/libp2p/go-libp2p/core/connmgr" "github.com/libp2p/go-libp2p/core/crypto" - "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/p2p/muxer/yamux" @@ -38,6 +36,7 @@ import ( "github.com/dennis-tra/nebula-crawler/config" "github.com/dennis-tra/nebula-crawler/core" "github.com/dennis-tra/nebula-crawler/db" + nebp2p "github.com/dennis-tra/nebula-crawler/libp2p" "github.com/dennis-tra/nebula-crawler/utils" ) @@ -164,7 +163,6 @@ type CrawlDriverConfig struct { BootstrapPeers []*enode.Node CrawlWorkerCount int AddrDialType config.AddrType - AddrTrackType config.AddrType KeepENR bool MeterProvider metric.MeterProvider TracerProvider trace.TracerProvider @@ -190,15 +188,13 @@ func (cfg *CrawlDriverConfig) CrawlerConfig() *CrawlerConfig { } func (cfg *CrawlDriverConfig) WriterConfig() *core.CrawlWriterConfig { - return &core.CrawlWriterConfig{ - AddrTrackType: cfg.AddrTrackType, - } + return &core.CrawlWriterConfig{} } type CrawlDriver struct { cfg *CrawlDriverConfig dbc db.Client - hosts []host.Host + hosts []*nebp2p.Host tasksChan chan PeerInfo peerstore *enode.DB crawlerCount int @@ -210,9 +206,9 @@ var _ core.Driver[PeerInfo, core.CrawlResult[PeerInfo]] = (*CrawlDriver)(nil) func NewCrawlDriver(dbc db.Client, cfg *CrawlDriverConfig) (*CrawlDriver, error) { // create a libp2p host per CPU core to distribute load - hosts := make([]host.Host, 0, runtime.NumCPU()) + hosts := make([]*nebp2p.Host, 0, runtime.NumCPU()) for i := 0; i < runtime.NumCPU(); i++ { - h, err := newLibp2pHost(cfg) + h, err := newHost(cfg.Network, cfg.Version) if err != nil { return nil, fmt.Errorf("new libp2p host: %w", err) } @@ -302,7 +298,7 @@ func (d *CrawlDriver) NewWorker() (core.Worker[PeerInfo, core.CrawlResult[PeerIn c := &Crawler{ id: fmt.Sprintf("crawler-%02d", d.crawlerCount), cfg: d.cfg.CrawlerConfig(), - host: h.(*libp2pconfig.ClosableBasicHost).BasicHost, + host: h, listener: listener, done: make(chan struct{}), } @@ -340,7 +336,7 @@ func (d *CrawlDriver) Close() { } } -func newLibp2pHost(cfg *CrawlDriverConfig) (host.Host, error) { +func newHost(net config.Network, version string) (*nebp2p.Host, error) { cm := connmgr.NullConnMgr{} rm := network.NullResourceManager{} @@ -359,7 +355,7 @@ func newLibp2pHost(cfg *CrawlDriverConfig) (host.Host, error) { libp2p.ResourceManager(&rm), libp2p.Identity(secpKey), libp2p.Security(noise.ID, noise.New), - libp2p.UserAgent("nebula/"+cfg.Version), + libp2p.UserAgent("nebula/"+version), libp2p.Transport(tcp.NewTCPTransport), libp2p.Transport(quic.NewTransport), libp2p.Muxer(mplex.ID, mplex.DefaultTransport), @@ -375,7 +371,7 @@ func newLibp2pHost(cfg *CrawlDriverConfig) (host.Host, error) { return nil, fmt.Errorf("new libp2p host: %w", err) } - switch cfg.Network { + switch net { case config.NetworkEthCons, config.NetworkHolesky, config.NetworkPortal: // According to Diva, these are required protocols. Some of them are just // assumed to be required. We just read from the stream indefinitely to @@ -392,5 +388,5 @@ func newLibp2pHost(cfg *CrawlDriverConfig) (host.Host, error) { } log.WithField("peerID", h.ID().String()).Infoln("Started libp2p host") - return h, nil + return nebp2p.WrapHost(h) } diff --git a/libp2p/crawler.go b/libp2p/crawler.go index 99276303..7c254cfb 100644 --- a/libp2p/crawler.go +++ b/libp2p/crawler.go @@ -44,7 +44,7 @@ func DefaultCrawlerConfig() *CrawlerConfig { type Crawler struct { id string cfg *CrawlerConfig - host Host + host *Host pm *pb.ProtocolMessenger psTopics map[string]struct{} crawledPeers int @@ -184,14 +184,29 @@ func mergeResults(r *core.CrawlResult[PeerInfo], p2pRes P2PResult, apiRes APIRes knownMaddrs[string(maddr.Bytes())] = struct{}{} } + // map to deduplicate the listen address lists from the api and p2p responses. + listenMaddrsMap := map[string]ma.Multiaddr{} + r.ExtraMaddrs = []ma.Multiaddr{} for _, maddr := range slices.Concat(apiRes.ListenMaddrs(), p2pRes.ListenMaddrs) { + // build up deduplicated map + listenMaddrsMap[string(maddr.Bytes())] = maddr + + // if we know this listen address already, skip it. if _, ok := knownMaddrs[string(maddr.Bytes())]; ok { continue } + + // we didn't know this address at dial-time, so we add it to the extra list. r.ExtraMaddrs = append(r.ExtraMaddrs, maddr) } + // transform the map of deduplicated listen addresses into a slice of multi addresses. + r.ListenMaddrs = make([]ma.Multiaddr, 0, len(listenMaddrsMap)) + for _, maddr := range listenMaddrsMap { + r.ListenMaddrs = append(r.ListenMaddrs, maddr) + } + if len(r.RoutingTable.Neighbors) == 0 && apiRes.RoutingTable != nil { // construct routing table struct from API response rt := &core.RoutingTable[PeerInfo]{ diff --git a/libp2p/crawler_p2p.go b/libp2p/crawler_p2p.go index 343db618..7a3c2976 100644 --- a/libp2p/crawler_p2p.go +++ b/libp2p/crawler_p2p.go @@ -83,6 +83,10 @@ func (c *Crawler) crawlP2P(ctx context.Context, pi PeerInfo) <-chan P2PResult { RoutingTable: &core.RoutingTable[PeerInfo]{PeerID: pi.ID()}, } + // register the given peer (before connecting) to receive + // the identify result on the returned channel + identifyChan := c.host.RegisterIdentify(pi.ID()) + var conn network.Conn result.ConnectStartTime = time.Now() conn, result.ConnectError = c.connect(ctx, pi.AddrInfo) // use filtered addr list @@ -110,29 +114,20 @@ func (c *Crawler) crawlP2P(ctx context.Context, pi PeerInfo) <-chan P2PResult { select { case <-timeoutCtx.Done(): // identification timed out. - case <-c.host.IDService().IdentifyWait(conn): + case identify, more := <-identifyChan: // identification may have succeeded. - } - - // Extract information from peer store - ps := c.host.Peerstore() - - // Extract agent - if agent, err := ps.Get(pi.ID(), "AgentVersion"); err == nil { - result.Agent = agent.(string) - } + if !more { + break + } - // Extract protocols - if protocols, err := ps.GetProtocols(pi.ID()); err == nil { - result.Protocols = make([]string, len(protocols)) - for i := range protocols { - result.Protocols[i] = string(protocols[i]) + result.Agent = identify.AgentVersion + result.ListenMaddrs = identify.ListenAddrs + result.Protocols = make([]string, len(identify.Protocols)) + for i := range identify.Protocols { + result.Protocols[i] = string(identify.Protocols[i]) } } - // Extract listen addresses - result.ListenMaddrs = ps.Addrs(pi.ID()) - if c.cfg.GossipSubPX { // give the other peer a chance to open a stream to and prune us streams := openInboundGossipSubStreams(c.host, pi.ID()) @@ -199,6 +194,9 @@ func (c *Crawler) crawlP2P(ctx context.Context, pi PeerInfo) <-chan P2PResult { result.ConnectErrorStr = db.NetError(result.ConnectError) } + // deregister peer from identify messages + c.host.DeregisterIdentify(pi.ID()) + // send the result back and close channel select { case resultCh <- result: diff --git a/libp2p/driver_crawler.go b/libp2p/driver_crawler.go index 32bbf653..7a59a5d6 100644 --- a/libp2p/driver_crawler.go +++ b/libp2p/driver_crawler.go @@ -21,7 +21,6 @@ import ( "github.com/libp2p/go-libp2p/core/protocol" "github.com/libp2p/go-libp2p/core/record" "github.com/libp2p/go-libp2p/p2p/net/swarm" - "github.com/libp2p/go-libp2p/p2p/protocol/identify" "github.com/libp2p/go-msgio" ma "github.com/multiformats/go-multiaddr" log "github.com/sirupsen/logrus" @@ -36,14 +35,6 @@ import ( "github.com/dennis-tra/nebula-crawler/utils" ) -// Host is the interface that's required for crawling libp2p peers. Actually -// the *basichost.Host is required but to allow testing we define this interface -// here. That allows us to inject a mock host. -type Host interface { - host.Host - IDService() identify.IDService -} - type PeerInfo struct { peer.AddrInfo } @@ -88,7 +79,6 @@ type CrawlDriverConfig struct { DialTimeout time.Duration CheckExposed bool BootstrapPeers []peer.AddrInfo - AddrTrackType config.AddrType AddrDialType config.AddrType MeterProvider metric.MeterProvider TracerProvider trace.TracerProvider @@ -107,14 +97,12 @@ func (cfg *CrawlDriverConfig) CrawlerConfig() *CrawlerConfig { } func (cfg *CrawlDriverConfig) WriterConfig() *core.CrawlWriterConfig { - return &core.CrawlWriterConfig{ - AddrTrackType: cfg.AddrTrackType, - } + return &core.CrawlWriterConfig{} } type CrawlDriver struct { cfg *CrawlDriverConfig - hosts map[peer.ID]Host + hosts map[peer.ID]*Host dbc db.Client // pxPeersChan receives peers that we get to know @@ -151,7 +139,7 @@ func NewCrawlDriver(dbc db.Client, cfg *CrawlDriverConfig) (*CrawlDriver, error) userAgent = "avail-light-client/light-client/1.12.13/rust-client" } - hosts := make(map[peer.ID]Host, runtime.NumCPU()) + hosts := make(map[peer.ID]*Host, runtime.NumCPU()) for i := 0; i < runtime.NumCPU(); i++ { h, err := newLibp2pHost(userAgent) if err != nil { @@ -201,7 +189,7 @@ func (d *CrawlDriver) NewWorker() (core.Worker[PeerInfo, core.CrawlResult[PeerIn hostID := peer.ID(hostsList[d.crawlerCount%len(d.hosts)]) ms := &msgSender{ - h: d.hosts[hostID], + h: d.hosts[hostID].Host, protocols: protocol.ConvertFromStrings(d.cfg.Protocols), timeout: d.cfg.DialTimeout, } @@ -254,7 +242,7 @@ func (d *CrawlDriver) shutdown() { var wgHostClose sync.WaitGroup for _, h := range d.hosts { wgHostClose.Add(1) - go func(h Host) { + go func(h *Host) { defer wgHostClose.Done() if err := h.Close(); err != nil { log.WithError(err).Warnln("failed to close host") @@ -313,7 +301,7 @@ LOOP: } } -func newLibp2pHost(userAgent string) (Host, error) { +func newLibp2pHost(userAgent string) (*Host, error) { // Configure the resource manager to not limit anything // Don't use a connection manager that could potentially // prune any connections. We clean up after ourselves. @@ -330,8 +318,11 @@ func newLibp2pHost(userAgent string) (Host, error) { libp2p.UDPBlackHoleSuccessCounter(nil), libp2p.IPv6BlackHoleSuccessCounter(nil), ) + if err != nil { + return nil, fmt.Errorf("new libp2p host: %w", err) + } - return h.(Host), err + return WrapHost(h) } // handleGossipSubStream manages a GossipSub stream between two peers. It first diff --git a/libp2p/host.go b/libp2p/host.go new file mode 100644 index 00000000..3b9d8cfa --- /dev/null +++ b/libp2p/host.go @@ -0,0 +1,123 @@ +package libp2p + +import ( + "fmt" + "sync" + "time" + + "github.com/libp2p/go-libp2p/core/event" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/peer" + log "github.com/sirupsen/logrus" +) + +// Host wraps a regular libp2p host and exposes additional methods to +// register peers to receive Identify exchange events. +// TODO: I'm not quite happy with all the locking going on here... +type Host struct { + host.Host + + // the event bus subscription for Identify events + sub event.Subscription + subDone chan struct{} + + // a map of peer registrations ot the respective channel on which the events + // should be emitted. + regsMu sync.Mutex + regs map[peer.ID]chan event.EvtPeerIdentificationCompleted +} + +// WrapHost wraps the given host to allow for registering peers for Identify +// events. +func WrapHost(h host.Host) (*Host, error) { + sub, err := h.EventBus().Subscribe([]interface{}{new(event.EvtPeerIdentificationCompleted), new(event.EvtPeerIdentificationFailed)}) + if err != nil { + return nil, fmt.Errorf("establish event subscription: %w", err) + } + + wrapped := &Host{ + Host: h, + sub: sub, + subDone: make(chan struct{}), + regs: make(map[peer.ID]chan event.EvtPeerIdentificationCompleted), + } + + go wrapped.consumeEvents() + + return wrapped, nil +} + +// RegisterIdentify registers the given peer to receive the corresponding Identify +// events. The returned channel will emit the +// [event.EvtPeerIdentificationCompleted] event or be closed without an event in +// case of an error. The returned channel will emit at most one event. +func (h *Host) RegisterIdentify(pid peer.ID) <-chan event.EvtPeerIdentificationCompleted { + h.regsMu.Lock() + defer h.regsMu.Unlock() + + if resultChan, ok := h.regs[pid]; ok { + return resultChan + } + + h.regs[pid] = make(chan event.EvtPeerIdentificationCompleted, 1) + + return h.regs[pid] +} + +// DeregisterIdentify deregisters the given peer and closes the channel which +// was previously returned in [RegisterIdentify] +func (h *Host) DeregisterIdentify(pid peer.ID) { + h.regsMu.Lock() + defer h.regsMu.Unlock() + + if resultChan, ok := h.regs[pid]; ok { + delete(h.regs, pid) + close(resultChan) + } +} + +func (h *Host) consumeEvents() { + defer close(h.subDone) + for e := range h.sub.Out() { + + h.regsMu.Lock() + + switch evt := e.(type) { + case event.EvtPeerIdentificationCompleted: + resultChan, found := h.regs[evt.Peer] + if !found { + break + } + + resultChan <- evt + delete(h.regs, evt.Peer) + close(resultChan) + case event.EvtPeerIdentificationFailed: + resultChan, found := h.regs[evt.Peer] + if !found { + break + } + + delete(h.regs, evt.Peer) + close(resultChan) + default: + panic(fmt.Sprintf("received unexpected event")) + } + + h.regsMu.Unlock() + } +} + +func (h *Host) Close() error { + if err := h.sub.Close(); err != nil { + log.WithError(err).Warnln("Failed closing event subscription") + } else { + select { + case <-h.subDone: + case <-time.After(time.Second): + log.Warnln("Event subscription did not close within 1s") + } + } + + return h.Host.Close() +}