[Networking] logging dial addresses (#1492)

* adds addresses as return value for creating streams

* fixes lint in middleware

* adds dial addresses in libp2p node

* removes chainid from network

* reverts back dht logs

* refactors log

* Update network/p2p/libp2pNode.go

Co-authored-by: Vishal <[email protected]>

* refactors invocation of peerstore

Co-authored-by: Vishal <[email protected]>

Author: thep2p

cmd/scaffold.go

@@ -199,12 +199,11 @@ func (fnb *FlowNodeBuilder) EnqueueNetworkInit() {
 		}
 
 		libP2PNodeFactory, err := p2p.DefaultLibP2PNodeFactory(
-			fnb.Logger.Level(zerolog.ErrorLevel),
+			fnb.Logger,
 			fnb.Me.NodeID(),
 			myAddr,
 			fnb.NetworkKey,
 			fnb.RootBlock.ID(),
-			fnb.RootChainID,
 			fnb.IdentityProvider,
 			p2p.DefaultMaxPubSubMsgSize,
 			fnb.Metrics.Network,

network/p2p/libp2pNode.go

@@ -99,7 +99,6 @@ func DefaultLibP2PNodeFactory(
 	address string,
 	flowKey fcrypto.PrivateKey,
 	rootBlockID flow.Identifier,
-	chainID flow.ChainID,
 	idProvider id.IdentityProvider,
 	maxPubSubMsgSize int,
 	metrics module.NetworkMetrics,
@@ -445,11 +444,13 @@ func (n *Node) RemovePeer(ctx context.Context, peerID peer.ID) error {
 
 // CreateStream returns an existing stream connected to the peer if it exists, or creates a new stream with it.
 func (n *Node) CreateStream(ctx context.Context, peerID peer.ID) (libp2pnet.Stream, error) {
+	lg := n.logger.With().Str("peer_id", peerID.Pretty()).Logger()
+
 	// If we do not currently have any addresses for the given peer, stream creation will almost
-	// certainly fail. If this Node was configure with a DHT, we can try to lookup the address of
+	// certainly fail. If this Node was configured with a DHT, we can try to look up the address of
 	// the peer in the DHT as a last resort.
 	if len(n.host.Peerstore().Addrs(peerID)) == 0 && n.dht != nil {
-		n.logger.Info().Str("peerID", peerID.Pretty()).Msg("address not found in peerstore, searching for peer in dht")
+		lg.Info().Msg("address not found in peer store, searching for peer in dht")
 
 		var err error
 		func() {
@@ -460,36 +461,41 @@ func (n *Node) CreateStream(ctx context.Context, peerID peer.ID) (libp2pnet.Stre
 		}()
 
 		if err != nil {
-			n.logger.Warn().Err(err).Str("peerID", peerID.Pretty()).Msg("could not find addresses")
+			lg.Warn().Err(err).Msg("address not found in both peer store and dht")
 		} else {
-			n.logger.Info().Str("peerID", peerID.Pretty()).Msg("addresses found")
+			lg.Debug().Msg("address not found in peer store, but found in dht search")
 		}
 	}
-	// Open libp2p Stream with the remote peer (will use an existing TCP connection underneath if it exists)
-	stream, err := n.tryCreateNewStream(ctx, peerID, maxConnectAttempt)
+	stream, dialAddrs, err := n.tryCreateNewStream(ctx, peerID, maxConnectAttempt)
 	if err != nil {
-		return nil, flownet.NewPeerUnreachableError(fmt.Errorf("could not create stream (peer_id: %s): %w", peerID, err))
+		return nil, flownet.NewPeerUnreachableError(fmt.Errorf("could not create stream (peer_id: %s, dialing address(s): %v): %w", peerID,
+			dialAddrs, err))
 	}
+
+	lg.Debug().Str("dial_address", fmt.Sprintf("%v", dialAddrs)).Msg("stream successfully created to remote peer")
 	return stream, nil
 }
 
-// tryCreateNewStream makes at most maxAttempts to create a stream with the peer.
+// tryCreateNewStream makes at most `maxAttempts` to create a stream with the peer.
 // This was put in as a fix for #2416. PubSub and 1-1 communication compete with each other when trying to connect to
-// remote nodes and once in a while NewStream returns an error 'both yamux endpoints are clients'
-func (n *Node) tryCreateNewStream(ctx context.Context, peerID peer.ID, maxAttempts int) (libp2pnet.Stream, error) {
+// remote nodes and once in a while NewStream returns an error 'both yamux endpoints are clients'.
+//
+// Note that in case an existing TCP connection underneath to `peerID` exists, that connection is utilized for creating a new stream.
+// The multiaddr.Multiaddr return value represents the addresses of `peerID` we dial while trying to create a stream to it.
+func (n *Node) tryCreateNewStream(ctx context.Context, peerID peer.ID, maxAttempts int) (libp2pnet.Stream, []multiaddr.Multiaddr, error) {
 	// protect the underlying connection from being inadvertently pruned by the peer manager while the stream and
-	// connection creation is being attempted
+	// connection creation is being attempted, and remove it from protected list once stream created.
 	n.connMgr.ProtectPeer(peerID)
-	// unprotect it once done
 	defer n.connMgr.UnprotectPeer(peerID)
 
 	var errs error
 	var s libp2pnet.Stream
 	var retries = 0
+	var dialAddr []multiaddr.Multiaddr // address on which we dial peerID
 	for ; retries < maxAttempts; retries++ {
 		select {
 		case <-ctx.Done():
-			return nil, fmt.Errorf("context done before stream could be created (retry attempt: %d, errors: %w)", retries, errs)
+			return nil, nil, fmt.Errorf("context done before stream could be created (retry attempt: %d, errors: %w)", retries, errs)
 		default:
 		}
 
@@ -501,6 +507,7 @@ func (n *Node) tryCreateNewStream(ctx context.Context, peerID peer.ID, maxAttemp
 
 		// cancel the dial back off (if any), since we want to connect immediately
 		network := n.host.Network()
+		dialAddr = network.Peerstore().Addrs(peerID)
 		if swm, ok := network.(*swarm.Swarm); ok {
 			swm.Backoff().Clear(peerID)
 		}
@@ -518,12 +525,12 @@ func (n *Node) tryCreateNewStream(ctx context.Context, peerID peer.ID, maxAttemp
 
 			// if the connection was rejected due to invalid node id, skip the re-attempt
 			if strings.Contains(err.Error(), "failed to negotiate security protocol") {
-				return s, fmt.Errorf("invalid node id: %w", err)
+				return s, dialAddr, fmt.Errorf("invalid node id: %w", err)
 			}
 
 			// if the connection was rejected due to allowlisting, skip the re-attempt
 			if errors.Is(err, swarm.ErrGaterDisallowedConnection) {
-				return s, fmt.Errorf("target node is not on the approved list of nodes: %w", err)
+				return s, dialAddr, fmt.Errorf("target node is not on the approved list of nodes: %w", err)
 			}
 
 			errs = multierror.Append(errs, err)
@@ -535,7 +542,8 @@ func (n *Node) tryCreateNewStream(ctx context.Context, peerID peer.ID, maxAttemp
 		if err != nil {
 			// if the stream creation failed due to invalid protocol id, skip the re-attempt
 			if strings.Contains(err.Error(), "protocol not supported") {
-				return nil, fmt.Errorf("remote node is running on a different spork: %w, protocol attempted: %s", err, n.flowLibP2PProtocolID)
+				return nil, dialAddr, fmt.Errorf("remote node is running on a different spork: %w, protocol attempted: %s", err,
+					n.flowLibP2PProtocolID)
 			}
 			errs = multierror.Append(errs, err)
 			continue
@@ -545,15 +553,15 @@ func (n *Node) tryCreateNewStream(ctx context.Context, peerID peer.ID, maxAttemp
 	}
 
 	if retries == maxAttempts {
-		return s, errs
+		return s, dialAddr, errs
 	}
 
 	s, err := n.compressedStream(s)
 	if err != nil {
-		return nil, fmt.Errorf("could not create compressed stream: %w", err)
+		return nil, dialAddr, fmt.Errorf("could not create compressed stream: %w", err)
 	}
 
-	return s, nil
+	return s, dialAddr, nil
 }
 
 // GetIPPort returns the IP and Port the libp2p node is listening on.

network/p2p/middleware.go

@@ -335,10 +335,10 @@ func (m *Middleware) SendDirect(msg *message.Message, targetID flow.Identifier)
 	// create new stream
 	// (streams don't need to be reused and are fairly inexpensive to be created for each send.
 	// A stream creation does NOT incur an RTT as stream negotiation happens as part of the first message
-	// sent out the the receiver
+	// sent out the receiver
 	stream, err := m.libP2PNode.CreateStream(ctx, peerID)
 	if err != nil {
-		return fmt.Errorf("failed to create stream for %s :%w", targetID, err)
+		return fmt.Errorf("failed to create stream for %s: %w", targetID, err)
 	}
 
 	// create a gogo protobuf writer