improve logging and error reporting in libp2p and handshake

cmd/bee/cmd/start.go

@@ -93,6 +93,7 @@ func (c *command) initStartCmd() (err error) {
 					ConnectionsLow:   c.config.GetInt(optionNameConnectionsLow),
 					ConnectionsHigh:  c.config.GetInt(optionNameConnectionsHigh),
 					ConnectionsGrace: c.config.GetDuration(optionNameConnectionsGrace),
+					Logger:           logger,
 				},
 				Logger: logger,
 			})

pkg/p2p/libp2p/internal/handshake/handshake.go

@@ -8,8 +8,6 @@ package handshake
 
 import (
 	"fmt"
-	"io"
-	"log"
 
 	"github.com/janos/bee/pkg/p2p"
 	"github.com/janos/bee/pkg/p2p/protobuf"
@@ -23,52 +21,52 @@ const (
 
 type Service struct {
 	overlay string
+	logger  Logger
 }
 
-func New(overlay string) *Service {
-	return &Service{overlay: overlay}
+func New(overlay string, logger Logger) *Service {
+	return &Service{
+		overlay: overlay,
+		logger:  logger,
+	}
+}
+
+type Logger interface {
+	Tracef(format string, args ...interface{})
 }
 
 func (s *Service) Handshake(stream p2p.Stream) (overlay string, err error) {
 	w, r := protobuf.NewWriterAndReader(stream)
 	var resp ShakeHand
 	if err := w.WriteMsg(&ShakeHand{Address: s.overlay}); err != nil {
-		return "", fmt.Errorf("handshake handler: write message: %v\n", err)
+		return "", fmt.Errorf("handshake handler: write message: %w", err)
 	}
 
-	log.Printf("sent handshake req %s\n", s.overlay)
+	s.logger.Tracef("handshake: sent request %s", s.overlay)
 	if err := r.ReadMsg(&resp); err != nil {
-		if err == io.EOF {
-			return "", nil
+		return "", fmt.Errorf("handshake handler: read message: %w", err)
 	}
 
-		return "", fmt.Errorf("handshake handler: read message: %v\n", err)
-	}
-
-	log.Printf("read handshake resp: %s\n", resp.Address)
+	s.logger.Tracef("handshake: read response: %s", resp.Address)
 	return resp.Address, nil
 }
 
-func (s *Service) Handler(stream p2p.Stream) string {
+func (s *Service) Handle(stream p2p.Stream) (overlay string, err error) {
 	w, r := protobuf.NewWriterAndReader(stream)
 	defer stream.Close()
 
 	var req ShakeHand
 	if err := r.ReadMsg(&req); err != nil {
-		if err == io.EOF {
-			return ""
-		}
-		log.Printf("handshake handler: read message: %v\n", err)
-		return ""
+		return "", fmt.Errorf("read message: %w", err)
 	}
 
-	log.Printf("received handshake req %s\n", req.Address)
+	s.logger.Tracef("handshake: received request %s", req.Address)
 	if err := w.WriteMsg(&ShakeHand{
 		Address: s.overlay,
 	}); err != nil {
-		log.Printf("handshake handler: write message: %v\n", err)
+		return "", fmt.Errorf("write message: %w", err)
 	}
 
-	log.Printf("sent handshake resp: %s\n", s.overlay)
-	return req.Address
+	s.logger.Tracef("handshake: handled response: %s", s.overlay)
+	return req.Address, nil
 }

pkg/p2p/libp2p/libp2p.go

@@ -42,6 +42,7 @@ type Service struct {
 	metrics          metrics
 	handshakeService *handshake.Service
 	peers            *peerRegistry
+	logger           Logger
 }
 
 type Options struct {
@@ -54,6 +55,13 @@ type Options struct {
 	ConnectionsLow   int
 	ConnectionsHigh  int
 	ConnectionsGrace time.Duration
+	Logger           Logger
+}
+
+type Logger interface {
+	Tracef(format string, args ...interface{})
+	Infof(format string, args ...interface{})
+	Errorf(format string, args ...interface{})
 }
 
 func New(ctx context.Context, o Options) (*Service, error) {
@@ -182,8 +190,9 @@ func New(ctx context.Context, o Options) (*Service, error) {
 	s := &Service{
 		host:             h,
 		metrics:          newMetrics(),
-		handshakeService: handshake.New(overlay),
+		handshakeService: handshake.New(overlay, o.Logger),
 		peers:            newPeerRegistry(),
+		logger:           o.Logger,
 	}
 
 	// Construct protocols.
@@ -195,9 +204,14 @@ func New(ctx context.Context, o Options) (*Service, error) {
 	}
 
 	s.host.SetStreamHandlerMatch(id, matcher, func(stream network.Stream) {
+		peerID := stream.Conn().RemotePeer()
+		overlay, err := s.handshakeService.Handle(stream)
+		if err != nil {
+			s.logger.Errorf("handshake with peer %s: %w", peerID, err)
+		}
+		s.peers.add(peerID, overlay)
 		s.metrics.HandledStreamCount.Inc()
-		overlay := s.handshakeService.Handler(stream)
-		s.peers.add(stream.Conn().RemotePeer(), overlay)
+		s.logger.Infof("peer %q connected", overlay)
 	})
 
 	// TODO: be more resilient on connection errors and connect in parallel
@@ -233,7 +247,7 @@ func (s *Service) AddProtocol(p p2p.ProtocolSpec) (err error) {
 			overlay, found := s.peers.overlay(peerID)
 			if !found {
 				// todo: handle better
-				fmt.Printf("Could not fetch handshake for peerID %s\n", stream)
+				s.logger.Errorf("overlay address for peer %q not found", peerID)
 				return
 			}
 
@@ -283,7 +297,7 @@ func (s *Service) Connect(ctx context.Context, addr ma.Multiaddr) (err error) {
 
 	s.peers.add(info.ID, overlay)
 	s.metrics.CreatedConnectionCount.Inc()
-	fmt.Println("handshake handshake finished")
+	s.logger.Infof("peer %q connected", overlay)
 	return nil
 }
 func (s *Service) NewStream(ctx context.Context, overlay, protocolName, streamName, version string) (p2p.Stream, error) {