Migrate to log/slog

Author: MarcoPolo

comm.go

@@ -47,7 +47,7 @@ func (p *PubSub) handleNewStream(s network.Stream) {
 	p.inboundStreamsMx.Lock()
 	other, dup := p.inboundStreams[peer]
 	if dup {
-		log.Debugf("duplicate inbound stream from %s; resetting other stream", peer)
+		p.logger.Debug("duplicate inbound stream from; resetting other stream", "peer", peer)
 		other.Reset()
 	}
 	p.inboundStreams[peer] = s
@@ -63,12 +63,16 @@ func (p *PubSub) handleNewStream(s network.Stream) {
 
 	r := msgio.NewVarintReaderSize(s, p.maxMessageSize)
 	for {
+		// Peek at the message length to know when we should mark the start time
+		// for measuring how long it took to receive a message.
+		_, _ = r.NextMsgLen()
+		start := time.Now()
 		msgbytes, err := r.ReadMsg()
 		if err != nil {
 			r.ReleaseMsg(msgbytes)
 			if err != io.EOF {
 				s.Reset()
-				log.Debugf("error reading rpc from %s: %s", s.Conn().RemotePeer(), err)
+				p.rpcLogger.Debug("error reading rpc", "from", s.Conn().RemotePeer(), "err", err)
 			} else {
 				// Just be nice. They probably won't read this
 				// but it doesn't hurt to send it.
@@ -86,10 +90,14 @@ func (p *PubSub) handleNewStream(s network.Stream) {
 		r.ReleaseMsg(msgbytes)
 		if err != nil {
 			s.Reset()
-			log.Warnf("bogus rpc from %s: %s", s.Conn().RemotePeer(), err)
+
+			p.rpcLogger.Warn("bogus rpc from", "peer", s.Conn().RemotePeer(), "err", err)
 			return
 		}
 
+		timeToReceive := time.Since(start)
+		p.rpcLogger.Debug("received", "peer", s.Conn().RemotePeer(), "duration_s", timeToReceive.Seconds(), "rpc", rpc)
+
 		rpc.from = peer
 		select {
 		case p.incoming <- rpc:
@@ -117,7 +125,7 @@ func (p *PubSub) notifyPeerDead(pid peer.ID) {
 func (p *PubSub) handleNewPeer(ctx context.Context, pid peer.ID, outgoing *rpcQueue) {
 	s, err := p.host.NewStream(p.ctx, pid, p.rt.Protocols()...)
 	if err != nil {
-		log.Debug("opening new stream to peer: ", err, pid)
+		p.logger.Debug("error opening new stream to peer", "err", err, "peer", pid)
 
 		select {
 		case p.newPeerError <- pid:
@@ -149,7 +157,7 @@ func (p *PubSub) handlePeerDead(s network.Stream) {
 
 	_, err := s.Read([]byte{0})
 	if err == nil {
-		log.Debugf("unexpected message from %s", pid)
+		p.logger.Debug("unexpected message from peer", "peer", pid)
 	}
 
 	s.Reset()
@@ -170,21 +178,26 @@ func (p *PubSub) handleSendingMessages(ctx context.Context, s network.Stream, ou
 		}
 
 		_, err = s.Write(buf)
-		return err
+		if err != nil {
+			p.rpcLogger.Debug("failed to send message", "peer", s.Conn().RemotePeer(), "rpc", rpc, "err", err)
+			return err
+		}
+		p.rpcLogger.Debug("sent", "peer", s.Conn().RemotePeer(), "rpc", rpc)
+		return nil
 	}
 
 	defer s.Close()
 	for ctx.Err() == nil {
 		rpc, err := outgoing.Pop(ctx)
 		if err != nil {
-			log.Debugf("popping message from the queue to send to %s: %s", s.Conn().RemotePeer(), err)
+			p.logger.Debug("error popping message from the queue to send to peer", "peer", s.Conn().RemotePeer(), "err", err)
 			return
 		}
 
 		err = writeRpc(rpc)
 		if err != nil {
 			s.Reset()
-			log.Debugf("writing message to %s: %s", s.Conn().RemotePeer(), err)
+			p.logger.Debug("error writing message to peer", "peer", s.Conn().RemotePeer(), "err", err)
 			return
 		}
 	}

discovery.go

@@ -190,7 +190,7 @@ func (d *discover) Advertise(topic string) {
 	go func() {
 		next, err := d.discovery.Advertise(advertisingCtx, topic)
 		if err != nil {
-			log.Warnf("bootstrap: error providing rendezvous for %s: %s", topic, err.Error())
+			d.p.logger.Warn("bootstrap: error providing rendezvous for topic", "topic", topic, "err", err)
 			if next == 0 {
 				next = discoveryAdvertiseRetryInterval
 			}
@@ -204,7 +204,7 @@ func (d *discover) Advertise(topic string) {
 			case <-t.C:
 				next, err = d.discovery.Advertise(advertisingCtx, topic)
 				if err != nil {
-					log.Warnf("bootstrap: error providing rendezvous for %s: %s", topic, err.Error())
+					d.p.logger.Warn("bootstrap: error providing rendezvous for topic", "topic", topic, "err", err)
 					if next == 0 {
 						next = discoveryAdvertiseRetryInterval
 					}
@@ -302,7 +302,7 @@ func (d *discover) handleDiscovery(ctx context.Context, topic string, opts []dis
 
 	peerCh, err := d.discovery.FindPeers(discoverCtx, topic, opts...)
 	if err != nil {
-		log.Debugf("error finding peers for topic %s: %v", topic, err)
+		d.p.logger.Debug("error finding peers for topic", "topic", topic, "err", err)
 		return
 	}
 

floodsub.go

@@ -92,7 +92,7 @@ func (fs *FloodSubRouter) Publish(msg *Message) {
 
 		err := q.Push(out, false)
 		if err != nil {
-			log.Infof("dropping message to peer %s: queue full", pid)
+			fs.p.logger.Info("dropping message to peer: queue full", "peer", pid)
 			fs.tracer.DropRPC(out, pid)
 			// Drop it. The peer is too slow.
 			continue

go.mod

@@ -5,14 +5,12 @@ go 1.23
 require (
 	github.com/benbjohnson/clock v1.3.5
 	github.com/gogo/protobuf v1.3.2
-	github.com/ipfs/go-log/v2 v2.5.1
 	github.com/libp2p/go-buffer-pool v0.1.0
 	github.com/libp2p/go-libp2p v0.39.1
 	github.com/libp2p/go-libp2p-testing v0.12.0
 	github.com/libp2p/go-msgio v0.3.0
 	github.com/multiformats/go-multiaddr v0.14.0
 	github.com/multiformats/go-varint v0.0.7
-	go.uber.org/zap v1.27.0
 )
 
 require (
@@ -36,6 +34,7 @@ require (
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/huin/goupnp v1.3.0 // indirect
 	github.com/ipfs/go-cid v0.5.0 // indirect
+	github.com/ipfs/go-log/v2 v2.5.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
@@ -105,6 +104,7 @@ require (
 	go.uber.org/fx v1.23.0 // indirect
 	go.uber.org/mock v0.5.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.32.0 // indirect
 	golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c // indirect
 	golang.org/x/mod v0.23.0 // indirect