fix: wait for blobfuse mount

Author: cvvz

pkg/blob/nodeserver.go

@@ -46,6 +46,11 @@ import (
 	mount_azure_blob "sigs.k8s.io/blob-csi-driver/pkg/blobfuse-proxy/pb"
 )
 
+const (
+	waitForMountInterval = 20 * time.Millisecond
+	waitForMountTimeout  = 3 * time.Second
+)
+
 type MountClient struct {
 	service mount_azure_blob.MountServiceClient
 }
@@ -181,11 +186,7 @@ func (d *Driver) mountBlobfuseInsideDriver(args string, protocol string, authEnv
 	cmd.Env = append(os.Environ(), authEnv...)
 	output, err := cmd.CombinedOutput()
 	klog.V(2).Infof("mount output: %s\n", string(output))
-	if err == nil && protocol == Fuse2 {
-		// todo: remove this when https://github.com/Azure/azure-storage-fuse/issues/1079 is fixed
-		klog.V(2).Infof("sleep 2s, waiting for blobfuse2 mount complete")
-		time.Sleep(2 * time.Second)
-	}
+
 	return string(output), err
 }
 
@@ -406,6 +407,12 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, err
 	}
 
+	// wait a few seconds to make sure blobfuse mount is successful
+	// please refer to https://github.com/Azure/azure-storage-fuse/pull/1088 for more details
+	if err := waitForMount(targetPath, waitForMountInterval, waitForMountTimeout); err != nil {
+		return nil, fmt.Errorf("failed to wait for mount: %w", err)
+	}
+
 	klog.V(2).Infof("volume(%s) mount on %q succeeded", volumeID, targetPath)
 	return &csi.NodeStageVolumeResponse{}, nil
 }
@@ -575,3 +582,24 @@ func (d *Driver) ensureMountPoint(target string, perm os.FileMode) (bool, error)
 	}
 	return !notMnt, nil
 }
+
+func waitForMount(path string, intervel, timeout time.Duration) error {
+	timeAfter := time.After(timeout)
+	timeTick := time.Tick(intervel)
+
+	for {
+		select {
+		case <-timeTick:
+			notMount, err := mount.New("").IsLikelyNotMountPoint(path)
+			if err != nil {
+				return err
+			}
+			if !notMount {
+				klog.V(2).Infof("blobfuse mount at %s success", path)
+				return nil
+			}
+		case <-timeAfter:
+			return fmt.Errorf("timeout waiting for mount %s", path)
+		}
+	}
+}

pkg/blob/nodeserver_test.go

@@ -24,8 +24,10 @@ import (
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"strings"
 	"syscall"
 	"testing"
+	"time"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -754,3 +756,59 @@ func TestMountBlobfuseInsideDriver(t *testing.T) {
 	// the error should be of type exec.ExitError
 	assert.NotNil(t, err)
 }
+
+func Test_waitForMount(t *testing.T) {
+	if runtime.GOOS == "windows" || runtime.GOOS == "darwin" {
+		t.Skip("Skipping test on ", runtime.GOOS)
+	}
+
+	tmpDir, err := os.MkdirTemp("", "")
+	assert.NoError(t, err)
+	defer os.RemoveAll(tmpDir)
+
+	type args struct {
+		path     string
+		intervel time.Duration
+		timeout  time.Duration
+	}
+
+	tests := []struct {
+		name      string
+		args      args
+		wantErr   bool
+		subErrMsg string
+	}{
+		{
+			name: "test error timeout",
+			args: args{
+				path:     tmpDir,
+				intervel: 1 * time.Millisecond,
+				timeout:  10 * time.Millisecond,
+			},
+			wantErr:   true,
+			subErrMsg: "timeout",
+		},
+		{
+			name: "test error no such file or directory",
+			args: args{
+				path:     "/no/such/file/or/directory",
+				intervel: 1 * time.Millisecond,
+				timeout:  10 * time.Millisecond,
+			},
+			wantErr:   true,
+			subErrMsg: "no such file or directory",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := waitForMount(tt.args.path, tt.args.intervel, tt.args.timeout)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("waitForMount() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if err != nil && !strings.Contains(err.Error(), tt.subErrMsg) {
+				t.Errorf("waitForMount() error = %v, wantErr %v", err, tt.subErrMsg)
+			}
+		})
+	}
+}

pkg/blobfuse-proxy/server/server.go

@@ -23,7 +23,6 @@ import (
 	"os/exec"
 	"strings"
 	"sync"
-	"time"
 
 	"google.golang.org/grpc"
 	"k8s.io/klog/v2"
@@ -96,11 +95,6 @@ func (server *MountServer) MountAzureBlob(ctx context.Context,
 	if err != nil {
 		return &result, fmt.Errorf("%w %s", err, result.Output)
 	}
-	if protocol == blob.Fuse2 || server.blobfuseVersion == BlobfuseV2 {
-		// todo: remove this when https://github.com/Azure/azure-storage-fuse/issues/1079 is fixed
-		klog.V(2).Infof("sleep 2s, waiting for blobfuse2 mount complete\n")
-		time.Sleep(2 * time.Second)
-	}
 	return &result, nil
 }