Block RWX volume creation for VMFS datastores if policy is not EZT

Author: skogta

pkg/csi/service/common/util.go

@@ -37,6 +37,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	crconfig "sigs.k8s.io/controller-runtime/pkg/client/config"
 	cnsvolume "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/cns-lib/volume"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/cns-lib/vsphere"
 	cnsvsphere "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/cns-lib/vsphere"
 	cnsconfig "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/config"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/logger"
@@ -45,6 +46,8 @@ import (
 const (
 	defaultK8sCloudOperatorServicePort = 10000
 	MissingSnapshotAggregatedCapacity  = "csi.vsphere.missing-snapshot-aggregated-capacity"
+	vmfsNamespace                      = "com.vmware.storage.volumeallocation"
+	vmfsNamespaceEztValue              = "Fully initialized"
 )
 
 var ErrAvailabilityZoneCRNotRegistered = errors.New("AvailabilityZone custom resource not registered")
@@ -535,3 +538,69 @@ func GetCNSVolumeInfoPatch(ctx context.Context, CapacityInMb int64, volumeId str
 	}
 	return patch, nil
 }
+
+// ValidateStoragePolicyForRWXVolume returns an error if the storagepolicy is not compatible
+// for RWX volumes.
+// Currently it returns an error if policy is for VMFS datastores but it is not EagerZeroedThick.
+func ValidateStoragePolicyForRWXVolume(ctx context.Context,
+	vc *vsphere.VirtualCenter, storagePolicyId string) error {
+	log := logger.GetLogger(ctx)
+
+	policies, err := vc.PbmRetrieveContent(ctx, []string{storagePolicyId})
+	if err != nil {
+		log.Errorf("failed to retrieve policy %s. Err %s", storagePolicyId, err)
+		return err
+	}
+
+	if len(policies) != 1 {
+		msg := fmt.Sprintf("Retrieved %d polciies for policyID %s", len(policies), storagePolicyId)
+		log.Errorf(msg)
+		return errors.New(msg)
+	}
+
+	return verifyStoragePolicyForVmfsWithEagerZeroedThick(ctx, policies[0], storagePolicyId)
+}
+
+// verifyStoragePolicyForVmfsWithEagerZeroedThick goes through each rule in the policy to
+// find out if it is fully intialized for VMFS datastores.
+// This check is required for RWX shared block volumes as for VMFS datastores, the policy must be EZT.
+func verifyStoragePolicyForVmfsWithEagerZeroedThick(
+	ctx context.Context,
+	policy vsphere.SpbmPolicyContent,
+	storagePolicyID string,
+) error {
+	log := logger.GetLogger(ctx)
+
+	log.Infof("Validating policy %s", storagePolicyID)
+
+	for _, profile := range policy.Profiles {
+		isVmfs, isEzt := isVmfsEagerZeroed(profile)
+		if !isVmfs {
+			continue
+		}
+		if !isEzt {
+			msg := fmt.Sprintf(
+				"Policy %s is for VMFS datastores. It must be Thick Provision Eager Zero for RWX block volumes.",
+				storagePolicyID,
+			)
+			log.Errorf(msg)
+			return errors.New(msg)
+		}
+		log.Infof("Policy %s is for VMFS and is fully initialized", storagePolicyID)
+		return nil
+	}
+
+	return nil
+}
+
+// isVmfsEagerZeroed returns two boolean values:
+// 1. First indicates if the policy is for a VMFS datastores.
+// 2. Second indicates if the policy is eager zeroed thick or fully initialised.
+func isVmfsEagerZeroed(profile vsphere.SpbmPolicySubProfile) (bool, bool) {
+	for _, rule := range profile.Rules {
+		if rule.Ns == vmfsNamespace {
+			return true, rule.Value == vmfsNamespaceEztValue
+		}
+	}
+	return false, false
+}

pkg/csi/service/common/util_test.go

