feature: add piece error API

Signed-off-by: Starnop <starnop@163.com>

apis/swagger.yml

@@ -414,6 +414,13 @@ paths:
           required: true
           description: "ID of task"
           type: string
+        - name: full
+          in: query
+          type: "boolean"
+          default: false
+          description: |
+            supernode will also delete the cdn files when the value of full equals true.
+
       responses:
         204:
           description: "no error"
@@ -513,6 +520,43 @@ paths:
         500:
             $ref: "#/responses/500ErrorResponse"
 
+  /tasks/{id}/pieces/{pieceRange}/error:
+    post:
+      summary: "report a piece error"
+      description: |
+        When a peer failed to download a piece from supernode or
+        failed to validate the pieceMD5,
+        and then dfget should report the error info to supernode.
+      consumes:
+        - "application/json"
+      produces:
+        - "application/json"
+      parameters:
+        - name: id
+          in: path
+          required: true
+          description: "ID of task"
+          type: string
+        - name: pieceRange
+          in: path
+          required: true
+          description: |
+            the range of specific piece in the task, example "0-45565".
+          type: string
+        - name: "PieceErrorRequest"
+          in: "body"
+          description: |
+            request body which contains piece error information.
+          schema:
+            $ref: "#/definitions/PieceErrorRequest"
+      responses:
+        200:
+          description: "no error"
+        404:
+            $ref: "#/responses/404ErrorResponse"
+        500:
+            $ref: "#/responses/500ErrorResponse"
+
   /preheats:
     post:
       summary: "Create a Preheat Task"
@@ -1095,6 +1139,47 @@ definitions:
         description: |
           the range of specific piece in the task, example "0-45565".
 
+  PieceErrorRequest:
+    type: "object"
+    description: "Peer's detailed information in supernode."
+    properties:
+      taskId:
+        type: "string"
+        description: |
+            the taskID of the piece.
+      srcCid:
+        type: "string"
+        description: |
+          the CID of the src Peer.
+      dstPid:
+        type: "string"
+        description: |
+          the peer ID of the target Peer.
+      dstIP:
+        type: "string"
+        description: |
+          the peer ID of the target Peer.
+      range:
+        type: "string"
+        description: |
+          the range of specific piece in the task, example "0-45565".
+      realMd5:
+        type: "string"
+        description: |
+          the MD5 information of piece which calculated by the piece content
+          which downloaded from the target peer.
+      expectedMd5:
+        type: "string"
+        description: |
+          the MD5 value of piece which returned by the supernode that
+          in order to verify the correctness of the piece content which
+          downloaded from the other peers.
+      errorType:
+        type: "string"
+        description: |
+          the error type when failed to download from supernode that dfget will report to supernode
+        enum: ["FILE_NOT_EXIST", "FILE_MD5_NOT_MATCH"]
+
   PreheatInfo:
     type: "object"
     description: |
@@ -1283,7 +1368,6 @@ definitions:
       message:
         type: "string"
         description: "detailed error message"
-
 
 responses:
   401ErrorResponse:

pkg/fileutils/fileutils.go

@@ -20,6 +20,7 @@ import (
 	"bufio"
 	"crypto/md5"
 	"fmt"
+	"hash"
 	"io"
 	"io/ioutil"
 	"os"
@@ -208,7 +209,12 @@ func Md5Sum(name string) string {
 		return ""
 	}
 
-	return fmt.Sprintf("%x", h.Sum(nil))
+	return GetMd5Sum(h, nil)
+}
+
+// GetMd5Sum gets md5 sum as a string and appends the current hash to b.
+func GetMd5Sum(md5 hash.Hash, b []byte) string {
+	return fmt.Sprintf("%x", md5.Sum(b))
 }
 
 // GetSys returns the underlying data source of the os.FileInfo.

pkg/limitreader/limit_reader.go

@@ -18,10 +18,10 @@ package limitreader
 
 import (
 	"crypto/md5"
-	"fmt"
 	"hash"
 	"io"
 
+	"github.com/dragonflyoss/Dragonfly/pkg/fileutils"
 	"github.com/dragonflyoss/Dragonfly/pkg/ratelimiter"
 )
 
@@ -93,7 +93,7 @@ func (lr *LimitReader) Read(p []byte) (n int, err error) {
 // Md5 calculates the md5 of all contents read
 func (lr *LimitReader) Md5() string {
 	if lr.md5sum != nil {
-		return fmt.Sprintf("%x", lr.md5sum.Sum(nil))
+		return fileutils.GetMd5Sum(lr.md5sum, nil)
 	}
 	return ""
 }

supernode/daemon/mgr/cdn/cache_detector.go

@@ -30,14 +30,14 @@ import (
 type cacheDetector struct {
 	cacheStore      *store.Store
 	metaDataManager *fileMetaDataManager
-	OriginClient    httpclient.OriginHTTPClient
+	originClient    httpclient.OriginHTTPClient
 }
 
 func newCacheDetector(cacheStore *store.Store, metaDataManager *fileMetaDataManager, originClient httpclient.OriginHTTPClient) *cacheDetector {
 	return &cacheDetector{
 		cacheStore:      cacheStore,
 		metaDataManager: metaDataManager,
-		OriginClient:    originClient,
+		originClient:    originClient,
 	}
 }
 
@@ -68,7 +68,7 @@ func (cd *cacheDetector) detectCache(ctx context.Context, task *types.TaskInfo)
 }
 
 func (cd *cacheDetector) parseBreakNum(ctx context.Context, task *types.TaskInfo, metaData *fileMetaData) int {
-	expired, err := cd.OriginClient.IsExpired(task.RawURL, task.Headers, metaData.LastModified, metaData.ETag)
+	expired, err := cd.originClient.IsExpired(task.RawURL, task.Headers, metaData.LastModified, metaData.ETag)
 	if err != nil {
 		logrus.Errorf("failed to check whether the task(%s) has expired: %v", task.ID, err)
 	}
@@ -85,7 +85,7 @@ func (cd *cacheDetector) parseBreakNum(ctx context.Context, task *types.TaskInfo
 		return 0
 	}
 
-	supportRange, err := cd.OriginClient.IsSupportRange(task.TaskURL, task.Headers)
+	supportRange, err := cd.originClient.IsSupportRange(task.TaskURL, task.Headers)
 	if err != nil {
 		logrus.Errorf("failed to check whether the task(%s) supports partial requests: %v", task.ID, err)
 	}