UpdateCnsRegisterApi for multiwriter

Author: skogta

pkg/apis/cnsoperator/cnsregistervolume/v1alpha1/cnsregistervolume_types.go

@@ -54,6 +54,18 @@ type CnsRegisterVolumeSpec struct {
 	// This is for a 34a9c05d-5f03-e254-e692-02004479cb91/vm2_1.vmdk
 	// file under datacenter "Datacenter-1" and datastore "vsanDatastore".
 	DiskURLPath string `json:"diskURLPath,omitempty"`
+
+	// VolumeMode can either be Block (for raw block volume) or
+	// Filesystem. Default values will be assumed if it is missing:
+	// With RWO accessmode, default is Filesystem.
+	// With RWX accessmode, default is Block.
+	VolumeMode v1.PersistentVolumeMode `json:"volumeMode,omitempty"`
+
+	// FsType represents the filesystem type.
+	// Default values will be assumed if it is missing:
+	// With RWO, it will be ext4.
+	// With RWX with volumeMode as Filsystem, it will be ntf4.
+	FsType string `json:"fsType,omitempty"`
 }
 
 // CnsRegisterVolumeStatus defines the observed state of CnsRegisterVolume

pkg/syncer/cnsoperator/controller/cnsregistervolume/cnsregistervolume_controller.go

@@ -247,7 +247,8 @@ func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 	// Currently file volume registration is not supported.
 	ok := isBlockVolumeRegisterRequest(ctx, instance)
 	if !ok {
-		msg := fmt.Sprintf("AccessMode: %s is not supported", instance.Spec.AccessMode)
+		msg := fmt.Sprintf("AccessMode: %s is not supported with volume %s is not supported",
+			instance.Spec.AccessMode, instance.Spec.VolumeMode)
 		log.Error(msg)
 		setInstanceError(ctx, r, instance, msg)
 		return reconcile.Result{RequeueAfter: timeout}, nil
@@ -482,7 +483,7 @@ func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 				Name:       instance.Spec.PvcName,
 			}
 			pvSpec := getPersistentVolumeSpec(pvName, volumeID, capacityInMb,
-				accessMode, storageClassName, claimRef)
+				accessMode, instance.Spec.VolumeMode, storageClassName, claimRef, instance.Spec.FsType)
 			pvSpec.Spec.NodeAffinity = pvNodeAffinity
 			log.Debugf("PV spec is: %+v", pvSpec)
 			pv, err = k8sclient.CoreV1().PersistentVolumes().Create(ctx, pvSpec, metav1.CreateOptions{})
