Add Block volume support for CSI provisioner

In CSI provisioner, below three logics need to be implemented,
to add Block volume support to CSI provisioner:

  1. Add SupportsBlock that always return true,
  2. Pass BlockVolume instead of MountVolume to CreateVolume
     if volumeMode is set to be Block on Provision,
  3. Set volumeMode and skip adding FsType to PV returned by Provision,
     if volumeMode is set to be Block on Provision.

Also, below 2 test cases for TestProvision are added.
  1. volumeMode=Filesystem PVC case: return Filesystem PV expected
  2. volumeMode=Block PVC case: return Block PV expected

Fixes #110

Author: mkimuram

pkg/controller/controller.go

@@ -32,6 +32,7 @@ import (
 	"github.com/golang/glog"
 
 	"github.com/kubernetes-sigs/sig-storage-lib-external-provisioner/controller"
+	"github.com/kubernetes-sigs/sig-storage-lib-external-provisioner/util"
 
 	snapapi "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1"
 	snapclientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
@@ -108,6 +109,7 @@ const (
 )
 
 var _ controller.Provisioner = &csiProvisioner{}
+var _ controller.BlockProvisioner = &csiProvisioner{}
 
 var (
 	// Each provisioner have a identify string to distinguish with others. This
@@ -332,6 +334,58 @@ func makeVolumeName(prefix, pvcUID string, volumeNameUUIDLength int) (string, er
 
 }
 
+func getAccessTypeBlock() *csi.VolumeCapability_Block {
+	return &csi.VolumeCapability_Block{
+		Block: &csi.VolumeCapability_BlockVolume{},
+	}
+}
+
+func getAccessTypeMount(fsType string, mountFlags []string) *csi.VolumeCapability_Mount {
+	return &csi.VolumeCapability_Mount{
+		Mount: &csi.VolumeCapability_MountVolume{
+			FsType:     fsType,
+			MountFlags: mountFlags,
+		},
+	}
+}
+
+func getAccessMode(pvcAccessMode v1.PersistentVolumeAccessMode) *csi.VolumeCapability_AccessMode {
+	switch pvcAccessMode {
+	case v1.ReadWriteOnce:
+		return &csi.VolumeCapability_AccessMode{
+			Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+		}
+	case v1.ReadWriteMany:
+		return &csi.VolumeCapability_AccessMode{
+			Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
+		}
+	case v1.ReadOnlyMany:
+		return &csi.VolumeCapability_AccessMode{
+			Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+		}
+	default:
+		return nil
+	}
+}
+
+func getVolumeCapability(
+	pvcOptions controller.VolumeOptions,
+	pvcAccessMode v1.PersistentVolumeAccessMode,
+	fsType string,
+) *csi.VolumeCapability {
+	if util.CheckPersistentVolumeClaimModeBlock(pvcOptions.PVC) {
+		return &csi.VolumeCapability{
+			AccessType: getAccessTypeBlock(),
+			AccessMode: getAccessMode(pvcAccessMode),
+		}
+	} else {
+		return &csi.VolumeCapability{
+			AccessType: getAccessTypeMount(fsType, pvcOptions.MountOptions),
+			AccessMode: getAccessMode(pvcAccessMode),
+		}
+	}
+}
+
 func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.PersistentVolume, error) {
 	if options.PVC.Spec.Selector != nil {
 		return nil, fmt.Errorf("claim Selector is not supported")
@@ -375,44 +429,14 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 	capacity := options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
 	volSizeBytes := capacity.Value()
 
-	accessType := &csi.VolumeCapability_Mount{
-		Mount: &csi.VolumeCapability_MountVolume{
-			FsType:     fsType,
-			MountFlags: options.MountOptions,
-		},
-	}
-
 	// Get access mode
 	volumeCaps := make([]*csi.VolumeCapability, 0)
-	for _, cap := range options.PVC.Spec.AccessModes {
-		switch cap {
-		case v1.ReadWriteOnce:
-			volumeCaps = append(volumeCaps, &csi.VolumeCapability{
-				AccessMode: &csi.VolumeCapability_AccessMode{
-					Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-				},
-				AccessType: accessType,
-			})
-		case v1.ReadWriteMany:
-			volumeCaps = append(volumeCaps, &csi.VolumeCapability{
-				AccessMode: &csi.VolumeCapability_AccessMode{
-					Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
-				},
-				AccessType: accessType,
-			})
-		case v1.ReadOnlyMany:
-			volumeCaps = append(volumeCaps, &csi.VolumeCapability{
-				AccessMode: &csi.VolumeCapability_AccessMode{
-					Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
-				},
-				AccessType: accessType,
-			})
-		}
+	for _, pvcAccessMode := range options.PVC.Spec.AccessModes {
+		volumeCaps = append(volumeCaps, getVolumeCapability(options, pvcAccessMode, fsType))
 	}
 
 	// Create a CSI CreateVolumeRequest and Response
 	req := csi.CreateVolumeRequest{
-
 		Name:               pvName,
 		Parameters:         options.Parameters,
 		VolumeCapabilities: volumeCaps,
@@ -538,7 +562,6 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 				CSI: &v1.CSIPersistentVolumeSource{
 					Driver:                     driverState.driverName,
 					VolumeHandle:               p.volumeIdToHandle(rep.Volume.VolumeId),
-					FSType:                     fsType,
 					VolumeAttributes:           volumeAttributes,
 					ControllerPublishSecretRef: controllerPublishSecretRef,
 					NodeStageSecretRef:         nodeStageSecretRef,
@@ -553,6 +576,15 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		pv.Spec.NodeAffinity = GenerateVolumeNodeAffinity(rep.Volume.AccessibleTopology)
 	}
 
+	// Set VolumeMode to PV if it is passed via PVC spec when Block feature is enabled
+	if options.PVC.Spec.VolumeMode != nil {
+		pv.Spec.VolumeMode = options.PVC.Spec.VolumeMode
+	}
+	// Set FSType if PV is not Block Volume
+	if !util.CheckPersistentVolumeClaimModeBlock(options.PVC) {
+		pv.Spec.PersistentVolumeSource.CSI.FSType = fsType
+	}
+
 	glog.Infof("successfully created PV %+v", pv.Spec.PersistentVolumeSource)
 
 	return pv, nil
@@ -653,6 +685,14 @@ func (p *csiProvisioner) Delete(volume *v1.PersistentVolume) error {
 	return err
 }
 
+func (p *csiProvisioner) SupportsBlock() bool {
+	// SupportsBlock always return true, because current CSI spec doesn't allow checking
+	// drivers' capability of block volume before creating volume.
+	// Drivers that don't support block volume should return error for CreateVolume called
+	// by Provision if block AccessType is specified.
+	return true
+}
+
 //TODO use a unique volume handle from and to Id
 func (p *csiProvisioner) volumeIdToHandle(id string) string {
 	return id

pkg/controller/controller_test.go

@@ -585,6 +585,13 @@ func createFakePVC(requestBytes int64) *v1.PersistentVolumeClaim {
 	}
 }
 
+// createFakePVCWithVolumeMode returns PVC with VolumeMode
+func createFakePVCWithVolumeMode(requestBytes int64, volumeMode v1.PersistentVolumeMode) *v1.PersistentVolumeClaim {
+	claim := createFakePVC(requestBytes)
+	claim.Spec.VolumeMode = &volumeMode
+	return claim
+}
+
 func TestGetSecretReference(t *testing.T) {
 	testcases := map[string]struct {
 		nameKey      string
@@ -709,12 +716,18 @@ func TestGetSecretReference(t *testing.T) {
 }
 
 func TestProvision(t *testing.T) {
-	var requestedBytes int64 = 100
+
+	var (
+		requestedBytes       int64 = 100
+		volumeModeFileSystem       = v1.PersistentVolumeFilesystem
+		volumeModeBlock            = v1.PersistentVolumeBlock
+	)
 
 	type pvSpec struct {
 		Name          string
 		ReclaimPolicy v1.PersistentVolumeReclaimPolicy
 		AccessModes   []v1.PersistentVolumeAccessMode
+		VolumeMode    *v1.PersistentVolumeMode
 		Capacity      v1.ResourceList
 		CSIPVS        *v1.CSIPersistentVolumeSource
 	}
@@ -989,6 +1002,49 @@ func TestProvision(t *testing.T) {
 				},
 			},
 		},
+		"provision with volume mode(Filesystem)": {
+			volOpts: controller.VolumeOptions{
+				PVName:     "test-name",
+				PVC:        createFakePVCWithVolumeMode(requestedBytes, volumeModeFileSystem),
+				Parameters: map[string]string{},
+			},
+			expectedPVSpec: &pvSpec{
+				Name: "test-testi",
+				Capacity: v1.ResourceList{
+					v1.ResourceName(v1.ResourceStorage): bytesToGiQuantity(requestedBytes),
+				},
+				VolumeMode: &volumeModeFileSystem,
+				CSIPVS: &v1.CSIPersistentVolumeSource{
+					Driver:       "test-driver",
+					VolumeHandle: "test-volume-id",
+					FSType:       "ext4",
+					VolumeAttributes: map[string]string{
+						"storage.kubernetes.io/csiProvisionerIdentity": "test-provisioner",
+					},
+				},
+			},
+		},
+		"provision with volume mode(Block)": {
+			volOpts: controller.VolumeOptions{
+				PVName:     "test-name",
+				PVC:        createFakePVCWithVolumeMode(requestedBytes, volumeModeBlock),
+				Parameters: map[string]string{},
+			},
+			expectedPVSpec: &pvSpec{
+				Name: "test-testi",
+				Capacity: v1.ResourceList{
+					v1.ResourceName(v1.ResourceStorage): bytesToGiQuantity(requestedBytes),
+				},
+				VolumeMode: &volumeModeBlock,
+				CSIPVS: &v1.CSIPersistentVolumeSource{
+					Driver:       "test-driver",
+					VolumeHandle: "test-volume-id",
+					VolumeAttributes: map[string]string{
+						"storage.kubernetes.io/csiProvisionerIdentity": "test-provisioner",
+					},
+				},
+			},
+		},
 		"fail to get secret reference": {
 			volOpts: controller.VolumeOptions{
 				PVName:     "test-name",
@@ -1196,6 +1252,10 @@ func TestProvision(t *testing.T) {
 				t.Errorf("test %q: expected access modes: %v, got: %v", k, tc.expectedPVSpec.AccessModes, pv.Spec.AccessModes)
 			}
 
+			if !reflect.DeepEqual(pv.Spec.VolumeMode, tc.expectedPVSpec.VolumeMode) {
+				t.Errorf("test %q: expected volumeMode: %v, got: %v", k, tc.expectedPVSpec.VolumeMode, pv.Spec.VolumeMode)
+			}
+
 			if !reflect.DeepEqual(pv.Spec.Capacity, tc.expectedPVSpec.Capacity) {
 				t.Errorf("test %q: expected capacity: %v, got: %v", k, tc.expectedPVSpec.Capacity, pv.Spec.Capacity)
 			}