feat: support NFS

feat: add Dockerfile change

feat: add nodeserver change to support NFS

fix test failure

fix: rename fsType to protocol

Author: andyzhangx

deploy/example/storageclass-blob-nfs.yaml

@@ -0,0 +1,10 @@
+---
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: blob
+provisioner: blobfuse.csi.azure.com
+parameters:
+  storageAccount: EXISTING_STORAGE_ACCOUNT_NAME
+  protocol: nfs
+volumeBindingMode: Immediate

pkg/blobfuse/blobfuse.go

@@ -45,6 +45,9 @@ const (
 	defaultVers      = "3.0"
 	serverNameField  = "server"
 	tagsField        = "tags"
+	protocolField    = "protocol"
+	fuse             = "fuse"
+	nfs              = "nfs"
 
 	// See https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names
 	containerNameMinLength = 3
@@ -58,7 +61,8 @@ const (
 )
 
 var (
-	retriableErrors = []string{accountNotProvisioned, tooManyRequests, shareNotFound, shareBeingDeleted, clientThrottled}
+	supportedProtocolList = []string{fuse, nfs}
+	retriableErrors       = []string{accountNotProvisioned, tooManyRequests, shareNotFound, shareBeingDeleted, clientThrottled}
 )
 
 // Driver implements all interfaces of CSI drivers
@@ -417,3 +421,15 @@ func isRetriableError(err error) bool {
 	}
 	return false
 }
+
+func isSupportedProtocol(protocol string) bool {
+	if protocol == "" {
+		return true
+	}
+	for _, v := range supportedProtocolList {
+		if protocol == v {
+			return true
+		}
+	}
+	return false
+}

pkg/blobfuse/blobfuse_test.go

@@ -314,3 +314,34 @@ func TestIsCorruptedDir(t *testing.T) {
 		assert.Equal(t, test.expectedResult, isCorruptedDir, "TestCase[%d]: %s", i, test.desc)
 	}
 }
+
+func TestIsSupportedProtocol(t *testing.T) {
+	tests := []struct {
+		protocol       string
+		expectedResult bool
+	}{
+		{
+			protocol:       "",
+			expectedResult: true,
+		},
+		{
+			protocol:       "fuse",
+			expectedResult: true,
+		},
+		{
+			protocol:       "nfs",
+			expectedResult: true,
+		},
+		{
+			protocol:       "invalid",
+			expectedResult: false,
+		},
+	}
+
+	for _, test := range tests {
+		result := isSupportedProtocol(test.protocol)
+		if result != test.expectedResult {
+			t.Errorf("isSupportedProtocol(%s) returned with %v, not equal to %v", test.protocol, result, test.expectedResult)
+		}
+	}
+}

pkg/blobfuse/controllerserver.go

@@ -55,7 +55,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	requestGiB := int(util.RoundUpGiB(volSizeBytes))
 
 	parameters := req.GetParameters()
-	var storageAccountType, resourceGroup, location, account, containerName, customTags string
+	var storageAccountType, resourceGroup, location, account, containerName, protocol, customTags string
 
 	// Apply ProvisionerParameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
@@ -73,6 +73,8 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 			resourceGroup = v
 		case "containername":
 			containerName = v
+		case protocolField:
+			protocol = v
 		case tagsField:
 			customTags = v
 		}
@@ -82,6 +84,16 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		resourceGroup = d.cloud.ResourceGroup
 	}
 
+	if !isSupportedProtocol(protocol) {
+		return nil, status.Errorf(codes.InvalidArgument, "protocol(%s) is not supported, supported protocol list: %v", protocol, supportedProtocolList)
+	}
+
+	if protocol == nfs {
+		if account == "" {
+			return nil, status.Errorf(codes.InvalidArgument, "storage account must be specified when provisioning nfs file share")
+		}
+	}
+
 	accountKind := string(storage.StorageV2)
 	if strings.HasPrefix(strings.ToLower(storageAccountType), "premium") {
 		accountKind = string(storage.BlockBlobStorage)

pkg/blobfuse/controllerserver_test.go

@@ -135,7 +135,7 @@ func TestCreateVolume(t *testing.T) {
 				}
 				mockStorageAccountsClient.EXPECT().ListByResourceGroup(gomock.Any(), gomock.Any()).Return(nil, rerr).AnyTimes()
 				_, err := d.CreateVolume(context.Background(), req)
-				expectedErr := fmt.Errorf("could not get storage key for storage account : could not list storage accounts for account type : Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: test")
+				expectedErr := status.Errorf(codes.Internal, "failed to ensure storage account: could not list storage accounts for account type : Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: test")
 				if !reflect.DeepEqual(err, expectedErr) {
 					t.Errorf("actualErr: (%v), expectedErr: (%v)", err, expectedErr)
 				}

pkg/blobfuse/nodeserver.go

@@ -22,11 +22,13 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"time"
 
 	volumehelper "sigs.k8s.io/blobfuse-csi-driver/pkg/util"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
 
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/utils/mount"
@@ -119,7 +121,6 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 	}
 
 	volumeID := req.GetVolumeId()
-	fsType := req.GetVolumeCapability().GetMount().GetFsType()
 	mountFlags := req.GetVolumeCapability().GetMount().GetMountFlags()
 	attrib := req.GetVolumeContext()
 	secrets := req.GetSecrets()
@@ -129,29 +130,53 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, err
 	}
 
