test(dragonfly-client-storage): add e2e test for cache upload/download.

.github/workflows/compatibility-e2e-v2.yml

@@ -33,12 +33,12 @@ jobs:
             image: manager
             image-tag: v2.2.1
             chart-name: manager
-            skip: "Rate Limit"
+            skip: "Rate Limit | preheat files in cache"
           - module: scheduler
             image: scheduler
             image-tag: v2.2.1
             chart-name: scheduler
-            skip: "Rate Limit"
+            skip: "Rate Limit | preheat files in cache"
           - module: client
             image: client
             image-tag: v0.2.18
@@ -146,7 +146,7 @@ jobs:
 
       - name: Run E2E test
         run: |
-          ginkgo -v -r --race --fail-fast --cover --trace --show-node-events --skip=${{ matrix.skip }} test/e2e/v2
+          ginkgo -v -r --race --fail-fast --cover --trace --show-node-events --skip="${{ matrix.skip }}" test/e2e/v2
           cat coverprofile.out >> coverage.txt
 
       - name: Move cache

internal/job/types.go

@@ -34,6 +34,7 @@ type PreheatRequest struct {
 	CertificateChain    [][]byte          `json:"certificate_chain" validate:"omitempty"`
 	InsecureSkipVerify  bool              `json:"insecure_skip_verify" validate:"omitempty"`
 	Timeout             time.Duration     `json:"timeout" validate:"omitempty"`
+	LoadToCache         bool              `json:"load_to_cache" validate:"omitempty"`
 }
 
 // PreheatResponse defines the response parameters for preheating.

manager/job/preheat.go

@@ -133,6 +133,7 @@ func (p *preheat) CreatePreheat(ctx context.Context, schedulers []models.Schedul
 				CertificateChain:    p.certificateChain,
 				InsecureSkipVerify:  p.insecureSkipVerify,
 				Timeout:             json.Timeout,
+				LoadToCache:         json.LoadToCache,
 			},
 		}
 	default:

manager/types/job.go

@@ -143,6 +143,9 @@ type PreheatArgs struct {
 
 	// Timeout is the timeout for preheating, default is 30 minutes.
 	Timeout time.Duration `json:"timeout" binding:"omitempty"`
+
+	// LoadToCache is the flag for preheating content in cache storage, default is false.
+	LoadToCache bool `json:"load_to_cache" binding:"omitempty"`
 }
 
 type CreateSyncPeersJobRequest struct {

scheduler/job/job.go

@@ -308,6 +308,7 @@ func (j *job) preheatAllSeedPeers(ctx context.Context, taskID string, req *inter
 					RequestHeader:       req.Headers,
 					Timeout:             durationpb.New(req.Timeout),
 					CertificateChain:    req.CertificateChain,
+					LoadToCache:         req.LoadToCache,
 				}})
 			if err != nil {
 				log.Errorf("preheat failed: %s", err.Error())
@@ -450,6 +451,7 @@ func (j *job) preheatAllPeers(ctx context.Context, taskID string, req *internalj
 					RequestHeader:       req.Headers,
 					Timeout:             durationpb.New(req.Timeout),
 					CertificateChain:    req.CertificateChain,
+					LoadToCache:         req.LoadToCache,
 				}})
 			if err != nil {
 				log.Errorf("preheat failed: %s", err.Error())
@@ -593,6 +595,7 @@ func (j *job) preheatV2(ctx context.Context, taskID string, req *internaljob.Pre
 			FilteredQueryParams: filteredQueryParams,
 			RequestHeader:       req.Headers,
 			CertificateChain:    req.CertificateChain,
+			LoadToCache:         req.LoadToCache,
 		}})
 	if err != nil {
 		log.Errorf("preheat failed: %s", err.Error())

test/e2e/v2/manager/preheat.go

@@ -19,6 +19,8 @@ package manager
 import (
 	"encoding/json"
 	"fmt"
+	"regexp"
+	"strings"
 
 	. "github.com/onsi/ginkgo/v2" //nolint
 	. "github.com/onsi/gomega"    //nolint
@@ -205,6 +207,116 @@ var _ = Describe("Preheat with Manager", func() {
 		})
 	})
 
