From 7e0ce640df19483ed8954077104de655b167494c Mon Sep 17 00:00:00 2001 From: Saloni Kogta Date: Thu, 6 Nov 2025 13:58:42 +0530 Subject: [PATCH] Block RWX volume creation for VMFS datastores if policy is not EZT --- pkg/csi/service/common/util.go | 65 +++++ pkg/csi/service/common/util_test.go | 104 +++++++ pkg/csi/service/wcp/controller.go | 10 + pkg/csi/service/wcp/controller_test.go | 261 ++++++++++++++++++ .../cnsregistervolume_controller.go | 24 +- 5 files changed, 459 insertions(+), 5 deletions(-) diff --git a/pkg/csi/service/common/util.go b/pkg/csi/service/common/util.go index c5affe97c2..344fd421b5 100644 --- a/pkg/csi/service/common/util.go +++ b/pkg/csi/service/common/util.go @@ -45,6 +45,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 +537,66 @@ 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 *cnsvsphere.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 cnsvsphere.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 { + return fmt.Errorf( + "policy %s is for VMFS datastores. It must be Thick Provision Eager Zero for RWX block volumes", + storagePolicyID) + } + 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 cnsvsphere.SpbmPolicySubProfile) (bool, bool) { + for _, rule := range profile.Rules { + if rule.Ns == vmfsNamespace { + return true, rule.Value == vmfsNamespaceEztValue + } + } + return false, false +} diff --git a/pkg/csi/service/common/util_test.go b/pkg/csi/service/common/util_test.go index ace87bdbdc..d01ffff3fd 100644 --- a/pkg/csi/service/common/util_test.go +++ b/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} +} diff --git a/pkg/csi/service/wcp/controller.go b/pkg/csi/service/wcp/controller.go index 17ea1298eb..a7c08b2220 100644 --- a/pkg/csi/service/wcp/controller.go +++ b/pkg/csi/service/wcp/controller.go @@ -527,6 +527,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) diff --git a/pkg/csi/service/wcp/controller_test.go b/pkg/csi/service/wcp/controller_test.go index 679ab5df90..7eb524e305 100644 --- a/pkg/csi/service/wcp/controller_test.go +++ b/pkg/csi/service/wcp/controller_test.go @@ -30,6 +30,8 @@ import ( "github.com/google/uuid" "github.com/vmware/govmomi/find" "github.com/vmware/govmomi/pbm" + pbmtypes "github.com/vmware/govmomi/pbm/types" + v1 "k8s.io/api/core/v1" cnstypes "github.com/vmware/govmomi/cns/types" @@ -257,6 +259,265 @@ func TestWCPCreateVolumeWithStoragePolicy(t *testing.T) { } } +func TestWCPCreateRWXVolumeWithVMFSNonEZTStoragePolicy(t *testing.T) { + ct := getControllerTest(t) + ctx := context.Background() + + params := make(map[string]string) + + profileID := os.Getenv("VSPHERE_STORAGE_POLICY_ID") + if profileID == "" { + storagePolicyName := os.Getenv("VSPHERE_STORAGE_POLICY_NAME") + if storagePolicyName == "" { + // Use our custom VMFS policy for PBM simulator + storagePolicyName = "VMFS Storage Policy" + } + + pc, err := pbm.NewClient(ctx, ct.vcenter.Client.Client) + if err != nil { + t.Fatal(err) + } + + // Check if the simulator already has it + profileID, err = pc.ProfileIDByName(ctx, storagePolicyName) + if err != nil { + // Create a fake VMFS policy in PBM simulator + spec := pbmtypes.PbmCapabilityProfileCreateSpec{ + Name: storagePolicyName, + Description: "Simulated VMFS storage policy", + ResourceType: pbmtypes.PbmProfileResourceType{ResourceType: "STORAGE"}, + Constraints: &pbmtypes.PbmCapabilitySubProfileConstraints{ + SubProfiles: []pbmtypes.PbmCapabilitySubProfile{ + { + Name: "VMFS-SubProfile", + Capability: []pbmtypes.PbmCapabilityInstance{ + { + Id: pbmtypes.PbmCapabilityMetadataUniqueId{ + Id: "vmfs.eagerzeroedthick", + Namespace: "com.vmware.storage.volumeallocation", + }, + Constraint: []pbmtypes.PbmCapabilityConstraintInstance{ + { + PropertyInstance: []pbmtypes.PbmCapabilityPropertyInstance{ + { + Id: "property-id", // must be string + Value: "testval", + }, + }, + }, + }, + }, + }, + }, + }, + }, + } + + id, err := pc.CreateProfile(ctx, spec) + if err != nil { + t.Fatalf("failed to create VMFS policy: %v", err) + } + profileID = id.UniqueId + } + } + + params[common.AttributeStoragePolicyID] = profileID + + reqCreate := &csi.CreateVolumeRequest{ + Name: testVolumeName + "-" + uuid.New().String(), + CapacityRange: &csi.CapacityRange{ + RequiredBytes: 1 * common.GbInBytes, + }, + Parameters: params, + VolumeCapabilities: []*csi.VolumeCapability{ + { + AccessMode: &csi.VolumeCapability_AccessMode{ + Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER, + }, + AccessType: &csi.VolumeCapability_Block{ + Block: &csi.VolumeCapability_BlockVolume{}, + }, + }, + }, + AccessibilityRequirements: &csi.TopologyRequirement{}, + } + + err := commonco.ContainerOrchestratorUtility.EnableFSS(ctx, "supports_shared_disks_with_VM_service_VMs") + if err != nil { + t.Fatal("failed to enable Workload_Domain_Isolation_Supported FSS") + } + + defer func() { + err := commonco.ContainerOrchestratorUtility.DisableFSS(ctx, "supports_shared_disks_with_VM_service_VMs") + if err != nil { + t.Fatal("failed to disable Workload_Domain_Isolation_Supported FSS") + } + }() + + // Create volume + _, err = ct.controller.CreateVolume(ctx, reqCreate) + if err == nil { + t.Fatalf("Error Expected") + } + + expectedSubstring := "It must be Thick Provision Eager Zero for RWX block volumes" + if !strings.Contains(err.Error(), expectedSubstring) { + t.Fatalf("expected error to contain %q, got %q", expectedSubstring, err.Error()) + } +} + +func TestWCPCreateRWXVolumeWithVMFSEZTStoragePolicy(t *testing.T) { + ct := getControllerTest(t) + ctx := context.Background() + + params := make(map[string]string) + + profileID := os.Getenv("VSPHERE_STORAGE_POLICY_ID") + if profileID == "" { + storagePolicyName := os.Getenv("VSPHERE_STORAGE_POLICY_NAME") + if storagePolicyName == "" { + // Use our custom VMFS policy for PBM simulator + storagePolicyName = "VMFS Storage Policy" + } + + pc, err := pbm.NewClient(ctx, ct.vcenter.Client.Client) + if err != nil { + t.Fatal(err) + } + + // Check if the simulator already has it + profileID, err = pc.ProfileIDByName(ctx, storagePolicyName) + if err != nil { + // Create a fake VMFS policy in PBM simulator + spec := pbmtypes.PbmCapabilityProfileCreateSpec{ + Name: storagePolicyName, + Description: "Simulated VMFS storage policy", + ResourceType: pbmtypes.PbmProfileResourceType{ResourceType: "STORAGE"}, + Constraints: &pbmtypes.PbmCapabilitySubProfileConstraints{ + SubProfiles: []pbmtypes.PbmCapabilitySubProfile{ + { + Name: "VMFS-SubProfile", + Capability: []pbmtypes.PbmCapabilityInstance{ + { + Id: pbmtypes.PbmCapabilityMetadataUniqueId{ + Id: "vmfs.eagerzeroedthick", + Namespace: "com.vmware.storage.volumeallocation", + }, + Constraint: []pbmtypes.PbmCapabilityConstraintInstance{ + { + PropertyInstance: []pbmtypes.PbmCapabilityPropertyInstance{ + { + Id: "property-id", + Value: "Fully initialized", + }, + }, + }, + }, + }, + }, + }, + }, + }, + } + + id, err := pc.CreateProfile(ctx, spec) + if err != nil { + t.Fatalf("failed to create VMFS policy: %v", err) + } + profileID = id.UniqueId + } + } + + params[common.AttributeStoragePolicyID] = profileID + + reqCreate := &csi.CreateVolumeRequest{ + Name: testVolumeName + "-" + uuid.New().String(), + CapacityRange: &csi.CapacityRange{ + RequiredBytes: 1 * common.GbInBytes, + }, + Parameters: params, + VolumeCapabilities: []*csi.VolumeCapability{ + { + AccessMode: &csi.VolumeCapability_AccessMode{ + Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER, + }, + AccessType: &csi.VolumeCapability_Block{ + Block: &csi.VolumeCapability_BlockVolume{}, + }, + }, + }, + AccessibilityRequirements: &csi.TopologyRequirement{}, + } + + err := commonco.ContainerOrchestratorUtility.EnableFSS(ctx, "supports_shared_disks_with_VM_service_VMs") + if err != nil { + t.Fatal("failed to enable Workload_Domain_Isolation_Supported FSS") + } + + // Create volume + respCreate, err := ct.controller.CreateVolume(ctx, reqCreate) + if err != nil { + t.Fatal(err) + } + volID := respCreate.Volume.VolumeId + queryFilter := cnstypes.CnsQueryFilter{ + VolumeIds: []cnstypes.CnsVolumeId{ + { + Id: volID, + }, + }, + } + queryResult, err := ct.vcenter.CnsClient.QueryVolume(ctx, &queryFilter) + if err != nil { + t.Fatal(err) + } + if len(queryResult.Volumes) != 1 && queryResult.Volumes[0].VolumeId.Id != volID { + t.Fatalf("failed to find the newly created volume with ID: %s", volID) + } + + if queryResult.Volumes[0].StoragePolicyId != profileID { + t.Fatalf("failed to match volume policy ID: %s", profileID) + } + + // QueryAll. + queryFilter = cnstypes.CnsQueryFilter{ + VolumeIds: []cnstypes.CnsVolumeId{ + { + Id: volID, + }, + }, + } + querySelection := cnstypes.CnsQuerySelection{} + queryResult, err = ct.vcenter.CnsClient.QueryAllVolume(ctx, queryFilter, querySelection) + if err != nil { + t.Fatal(err) + } + + if len(queryResult.Volumes) != 1 && queryResult.Volumes[0].VolumeId.Id != volID { + t.Fatalf("failed to find the newly created volume with ID: %s", volID) + } + + // Delete. + reqDelete := &csi.DeleteVolumeRequest{ + VolumeId: volID, + } + _, err = ct.controller.DeleteVolume(ctx, reqDelete) + if err != nil { + t.Fatal(err) + } + + // Varify the volume has been deleted. + queryResult, err = ct.vcenter.CnsClient.QueryVolume(ctx, &queryFilter) + if err != nil { + t.Fatal(err) + } + + if len(queryResult.Volumes) != 0 { + t.Fatalf("Volume should not exist after deletion with ID: %s", volID) + } + +} + // TestWCPCreateVolumeWithZonalLabelPresentButNoStorageTopoType creates volume with zonal label present // but not storage topology type. It is a negative case. func TestWCPCreateVolumeWithZonalLabelPresentButNoStorageTopoType(t *testing.T) { diff --git a/pkg/syncer/cnsoperator/controller/cnsregistervolume/cnsregistervolume_controller.go b/pkg/syncer/cnsoperator/controller/cnsregistervolume/cnsregistervolume_controller.go index f4b5c0e897..970862eeab 100644 --- a/pkg/syncer/cnsoperator/controller/cnsregistervolume/cnsregistervolume_controller.go +++ b/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) {