-	// Get mountOptions that the volume will be formatted and mounted with
-	options := []string{"--use-https=true"}
-	mountOptions := util.JoinMountOptions(mountFlags, options)
-
-	args := targetPath + " " + "--tmp-path=/mnt/" + volumeID + " " + "--container-name=" + containerName
-	for _, opt := range mountOptions {
-		args = args + " " + opt
-	}
-
-	var blobStorageEndPoint string
+	var blobStorageEndPoint, protocol string
 	for k, v := range attrib {
 		switch strings.ToLower(k) {
 		case serverNameField:
 			blobStorageEndPoint = v
+		case protocolField:
+			protocol = v
 		}
 	}
 	if strings.TrimSpace(blobStorageEndPoint) == "" {
 		// server address is "accountname.blob.core.windows.net" by default
 		blobStorageEndPoint = fmt.Sprintf("%s.blob.%s", accountName, d.cloud.Environment.StorageEndpointSuffix)
 	}
 
-	klog.V(2).Infof("target %v\nfstype %v\n\nvolumeId %v\ncontext %v\nmountflags %v\nmountOptions %v\nargs %v\nblobStorageEndPoint %v",
-		targetPath, fsType, volumeID, attrib, mountFlags, mountOptions, args, blobStorageEndPoint)
+	if protocol == nfs {
+		klog.V(2).Infof("target %v\nprotocol %v\n\nvolumeId %v\ncontext %v\nmountflags %v\nblobStorageEndPoint %v",
+			targetPath, protocol, volumeID, attrib, mountFlags, blobStorageEndPoint)
+
+		source := fmt.Sprintf("%s:/%s/%s", blobStorageEndPoint, accountName, containerName)
+		mountOptions := util.JoinMountOptions(mountFlags, []string{"sec=sys,vers=3,nolock"})
+		mountComplete := false
+		err := wait.PollImmediate(1*time.Second, 2*time.Minute, func() (bool, error) {
+			err := d.mounter.MountSensitive(source, targetPath, nfs, mountOptions, []string{})
+			mountComplete = true
+			return true, err
+		})
+		if !mountComplete {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("volume(%s) mount %q on %q failed with timeout(2m)", volumeID, source, targetPath))
+		}
+		if err != nil {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("volume(%s) mount %q on %q failed with %v", volumeID, source, targetPath, err))
+		}
+		klog.V(2).Infof("volume(%s) mount %q on %q succeeded", volumeID, source, targetPath)
+
+		return &csi.NodeStageVolumeResponse{}, nil
+	}
+
+	// Get mountOptions that the volume will be formatted and mounted with
+	mountOptions := util.JoinMountOptions(mountFlags, []string{"--use-https=true"})
+
+	args := targetPath + " " + "--tmp-path=/mnt/" + volumeID + " " + "--container-name=" + containerName
+	for _, opt := range mountOptions {
+		args = args + " " + opt
+	}
+
+	klog.V(2).Infof("target %v\nprotocol %v\n\nvolumeId %v\ncontext %v\nmountflags %v\nmountOptions %v\nargs %v\nblobStorageEndPoint %v",
+		targetPath, protocol, volumeID, attrib, mountFlags, mountOptions, args, blobStorageEndPoint)
 	cmd := exec.Command("blobfuse", strings.Split(args, " ")...)
 
 	cmd.Env = append(os.Environ(), "AZURE_STORAGE_ACCOUNT="+accountName)
@@ -187,6 +212,7 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return nil, err
 	}
 
+	klog.V(2).Infof("volume(%s) mount on %q succeeded", volumeID, targetPath)
 	return &csi.NodeStageVolumeResponse{}, nil
 }
 

pkg/blobfuseplugin/Dockerfile

@@ -25,6 +25,8 @@ COPY --from=builder /go/src/sigs.k8s.io/blobfuse-csi-driver/_output/blobfuseplug
 RUN apt-get update && clean-install ca-certificates pkg-config libfuse-dev cmake libcurl4-gnutls-dev libgnutls28-dev uuid-dev libgcrypt20-dev wget
 RUN wget -O /tmp/packages-microsoft-prod.deb https://packages.microsoft.com/config/ubuntu/16.04/packages-microsoft-prod.deb && dpkg -i /tmp/packages-microsoft-prod.deb && apt-get update && apt install blobfuse fuse -y && rm -f /tmp/packages-microsoft-prod.deb
 RUN apt remove wget -y
+# this is a workaround to install nfs-common and don't quit with error
+RUN apt install nfs-common -y || true
 LABEL maintainers="andyzhangx"
 LABEL description="Blobfuse CSI Driver"
 

pkg/blobfuseplugin/dev.Dockerfile

@@ -14,7 +14,7 @@
 
 FROM mcr.microsoft.com/aks/fundamental/base-ubuntu:v0.0.5
 RUN wget -O /tmp/packages-microsoft-prod.deb https://packages.microsoft.com/config/ubuntu/16.04/packages-microsoft-prod.deb
-RUN dpkg -i /tmp/packages-microsoft-prod.deb && apt-get update && apt-get install -y ca-certificates pkg-config libfuse-dev cmake libcurl4-gnutls-dev libgnutls28-dev uuid-dev libgcrypt20-dev blobfuse
+RUN dpkg -i /tmp/packages-microsoft-prod.deb && apt-get update && apt-get install -y ca-certificates pkg-config libfuse-dev cmake libcurl4-gnutls-dev libgnutls28-dev uuid-dev libgcrypt20-dev blobfuse nfs-common
 LABEL maintainers="andyzhangx"
 LABEL description="Blobfuse CSI Driver"