create volume with transaction support (#3350)

Author: divyenpatel

go.mod

@@ -25,7 +25,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	github.com/vmware-tanzu/vm-operator/api v1.8.7-0.20250509154507-b93e51fc90fa
 	github.com/vmware-tanzu/vm-operator/external/byok v0.0.0-20250509154507-b93e51fc90fa
-	github.com/vmware/govmomi v0.51.0
+	github.com/vmware/govmomi v0.52.0-alpha.0.0.20250624230355-39c6f0f94465
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.36.0
 	golang.org/x/sync v0.14.0

go.sum

@@ -293,8 +293,8 @@ github.com/vmware-tanzu/vm-operator/api v1.8.7-0.20250509154507-b93e51fc90fa h1:
 github.com/vmware-tanzu/vm-operator/api v1.8.7-0.20250509154507-b93e51fc90fa/go.mod h1:V0JbH4beGCU+q7yqnWUYYOuDij0ut5i1iBO4cyzg+tM=
 github.com/vmware-tanzu/vm-operator/external/byok v0.0.0-20250509154507-b93e51fc90fa h1:4MKu14YJ7J54O6QKmT4ds5EUpysWLLtQRMff73cVkmU=
 github.com/vmware-tanzu/vm-operator/external/byok v0.0.0-20250509154507-b93e51fc90fa/go.mod h1:8tiuyYslzjLIUmOlXZuGKQdQP2ZgWGCVhVeyptmZYnk=
-github.com/vmware/govmomi v0.51.0 h1:n3RLS9aw/irTOKbiIyJzAb6rOat4YOVv/uDoRsNTSQI=
-github.com/vmware/govmomi v0.51.0/go.mod h1:3ywivawGRfMP2SDCeyKqxTl2xNIHTXF0ilvp72dot5A=
+github.com/vmware/govmomi v0.52.0-alpha.0.0.20250624230355-39c6f0f94465 h1:BV3/IrRGRMmGWra0T4bK6xpWePsCiwLPOWns7SFREFI=
+github.com/vmware/govmomi v0.52.0-alpha.0.0.20250624230355-39c6f0f94465/go.mod h1:3ywivawGRfMP2SDCeyKqxTl2xNIHTXF0ilvp72dot5A=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/xiang90/probing v0.0.0-20221125231312-a49e3df8f510 h1:S2dVYn90KE98chqDkyE9Z4N61UnQd+KOfgp5Iu53llk=

manifests/supervisorcluster/1.29/cns-csi.yaml

@@ -559,6 +559,7 @@ data:
   "workload-domain-isolation": "false"
   "WCP_VMService_BYOK": "false"
   "file-volume-with-vm-service": "false"
+  "csi-transaction-support": "false"
 kind: ConfigMap
 metadata:
   name: csi-feature-states

manifests/supervisorcluster/1.30/cns-csi.yaml

@@ -567,6 +567,7 @@ data:
   "WCP_VMService_BYOK": "false"
   "sv-pvc-snapshot-protection-finalizer": "false"
   "file-volume-with-vm-service": "false"
+  "csi-transaction-support": "false"
 kind: ConfigMap
 metadata:
   name: csi-feature-states

manifests/supervisorcluster/1.31/cns-csi.yaml

@@ -567,6 +567,7 @@ data:
   "WCP_VMService_BYOK": "false"
   "sv-pvc-snapshot-protection-finalizer": "false"
   "file-volume-with-vm-service": "false"
+  "csi-transaction-support": "false"
 kind: ConfigMap
 metadata:
   name: csi-feature-states

manifests/vanilla/vsphere-csi-driver.yaml

@@ -150,6 +150,7 @@ apiVersion: v1
 data:
   "trigger-csi-fullsync": "false"
   "pv-to-backingdiskobjectid-mapping": "false"
+  "csi-transaction-support": "false"
 kind: ConfigMap
 metadata:
   name: internal-feature-states.csi.vsphere.vmware.com

pkg/common/cns-lib/volume/manager.go

@@ -386,6 +386,7 @@ func (m *defaultManager) MonitorCreateVolumeTask(ctx context.Context,
 		vCenterServerForVolumeOperationCR string
 	)
 	log := logger.GetLogger(ctx)
+	log.Debugf("Invoked MonitorCreateVolumeTask for task: %v", task.Reference())
 	if m.multivCenterTopologyDeployment {
 		vCenterServerForVolumeOperationCR = m.virtualCenter.Config.Host
 	}
@@ -467,21 +468,33 @@ func (m *defaultManager) MonitorCreateVolumeTask(ctx context.Context,
 	}
 	volumeOperationRes := taskResult.GetCnsVolumeOperationResult()
 	if volumeOperationRes.Fault != nil {
-		// Validate if the volume is already registered.
-		faultType = ExtractFaultTypeFromVolumeResponseResult(ctx, volumeOperationRes)
-		resp, err := validateCreateVolumeResponseFault(ctx, volNameFromInputSpec, volumeOperationRes)
+		if IsNotSupportedFault(ctx, volumeOperationRes.Fault) {
+			faultType = "vim25:NotSupported"
+			err = fmt.Errorf("failed to create volume with fault: %q", faultType)
+		} else if _, ok := volumeOperationRes.Fault.Fault.(*cnstypes.CnsVolumeAlreadyExistsFault); ok {
+			faultType = "vim.fault.CnsVolumeAlreadyExistsFault"
+			err = fmt.Errorf("failed to create volume with fault: %q", faultType)
+		} else {
+			// Validate if the volume is already registered.
+			faultType = ExtractFaultTypeFromVolumeResponseResult(ctx, volumeOperationRes)
+			resp, err := validateCreateVolumeResponseFault(ctx, volNameFromInputSpec, volumeOperationRes)
+			if err == nil {
+				*volumeOperationDetails = createRequestDetails(volNameFromInputSpec, resp.VolumeID.Id, "", 0,
+					(*volumeOperationDetails).QuotaDetails,
+					(*volumeOperationDetails).OperationDetails.TaskInvocationTimestamp, task.Reference().Value,
+					vCenterServerForVolumeOperationCR, taskInfo.ActivationId, taskInvocationStatusSuccess, "")
+				return resp, faultType, err
+			}
+		}
 		if err != nil {
+			log.Errorf("err:%v observed for task: %v, volume name: %q, OpID: %q",
+				err.Error(), task.Reference().Value, volNameFromInputSpec, taskInfo.ActivationId)
 			*volumeOperationDetails = createRequestDetails(volNameFromInputSpec, "", "", 0,
 				(*volumeOperationDetails).QuotaDetails, (*volumeOperationDetails).OperationDetails.TaskInvocationTimestamp,
 				task.Reference().Value, vCenterServerForVolumeOperationCR, taskInfo.ActivationId,
 				taskInvocationStatusError, err.Error())
-		} else {
-			*volumeOperationDetails = createRequestDetails(volNameFromInputSpec, resp.VolumeID.Id, "", 0,
-				(*volumeOperationDetails).QuotaDetails,
-				(*volumeOperationDetails).OperationDetails.TaskInvocationTimestamp, task.Reference().Value,
-				vCenterServerForVolumeOperationCR, taskInfo.ActivationId, taskInvocationStatusSuccess, "")
+			return nil, faultType, err
 		}
-		return resp, faultType, err
 	}
 	// Extract the CnsVolumeInfo from the taskResult.
 	resp, faultType, err := getCnsVolumeInfoFromTaskResult(ctx, m.virtualCenter, volNameFromInputSpec,
@@ -658,6 +671,125 @@ func (m *defaultManager) createVolumeWithImprovedIdempotency(ctx context.Context
 		spec.Metadata.ContainerClusterArray[0].ClusterId)
 }
 
+// createVolumeWithTransaction creates volume with supplied PVC UUID as VolumeID in the CreateVolumeSpec
+func (m *defaultManager) createVolumeWithTransaction(ctx context.Context, spec *cnstypes.CnsVolumeCreateSpec,
+	extraParams interface{}) (resp *CnsVolumeInfo, faultType string, finalErr error) {
+	log := logger.GetLogger(ctx)
+	var (
+		// Store the volume name passed in by input spec, this
+		// name may exceed 80 characters.
+		volNameFromInputSpec = spec.Name
+		// Reference to the CreateVolume task on CNS.
+		task *object.Task
+		// Local instance of CreateVolume details that needs to
+		// be persisted.
+		volumeOperationDetails *cnsvolumeoperationrequest.VolumeOperationRequestDetails
+		// quotaInfo consists of values required to populate QuotaDetails in CnsVolumeOperationRequest CR.
+		quotaInfo                            *cnsvolumeoperationrequest.QuotaDetails
+		isPodVMOnStretchSupervisorFSSEnabled bool
+	)
+
+	if extraParams != nil {
+		createVolParams, ok := extraParams.(*CreateVolumeExtraParams)
+		if !ok {
+			return nil, csifault.CSIInternalFault,
+				logger.LogNewErrorf(log, "unrecognised type for CreateVolume params: %+v", extraParams)
+		}
+		log.Debugf("Received CreateVolume extraParams: %+v", *createVolParams)
+
+		isPodVMOnStretchSupervisorFSSEnabled = createVolParams.IsPodVMOnStretchSupervisorFSSEnabled
+		if isPodVMOnStretchSupervisorFSSEnabled && m.clusterFlavor == cnstypes.CnsClusterFlavorWorkload {
+			var storagePolicyID string
+			if len(spec.Profile) >= 1 {
+				storagePolicyID = spec.Profile[0].(*vim25types.VirtualMachineDefinedProfileSpec).ProfileId
+			}
+			reservedQty := resource.NewQuantity(createVolParams.VolSizeBytes, resource.BinarySI)
+			quotaInfo = &cnsvolumeoperationrequest.QuotaDetails{
+				Reserved:         reservedQty,
+				StoragePolicyId:  storagePolicyID,
+				StorageClassName: createVolParams.StorageClassName,
+				Namespace:        createVolParams.Namespace,
+			}
+			log.Infof("QuotaInfo during CreateVolume call: %+v", *quotaInfo)
+		}
+	}
+
+	if m.operationStore == nil {
+		return nil, csifault.CSIInternalFault, logger.LogNewError(log, "operation store cannot be nil")
+	}
+	var vCenterServerForVolumeOperationCR string
+	if m.multivCenterTopologyDeployment {
+		vCenterServerForVolumeOperationCR = m.virtualCenter.Config.Host
+	}
+
+	volumeOperationDetails, finalErr = m.operationStore.GetRequestDetails(ctx, volNameFromInputSpec)
+	if !apierrors.IsNotFound(finalErr) {
+		return nil, csifault.CSIInternalFault, finalErr
+	}
+	defer func() {
+		// Persist the operation details before returning. Only success or error
+		// needs to be stored as InProgress details are stored when the task is
+		// created on CNS.
+		if volumeOperationDetails != nil && volumeOperationDetails.OperationDetails != nil &&
+			volumeOperationDetails.OperationDetails.TaskStatus != taskInvocationStatusInProgress {
+
+			if m.clusterFlavor == cnstypes.CnsClusterFlavorWorkload && isPodVMOnStretchSupervisorFSSEnabled {
+				// Decrease the reserved field in QuotaDetails when the CreateVolume task is
+				// successful or has errored out.
+				taskStatus := volumeOperationDetails.OperationDetails.TaskStatus
+				if (taskStatus == taskInvocationStatusSuccess || taskStatus == taskInvocationStatusError) &&
+					volumeOperationDetails.QuotaDetails != nil {
+					volumeOperationDetails.QuotaDetails.Reserved = resource.NewQuantity(0,
+						resource.BinarySI)
+					log.Infof("Setting the reserved field for VolumeOperationDetails instance %s to 0",
+						volumeOperationDetails.Name)
+				}
+				tempErr := m.operationStore.StoreRequestDetails(ctx, volumeOperationDetails)
+				if finalErr == nil && tempErr != nil {
+					log.Errorf("failed to store CreateVolume details with error: %v", tempErr)
+					finalErr = tempErr
+				}
+			} else {
+				err := m.operationStore.StoreRequestDetails(ctx, volumeOperationDetails)
+				if err != nil {
+					log.Warnf("failed to store CreateVolume details with error: %v", err)
+				}
+			}
+		}
+	}()
+	volumeOperationDetails = createRequestDetails(volNameFromInputSpec, "", "", 0,
+		quotaInfo, metav1.Now(), "", vCenterServerForVolumeOperationCR, "",
+		taskInvocationStatusInProgress, "")
+	err := m.operationStore.StoreRequestDetails(ctx, volumeOperationDetails)
+	if err != nil {
+		// Don't return if CreateVolume details can't be stored.
+		log.Warnf("failed to store CreateVolume details with error: %v", err)
+	}
+	task, finalErr = invokeCNSCreateVolume(ctx, m.virtualCenter, spec)
+	if finalErr != nil {
+		log.Errorf("failed to create volume with error: %v", finalErr)
+		volumeOperationDetails = createRequestDetails(volNameFromInputSpec, "", "", 0,
+			quotaInfo, volumeOperationDetails.OperationDetails.TaskInvocationTimestamp, "",
+			vCenterServerForVolumeOperationCR, "", taskInvocationStatusError, finalErr.Error())
+		faultType = ExtractFaultTypeFromErr(ctx, finalErr)
+		return nil, faultType, finalErr
+	}
+	if !isStaticallyProvisioned(spec) {
+		// Persist task details only for dynamically provisioned volumes.
+		volumeOperationDetails = createRequestDetails(volNameFromInputSpec, "", "", 0,
+			quotaInfo, metav1.Now(), task.Reference().Value, vCenterServerForVolumeOperationCR, "",
+			taskInvocationStatusInProgress, "")
+		err := m.operationStore.StoreRequestDetails(ctx, volumeOperationDetails)
+		if err != nil {
+			// Don't return if CreateVolume details can't be stored.
+			log.Warnf("failed to store CreateVolume details with error: %v", err)
+		}
+	}
+
+	return m.MonitorCreateVolumeTask(ctx, &volumeOperationDetails, task, volNameFromInputSpec,
+		spec.Metadata.ContainerClusterArray[0].ClusterId)
+}
+
 // IsTaskPending returns true in two cases -
 // 1. if the task status was in progress
 // 2. if the status was an error but the error was for adding the task to the listview
@@ -853,7 +985,29 @@ func (m *defaultManager) CreateVolume(ctx context.Context, spec *cnstypes.CnsVol
 		}
 		// Call CreateVolume implementation based on FSS value.
 		if m.idempotencyHandlingEnabled {
-			return m.createVolumeWithImprovedIdempotency(ctx, spec, extraParams)
+			if spec.VolumeId != nil && spec.VolumeId.Id != "" {
+				var cnsVolumeInfo *CnsVolumeInfo
+				cnsVolumeInfo, faultType, err = m.createVolumeWithTransaction(ctx, spec, extraParams)
+				if err != nil {
+					log.Errorf("failed to create volume with VolumeID: %q, faultType: %q, err: %v",
+						spec.VolumeId.Id, faultType, err)
+					if IsCnsVolumeAlreadyExistsFault(ctx, faultType) {
+						log.Infof("Observed volume with Id: %q is already Exists. Deleting Volume.", spec.VolumeId.Id)
+						deleteFaultType, deleteError := m.deleteVolume(ctx, spec.VolumeId.Id, true)
+						if deleteError != nil {
+							log.Errorf("failed to delete volume: %q to handle CnsVolumeAlreadyExistsFault , err :%v", spec.VolumeId.Id, err)
+							return nil, deleteFaultType, deleteError
+						}
+						log.Infof("Attempt to re-create volume with Id: %q", spec.VolumeId.Id)
+						cnsVolumeInfo, faultType, err = m.createVolumeWithTransaction(ctx, spec, extraParams)
+						return cnsVolumeInfo, faultType, err
+					}
+					return nil, faultType, err
+				}
+				return cnsVolumeInfo, faultType, err
+			} else {
+				return m.createVolumeWithImprovedIdempotency(ctx, spec, extraParams)
+			}
 		}
 		return m.createVolume(ctx, spec)
 	}
