Feat: Allow prefetching of data to leverage GetContentStream in blobfs

.gitignore

@@ -1,10 +1,13 @@
 *.tgz
 .DS_Store
 bin/blobcache
+bin/throughput
+bin/fs
 bin/testclient
 build.sh
 tmp/
 config.yaml
-e2e/testclient/testdata/*.bin
+e2e/throughput/testdata/*.bin
+e2e/fs/testdata/*.bin
 daemonset.yaml
 output.bin

Makefile

@@ -24,5 +24,6 @@ publish-chart:
 	helm push beam-blobcache-v2-chart-$(chartVersion).tgz oci://public.ecr.aws/n4e0e1y0
 	rm beam-blobcache-v2-chart-$(chartVersion).tgz
 
-testclient:
-	GOOS=linux GOARCH=amd64 go build -o bin/testclient e2e/testclient/main.go
+testclients:
+	GOOS=linux GOARCH=amd64 go build -o bin/throughput e2e/throughput/main.go
+	GOOS=linux GOARCH=amd64 go build -o bin/fs e2e/fs/main.go

cmd/main.go

@@ -10,7 +10,7 @@ import (
 func main() {
 	configManager, err := blobcache.NewConfigManager[blobcache.BlobCacheConfig]()
 	if err != nil {
-		log.Fatalf("Failed to load config: %v\n", err)
+		log.Fatalf("Failed to load config: %v", err)
 	}
 
 	ctx := context.Background()

e2e/fs/main.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	blobcache "github.com/beam-cloud/blobcache-v2/pkg"
+)
+
+var (
+	totalIterations int
+	checkContent    bool
+)
+
+type TestResult struct {
+	ElapsedTime        float64
+	ContentCheckPassed bool
+}
+
+func main() {
+	flag.IntVar(&totalIterations, "iterations", 3, "Number of iterations to run the tests")
+	flag.BoolVar(&checkContent, "checkcontent", true, "Check the content hash after receiving data")
+	flag.Parse()
+
+	configManager, err := blobcache.NewConfigManager[blobcache.BlobCacheConfig]()
+	if err != nil {
+		log.Fatalf("Failed to load config: %v", err)
+	}
+
+	cfg := configManager.GetConfig()
+
+	// Initialize logger
+	blobcache.InitLogger(cfg.DebugMode)
+
+	ctx := context.Background()
+
+	_, err = blobcache.NewBlobCacheClient(ctx, cfg)
+	if err != nil {
+		log.Fatalf("Unable to create client: %v", err)
+	}
+
+	// Block until Ctrl+C (SIGINT) or SIGTERM is received
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+	<-sigChan
+
+	log.Println("Received interrupt or termination signal, exiting.")
+}

e2e/testclient/main.go → e2e/throughput/main.go

@@ -30,7 +30,7 @@ func main() {
 
 	configManager, err := blobcache.NewConfigManager[blobcache.BlobCacheConfig]()
 	if err != nil {
-		log.Fatalf("Failed to load config: %v\n", err)
+		log.Fatalf("Failed to load config: %v", err)
 	}
 
 	cfg := configManager.GetConfig()
@@ -42,10 +42,10 @@ func main() {
 
 	client, err := blobcache.NewBlobCacheClient(ctx, cfg)
 	if err != nil {
-		log.Fatalf("Unable to create client: %v\n", err)
+		log.Fatalf("Unable to create client: %v", err)
 	}
 
-	filePath := "e2e/testclient/testdata/test3.bin"
+	filePath := "e2e/throughput/testdata/test3.bin"
 	b, err := os.ReadFile(filePath)
 	if err != nil {
 		log.Fatalf("Unable to read input file: %v\n", err)
@@ -101,7 +101,7 @@ func storeFile(client *blobcache.BlobCacheClient, filePath string) (string, erro
 			n, err := file.Read(buf)
 
 			if err != nil && err != io.EOF {
-				log.Fatalf("err reading file: %v\n", err)
+				log.Fatalf("err reading file: %v", err)
 			}
 
 			if n == 0 {
@@ -155,7 +155,7 @@ func TestGetContentStream(client *blobcache.BlobCacheClient, hash string, fileSi
 
 	// Verify received content's hash
 	if checkContent {
-		log.Printf("Verifying hash for GetContentStream\n")
+		log.Printf("Verifying hash for GetContentStream")
 
 		hashBytes := sha256.Sum256(contentStream)
 		retrievedHash := hex.EncodeToString(hashBytes[:])

pkg/blobfs.go

@@ -6,7 +6,6 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
-	"log"
 	"os"
 	"os/exec"
 	"strings"
@@ -67,17 +66,18 @@ type BlobFsSystemOpts struct {
 }
 
 type BlobFs struct {
-	ctx      context.Context
-	root     *FSNode
-	verbose  bool
-	Metadata *BlobCacheMetadata
-	Client   *BlobCacheClient
-	Config   BlobCacheConfig
+	ctx             context.Context
+	root            *FSNode
+	verbose         bool
+	Metadata        *BlobCacheMetadata
+	Client          *BlobCacheClient
+	Config          BlobCacheConfig
+	PrefetchManager *PrefetchManager
 }
 
 func Mount(ctx context.Context, opts BlobFsSystemOpts) (func() error, <-chan error, *fuse.Server, error) {
 	mountPoint := opts.Config.BlobFs.MountPoint
-	Logger.Infof("Mounting to %s\n", mountPoint)
+	Logger.Infof("Mounting to %s", mountPoint)
 
 	if _, err := os.Stat(mountPoint); os.IsNotExist(err) {
 		err = os.MkdirAll(mountPoint, 0755)
@@ -170,19 +170,24 @@ func NewFileSystem(ctx context.Context, opts BlobFsSystemOpts) (*BlobFs, error)
 		Metadata: metadata,
 	}
 
+	if opts.Config.BlobFs.Prefetch.Enabled {
+		bfs.PrefetchManager = NewPrefetchManager(ctx, opts.Config, opts.Client)
+		bfs.PrefetchManager.Start()
+	}
+
 	rootID := GenerateFsID("/")
 	rootPID := "" // Root node has no parent
 	rootPath := "/"
 
 	dirMeta, err := metadata.GetFsNode(bfs.ctx, rootID)
 	if err != nil || dirMeta == nil {
-		log.Printf("Root node metadata not found, creating it now...\n")
+		Logger.Infof("Root node metadata not found, creating it now...")
 
 		dirMeta = &BlobFsMetadata{PID: rootPID, ID: rootID, Path: rootPath, Ino: 1, Mode: fuse.S_IFDIR | 0755}
 
 		err := metadata.SetFsNode(bfs.ctx, rootID, dirMeta)
 		if err != nil {
-			log.Fatalf("Unable to create blobfs root node dir metdata: %+v\n", err)
+			Logger.Fatalf("Unable to create blobfs root node dir metdata: %+v", err)
 		}
 	}
 

pkg/blobfs_node.go

@@ -12,13 +12,14 @@ import (
 )
 
 type BlobFsNode struct {
-	Path   string
-	ID     string
-	PID    string
-	Name   string
-	Target string
-	Hash   string
-	Attr   fuse.Attr
+	Path     string
+	ID       string
+	PID      string
+	Name     string
+	Target   string
+	Hash     string
+	Attr     fuse.Attr
+	Prefetch *bool
 }
 type FSNode struct {
 	fs.Inode
@@ -94,13 +95,14 @@ func (n *FSNode) inodeFromFsId(ctx context.Context, fsId string) (*fs.Inode, *fu
 	// Create a new Inode on lookup
 	node := n.NewInode(ctx,
 		&FSNode{filesystem: n.filesystem, bfsNode: &BlobFsNode{
-			Path:   metadata.Path,
-			ID:     metadata.ID,
-			PID:    metadata.PID,
-			Name:   metadata.Name,
-			Hash:   metadata.Hash,
-			Attr:   attr,
-			Target: "",
+			Path:     metadata.Path,
+			ID:       metadata.ID,
+			PID:      metadata.PID,
+			Name:     metadata.Name,
+			Hash:     metadata.Hash,
+			Attr:     attr,
+			Target:   "",
+			Prefetch: nil,
 		}, attr: attr},
 		fs.StableAttr{Mode: metadata.Mode, Ino: metadata.Ino, Gen: metadata.Gen},
 	)
@@ -128,9 +130,6 @@ func (n *FSNode) Lookup(ctx context.Context, name string, out *fuse.EntryOut) (*
 			return nil, syscall.ENOENT
 		}
 
-		// TODO: stream file to a temp file in the container somewhere
-		// /tmp/cache/path/to/file
-
 		out.Attr = *attr
 		return node, fs.OK
 	}
@@ -162,6 +161,30 @@ func (n *FSNode) Open(ctx context.Context, flags uint32) (fh fs.FileHandle, fuse
 	return nil, 0, fs.OK
 }
 
+func (n *FSNode) shouldPrefetch(node *BlobFsNode) bool {
+	if node.Prefetch != nil {
+		return *node.Prefetch
+	}
+
+	if !n.filesystem.Config.BlobFs.Prefetch.Enabled {
+		return false
+	}
+
+	if n.bfsNode.Attr.Size < n.filesystem.Config.BlobFs.Prefetch.MinFileSizeBytes {
+		return false
+	}
+
+	for _, ext := range n.filesystem.Config.BlobFs.Prefetch.IgnoreFileExt {
+		if strings.HasSuffix(node.Name, ext) {
+			return false
+		}
+	}
+
+	prefetch := true
+	node.Prefetch = &prefetch
+	return true
+}
+
 func (n *FSNode) Read(ctx context.Context, f fs.FileHandle, dest []byte, off int64) (fuse.ReadResult, syscall.Errno) {
 	n.log("Read called with offset: %v", off)
 
@@ -170,6 +193,17 @@ func (n *FSNode) Read(ctx context.Context, f fs.FileHandle, dest []byte, off int
 		return fuse.ReadResultData(dest[:0]), fs.OK
 	}
 
+	// Attempt to prefetch the file
+	if n.shouldPrefetch(n.bfsNode) {
+		buffer := n.filesystem.PrefetchManager.GetPrefetchBuffer(n.bfsNode.Hash, n.bfsNode.Attr.Size)
+		if buffer != nil {
+			data, err := buffer.GetRange(uint64(off), uint64(len(dest)))
+			if err == nil {
+				return fuse.ReadResultData(data), fs.OK
+			}
+		}
+	}
+
 	buffer, err := n.filesystem.Client.GetContent(n.bfsNode.Hash, off, int64(len(dest)))
 	if err != nil {
 		return nil, syscall.EIO