pushsync: use logger with trace (#1164)

pkg/pusher/pusher.go

@@ -7,6 +7,7 @@ package pusher
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -18,6 +19,7 @@ import (
 	"github.com/ethersphere/bee/pkg/topology"
 	"github.com/ethersphere/bee/pkg/tracing"
 	"github.com/opentracing/opentracing-go"
+	"github.com/sirupsen/logrus"
 )
 
 type Service struct {
@@ -65,6 +67,7 @@ func (s *Service) chunksWorker() {
 		inflight      = make(map[string]struct{})
 		mtx           sync.Mutex
 		span          opentracing.Span
+		logger        *logrus.Entry
 	)
 	defer timer.Stop()
 	defer close(s.chunksWorkerQuitC)
@@ -90,7 +93,7 @@ LOOP:
 			}
 
 			if span == nil {
-				span, _, ctx = s.tracer.StartSpanFromContext(cctx, "pusher-sync-batch", s.logger)
+				span, logger, ctx = s.tracer.StartSpanFromContext(cctx, "pusher-sync-batch", s.logger)
 			}
 
 			// postpone a retry only after we've finished processing everything in index
@@ -130,7 +133,7 @@ LOOP:
 						s.metrics.TotalSynced.Inc()
 						s.metrics.SyncTime.Observe(time.Since(startTime).Seconds())
 						// only print this if there was no error while sending the chunk
-						s.logger.Tracef("pusher pushed chunk %s", ch.Address().String())
+						logger.Tracef("pusher pushed chunk %s", ch.Address().String())
 					} else {
 						s.metrics.TotalErrors.Inc()
 						s.metrics.ErrorTime.Observe(time.Since(startTime).Seconds())
@@ -145,13 +148,13 @@ LOOP:
 				_, err = s.pushSyncer.PushChunkToClosest(ctx, ch)
 				if err != nil {
 					if !errors.Is(err, topology.ErrNotFound) {
-						s.logger.Debugf("pusher: error while sending chunk or receiving receipt: %v", err)
+						logger.Debugf("pusher: error while sending chunk or receiving receipt: %v", err)
 					}
 					return
 				}
 				err = s.setChunkAsSynced(ctx, ch)
 				if err != nil {
-					s.logger.Debugf("pusher: error setting chunk as synced: %v", err)
+					logger.Debugf("pusher: error setting chunk as synced: %v", err)
 					return
 				}
 			}(ctx, ch)
@@ -208,7 +211,7 @@ LOOP:
 
 func (s *Service) setChunkAsSynced(ctx context.Context, ch swarm.Chunk) error {
 	if err := s.storer.Set(ctx, storage.ModeSetSync, ch.Address()); err != nil {
-		s.logger.Errorf("pusher: error setting chunk as synced: %v", err)
+		return fmt.Errorf("set synced: %w", err)
 	}
 
 	t, err := s.tagg.Get(ch.TagID())

pkg/pushsync/pushsync.go

@@ -217,8 +217,8 @@ func (ps *PushSync) receiveReceipt(ctx context.Context, r protobuf.Reader) (rece
 // PushChunkToClosest sends chunk to the closest peer by opening a stream. It then waits for
 // a receipt from that peer and returns error or nil based on the receiving and
 // the validity of the receipt.
-func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Receipt, error) {
-	span, _, ctx := ps.tracer.StartSpanFromContext(ctx, "pushsync-push", ps.logger, opentracing.Tag{Key: "address", Value: ch.Address().String()})
+func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (r *Receipt, reterr error) {
+	span, logger, ctx := ps.tracer.StartSpanFromContext(ctx, "pushsync-push", ps.logger, opentracing.Tag{Key: "address", Value: ch.Address().String()})
 	defer span.Finish()
 
 	var (
@@ -289,7 +289,7 @@ func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Re
 		if err != nil {
 			ps.metrics.TotalErrors.Inc()
 			lastErr = fmt.Errorf("new stream for peer %s: %w", peer.String(), err)
-			ps.logger.Debugf("pushsync-push: %v", lastErr)
+			logger.Debugf("pushsync-push: %v", lastErr)
 			continue
 		}
 		deferFuncs = append(deferFuncs, func() { go streamer.FullClose() })
@@ -299,7 +299,7 @@ func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Re
 			ps.metrics.TotalErrors.Inc()
 			_ = streamer.Reset()
 			lastErr = fmt.Errorf("chunk %s deliver to peer %s: %w", ch.Address().String(), peer.String(), err)
-			ps.logger.Debugf("pushsync-push: %v", lastErr)
+			logger.Debugf("pushsync-push: %v", lastErr)
 			continue
 		}
 
@@ -319,7 +319,7 @@ func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Re
 			ps.metrics.TotalErrors.Inc()
 			_ = streamer.Reset()
 			lastErr = fmt.Errorf("chunk %s receive receipt from peer %s: %w", ch.Address().String(), peer.String(), err)
-			ps.logger.Debugf("pushsync-push: %v", lastErr)
+			logger.Debugf("pushsync-push: %v", lastErr)
 			continue
 		}
 
@@ -341,7 +341,7 @@ func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Re
 		return rec, nil
 	}
 
-	ps.logger.Tracef("pushsync-push: failed to push chunk %s: reached max peers of %v", ch.Address(), maxPeers)
+	logger.Tracef("pushsync-push: failed to push chunk %s: reached max peers of %v", ch.Address(), maxPeers)
 
 	if lastErr != nil {
 		return nil, lastErr
@@ -364,10 +364,5 @@ func (ps *PushSync) handleDeliveryResponse(ctx context.Context, w protobuf.Write
 		return fmt.Errorf("send receipt to peer %s: %w", p.Address.String(), err)
 	}
 
-	err = ps.accounting.Debit(p.Address, ps.pricer.Price(chunk.Address()))
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return ps.accounting.Debit(p.Address, ps.pricer.Price(chunk.Address()))
 }