@@ -1854,7 +2008,7 @@ func (m *defaultManager) expandVolumeWithImprovedIdempotency(ctx context.Context
 
 	volumeOperationRes := taskResult.GetCnsVolumeOperationResult()
 	if volumeOperationRes.Fault != nil {
-		if _, ok := volumeOperationRes.Fault.Fault.(cnstypes.CnsFault); ok {
+		if _, ok := volumeOperationRes.Fault.Fault.(*cnstypes.CnsFault); ok {
 			log.Debugf("ExtendVolume task %s returned with CnsFault. Querying CNS to "+
 				"determine if volume with ID %s was successfully expanded.",
 				task.Reference().Value, volumeID)
@@ -2562,7 +2716,7 @@ func (m *defaultManager) createSnapshotWithImprovedIdempotencyCheck(ctx context.
 				errMsg = fmt.Sprintf("snapshot %q on volume %q got created, but post-processing failed. "+
 					"Fault: %q, opID: %q", instanceName, volumeID, spew.Sdump(createSnapshotsOperationRes.Fault),
 					createSnapshotsTaskInfo.ActivationId)
-				snapshotId := createSnapshotsOperationRes.Fault.Fault.(cnstypes.CnsSnapshotCreatedFault).SnapshotId.Id
+				snapshotId := createSnapshotsOperationRes.Fault.Fault.(*cnstypes.CnsSnapshotCreatedFault).SnapshotId.Id
 				volumeOperationDetails = createRequestDetails(instanceName, volumeID, snapshotId, 0, nil,
 					volumeOperationDetails.OperationDetails.TaskInvocationTimestamp, createSnapshotsTask.Reference().Value,
 					"", createSnapshotsTaskInfo.ActivationId, taskInvocationStatusPartiallyFailed, errMsg)

pkg/common/cns-lib/volume/util.go

@@ -284,7 +284,7 @@ func getTaskResultFromTaskInfo(ctx context.Context, taskInfo *types.TaskInfo) (c
 func validateCreateVolumeResponseFault(ctx context.Context, name string,
 	resp *cnstypes.CnsVolumeOperationResult) (*CnsVolumeInfo, error) {
 	log := logger.GetLogger(ctx)
-	fault, ok := resp.Fault.Fault.(cnstypes.CnsAlreadyRegisteredFault)
+	fault, ok := resp.Fault.Fault.(*cnstypes.CnsAlreadyRegisteredFault)
 	if ok {
 		log.Infof("Volume is already registered with CNS. VolumeName: %q, volumeID: %q",
 			name, fault.VolumeId.Id)
@@ -374,11 +374,13 @@ func ExtractFaultTypeFromVolumeResponseResult(ctx context.Context,
 				faultType, fault.Fault, fault, resp)
 			slice := strings.Split(faultType, ".")
 			vimFaultType := csifault.VimFaultPrefix + slice[1]
+			log.Infof("returning fault: %q", vimFaultType)
 			return vimFaultType
 		} else {
 			faultType = reflect.TypeOf(fault).String()
 			log.Infof("Extract fault: %q from resp: %+v",
 				faultType, resp)
+			log.Infof("returning fault: %q", faultType)
 			return faultType
 		}
 	}
@@ -557,3 +559,37 @@ func IsNotFoundFault(ctx context.Context, faultType string) bool {
 	return faultType == "vim.fault.NotFound"
 
 }
+
+// IsNotSupportedFault returns true if a given fault is NotSupported fault
+func IsNotSupportedFault(ctx context.Context, fault *types.LocalizedMethodFault) bool {
+	log := logger.GetLogger(ctx)
+	if cnsFault, ok := fault.Fault.(*cnstypes.CnsFault); ok {
+		if cause := cnsFault.FaultCause; cause != nil {
+			if innerfault, ok := cause.Fault.(*types.NotSupported); ok {
+				log.Info("observed NotSupported fault")
+				return true
+			} else {
+				log.Infof("observed fault: %T", innerfault)
+				return false
+			}
+		} else {
+			log.Errorf("observed fault with nil cause")
+		}
+	} else {
+		log.Errorf("can not typecast fault to CnsFault")
+	}
+	return false
+}
+
+func IsNotSupportedFaultType(ctx context.Context, faultType string) bool {
+	log := logger.GetLogger(ctx)
+	log.Infof("Checking fault type: %q is vim25:NotSupported", faultType)
+	return faultType == "vim25:NotSupported"
+}
+
+// IsCnsVolumeAlreadyExistsFault returns true if a given faultType value is vim.fault.CnsVolumeAlreadyExistsFault
+func IsCnsVolumeAlreadyExistsFault(ctx context.Context, faultType string) bool {
+	log := logger.GetLogger(ctx)
+	log.Infof("Checking fault type: %q is vim.fault.CnsVolumeAlreadyExistsFault", faultType)
+	return faultType == "vim.fault.CnsVolumeAlreadyExistsFault"
+}

pkg/common/cns-lib/vsphere/utils.go

@@ -97,7 +97,7 @@ func IsVimFaultNotFoundError(err error) bool {
 func IsCnsSnapshotCreatedFaultError(err error) bool {
 	isCnsSnapshotCreatedFaultError := false
 	if soap.IsVimFault(err) {
-		_, isCnsSnapshotCreatedFaultError = soap.ToVimFault(err).(cnstypes.CnsSnapshotCreatedFault)
+		_, isCnsSnapshotCreatedFaultError = soap.ToVimFault(err).(*cnstypes.CnsSnapshotCreatedFault)
 	}
 	return isCnsSnapshotCreatedFaultError
 }
@@ -106,7 +106,7 @@ func IsCnsSnapshotCreatedFaultError(err error) bool {
 func IsCnsSnapshotNotFoundError(err error) bool {
 	isCnsSnapshotNotFoundError := false
 	if soap.IsVimFault(err) {
-		_, isCnsSnapshotNotFoundError = soap.ToVimFault(err).(cnstypes.CnsSnapshotNotFoundFault)
+		_, isCnsSnapshotNotFoundError = soap.ToVimFault(err).(*cnstypes.CnsSnapshotNotFoundFault)
 	}
 	return isCnsSnapshotNotFoundError
 }

pkg/csi/service/common/constants.go

@@ -440,6 +440,8 @@ const (
 	FileVolumesWithVmService = "file-volume-with-vm-service"
 	// SharedDiskFss is an FSS that tells whether shared disks are supported or not
 	SharedDiskFss = "supports_shared_disks"
+	// CSITranSactionSupport is an FSS for transaction support
+	CSITranSactionSupport = "csi-transaction-support"
 )
 
 var WCPFeatureStates = map[string]struct{}{