Fix CnsVolumeOperationRequest update logic when transaction support is enabled (#3623)

Signed-off-by: Deepak Kinni <[email protected]>

Author: deepakkinni

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

@@ -2910,6 +2910,14 @@ func (m *defaultManager) createSnapshotWithTransaction(ctx context.Context, volu
 		faultType := ExtractFaultTypeFromErr(ctx, err)
 		return nil, faultType, logger.LogNewErrorf(log, "failed to create snapshot with error: %v", err)
 	}
+	// Persist the volume operation details.
+	volumeOperationDetails = createRequestDetails(instanceName, volumeID, "", 0, quotaInfo,
+		volumeOperationDetails.OperationDetails.TaskInvocationTimestamp,
+		createSnapshotsTask.Reference().Value, "", "", taskInvocationStatusInProgress, "")
+	if err := m.operationStore.StoreRequestDetails(ctx, volumeOperationDetails); err != nil {
+		// Don't return if CreateSnapshot details can't be stored.
+		log.Warnf("failed to store CreateSnapshot details with error: %v", err)
+	}
 
 	var createSnapshotsTaskInfo *vim25types.TaskInfo
 	var faultType string

pkg/csi/service/vanilla/controller.go

@@ -187,11 +187,13 @@ func (c *controller) Init(config *cnsconfig.Config, version string) error {
 	log.Infof("Initializing CNS controller")
 	var err error
 	var operationStore cnsvolumeoperationrequest.VolumeOperationRequest
+
 	operationStore, err = cnsvolumeoperationrequest.InitVolumeOperationRequestInterface(ctx,
 		config.Global.CnsVolumeOperationRequestCleanupIntervalInMin,
 		func() bool {
 			return commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.BlockVolumeSnapshot)
-		}, false)
+		}, false,
+		false)
 	if err != nil {
 		log.Errorf("failed to initialize VolumeOperationRequestInterface with error: %v", err)
 		return err
@@ -202,7 +204,6 @@ func (c *controller) Init(config *cnsconfig.Config, version string) error {
 		common.CnsMgrSuspendCreateVolume)
 	isTopologyAwareFileVolumeEnabled = commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
 		common.TopologyAwareFileVolume)
-	isCSITransactionSupportEnabled = commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.CSITransactionSupport)
 
 	vcManager := cnsvsphere.GetVirtualCenterManager(ctx)
 	// Multi vCenter feature enabled
@@ -260,7 +261,8 @@ func (c *controller) Init(config *cnsconfig.Config, version string) error {
 	for _, authMgr := range authMgrs {
 		go common.ComputeDatastoreMapForBlockVolumes(authMgr, config.Global.CSIAuthCheckIntervalInMin)
 	}
-	var vsanFileServiceNotSupported bool
+	allVCentersSupportvSANFileService := true
+	allVCentersSupportCnsTransaction := true
 	for _, vcconfig := range c.managers.VcenterConfigs {
 		isvSANFileServicesSupported, err := c.managers.VcenterManager.IsvSANFileServicesSupported(ctx,
 			vcconfig.Host)
@@ -269,13 +271,29 @@ func (c *controller) Init(config *cnsconfig.Config, version string) error {
 				"Error:%+v", vcconfig.Host, err)
 		}
 		if !isvSANFileServicesSupported {
-			vsanFileServiceNotSupported = true
-			break
+			allVCentersSupportvSANFileService = false
+		}
+
+		isCnsTransactionSupported, err := c.managers.VcenterManager.IsCnsTransactionSupported(ctx,
+			vcconfig.Host)
+		if err != nil {
+			return logger.LogNewErrorf(log, "failed to verify if CNS transaction is supported or not for vCenter: %s. "+
+				"Error:%+v", vcconfig.Host, err)
+		}
+		if !isCnsTransactionSupported {
+			allVCentersSupportCnsTransaction = false
 		}
 	}
