Merge pull request #788 from cvvz/support-blobfuse2

feat: support fuse2 protocol in storageclass

Author: k8s-ci-robot

docs/driver-parameters.md

@@ -12,7 +12,7 @@ skuName | Azure storage account type (alias: `storageAccountType`) | `Standard_L
 location | Azure location | `eastus`, `westus`, etc. | No | if empty, driver will use the same location name as current k8s cluster
 resourceGroup | Azure resource group name | existing resource group name | No | if empty, driver will use the same resource group name as current k8s cluster
 storageAccount | specify Azure storage account name| STORAGE_ACCOUNT_NAME | - No for blobfuse mount </br> - Yes for NFSv3 mount |  - For blobfuse mount: if empty, driver will find a suitable storage account that matches `skuName` in the same resource group; if a storage account name is provided, storage account must exist. </br>  - For NFSv3 mount, storage account name must be provided
-protocol | specify blobfuse mount or NFSv3 mount | `fuse`, `nfs` | No | `fuse`
+protocol | specify blobfuse, blobfuse2 or NFSv3 mount (blobfuse2 is still in Preview) | `fuse`, `fuse2`, `nfs` | No | `fuse`
 containerName | specify the existing container(directory) name | existing container name | No | if empty, driver will create a new container name, starting with `pvc-fuse` for blobfuse or `pvc-nfs` for NFSv3
 containerNamePrefix | specify Azure storage directory prefix created by driver | can only contain lowercase letters, numbers, hyphens, and length should be less than 21 | No |
 server | specify Azure storage account server address | existing server address, e.g. `accountname.privatelink.blob.core.windows.net` | No | if empty, driver will use default `accountname.blob.core.windows.net` or other sovereign cloud account address
@@ -70,7 +70,7 @@ Name | Meaning | Available Value | Mandatory | Default value
 volumeAttributes.resourceGroup | Azure resource group name | existing resource group name | No | if empty, driver will use the same resource group name as current k8s cluster
 volumeAttributes.storageAccount | existing storage account name | existing storage account name | Yes |
 volumeAttributes.containerName | existing container name | existing container name | Yes |
-volumeAttributes.protocol | specify blobfuse mount or NFSv3 mount | `fuse`, `nfs` | No | `fuse`
+volumeAttributes.protocol | specify blobfuse, blobfuse2 or NFSv3 mount (blobfuse2 is still in Preview) | `fuse`, `fuse2`, `nfs` | No | `fuse`
 --- | **Following parameters are only for blobfuse** | --- | --- |
 volumeAttributes.secretName | secret name that stores storage account name and key(only applies for SMB) | | No |
 volumeAttributes.secretNamespace | secret namespace | `default`,`kube-system`, etc | No | pvc namespace

pkg/blob/blob.go

@@ -85,8 +85,9 @@ const (
 	trueValue                    = "true"
 	defaultSecretAccountName     = "azurestorageaccountname"
 	defaultSecretAccountKey      = "azurestorageaccountkey"
-	fuse                         = "fuse"
-	nfs                          = "nfs"
+	Fuse                         = "fuse"
+	Fuse2                        = "fuse2"
+	NFS                          = "nfs"
 	vnetResourceGroupField       = "vnetresourcegroup"
 	vnetNameField                = "vnetname"
 	subnetNameField              = "subnetname"
@@ -124,7 +125,7 @@ const (
 )
 
 var (
-	supportedProtocolList = []string{fuse, nfs}
+	supportedProtocolList = []string{Fuse, Fuse2, NFS}
 	retriableErrors       = []string{accountNotProvisioned, tooManyRequests, statusCodeNotFound, containerBeingDeletedDataplaneAPIError, containerBeingDeletedManagementAPIError, clientThrottled}
 )
 
@@ -409,7 +410,7 @@ func (d *Driver) GetAuthEnv(ctx context.Context, volumeID, protocol string, attr
 	}
 	klog.V(2).Infof("volumeID(%s) authEnv: %s", volumeID, authEnv)
 
-	if protocol == nfs {
+	if protocol == NFS {
 		// nfs protocol does not need account key, return directly
 		return rgName, accountName, accountKey, containerName, authEnv, err
 	}

pkg/blob/blob_test.go

@@ -585,7 +585,7 @@ func TestGetAuthEnv(t *testing.T) {
 				volumeID := "unique-volumeid"
 				attrib[storageAccountField] = "accountname"
 				attrib[containerNameField] = "containername"
-				rg, accountName, accountkey, containerName, authEnv, err := d.GetAuthEnv(context.TODO(), volumeID, nfs, attrib, secret)
+				rg, accountName, accountkey, containerName, authEnv, err := d.GetAuthEnv(context.TODO(), volumeID, NFS, attrib, secret)
 				if err != nil {
 					t.Errorf("actualErr: (%v), expect no error", err)
 				}

pkg/blob/controllerserver.go

@@ -163,7 +163,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	}
 
 	if subsID != "" && subsID != d.cloud.SubscriptionID {
-		if protocol == nfs {
+		if protocol == NFS {
 			return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("NFS protocol is not supported in cross subscription(%s)", subsID))
 		}
 		if !storeAccountKey {
@@ -184,7 +184,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	}
 
 	if protocol == "" {
-		protocol = fuse
+		protocol = Fuse
 	}
 	if !isSupportedProtocol(protocol) {
 		return nil, status.Errorf(codes.InvalidArgument, "protocol(%s) is not supported, supported protocol list: %v", protocol, supportedProtocolList)
@@ -206,7 +206,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		vnetResourceIDs []string
 		enableNfsV3     *bool
 	)
-	if protocol == nfs {
+	if protocol == NFS {
 		isHnsEnabled = to.BoolPtr(true)
 		enableNfsV3 = to.BoolPtr(true)
 		// set VirtualNetworkResourceIDs for storage account firewall setting

pkg/blob/controllerserver_test.go

@@ -237,7 +237,7 @@ func TestCreateVolume(t *testing.T) {
 					controllerServiceCapability,
 				}
 				_, err := d.CreateVolume(context.Background(), req)
-				expectedErr := status.Errorf(codes.InvalidArgument, "protocol(unit-test) is not supported, supported protocol list: [fuse nfs]")
+				expectedErr := status.Errorf(codes.InvalidArgument, "protocol(unit-test) is not supported, supported protocol list: [fuse fuse2 nfs]")
 				if !reflect.DeepEqual(err, expectedErr) {
 					t.Errorf("actualErr: (%v), expectedErr: (%v)", err, expectedErr)
 				}

pkg/blob/nodeserver.go

@@ -140,7 +140,7 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 	return &csi.NodePublishVolumeResponse{}, nil
 }
 
-func (d *Driver) mountBlobfuseWithProxy(args string, authEnv []string) (string, error) {
+func (d *Driver) mountBlobfuseWithProxy(args string, protocol string, authEnv []string) (string, error) {
 	klog.V(2).Infof("mouting using blobfuse proxy")
 	var resp *mount_azure_blob.MountAzureBlobResponse
 	var output string
@@ -152,6 +152,7 @@ func (d *Driver) mountBlobfuseWithProxy(args string, authEnv []string) (string,
 		mountClient := NewMountClient(conn)
 		mountreq := mount_azure_blob.MountAzureBlobRequest{
 			MountArgs: args,
+			Protocol:  protocol,
 			AuthEnv:   authEnv,
 		}
 		klog.V(2).Infof("calling BlobfuseProxy: MountAzureBlob function")
@@ -164,11 +165,23 @@ func (d *Driver) mountBlobfuseWithProxy(args string, authEnv []string) (string,
 	return output, err
 }
 
-func (d *Driver) mountBlobfuseInsideDriver(args string, authEnv []string) (string, error) {
+func (d *Driver) mountBlobfuseInsideDriver(args string, protocol string, authEnv []string) (string, error) {
+	var cmd *exec.Cmd
+
 	klog.V(2).Infof("mounting blobfuse inside driver")
-	cmd := exec.Command("blobfuse", strings.Split(args, " ")...)
+	if protocol == Fuse2 {
+		klog.V(2).Infof("using blobfuse V2 to mount")
+		args = "mount " + args
+		cmd = exec.Command("blobfuse2", strings.Split(args, " ")...)
+	} else {
+		klog.V(2).Infof("using blobfuse V1 to mount")
+		cmd = exec.Command("blobfuse", strings.Split(args, " ")...)
+	}
+
 	cmd.Env = append(os.Environ(), authEnv...)
 	output, err := cmd.CombinedOutput()
+	klog.V(2).Infof("mount output: %s\n", string(output))
+
 	return string(output), err
 }
 
@@ -290,14 +303,14 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		serverAddress = fmt.Sprintf("%s.blob.%s", accountName, storageEndpointSuffix)
 	}
 
-	if protocol == nfs {
+	if protocol == NFS {
 		klog.V(2).Infof("target %v\nprotocol %v\n\nvolumeId %v\ncontext %v\nmountflags %v\nserverAddress %v",
 			targetPath, protocol, volumeID, attrib, mountFlags, serverAddress)
 
 		source := fmt.Sprintf("%s:/%s/%s", serverAddress, accountName, containerName)
 		mountOptions := util.JoinMountOptions(mountFlags, []string{"sec=sys,vers=3,nolock"})
 		if err := wait.PollImmediate(1*time.Second, 2*time.Minute, func() (bool, error) {
-			return true, d.mounter.MountSensitive(source, targetPath, nfs, mountOptions, []string{})
+			return true, d.mounter.MountSensitive(source, targetPath, NFS, mountOptions, []string{})
 		}); err != nil {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("volume(%s) mount %q on %q failed with %v", volumeID, source, targetPath, err))
 		}
@@ -348,9 +361,9 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 
 	var output string
 	if d.enableBlobfuseProxy {
-		output, err = d.mountBlobfuseWithProxy(args, authEnv)
+		output, err = d.mountBlobfuseWithProxy(args, protocol, authEnv)
 	} else {
-		output, err = d.mountBlobfuseInsideDriver(args, authEnv)
+		output, err = d.mountBlobfuseInsideDriver(args, protocol, authEnv)
 	}
 
 	if err != nil {

pkg/blob/nodeserver_test.go

@@ -741,7 +741,7 @@ func TestMountBlobfuseWithProxy(t *testing.T) {
 	args := "--tmp-path /tmp"
 	authEnv := []string{"username=blob", "authkey=blob"}
 	d := NewFakeDriver()
-	_, err := d.mountBlobfuseWithProxy(args, authEnv)
+	_, err := d.mountBlobfuseWithProxy(args, "fuse", authEnv)
 	// should be context.deadlineExceededError{} error
 	assert.NotNil(t, err)
 }
@@ -750,7 +750,7 @@ func TestMountBlobfuseInsideDriver(t *testing.T) {
 	args := "--tmp-path /tmp"
 	authEnv := []string{"username=blob", "authkey=blob"}
 	d := NewFakeDriver()
-	_, err := d.mountBlobfuseInsideDriver(args, authEnv)
+	_, err := d.mountBlobfuseInsideDriver(args, Fuse, authEnv)
 	// the error should be of type exec.ExitError
 	assert.NotNil(t, err)
 }

pkg/blobfuse-proxy/pb/azure_blob_mount.pb.go

@@ -5,6 +5,7 @@ option go_package = ".;pb";
 message MountAzureBlobRequest {
 	string mountArgs = 1;
 	repeated string authEnv = 2;
+	string protocol = 3;
 }
 
 message MountAzureBlobResponse {

pkg/blobfuse-proxy/proto/azure_blob_mount.proto

@@ -26,6 +26,7 @@ import (
 
 	"google.golang.org/grpc"
 	"k8s.io/klog/v2"
+	"sigs.k8s.io/blob-csi-driver/pkg/blob"
 	mount_azure_blob "sigs.k8s.io/blob-csi-driver/pkg/blobfuse-proxy/pb"
 	"sigs.k8s.io/blob-csi-driver/pkg/util"
 )
@@ -62,16 +63,18 @@ func (server *MountServer) MountAzureBlob(ctx context.Context,
 
 	args := req.GetMountArgs()
 	authEnv := req.GetAuthEnv()
-	klog.V(2).Infof("received mount request: Mounting with args %v \n", args)
+	protocol := req.GetProtocol()
+	klog.V(2).Infof("received mount request: Protocol: %s, server default blobfuseVersion: %v, Mounting with args %v \n", protocol, server.blobfuseVersion, args)
 
 	var cmd *exec.Cmd
 	var result mount_azure_blob.MountAzureBlobResponse
-	switch server.blobfuseVersion {
-	case BlobfuseV1:
-		cmd = exec.Command("blobfuse", strings.Split(args, " ")...)
-	case BlobfuseV2:
+	if protocol == blob.Fuse2 || server.blobfuseVersion == BlobfuseV2 {
+		klog.V(2).Infof("using blobfuse V2 to mount")
 		args = "mount " + args
 		cmd = exec.Command("blobfuse2", strings.Split(args, " ")...)
+	} else {
+		klog.V(2).Infof("using blobfuse V1 to mount")
+		cmd = exec.Command("blobfuse", strings.Split(args, " ")...)
 	}
 
 	cmd.Env = append(cmd.Env, authEnv...)
@@ -111,9 +114,10 @@ func getBlobfuseVersion() BlobfuseVersion {
 	}
 
 	if osinfo.Distro == "Ubuntu" && osinfo.Version >= "22.04" {
-		klog.V(2).Info("proxy using blobfuse V2 for mounting")
+		klog.V(2).Info("proxy default using blobfuse V2 for mounting")
 		return BlobfuseV2
 	}
 
+	klog.V(2).Info("proxy default using blobfuse V1 for mounting")
 	return BlobfuseV1
 }

pkg/blobfuse-proxy/server/server.go

@@ -571,4 +571,34 @@ var _ = ginkgo.Describe("[blob-csi-e2e] Dynamic Provisioning", func() {
 		}
 		test.Run(cs, ns)
 	})
+
+	ginkgo.It("should create a blobfuse2 volume on demand with mount options [fuse2]", func() {
+		if isAzureStackCloud {
+			ginkgo.Skip("test case is not available for Azure Stack")
+		}
+		pods := []testsuites.PodDetails{
+			{
+				Cmd: "echo 'hello world' > /mnt/test-1/data && grep 'hello world' /mnt/test-1/data",
+				Volumes: []testsuites.VolumeDetails{
+					{
+						ClaimSize:    "10Gi",
+						MountOptions: []string{},
+						VolumeMount: testsuites.VolumeMountDetails{
+							NameGenerate:      "test-volume-",
+							MountPathGenerate: "/mnt/test-",
+						},
+					},
+				},
+			},
+		}
+		test := testsuites.DynamicallyProvisionedCmdVolumeTest{
+			CSIDriver: testDriver,
+			Pods:      pods,
+			StorageClassParameters: map[string]string{
+				"skuName":  "Standard_LRS",
+				"protocol": "fuse2",
+			},
+		}
+		test.Run(cs, ns)
+	})
 })