@@ -510,7 +511,7 @@ func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 	// Create PVC mapping to above created PV.
 	log.Infof("Creating PVC: %s", instance.Spec.PvcName)
 	pvcSpec, err := getPersistentVolumeClaimSpec(ctx, instance.Spec.PvcName, instance.Namespace, capacityInMb,
-		storageClassName, accessMode, pvName, datastoreAccessibleTopology)
+		storageClassName, accessMode, instance.Spec.VolumeMode, pvName, datastoreAccessibleTopology)
 	if err != nil {
 		msg := fmt.Sprintf("Failed to create spec for PVC: %q. Error: %v", instance.Spec.PvcName, err)
 		log.Errorf(msg)
@@ -701,6 +702,16 @@ func validateCnsRegisterVolumeSpec(ctx context.Context, instance *cnsregistervol
 	} else if instance.Spec.DiskURLPath != "" && instance.Spec.AccessMode != "" &&
 		instance.Spec.AccessMode != v1.ReadWriteOnce {
 		msg = fmt.Sprintf("DiskURLPath cannot be used with accessMode: %q", instance.Spec.AccessMode)
+	} else if instance.Spec.AccessMode == v1.ReadWriteMany && instance.Spec.VolumeMode != "" && instance.Spec.VolumeMode != v1.PersistentVolumeBlock {
+		return errors.New(fmt.Sprintf("AccessMode %q is not allowed with VolumeMode %q", instance.Spec.AccessMode, instance.Spec.VolumeMode))
+	}
+	if instance.Spec.VolumeID != "" && instance.Spec.AccessMode == "" {
+		return errors.New(fmt.Sprintf("AccessMode cannot be empty when VolumeID is provided"))
+	}
+	if instance.Spec.AccessMode == v1.ReadWriteMany {
+		if instance.Spec.VolumeMode != "" || instance.Spec.VolumeMode != v1.PersistentVolumeFilesystem {
+			return errors.New(fmt.Sprintf("AccessMode %q is not allowed with VolumeMode %q", instance.Spec.AccessMode, instance.Spec.VolumeMode))
+		}
 	}
 	if instance.Spec.VolumeID != "" {
 		pvName, found := commonco.ContainerOrchestratorUtility.GetPVNameFromCSIVolumeID(instance.Spec.VolumeID)
@@ -721,7 +732,7 @@ func validateCnsRegisterVolumeSpec(ctx context.Context, instance *cnsregistervol
 // via CnsRegisterVolume instance.
 func isBlockVolumeRegisterRequest(ctx context.Context, instance *cnsregistervolumev1alpha1.CnsRegisterVolume) bool {
 	if instance.Spec.AccessMode != "" {
-		if instance.Spec.AccessMode == v1.ReadWriteOnce {
+		if instance.Spec.AccessMode == v1.ReadWriteOnce || (instance.Spec.AccessMode == v1.ReadWriteMany && instance.Spec.VolumeMode == v1.PersistentVolumeBlock) {
 			return true
 		}
 	} else {

pkg/syncer/cnsoperator/controller/cnsregistervolume/util.go

@@ -138,7 +138,7 @@ func constructCreateSpecForInstance(r *ReconcileCnsRegisterVolume,
 			BackingDiskUrlPath: instance.Spec.DiskURLPath,
 		}
 	}
-	if instance.Spec.AccessMode == v1.ReadWriteOnce || instance.Spec.AccessMode == "" {
+	if instance.Spec.AccessMode == v1.ReadWriteOnce || instance.Spec.AccessMode == "" || (instance.Spec.AccessMode == v1.ReadWriteMany && instance.Spec.VolumeMode == v1.PersistentVolumeBlock) {
 		createSpec.VolumeType = common.BlockVolumeType
 	} else {
 		createSpec.VolumeType = common.FileVolumeType
@@ -241,8 +241,20 @@ func getK8sStorageClassNameWithImmediateBindingModeForPolicy(ctx context.Context
 
 // getPersistentVolumeSpec to create PV volume spec for the given input params.
 func getPersistentVolumeSpec(volumeName string, volumeID string, capacity int64,
-	accessMode v1.PersistentVolumeAccessMode, scName string, claimRef *v1.ObjectReference) *v1.PersistentVolume {
+	accessMode v1.PersistentVolumeAccessMode, volumeMode v1.PersistentVolumeMode, scName string, claimRef *v1.ObjectReference, fsType string) *v1.PersistentVolume {
 	capacityInMb := strconv.FormatInt(capacity, 10) + "Mi"
+	if fsType == "" {
+		if accessMode == v1.ReadWriteOnce {
+			fsType = "ext4"
+		}
+	}
+	if volumeMode == "" {
+		if accessMode == v1.ReadWriteOnce {
+			volumeMode = v1.PersistentVolumeFilesystem
+		} else {
+			volumeMode = v1.PersistentVolumeBlock
+		}
+	}
 	pv := &v1.PersistentVolume{
 		TypeMeta: metav1.TypeMeta{},
 		ObjectMeta: metav1.ObjectMeta{
@@ -258,7 +270,7 @@ func getPersistentVolumeSpec(volumeName string, volumeID string, capacity int64,
 					Driver:       cnsoperatortypes.VSphereCSIDriverName,
 					VolumeHandle: volumeID,
 					ReadOnly:     false,
-					FSType:       "ext4",
+					FSType:       fsType,
 				},
 			},
 			AccessModes: []v1.PersistentVolumeAccessMode{
@@ -269,6 +281,9 @@ func getPersistentVolumeSpec(volumeName string, volumeID string, capacity int64,
 		},
 		Status: v1.PersistentVolumeStatus{},
 	}
+
+	*pv.Spec.VolumeMode = volumeMode
+
 	annotations := make(map[string]string)
 	annotations["pv.kubernetes.io/provisioned-by"] = cnsoperatortypes.VSphereCSIDriverName
 	pv.Annotations = annotations
@@ -278,7 +293,7 @@ func getPersistentVolumeSpec(volumeName string, volumeID string, capacity int64,
 // getPersistentVolumeClaimSpec return the PersistentVolumeClaim spec with
 // specified storage class.
 func getPersistentVolumeClaimSpec(ctx context.Context, name string, namespace string, capacity int64,
-	storageClassName string, accessMode v1.PersistentVolumeAccessMode, pvName string,
+	storageClassName string, accessMode v1.PersistentVolumeAccessMode, volumeMode v1.PersistentVolumeMode, pvName string,
 	datastoreAccessibleTopology []map[string]string) (*v1.PersistentVolumeClaim, error) {
 
 	log := logger.GetLogger(ctx)
@@ -318,6 +333,16 @@ func getPersistentVolumeClaimSpec(ctx context.Context, name string, namespace st
 			VolumeName:       pvName,
 		},
 	}
+
+	if volumeMode == "" {
+		if accessMode == v1.ReadWriteOnce {
+			volumeMode = v1.PersistentVolumeFilesystem
+		} else {
+			volumeMode = v1.PersistentVolumeBlock
+		}
+	}
+	*claim.Spec.VolumeMode = volumeMode
+
 	return claim, nil
 }