[WIP]Warmup data for ossfs2.0

pkg/mounter/proxy/server/alinas/driver.go

@@ -52,6 +52,9 @@ func (h *Driver) Mount(ctx context.Context, req *proxy.MountRequest) error {
 	return h.mounter.Mount(req.Source, req.Target, req.Fstype, options)
 }
 
+func (h *Driver) Warmup(targetPath, warmupDir string, workercount int, totalBytes int64) {
+}
+
 func (h *Driver) Init() {
 	go runCommandForever("aliyun-alinas-mount-watchdog")
 	go runCommandForever("aliyun-cpfs-mount-watchdog")

pkg/mounter/proxy/server/driver.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/proxy"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 )
 
 type Driver interface {
@@ -13,6 +14,7 @@ type Driver interface {
 	Init()
 	Terminate()
 	Mount(ctx context.Context, req *proxy.MountRequest) error
+	Warmup(string, string, int, int64)
 }
 
 var (
@@ -29,5 +31,10 @@ func handleMountRequest(ctx context.Context, req *proxy.MountRequest) error {
 	if h == nil {
 		return fmt.Errorf("fstype %q not supported", req.Fstype)
 	}
-	return h.Mount(ctx, req)
+	err := h.Mount(ctx, req)
+	if err != nil {
+		return err
+	}
+	h.Warmup(req.Target, "warmupPath", 10, 10 * utils.GiB)
+	return nil
 }

pkg/mounter/proxy/server/ossfs/driver.go

@@ -160,3 +160,6 @@ func (h *Driver) Terminate() {
 	h.wg.Wait()
 	klog.InfoS("All ossfs processes exited")
 }
+
+func (h *Driver) Warmup(targetPath, warmupDir string, workercount int, totalBytes int64) {
+}

pkg/mounter/proxy/server/ossfs2/driver.go

@@ -164,3 +164,56 @@ func (h *Driver) Terminate() {
 	h.wg.Wait()
 	klog.InfoS("All ossfs2 processes exited")
 }
+
+func (h *Driver) Warmup(targetPath, warmupDir string, workerCount int, totalBytes int64) {
+	klog.Infof("Starting FUSE mountpoint warmup for: %v, workercount: %v", targetPath, workerCount)
+	startTime := time.Now()
+
+	// Find the immediate entries (files and subdirectories) - our chunks
+	entries, err := os.ReadDir(targetPath)
+	if err != nil {
+		klog.Errorf("Error reading mountpoint directory %s: %v", targetPath, err)
+		return
+	}
+
+	if len(entries) == 0 {
+		klog.Errorf("No entries found in %s to process.", targetPath)
+		return
+	}
+
+	// Use a channel to send entry paths to workers
+	entryChan := make(chan string, len(entries))
+	// Use a channel to receive bytes read from workers
+	bytesReadChan := make(chan int64, workerCount)
+	var totalBytesRead int64
+	var totalBytesMu sync.Mutex
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		for bytes := range bytesReadChan {
+			totalBytesMu.Lock()
+			totalBytesRead += bytes
+			totalBytesMu.Unlock()
+		}
+		wg.Done()
+	}()
+
+	for i := 0; i < workerCount; i++ {
+		wg.Add(1)
+		go worker(i, entryChan, bytesReadChan, &wg)
+	}
+
+	for _, entry := range entries {
+		entryPath := filepath.Join(targetPath, entry.Name())
+		entryChan <- entryPath
+	}
+	close(entryChan)
+
+	wg.Wait()
+	// all works done
+	close(bytesReadChan) 
+
+	duration := time.Since(startTime)
+	klog.Infof("Finished FUSE mountpoint warmup for: %s, total bytes read: %d MiB, duration: %v", targetPath, totalBytesRead/(1024*1024), duration)
+}

pkg/mounter/proxy/server/ossfs2/warmup.go

@@ -0,0 +1,103 @@
+package ossfs2
+
+import (
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"k8s.io/klog/v2"
+	"sync"
+)
+
+// Read files in 1GB chunks to trigger caching without excessive memory usage
+const readChunkSize = 1024 * 1024 * 1024 
+var totalBytesRead int64
+
+
+func worker(id int, entryChan <-chan string, bytesReadChan chan<- int64, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	// Function to read a single file (used if the entry is a file)
+	readSingleFile := func(filePath string, workerBytesRead *int64) error {
+		klog.Infof("Worker %d: Reading single file: %s", id, filePath)
+		file, err := os.Open(filePath)
+		if err != nil {
+			klog.Errorf("Worker %d: Error opening file %s: %v", id, filePath, err)
+			return err
+		}
+		defer file.Close()
+
+		buffer := make([]byte, readChunkSize)
+		for {
+			n, readErr := file.Read(buffer)
+			if n > 0 {
+				*workerBytesRead += int64(n)
+			}
+			if readErr == io.EOF {
+				break
+			}
+			if readErr != nil {
+				klog.Errorf("Worker %d: Error reading file %s: %v", id, filePath, readErr)
+				return readErr
+			}
+		}
+		return nil
+	}
+
+
+	walkAndRead := func(startPath string, workerBytesRead *int64) error {
+		return filepath.WalkDir(startPath, func(path string, d fs.DirEntry, err error) error {
+			if err != nil {
+				klog.Errorf("Worker %d: Error accessing path %s: %v", id, path, err)
+				return nil
+			}
+
+			// Skip the starting directory itself, as we're processing its contents
+			if path == startPath && d.IsDir() {
+				return nil
+			}
+
+			if !d.IsDir() {
+				return readSingleFile(path, workerBytesRead)
+			} else {
+                klog.Infof("Worker %d: Visiting directory: %s", id, path)
+            }
+			return nil
+		})
+	}
+
+	for entryPath := range entryChan {
+		klog.Infof("Worker %d received task for: %s", id, entryPath)
+		var workerBytesRead int64
+
+		info, err := os.Stat(entryPath)
+		if err != nil {
+			klog.Errorf("Worker %d: Error stating entry %s: %v", id, entryPath, err)
+			// Report 0 bytes for this failed entry and continue
+			bytesReadChan <- 0
+			continue
+		}
+
+		if info.IsDir() {
+			// Process directory recursively
+			err = walkAndRead(entryPath, &workerBytesRead)
+			if err != nil {
+				klog.Errorf("Worker %d: Error during recursive walk for %s: %v", id, entryPath, err)
+			}
+			klog.Infof("Worker %d finished directory task: %s (Read %d MiB)", id, entryPath, workerBytesRead/(1024*1024))
+
+		} else {
+			// Process single file
+			err = readSingleFile(entryPath, &workerBytesRead)
+			if err != nil {
+                klog.Errorf("Worker %d: Error reading single file %s: %v", id, entryPath, err)
+            }
+            klog.Infof("Worker %d finished file task: %s (Read %d MiB)", id, entryPath, workerBytesRead/(1024*1024))
+		}
+
+		// Report bytes read for this task (either recursive walk or single file read)
+		bytesReadChan <- workerBytesRead
+	}
+
+	klog.Infof("Worker %d shutting down.", id)
+}