-	if vsanFileServiceNotSupported {
+	if !allVCentersSupportvSANFileService {
 		return logger.LogNewErrorf(log, "vSAN file service is not supported in one or more vCenter(s)")
 	}
+
+	// CSI Transaction Support requires both FSS to be enabled AND all vCenters to support CNS transactions
+	fssTransactionSupportEnabled := commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
+		common.CSITransactionSupport)
+	isCSITransactionSupportEnabled = fssTransactionSupportEnabled && allVCentersSupportCnsTransaction
+	cnsvolumeoperationrequest.SetCSITransactionSupport(isCSITransactionSupportEnabled)
+
 	for _, vcconfig := range c.managers.VcenterConfigs {
 		go common.ComputeFSEnabledClustersToDsMap(authMgrs[vcconfig.Host], config.Global.CSIAuthCheckIntervalInMin)
 	}

pkg/csi/service/wcp/controller.go

@@ -160,6 +160,8 @@ func (c *controller) Init(config *cnsconfig.Config, version string) error {
 		common.PodVMOnStretchedSupervisor)
 	idempotencyHandlingEnabled := commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
 		common.CSIVolumeManagerIdempotency)
+	isCSITransactionSupportEnabled := commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
+		common.FCDTransactionSupport)
 	IsMultipleClustersPerVsphereZoneFSSEnabled = commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
 		common.MultipleClustersPerVsphereZone)
 	if !IsMultipleClustersPerVsphereZoneFSSEnabled {
@@ -177,7 +179,8 @@ func (c *controller) Init(config *cnsconfig.Config, version string) error {
 			config.Global.CnsVolumeOperationRequestCleanupIntervalInMin,
 			func() bool {
 				return commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.BlockVolumeSnapshot)
-			}, isPodVMOnStretchSupervisorFSSEnabled)
+			}, isPodVMOnStretchSupervisorFSSEnabled,
+			isCSITransactionSupportEnabled)
 		if err != nil {
 			log.Errorf("failed to initialize VolumeOperationRequestInterface with error: %v", err)
 			return err
@@ -403,13 +406,16 @@ func (c *controller) ReloadConfiguration(reconnectToVCFromNewConfig bool) error
 		}
 		idempotencyHandlingEnabled := commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
 			common.CSIVolumeManagerIdempotency)