+	Context("10MiB file in cache", func() {
+		var (
+			testFile *util.File
+			err      error
+		)
+
+		BeforeEach(func() {
+			testFile, err = util.GetFileServer().GenerateFile(util.FileSize10MiB)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(testFile).NotTo(BeNil())
+		})
+
+		AfterEach(func() {
+			err = util.GetFileServer().DeleteFile(testFile.GetInfo())
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("preheat files in cache should be ok", Label("preheat", "file", "cache"), func() {
+			managerPod, err := util.ManagerExec(0)
+			fmt.Println(err)
+			Expect(err).NotTo(HaveOccurred())
+
+			req, err := structure.StructToMap(types.CreatePreheatJobRequest{
+				Type: internaljob.PreheatJob,
+				Args: types.PreheatArgs{
+					Type:        "file",
+					URL:         testFile.GetDownloadURL(),
+					Scope:       "single_seed_peer",
+					LoadToCache: true,
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			out, err := managerPod.CurlCommand("POST", map[string]string{"Content-Type": "application/json"}, req,
+				"http://dragonfly-manager.dragonfly-system.svc:8080/api/v1/jobs").CombinedOutput()
+			fmt.Println(err)
+			Expect(err).NotTo(HaveOccurred())
+			fmt.Println(string(out))
+
+			job := &models.Job{}
+			err = json.Unmarshal(out, job)
+			fmt.Println(err)
+			Expect(err).NotTo(HaveOccurred())
+
+			done := waitForDone(job, managerPod)
+			Expect(done).Should(BeTrue())
+
+			var preheatedSeedClient *util.PodExec
+
+			taskIDCmd := fmt.Sprintf("grep -a '%s' /var/log/dragonfly/dfdaemon/*", testFile.GetTaskID())
+			successCmd := fmt.Sprintf("%s | grep -a 'download task succeeded'", taskIDCmd)
+
+			for i := 0; i < 3; i++ {
+				seedClient, err := util.SeedClientExec(i)
+				fmt.Println(err)
+				Expect(err).NotTo(HaveOccurred())
+
+				out, err = seedClient.Command("bash", "-c", successCmd).CombinedOutput()
+				if err == nil && len(out) > 0 {
+					preheatedSeedClient = seedClient
+					fmt.Printf("Found preheated seed client: %d\n", i)
+					break
+				}
+			}
+
+			Expect(preheatedSeedClient).NotTo(BeNil())
+
+			sha256sum, err := util.CalculateSha256ByTaskID([]*util.PodExec{preheatedSeedClient}, testFile.GetTaskID())
+			Expect(err).NotTo(HaveOccurred())
+			Expect(testFile.GetSha256()).To(Equal(sha256sum))
+
+			clientPod, err := util.ClientExec()
+			fmt.Println(err)
+			Expect(err).NotTo(HaveOccurred())
+
+			out, err = clientPod.Command("sh", "-c", fmt.Sprintf("dfget %s --disable-back-to-source --output %s", testFile.GetDownloadURL(), testFile.GetOutputPath())).CombinedOutput()
+			fmt.Println(string(out), err)
+			Expect(err).NotTo(HaveOccurred())
+
+			sha256sum, err = util.CalculateSha256ByTaskID([]*util.PodExec{clientPod}, testFile.GetTaskID())
+			Expect(err).NotTo(HaveOccurred())
+			Expect(testFile.GetSha256()).To(Equal(sha256sum))
+
+			sha256sum, err = util.CalculateSha256ByOutput([]*util.PodExec{clientPod}, testFile.GetOutputPath())
+			Expect(err).NotTo(HaveOccurred())
+			Expect(testFile.GetSha256()).To(Equal(sha256sum))
+
+			out, err = preheatedSeedClient.Command("bash", "-c", taskIDCmd).CombinedOutput()
+			fmt.Println(err)
+			Expect(err).NotTo(HaveOccurred())
+			logs := string(out)
+
+			Expect(logs).To(ContainSubstring(fmt.Sprintf("put task to cache: %s", testFile.GetTaskID())))
+
+			pieceRegex := regexp.MustCompile(`pieces: \[([0-9, ]+)\]`)
+			pieceMatches := pieceRegex.FindStringSubmatch(logs)
+			Expect(pieceMatches).NotTo(BeNil())
+			pieceNumbers := strings.Split(strings.ReplaceAll(pieceMatches[1], " ", ""), ",")
+
+			for _, number := range pieceNumbers {
+				pieceID := fmt.Sprintf("%s-%s", testFile.GetTaskID(), number)
+				putPieceCacheLog := fmt.Sprintf("put piece to cache: %s", pieceID)
+				Expect(logs).To(ContainSubstring(putPieceCacheLog))
+
+				getPieceCacheLog := fmt.Sprintf("get piece from cache: %s", pieceID)
+				Expect(logs).To(ContainSubstring(getPieceCacheLog))
+			}
+		})
+	})
+
 	Context("ghcr.io/dragonflyoss/busybox:v1.35.0 image", func() {
 		It("preheat image should be ok", Label("preheat", "image"), func() {
 			managerPod, err := util.ManagerExec(0)