enhance clean up of CNSVolumeOperations instances to delete CnsVolumeOperationRequest instances (#3375)

with latest TaskInvocationTimestamp older than 15 minutes

Author: divyenpatel

pkg/common/config/config.go

@@ -80,8 +80,8 @@ const (
 	DefaultCSIFetchPreferredDatastoresIntervalInMin = 5
 	// DefaultCnsVolumeOperationRequestCleanupIntervalInMin is the default time
 	// interval after which stale CnsVSphereVolumeMigration CRs will be cleaned up.
-	// Current default value is set to 24 hours.
-	DefaultCnsVolumeOperationRequestCleanupIntervalInMin = 1440
+	// Current default value is set to 15 minutes.
+	DefaultCnsVolumeOperationRequestCleanupIntervalInMin = 15
 	// DefaultGlobalMaxSnapshotsPerBlockVolume is the default maximum number of block volume snapshots per volume.
 	DefaultGlobalMaxSnapshotsPerBlockVolume = 3
 	// MaxNumberOfTopologyCategories is the max number of topology domains/categories allowed.

pkg/internalapis/cnsvolumeoperationrequest/cnsvolumeoperationrequest.go

@@ -31,7 +31,6 @@ import (
 
 	csiconfig "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/config"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/logger"
-	csitypes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/types"
 	cnsvolumeoperationrequestconfig "sigs.k8s.io/vsphere-csi-driver/v3/pkg/internalapis/cnsvolumeoperationrequest/config"
 	cnsvolumeoprequestv1alpha1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/internalapis/cnsvolumeoperationrequest/v1alpha1"
 	k8s "sigs.k8s.io/vsphere-csi-driver/v3/pkg/kubernetes"
@@ -127,7 +126,7 @@ func InitVolumeOperationRequestInterface(ctx context.Context, cleanupInterval in
 		operationRequestStoreInstance = &operationRequestStore{
 			k8sclient: k8sclient,
 		}
-		go operationRequestStoreInstance.cleanupStaleInstances(cleanupInterval, isBlockVolumeSnapshotEnabled)
+		go operationRequestStoreInstance.cleanupStaleInstances(cleanupInterval)
 	}
 	// Store PodVMOnStretchedSupervisor FSS value for later use.
 	isPodVMOnStretchSupervisorFSSEnabled = isPodVMOnStretchSupervisorEnabled
@@ -352,109 +351,58 @@ func (or *operationRequestStore) DeleteRequestDetails(ctx context.Context, name
 	return nil
 }
 
-// cleanupStaleInstances cleans up CnsVolumeOperationRequest instances for
-// volumes that are no longer present in the kubernetes cluster.
-func (or *operationRequestStore) cleanupStaleInstances(cleanupInterval int, isBlockVolumeSnapshotEnabled func() bool) {
+// cleanupStaleInstances cleans up CnsVolumeOperationRequest instances
+// with latest TaskInvocationTimestamp older than 15 minutes
+func (or *operationRequestStore) cleanupStaleInstances(cleanupInterval int) {
 	ticker := time.NewTicker(time.Duration(cleanupInterval) * time.Minute)
 	ctx, log := logger.GetNewContextWithLogger()
 	log.Infof("CnsVolumeOperationRequest clean up interval is set to %d minutes", cleanupInterval)
 	for ; true; <-ticker.C {
+		cutoffTime := time.Now().Add(-15 * time.Minute)
+		continueToken := ""
 		log.Infof("Cleaning up stale CnsVolumeOperationRequest instances.")
-
-		instanceMap := make(map[string]bool)
-
-		cnsVolumeOperationRequestList := &cnsvolumeoprequestv1alpha1.CnsVolumeOperationRequestList{}
-		err := or.k8sclient.List(ctx, cnsVolumeOperationRequestList)
-		if err != nil {
-			log.Errorf("failed to list CnsVolumeOperationRequests with error %v. Abandoning "+
-				"CnsVolumeOperationRequests clean up ...", err)
-			continue
-		}
-
-		k8sclient, err := k8s.NewClient(ctx)
-		if err != nil {
-			log.Errorf("failed to get k8sclient with error: %v. Abandoning CnsVolumeOperationRequests "+
-				"clean up ...", err)
-			continue
-		}
-		pvList, err := k8sclient.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
-		if err != nil {
-			log.Errorf("failed to list PersistentVolumes with error %v. Abandoning "+
-				"CnsVolumeOperationRequests clean up ...", err)
-			continue
-		}
-
-		for _, pv := range pvList.Items {
-			if pv.Spec.CSI != nil && pv.Spec.CSI.Driver == csitypes.Name {
-				instanceMap[pv.Name] = true
-				volumeHandle := pv.Spec.CSI.VolumeHandle
-				if strings.Contains(volumeHandle, "file") {
-					volumeHandle = strings.ReplaceAll(volumeHandle, ":", "-")
-				}
-				instanceMap[volumeHandle] = true
+		for {
+			listOptions := &client.ListOptions{
+				Limit:    5000,
+				Continue: continueToken,
 			}
-		}
-		blockVolumeSnapshotEnabled := isBlockVolumeSnapshotEnabled()
-		// skip cleaning up of snapshot related CnsVolumeOperationRequests if FSS is not enabled.
-		if blockVolumeSnapshotEnabled {
-			snapshotterClient, err := k8s.NewSnapshotterClient(ctx)
+			cnsVolumeOperationRequestList := &cnsvolumeoprequestv1alpha1.CnsVolumeOperationRequestList{}
+			err := or.k8sclient.List(ctx, cnsVolumeOperationRequestList, listOptions)
 			if err != nil {
-				log.Errorf("failed to get snapshotterClient with error: %v. Abandoning "+
+				log.Errorf("failed to list CnsVolumeOperationRequests with error %v. Abandoning "+
 					"CnsVolumeOperationRequests clean up ...", err)
-				return
+				break
 			}
-
-			// the List API below ensures VolumeSnapshotContent CRD is installed and lists the existing
-			// VolumeSnapshotContent CRs in cluster.
-			vscList, err := snapshotterClient.SnapshotV1().VolumeSnapshotContents().List(ctx, metav1.ListOptions{})
-			if err != nil {
-				log.Errorf("failed to list VolumeSnapshotContents with error %v. Abandoning "+
-					"CnsVolumeOperationRequests clean up ...", err)
-				return
-			}
-
-			for _, vsc := range vscList.Items {
-				if vsc.Spec.Driver != csitypes.Name {
+			for _, instance := range cnsVolumeOperationRequestList.Items {
+				latestOperationDetailsLength := len(instance.Status.LatestOperationDetails)
+				// Skip if task is still in progress
+				if latestOperationDetailsLength != 0 &&
+					instance.Status.LatestOperationDetails[latestOperationDetailsLength-1].TaskStatus ==
+						TaskInvocationStatusInProgress {
 					continue
 				}
-				volumeHandle := vsc.Spec.Source.VolumeHandle
-				if volumeHandle != nil {
-					// CnsVolumeOperation instance for CreateSnapshot
-					instanceMap[strings.TrimPrefix(vsc.Name, "snapcontent-")+"-"+*volumeHandle] = true
-				}
-				if vsc.Status != nil && vsc.Status.SnapshotHandle != nil {
-					// CnsVolumeOperation instance for DeleteSnapshot
-					instanceMap[strings.Replace(*vsc.Status.SnapshotHandle, "+", "-", 1)] = true
+				// Delete instance if TaskInvocationTimestamp is older than 15 minutes
+				if latestOperationDetailsLength != 0 &&
+					instance.Status.LatestOperationDetails[latestOperationDetailsLength-1].
+						TaskInvocationTimestamp.Time.After(cutoffTime) {
+					log.Debugf("CnsVolumeOperationRequest instance %q is skipped for deletion", instance.Name)
+					continue
 				}
-			}
-		}
-
-		for _, instance := range cnsVolumeOperationRequestList.Items {
-			latestOperationDetailsLength := len(instance.Status.LatestOperationDetails)
-			if latestOperationDetailsLength != 0 &&
-				instance.Status.LatestOperationDetails[latestOperationDetailsLength-1].TaskStatus ==
-					TaskInvocationStatusInProgress {
-				continue
-			}
-			var trimmedName string
-			switch {
-			case strings.HasPrefix(instance.Name, "pvc"):
-				trimmedName = instance.Name
-			case strings.HasPrefix(instance.Name, "delete"):
-				trimmedName = strings.TrimPrefix(instance.Name, "delete-")
-			case strings.HasPrefix(instance.Name, "expand"):
-				trimmedName = strings.TrimPrefix(instance.Name, "expand-")
-			case blockVolumeSnapshotEnabled && strings.HasPrefix(instance.Name, "snapshot"):
-				trimmedName = strings.TrimPrefix(instance.Name, "snapshot-")
-			case blockVolumeSnapshotEnabled && strings.HasPrefix(instance.Name, "deletesnapshot"):
-				trimmedName = strings.TrimPrefix(instance.Name, "deletesnapshot-")
-			}
-			if _, ok := instanceMap[trimmedName]; !ok {
+				log.Debugf("Calling DeleteRequestDetails for %q", instance.Name)
 				err = or.DeleteRequestDetails(ctx, instance.Name)
 				if err != nil {
 					log.Errorf("failed to delete CnsVolumeOperationRequest instance %s with error %v",
 						instance.Name, err)
 				}
+				log.Debugf("CnsVolumeOperationRequest instance %q is deleted", instance.Name)
+			}
+			// Exit if there are no more pages
+			continueToken = cnsVolumeOperationRequestList.GetContinue()
+			log.Debugf("continueToken to process remaining CnsVolumeOperationRequest "+
+				"insances is %q", continueToken)
+			if continueToken == "" {
+				log.Infof("all CnsVolumeOperationRequest instances are processed")
+				break
 			}
 		}
 		log.Infof("Clean up of stale CnsVolumeOperationRequest complete.")