+		isCSITransactionSupportEnabled := commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx,
+			common.FCDTransactionSupport)
 		if idempotencyHandlingEnabled {
 			log.Info("CSI Volume manager idempotency handling feature flag is enabled.")
 			operationStore, err = cnsvolumeoperationrequest.InitVolumeOperationRequestInterface(ctx,
 				c.manager.CnsConfig.Global.CnsVolumeOperationRequestCleanupIntervalInMin,
 				func() bool {
 					return commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.BlockVolumeSnapshot)
-				}, isPodVMOnStretchSupervisorFSSEnabled)
+				}, isPodVMOnStretchSupervisorFSSEnabled,
+				isCSITransactionSupportEnabled)
 			if err != nil {
 				log.Errorf("failed to initialize VolumeOperationRequestInterface with error: %v", err)
 				return err
@@ -2551,7 +2557,7 @@ func (c *controller) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshot
 					StorageClassName:           cnsVolumeInfo.Spec.StorageClassName,
 					StoragePolicyID:            cnsVolumeInfo.Spec.StoragePolicyID,
 					Capacity:                   resource.NewQuantity(0, resource.BinarySI),
-					Namespace:                  cnsVolumeInfo.Namespace,
+					Namespace:                  cnsVolumeInfo.Spec.Namespace,
 					IsStorageQuotaM2FSSEnabled: isStorageQuotaM2FSSEnabled,
 				})
 			if err != nil {

pkg/internalapis/cnsvolumeoperationrequest/cnsvolumeoperationrequest.go

@@ -78,14 +78,16 @@ var (
 	operationRequestStoreInstance        *operationRequestStore
 	operationStoreInitLock               = &sync.Mutex{}
 	isPodVMOnStretchSupervisorFSSEnabled bool
+	isCSITransactionSupportEnabled       bool
 )
 
 // InitVolumeOperationRequestInterface creates the CnsVolumeOperationRequest
 // definition on the API server and returns an implementation of
 // VolumeOperationRequest interface. Clients are unaware of the implementation
 // details to read and persist volume operation details.
 func InitVolumeOperationRequestInterface(ctx context.Context, cleanupInterval int,
-	isBlockVolumeSnapshotEnabled func() bool, isPodVMOnStretchSupervisorEnabled bool) (
+	isBlockVolumeSnapshotEnabled func() bool, isPodVMOnStretchSupervisorEnabled bool,
+	csiTransactionSupportEnabled bool) (
 	VolumeOperationRequest, error) {
 	log := logger.GetLogger(ctx)
 	csiNamespace = getCSINamespace()
@@ -130,6 +132,8 @@ func InitVolumeOperationRequestInterface(ctx context.Context, cleanupInterval in
 	}
 	// Store PodVMOnStretchedSupervisor FSS value for later use.
 	isPodVMOnStretchSupervisorFSSEnabled = isPodVMOnStretchSupervisorEnabled
+	// Store CSI Transaction Support FSS value for later use.
+	isCSITransactionSupportEnabled = csiTransactionSupportEnabled
 
 	return operationRequestStoreInstance, nil
 }
@@ -261,6 +265,32 @@ func (or *operationRequestStore) StoreRequestDetails(
 	// Create a deep copy since we modify the object.
 	updatedInstance := instance.DeepCopy()
 
+	// If CSI Transaction Support is enabled and we're storing a new InProgress operation with empty TaskID,
+	// mark any existing InProgress entries as TrackingAborted since this indicates a retry scenario.
+	if isCSITransactionSupportEnabled && operationDetailsToStore.TaskStatus == TaskInvocationStatusInProgress &&
+		operationDetailsToStore.TaskID == "" {
+		// This is a new operation attempt (Phase 1: Intent Registration)
+		// Mark any existing InProgress entries as TrackingAborted since this is clearly a retry
+		for index := range updatedInstance.Status.LatestOperationDetails {
+			existingOp := &updatedInstance.Status.LatestOperationDetails[index]
+			if existingOp.TaskStatus == TaskInvocationStatusInProgress {
+				// This is a retry - mark the previous attempt as aborted
+				existingOp.TaskStatus = TaskInvocationStatusTrackingAborted
+				existingOp.Error = "Operation tracking aborted due to retry attempt"
+				log.Infof("Marked previous InProgress operation as TrackingAborted due to retry detection. Instance: %s",
+					operationToStore.Name)
+			}
+		}
+
+		// Also check FirstOperationDetails
+		if updatedInstance.Status.FirstOperationDetails.TaskStatus == TaskInvocationStatusInProgress {
+			updatedInstance.Status.FirstOperationDetails.TaskStatus = TaskInvocationStatusTrackingAborted
+			updatedInstance.Status.FirstOperationDetails.Error = "Operation tracking aborted due to retry attempt"
+			log.Infof("Marked FirstOperationDetails as TrackingAborted due to retry detection. Instance: %s",
+				operationToStore.Name)
+		}
+	}
+
 	// Modify VolumeID, SnapshotID and Capacity
 	updatedInstance.Status.VolumeID = operationToStore.VolumeID
 	updatedInstance.Status.SnapshotID = operationToStore.SnapshotID
@@ -409,6 +439,12 @@ func (or *operationRequestStore) cleanupStaleInstances(cleanupInterval int) {
 	}
 }
 
+// SetCSITransactionSupport sets the CSI Transaction Support feature flag.
+// This function allows runtime modification of the isCSITransactionSupportEnabled variable.
+func SetCSITransactionSupport(enabled bool) {
+	isCSITransactionSupportEnabled = enabled
+}
+
 func getCSINamespace() string {
 	csiNamespace := os.Getenv(EnvCSINamespace)
 	if strings.TrimSpace(csiNamespace) == "" {