refactor(client): improve error handling and logging in peer tasks

- Enhance error messages with more context and consistency
- Optimize logging format for errors in peer task conductor and related components
- Improve error recording in trace spans
- Simplify error handling in various peer task functions

Signed-off-by: “jinrong” <1460595002@qq.com>

Author: jinronga

client/daemon/peer/peertask_conductor.go

@@ -746,14 +746,14 @@ loop:
 			}
 			pt.Errorf("receive peer packet with error: %d", peerPacket.Code)
 			if pt.isExitPeerPacketCode(peerPacket) {
-				pt.Errorf(pt.failedReason)
+				pt.Errorf("peer task failed: %s", pt.failedReason)
 				pt.cancel(pt.failedCode, pt.failedReason)
 				if !firstPacketReceived {
-					firstPeerSpan.RecordError(fmt.Errorf(pt.failedReason))
+					firstPeerSpan.RecordError(fmt.Errorf("peer task failed: %s", pt.failedReason))
 				}
 				pt.span.AddEvent("receive exit peer packet",
 					trace.WithAttributes(config.AttributePeerPacketCode.Int(int(peerPacket.Code))))
-				pt.span.RecordError(fmt.Errorf(pt.failedReason))
+				pt.span.RecordError(fmt.Errorf("peer task failed: %s", pt.failedReason))
 				break
 			} else {
 				pt.span.AddEvent("receive not success peer packet",
@@ -896,7 +896,7 @@ func (pt *peerTaskConductor) isExitPeerPacketCode(pp *schedulerv1.PeerPacket) bo
 		st := status.Newf(codes.Aborted, "source response is not valid")
 		st, err := st.WithDetails(pp.GetSourceError())
 		if err != nil {
-			pt.Errorf("convert source error details error: %s", err.Error())
+			pt.Errorf("convert source error details error: %s", err)
 			return false
 		}
 
@@ -1035,7 +1035,7 @@ func (pt *peerTaskConductor) waitFirstPeerPacket(done chan bool) {
 			pt.cancel(commonv1.Code_ClientScheduleTimeout, reasonBackSourceDisabled)
 			err := fmt.Errorf("%s, auto back source disabled", pt.failedReason)
 			pt.span.RecordError(err)
-			pt.Errorf(err.Error())
+			pt.Errorf("%s, auto back source disabled", pt.failedReason)
 			return
 		}
 		pt.Warnf("start download from source due to %s", reasonScheduleTimeout)

client/daemon/peer/peertask_file.go

@@ -128,7 +128,7 @@ func (f *fileTask) syncProgress() {
 			f.storeToOutput()
 			return
 		case <-f.peerTaskConductor.failCh:
-			f.span.RecordError(fmt.Errorf(f.peerTaskConductor.failedReason))
+			f.span.RecordError(fmt.Errorf("peer task failed: %s", f.peerTaskConductor.failedReason))
 			f.sendFailProgress(f.peerTaskConductor.failedCode, f.peerTaskConductor.failedReason)
 			return
 		case <-f.ctx.Done():

client/daemon/peer/peertask_reuse.go

@@ -161,15 +161,15 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 		// KeepOriginalOffset case
 		if length > 0 && stat.Size() == 0 {
 			err = fmt.Errorf("reuse failed, output file size is zero, but target length %d is not zero", length)
-			log.Errorf(err.Error())
+			log.Errorf("reuse failed, output file size is zero, but target length %d is not zero", length)
 			span.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
 			span.RecordError(err)
 			return nil, false
 		}
 	} else if length != stat.Size() {
 		// normal case
 		err = fmt.Errorf("reuse failed, output file size %d is not same with target length %d", stat.Size(), length)
-		log.Errorf(err.Error())
+		log.Errorf("reuse failed, output file size %d is not same with target length %d", stat.Size(), length)
 		span.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
 		span.RecordError(err)
 		return nil, false

client/daemon/peer/peertask_stream.go

@@ -164,7 +164,7 @@ func (s *streamTask) Start(ctx context.Context) (io.ReadCloser, map[string]strin
 		return nil, attr, ctx.Err()
 	case <-s.peerTaskConductor.failCh:
 		err := s.peerTaskConductor.getFailedError()
-		s.Errorf("wait first piece failed due to %s", err.Error())
+		s.Errorf("wait first piece failed due to %s", err)
 		return nil, attr, err
 	case <-s.peerTaskConductor.successCh:
 		if s.peerTaskConductor.GetContentLength() != -1 {
@@ -174,7 +174,7 @@ func (s *streamTask) Start(ctx context.Context) (io.ReadCloser, map[string]strin
 		}
 		exa, err := s.peerTaskConductor.storage.GetExtendAttribute(ctx, nil)
 		if err != nil {
-			s.Errorf("read extend attribute error due to %s ", err.Error())
+			s.Errorf("read extend attribute error due to %s ", err)
 			return nil, attr, err
 		}
 		if exa != nil {
@@ -255,13 +255,14 @@ func (s *streamTask) writeToPipe(desired int32, piece *PieceInfo, pw *io.PipeWri
 			return
 		case <-s.ctx.Done():
 			err = fmt.Errorf("context done due to: %s", s.ctx.Err())
-			s.Errorf(err.Error())
+			s.Errorf("context done due to: %s", s.ctx.Err())
 			s.closeWithError(pw, err)
 			return
 		case <-s.peerTaskConductor.failCh:
 			err = fmt.Errorf("stream close with peer task fail: %d/%s",
 				s.peerTaskConductor.failedCode, s.peerTaskConductor.failedReason)
-			s.Errorf(err.Error())
+			s.Errorf("stream close with peer task fail: %d/%s",
+				s.peerTaskConductor.failedCode, s.peerTaskConductor.failedReason)
 			s.closeWithError(pw, err)
 			return
 		}
@@ -352,7 +353,7 @@ func (s *resumeStreamTask) Start(ctx context.Context) (io.ReadCloser, map[string
 			return nil, attr, ctx.Err()
 		case <-s.peerTaskConductor.failCh:
 			err := s.peerTaskConductor.getFailedError()
-			s.Errorf("wait next piece failed due to %s", err.Error())
+			s.Errorf("wait next piece failed due to %s", err)
 			return nil, attr, err
 		case <-s.peerTaskConductor.successCh:
 			goto pieceReady
@@ -366,7 +367,7 @@ func (s *resumeStreamTask) Start(ctx context.Context) (io.ReadCloser, map[string
 pieceReady:
 	exa, err := s.peerTaskConductor.storage.GetExtendAttribute(ctx, nil)
 	if err != nil {
-		s.Errorf("read extend attribute error due to %s ", err.Error())
+		s.Errorf("read extend attribute error due to %s ", err)
 		return nil, attr, err
 	}
 	if exa != nil {
@@ -400,13 +401,14 @@ pieceReady:
 			n, err := s.writePartialPiece(pw, nextPiece, skipBytesInNextPiece)
 			if err != nil {
 				err = fmt.Errorf("write partial piece %d to pipe failed: %s", nextPiece, err.Error())
-				s.Errorf(err.Error())
+				s.Errorf("write partial piece %d to pipe failed: %s", nextPiece, err)
 				s.closeWithError(pw, err)
 				return
 			} else if n < int64(pieceSize)-skipBytesInNextPiece {
 				err = fmt.Errorf("write partial piece %d to pipe failed: short write, desire: %d, actual: %d",
 					nextPiece, int64(pieceSize)-skipBytesInNextPiece, n)
-				s.Errorf(err.Error())
+				s.Errorf("write partial piece %d to pipe failed: short write, desire: %d, actual: %d",
+					nextPiece, int64(pieceSize)-skipBytesInNextPiece, n)
 				s.closeWithError(pw, err)
 				return
 			}