@@ -36,6 +36,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/cns-lib/vsphere"
 )
 
 var (
@@ -622,3 +623,106 @@ func TestGetClusterComputeResourceMoIds_SingleClusterPerAZ(t *testing.T) {
 	gomega.Expect(multiple).To(gomega.BeFalse())
 	gomega.Expect(moIDs).To(gomega.ContainElements("domain-c1", "domain-c2"))
 }
+
+func TestIsVmfsEagerZeroed(t *testing.T) {
+	tests := []struct {
+		name       string
+		profile    vsphere.SpbmPolicySubProfile
+		wantIsVmfs bool
+		wantIsEzt  bool
+	}{
+		{
+			name:       "VMFS eager zeroed",
+			profile:    makeProfile(makeRule(vmfsNamespace, vmfsNamespaceEztValue)),
+			wantIsVmfs: true,
+			wantIsEzt:  true,
+		},
+		{
+			name:       "VMFS not eager zeroed",
+			profile:    makeProfile(makeRule(vmfsNamespace, "THIN")),
+			wantIsVmfs: true,
+			wantIsEzt:  false,
+		},
+		{
+			name:       "Non-VMFS",
+			profile:    makeProfile(makeRule("NFS", "ANY")),
+			wantIsVmfs: false,
+			wantIsEzt:  false,
+		},
+		{
+			name:       "Empty rules",
+			profile:    makeProfile(),
+			wantIsVmfs: false,
+			wantIsEzt:  false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotVmfs, gotEzt := isVmfsEagerZeroed(tt.profile)
+			assert.Equal(t, tt.wantIsVmfs, gotVmfs)
+			assert.Equal(t, tt.wantIsEzt, gotEzt)
+		})
+	}
+}
+
+func TestVerifyStoragePolicyForVmfsWithEagerZeroedThick(t *testing.T) {
+	ctx := context.Background()
+
+	tests := []struct {
+		name        string
+		policy      vsphere.SpbmPolicyContent
+		expectError bool
+	}{
+		{
+			name: "VMFS eager zeroed",
+			policy: makePolicy(
+				makeProfile(makeRule(vmfsNamespace, vmfsNamespaceEztValue)),
+			),
+			expectError: false,
+		},
+		{
+			name: "VMFS but not eager zeroed",
+			policy: makePolicy(
+				makeProfile(makeRule(vmfsNamespace, "THIN")),
+			),
+			expectError: true,
+		},
+		{
+			name: "Non-VMFS policy",
+			policy: makePolicy(
+				makeProfile(makeRule("NFS", "ANY")),
+			),
+			expectError: false,
+		},
+		{
+			name:        "Empty profiles",
+			policy:      makePolicy(),
+			expectError: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := verifyStoragePolicyForVmfsWithEagerZeroedThick(ctx, tt.policy, "test-policy")
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "It must be Thick Provision Eager Zero for RWX block volumes.")
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}
+
+func makeRule(ns, value string) vsphere.SpbmPolicyRule {
+	return vsphere.SpbmPolicyRule{Ns: ns, Value: value}
+}
+
+func makeProfile(rules ...vsphere.SpbmPolicyRule) vsphere.SpbmPolicySubProfile {
+	return vsphere.SpbmPolicySubProfile{Rules: rules}
+}
+
+func makePolicy(profiles ...vsphere.SpbmPolicySubProfile) vsphere.SpbmPolicyContent {
+	return vsphere.SpbmPolicyContent{Profiles: profiles}
+}

pkg/csi/service/wcp/controller.go

@@ -522,6 +522,16 @@ func (c *controller) createBlockVolume(ctx context.Context, req *csi.CreateVolum
 			"failed to get vCenter from Manager. Error: %v", err)
 	}
 
+	if commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
+		common.SharedDiskFss) && isSharedRawBlockRequest(ctx, req.VolumeCapabilities) {
+		log.Infof("Volume request is for shared RWX volume. Validatig if policy is compatible for VMFS datastores.")
+		err := common.ValidateStoragePolicyForRWXVolume(ctx, vc, storagePolicyID)
+		if err != nil {
+			log.Errorf("failed validation for policy %s", storagePolicyID)
+			return nil, csifault.CSIInternalFault, err
+		}
+	}
+
 	// Fetch the accessibility requirements from the request.
 	topologyRequirement = req.GetAccessibilityRequirements()
 	filterSuspendedDatastores := commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.CnsMgrSuspendCreateVolume)

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

@@ -432,6 +432,25 @@ func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 		return reconcile.Result{RequeueAfter: timeout}, nil
 	}
 
+	accessMode := instance.Spec.AccessMode
+	// If accessMode is not provided, set it to the default value - ReadWriteOnce.
+	if accessMode == "" {
+		accessMode = v1.ReadWriteOnce
+	}
+
+	// Here AccessMode ReadWriteMany indicates that it is a shared block volume.
+	// Validation that it does not indicate a file volume has already been done as part of
+	// isBlockVolumeRegisterRequest.
+	if isSharedDiskEnabled && accessMode == v1.ReadWriteMany {
+		log.Infof("Volume request is for shared RWX volume. Validatig if policy is compatible for VMFS datastores.")
+		err := common.ValidateStoragePolicyForRWXVolume(ctx, vc, volume.StoragePolicyId)
+		if err != nil {
+			log.Errorf("failed validation for policy %s. Err: %s", volume.StoragePolicyId, err)
+			setInstanceError(ctx, r, instance, err.Error())
+			return reconcile.Result{RequeueAfter: timeout}, nil
+		}
+	}
+
 	// Get K8S storageclass name mapping the storagepolicy id with Immediate volume binding mode
 	storageClassName, err := getK8sStorageClassNameWithImmediateBindingModeForPolicy(ctx, k8sclient, r.client,
 		volume.StoragePolicyId, request.Namespace, syncer.IsPodVMOnStretchSupervisorFSSEnabled)
@@ -584,11 +603,6 @@ func (r *ReconcileCnsRegisterVolume) Reconcile(ctx context.Context,
 	}
 
 	capacityInMb := volume.BackingObjectDetails.GetCnsBackingObjectDetails().CapacityInMb
-	accessMode := instance.Spec.AccessMode
-	// Set accessMode to ReadWriteOnce if DiskURLPath is used for import.
-	if accessMode == "" && instance.Spec.DiskURLPath != "" {
-		accessMode = v1.ReadWriteOnce
-	}
 	pv, err := k8sclient.CoreV1().PersistentVolumes().Get(ctx, pvName, metav1.GetOptions{})
 	if err != nil {
 		if apierrors.IsNotFound(err) {