Enhancement - Cleanup Orphaned PVCs

Author: kolluria

pkg/common/config/config.go

@@ -68,6 +68,10 @@ const (
 	// interval after which successful CnsRegisterVolumes will be cleaned up.
 	// Current default value is set to 12 hours
 	DefaultCnsRegisterVolumesCleanupIntervalInMin = 720
+	// DefaultCnsPVCProtectionCleanupIntervalInMin is the default time interval after which
+	// orphaned PVCs will be cleaned up.
+	// Current default value is set to 12 hours
+	DefaultCnsPVCProtectionCleanupIntervalInMin = 720
 	// DefaultVolumeMigrationCRCleanupIntervalInMin is the default time interval
 	// after which stale CnsVSphereVolumeMigration CRs will be cleaned up.
 	// Current default value is set to 2 hours.
@@ -451,6 +455,9 @@ func validateConfig(ctx context.Context, cfg *Config) error {
 	if cfg.Global.CnsRegisterVolumesCleanupIntervalInMin == 0 {
 		cfg.Global.CnsRegisterVolumesCleanupIntervalInMin = DefaultCnsRegisterVolumesCleanupIntervalInMin
 	}
+	if cfg.Global.CnsPVCProtectionCleanupIntervalInMin == 0 {
+		cfg.Global.CnsPVCProtectionCleanupIntervalInMin = DefaultCnsPVCProtectionCleanupIntervalInMin
+	}
 	if cfg.Global.VolumeMigrationCRCleanupIntervalInMin == 0 {
 		cfg.Global.VolumeMigrationCRCleanupIntervalInMin = DefaultVolumeMigrationCRCleanupIntervalInMin
 	}

pkg/common/config/types.go

@@ -51,6 +51,9 @@ type Config struct {
 		// CnsRegisterVolumesCleanupIntervalInMin specifies the interval after which
 		// successful CnsRegisterVolumes will be cleaned up.
 		CnsRegisterVolumesCleanupIntervalInMin int `gcfg:"cnsregistervolumes-cleanup-intervalinmin"`
+		// CnsPVCProtectionCleanupIntervalInMin specifies the interval after which
+		// orphaned PVCs will be cleaned up.
+		CnsPVCProtectionCleanupIntervalInMin int `gcfg:"cnspvcprotection-cleanup-intervalinmin"`
 		// VolumeMigrationCRCleanupIntervalInMin specifies the interval after which
 		// stale CnsVSphereVolumeMigration CRs will be cleaned up.
 		VolumeMigrationCRCleanupIntervalInMin int `gcfg:"volumemigration-cr-cleanup-intervalinmin"`

pkg/common/unittestcommon/types.go

@@ -24,6 +24,7 @@ import (
 	cnstypes "github.com/vmware/govmomi/cns/types"
 	"github.com/vmware/govmomi/object"
 	"github.com/vmware/govmomi/vim25/types"
+	v1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/migration"
 	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"
@@ -47,6 +48,8 @@ type FakeK8SOrchestrator struct {
 	featureStates     map[string]string
 	// CSINodeTopology instances for topology testing
 	csiNodeTopologyInstances []interface{}
+	// PVCs for testing
+	pvcs []*v1.PersistentVolumeClaim
 }
 
 // volumeMigration holds mocked migrated volume information

pkg/common/unittestcommon/utils.go

@@ -543,6 +543,27 @@ func (c *FakeK8SOrchestrator) SetCSINodeTopologyInstances(instances []interface{
 	c.csiNodeTopologyInstances = instances
 }
 
+func (c *FakeK8SOrchestrator) ListPVCs(ctx context.Context, namespace string) []*v1.PersistentVolumeClaim {
+	if namespace == "" {
+		// Return all PVCs
+		return c.pvcs
+	}
+
+	// Filter by namespace
+	var filteredPVCs []*v1.PersistentVolumeClaim
+	for _, pvc := range c.pvcs {
+		if pvc.Namespace == namespace {
+			filteredPVCs = append(filteredPVCs, pvc)
+		}
+	}
+	return filteredPVCs
+}
+
+// SetPVCs sets the PVCs for testing
+func (c *FakeK8SOrchestrator) SetPVCs(pvcs []*v1.PersistentVolumeClaim) {
+	c.pvcs = pvcs
+}
+
 // configFromVCSim starts a vcsim instance and returns config for use against the
 // vcsim instance. The vcsim instance is configured with an empty tls.Config.
 func configFromVCSim(vcsimParams VcsimParams, isTopologyEnv bool) (*config.Config, func()) {

pkg/csi/service/common/commonco/coagnostic.go

@@ -128,6 +128,7 @@ type COCommonInterface interface {
 	// GetPVCNamespacedNameByUID returns the PVC's namespaced name (namespace/name) for the given UID.
 	// If the PVC is not found in the cache, it returns an empty string and false.
 	GetPVCNamespacedNameByUID(uid string) (k8stypes.NamespacedName, bool)
+	ListPVCs(ctx context.Context, namespace string) []*v1.PersistentVolumeClaim
 }
 
 // GetContainerOrchestratorInterface returns orchestrator object for a given

pkg/csi/service/common/commonco/k8sorchestrator/k8sorchestrator.go

@@ -30,6 +30,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc/codes"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/util/retry"
 
 	snapshotterClientSet "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned"
@@ -2286,3 +2287,15 @@ func GetPVCDataSource(ctx context.Context, claim *v1.PersistentVolumeClaim) (*v1
 	}
 	return &dataSource, nil
 }
+
+func (c *K8sOrchestrator) ListPVCs(ctx context.Context, namespace string) []*v1.PersistentVolumeClaim {
+	log := logger.GetLogger(ctx)
+
+	pvcs, err := c.informerManager.GetPVCLister().PersistentVolumeClaims(namespace).List(labels.Everything())
+	if err != nil {
+		log.With("namespace", namespace).Error("failed to list PVCs")
+		return []*v1.PersistentVolumeClaim{}
+	}
+
+	return pvcs
+}

pkg/syncer/admissionhandler/cnscsi_admissionhandler_test.go

@@ -47,6 +47,11 @@ type MockCOCommonInterface struct {
 	mock.Mock
 }
 
+func (m *MockCOCommonInterface) ListPVCs(ctx context.Context, namespace string) []*corev1.PersistentVolumeClaim {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (m *MockCOCommonInterface) GetPVCNamespacedNameByUID(uid string) (apitypes.NamespacedName, bool) {
 	//TODO implement me
 	panic("implement me")

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

@@ -224,6 +224,11 @@ func (m *mockVolumeManager) SyncVolume(ctx context.Context,
 
 type mockCOCommon struct{}
 
+func (m *mockCOCommon) ListPVCs(ctx context.Context, namespace string) []*corev1.PersistentVolumeClaim {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (m *mockCOCommon) GetPVCNamespacedNameByUID(uid string) (types.NamespacedName, bool) {
 	//TODO implement me
 	panic("implement me")

pkg/syncer/cnsoperator/manager/cleanupcustomresources.go

@@ -21,10 +21,15 @@ import (
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	cnsoperatorv1alpha1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator"
+	batchattachv1a1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator/cnsnodevmbatchattachment/v1alpha1"
 	cnsregistervolumev1alpha1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator/cnsregistervolume/v1alpha1"
 	cnsunregistervolumev1alpha1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator/cnsunregistervolume/v1alpha1"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common/commonco"
+	cnsoperatortypes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/types"
 
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/logger"
 	k8s "sigs.k8s.io/vsphere-csi-driver/v3/pkg/kubernetes"
@@ -109,3 +114,76 @@ func cleanUpCnsUnregisterVolumeInstances(ctx context.Context, restClientConfig *
 		}
 	}
 }
+
+var (
+	newClientForGroup = k8s.NewClientForGroup
+	newForConfig      = func(config *rest.Config) (kubernetes.Interface, error) {
+		return kubernetes.NewForConfig(config)
+	}
+)
+
+// cleanupPVCs removes the `CNSPvcFinalizer` from the PVCs in cases
+// where the CnsNodeVmBatchAttachment CR gets deleted before removing the finalizer.
+// This is EXTREMELY UNLIKELY to happen but still a possibility that has to be addressed.
+func cleanupPVCs(ctx context.Context, config rest.Config) {
+	log := logger.GetLogger(ctx)
+
+	pvcList := commonco.ContainerOrchestratorUtility.ListPVCs(ctx, "")
+	if len(pvcList) == 0 {
+		log.Info("no PVCs found. Exiting...")
+		return
+	}
+
+	// map to hold all the PVCs that have the finalizer added by the batch attach reconciler
+	pvcMap := make(map[string]map[string]struct{})
+	for _, pvc := range pvcList {
+		// Check if PVC has the CNS finalizer in metadata.finalizers
+		if !controllerutil.ContainsFinalizer(pvc, cnsoperatortypes.CNSPvcFinalizer) {
+			// not a PVC that is attached to or being attached to a VM. Can be ignored.
+			continue
+		}
+
+		if _, ok := pvcMap[pvc.Namespace]; !ok {
+			pvcMap[pvc.Namespace] = map[string]struct{}{}
+		}
+		pvcMap[pvc.Namespace][pvc.Name] = struct{}{}
+	}
+
+	cnsClient, err := newClientForGroup(ctx, &config, cnsoperatorv1alpha1.GroupName)
+	if err != nil {
+		log.Error("failed to create cns operator client")
+		return
+	}
+
+	batchAttachList := batchattachv1a1.CnsNodeVMBatchAttachmentList{}
+	err = cnsClient.List(ctx, &batchAttachList)
+	if err != nil {
+		log.With("kind", batchAttachList.Kind).Error("listing failed")
+		return
+	}
+
+	for _, cr := range batchAttachList.Items {
+		for _, vol := range cr.Spec.Volumes {
+			// Any PVCs that are still in the spec can be safely ignored to be processed by the reconciler.
+			delete(pvcMap[cr.Namespace], vol.PersistentVolumeClaim.ClaimName)
+		}
+	}
+
+	c, err := newForConfig(&config)
+	if err != nil {
+		log.Error("failed to create core API client")
+		return
+	}
+
+	// Remove the finalizer for the remaining PVCs
+	for namespace, pvcs := range pvcMap {
+		for name := range pvcs {
+			err := k8s.RemoveFinalizerFromPVC(ctx, c, name, namespace, cnsoperatortypes.CNSPvcFinalizer)
+			if err != nil {
+				log.With("name", name).With("namespace", namespace).
+					Error("failed to remove the finalizer")
+				continue
+			}
+		}
+	}
+}

pkg/syncer/cnsoperator/manager/init.go

@@ -137,6 +137,29 @@ func InitCnsOperator(ctx context.Context, clusterFlavor cnstypes.CnsClusterFlavo
 				log.Errorf("failed to create %q CRD. Err: %+v", crdNameNodeVmBatchAttachment, err)
 				return err
 			}
+
+			err = watcher(ctx, cnsOperator)
+			if err != nil {
+				log.Error("Failed to watch on config file for changes to "+
+					"CnsPVCProtectionCleanupIntervalInMin. Error: %+v", err)
+				return err
+			}
+
+			// Start cleanup routine to remove orphaned PVC finalizers.
+			// This handles cases where CnsNodeVMBatchAttachment CRs are deleted before
+			// removing finalizers from their associated PVCs.
+			log.Info("Starting go routine to cleanup orphaned PVC finalizers from batch attach.")
+			go func() {
+				for {
+					ctx, log = logger.GetNewContextWithLogger()
+					log.Info("Triggering PVC finalizer cleanup routine")
+					cleanupPVCs(ctx, *restConfig)
+					log.Info("Completed PVC finalizer cleanup")
+					for i := 1; i <= cnsOperator.configInfo.Cfg.Global.CnsPVCProtectionCleanupIntervalInMin; i++ {
+						time.Sleep(1 * time.Minute)
+					}
+				}
+			}()
 		}
 
 		// Create CnsVolumeMetadata CRD
@@ -162,7 +185,7 @@ func InitCnsOperator(ctx context.Context, clusterFlavor cnstypes.CnsClusterFlavo
 		if stretchedSupervisor {
 			log.Info("Observed stretchedSupervisor setup")
 		}
-		if !stretchedSupervisor || (stretchedSupervisor && syncer.IsPodVMOnStretchSupervisorFSSEnabled) {
+		if !stretchedSupervisor || syncer.IsPodVMOnStretchSupervisorFSSEnabled {
 			// Create CnsRegisterVolume CRD from manifest.
 			log.Infof("Creating %q CRD", cnsoperatorv1alpha1.CnsRegisterVolumePlural)
 			err = k8s.CreateCustomResourceDefinitionFromManifest(ctx, cnsoperatorconfig.EmbedCnsRegisterVolumeCRFile,
@@ -228,7 +251,7 @@ func InitCnsOperator(ctx context.Context, clusterFlavor cnstypes.CnsClusterFlavo
 			}
 		}
 
-		if !stretchedSupervisor || (stretchedSupervisor && syncer.IsWorkloadDomainIsolationSupported) {
+		if !stretchedSupervisor || syncer.IsWorkloadDomainIsolationSupported {
 			if cnsOperator.coCommonInterface.IsFSSEnabled(ctx, common.FileVolume) {
 				// Create CnsFileAccessConfig CRD from manifest if file volume feature
 				// is enabled.