Added add, get and remove methods on the pvcToSnapshotsMap

Added GetSnapshotsForPVC method on COCommonInterface interface and it's implementors
Updated GetSnapshotsForPVC method of K8sOrchestrator to return the snapshots associated with the given PVC

Author: kolluria

pkg/common/unittestcommon/utils.go

@@ -516,6 +516,11 @@ func (c *FakeK8SOrchestrator) SetCSINodeTopologyInstances(instances []interface{
 	c.csiNodeTopologyInstances = instances
 }
 
+func (c *FakeK8SOrchestrator) GetSnapshotsForPVC(ctx context.Context, pvcName, namespace string) []string {
+	//TODO implement me
+	panic("implement me")
+}
+
 // 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,8 @@ 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)
+	// GetSnapshotsForPVC returns the names of the snapshots for the given PVC
+	GetSnapshotsForPVC(ctx context.Context, pvcName, namespace string) []string
 }
 
 // GetContainerOrchestratorInterface returns orchestrator object for a given

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

@@ -256,6 +256,71 @@ type pvcToSnapshotsMap struct {
 	items map[namespacedName]map[namespacedName]struct{}
 }
 
+func (m *pvcToSnapshotsMap) add(pvc, snapshot, namespace string) {
+	m.Lock()
+	defer m.Unlock()
+
+	pvcKey := namespacedName{
+		namespace: namespace,
+		name:      pvc,
+	}
+	if _, ok := m.items[pvcKey]; !ok {
+		m.items[pvcKey] = make(map[namespacedName]struct{})
+	}
+	snapKey := namespacedName{
+		namespace: namespace,
+		name:      snapshot,
+	}
+	m.items[pvcKey][snapKey] = struct{}{}
+}
+
+func (m *pvcToSnapshotsMap) get(pvc, namespace string) []string {
+	m.RLock()
+	defer m.RUnlock()
+
+	pvcKey := namespacedName{
+		namespace: namespace,
+		name:      pvc,
+	}
+	snapMap, ok := m.items[pvcKey]
+	if !ok {
+		return []string{}
+	}
+
+	snaps := make([]string, 0, len(snapMap))
+	for k := range snapMap {
+		snaps = append(snaps, k.name)
+	}
+	return snaps
+}
+
+func (m *pvcToSnapshotsMap) delete(pvc, snapshot, namespace string) {
+	m.Lock()
+	defer m.Unlock()
+
+	pvcKey := namespacedName{
+		namespace: namespace,
+		name:      pvc,
+	}
+	snapMap, ok := m.items[pvcKey]
+	if !ok {
+		return
+	}
+
+	snapKey := namespacedName{
+		namespace: namespace,
+		name:      snapshot,
+	}
+	delete(snapMap, snapKey)
+	if len(snapMap) != 0 {
+		m.items[pvcKey] = snapMap
+		return
+	}
+
+	// delete pvc entry if no snapshots are present
+	delete(m.items, pvcKey)
+}
+
 // K8sOrchestrator defines set of properties specific to K8s.
 type K8sOrchestrator struct {
 	supervisorFSS        FSSConfigMapInfo
@@ -1968,6 +2033,74 @@ func nodeRemove(obj interface{}) {
 	k8sOrchestratorInstance.nodeIDToNameMap.remove(nodeMoID)
 }
 
+func initPVCToSnapshotsMap(ctx context.Context, controllerClusterFlavor cnstypes.CnsClusterFlavor) error {
+	log := logger.GetLogger(ctx)
+	// TODO: check if we need to check the FSS as well
+	if controllerClusterFlavor != cnstypes.CnsClusterFlavorWorkload {
+		// PVC to VolumeSnapshot cache is only required for WCP for now.
+		log.Info("non WCP cluster detected. skipping initialising PVC to Snapshot cache.")
+		return nil
+	}
+
+	k8sOrchestratorInstance.pvcToSnapshotsMap = pvcToSnapshotsMap{
+		RWMutex: &sync.RWMutex{},
+		items:   make(map[namespacedName]map[namespacedName]struct{}),
+	}
+	snapshotAdded := func(obj any) {
+		snap, ok := obj.(*snapshotv1.VolumeSnapshot)
+		if !ok || snap == nil {
+			log.Warnf("unrecognized object %+v", obj)
+			return
+		}
+
+		log.Infof("snapshotAdded: snapshot=%v", snap)
+		if snap.Spec.Source.PersistentVolumeClaimName == nil {
+			log.Warnf("snapshot is not associated with any PVC. Ignoring it...")
+			return
+		}
+
+		k8sOrchestratorInstance.pvcToSnapshotsMap.add(
+			*snap.Spec.Source.PersistentVolumeClaimName, snap.Name, snap.Namespace)
+		log.With("pvc", *snap.Spec.Source.PersistentVolumeClaimName).With("snapshot", snap.Name).
+			With("namespace", snap.Namespace).Debug("successfully added the snapshot to the cache")
+	}
+	snapshotDeleted := func(obj any) {
+		snap, ok := obj.(*snapshotv1.VolumeSnapshot)
+		if !ok || snap == nil {
+			log.Warnf("unrecognized object %+v", obj)
+			return
+		}
+
+		log.Infof("snapshotDeleted: snapshot=%v", snap)
+		if snap.Spec.Source.PersistentVolumeClaimName == nil {
+			log.Warnf("snapshot is not associated with any PVC. Ignoring it...")
+			return
+		}
+
+		k8sOrchestratorInstance.pvcToSnapshotsMap.delete(
+			*snap.Spec.Source.PersistentVolumeClaimName, snap.Name, snap.Namespace)
+		log.With("pvc", *snap.Spec.Source.PersistentVolumeClaimName).With("snapshot", snap.Name).
+			With("namespace", snap.Namespace).Debug("successfully removed the snapshot from the cache")
+	}
+
+	err := k8sOrchestratorInstance.informerManager.AddSnapshotListener(ctx,
+		func(obj any) {
+			snapshotAdded(obj)
+		},
+		func(oldObj, newObj any) {
+			// TODO: implement or remove as we probably don't need to take any action
+			log.Info("snapshotUpdated")
+		},
+		func(obj any) {
+			snapshotDeleted(obj)
+		})
+	if err != nil {
+		return logger.LogNewErrorf(log, "failed to listen on volumesnapshots. Error: %v", err)
+	}
+
+	return nil
+}
+
 // GetNodeIDtoNameMap returns a map containing the nodeID to node name
 func (c *K8sOrchestrator) GetNodeIDtoNameMap(ctx context.Context) map[string]string {
 	return c.nodeIDToNameMap.items
@@ -2282,6 +2415,11 @@ func (c *K8sOrchestrator) GetPVCNamespacedNameByUID(uid string) (k8stypes.Namesp
 	return nsName, true
 }
 
+// GetSnapshotsForPVC returns the names of the snapshots for the given PVC
+func (c *K8sOrchestrator) GetSnapshotsForPVC(ctx context.Context, pvc, namespace string) []string {
+	return c.pvcToSnapshotsMap.get(pvc, namespace)
+}
+
 // GetPVCDataSource Retrieves the VolumeSnapshot source when a PVC from VolumeSnapshot is being created.
 func GetPVCDataSource(ctx context.Context, claim *v1.PersistentVolumeClaim) (*v1.ObjectReference, error) {
 	var dataSource v1.ObjectReference
@@ -2309,98 +2447,3 @@ func GetPVCDataSource(ctx context.Context, claim *v1.PersistentVolumeClaim) (*v1
 	}
 	return &dataSource, nil
 }
-
-func initPVCToSnapshotsMap(ctx context.Context, controllerClusterFlavor cnstypes.CnsClusterFlavor) error {
-	log := logger.GetLogger(ctx)
-	if controllerClusterFlavor != cnstypes.CnsClusterFlavorWorkload {
-		// PVC to VolumeSnapshot cache is only required for WCP for now.
-		log.Info("non WCP cluster detected. skipping initialising PVC to Snapshot cache.")
-		return nil
-	}
-
-	k8sOrchestratorInstance.pvcToSnapshotsMap = pvcToSnapshotsMap{
-		RWMutex: &sync.RWMutex{},
-		items:   make(map[namespacedName]map[namespacedName]struct{}),
-	}
-
-	snapshotAdded := func(obj any) {
-		snap, ok := obj.(*snapshotv1.VolumeSnapshot)
-		if !ok || snap == nil {
-			log.Warnf("snapshotAdded: unrecognized object %+v", obj)
-			return
-		}
-
-		log.Infof("snapshotAdded: snapshot=%v", snap)
-		if snap.Spec.Source.PersistentVolumeClaimName == nil {
-			log.Warnf("snapshotAdded: snapshot is not associated with any PVC. Ignoring it...")
-			return
-		}
-
-		k8sOrchestratorInstance.pvcToSnapshotsMap.Lock()
-		defer k8sOrchestratorInstance.pvcToSnapshotsMap.Unlock()
-
-		pvcName := namespacedName{
-			namespace: snap.Namespace,
-			name:      *snap.Spec.Source.PersistentVolumeClaimName,
-		}
-		if _, ok := k8sOrchestratorInstance.pvcToSnapshotsMap.items[pvcName]; !ok {
-			k8sOrchestratorInstance.pvcToSnapshotsMap.items[pvcName] = make(map[namespacedName]struct{})
-		}
-		snapName := namespacedName{
-			namespace: snap.Namespace,
-			name:      snap.Name,
-		}
-		k8sOrchestratorInstance.pvcToSnapshotsMap.items[pvcName][snapName] = struct{}{}
-		log.With("pvc", pvcName).With("snapshot", snapName).Debug("successfully added the snapshot to the cache")
-	}
-
-	snapshotDeleted := func(obj any) {
-		snap, ok := obj.(*snapshotv1.VolumeSnapshot)
-		if !ok || snap == nil {
-			log.Warnf("snapshotDeleted: unrecognized object %+v", obj)
-			return
-		}
-
-		log.Infof("snapshotDeleted: snapshot=%v", snap)
-		if snap.Spec.Source.PersistentVolumeClaimName == nil {
-			log.Warnf("snapshotDeleted: snapshot is not associated with any PVC. Ignoring it...")
-			return
-		}
-
-		k8sOrchestratorInstance.pvcToSnapshotsMap.Lock()
-		defer k8sOrchestratorInstance.pvcToSnapshotsMap.Unlock()
-
-		pvcName := namespacedName{
-			namespace: snap.Namespace,
-			name:      *snap.Spec.Source.PersistentVolumeClaimName,
-		}
-		if _, ok := k8sOrchestratorInstance.pvcToSnapshotsMap.items[pvcName]; !ok {
-			log.Infof("snapshotDeleted: associated PVC %s not found in the map. Ignoring the delete event...", pvcName)
-			return
-		}
-
-		snapName := namespacedName{
-			namespace: snap.Namespace,
-			name:      snap.Name,
-		}
-		delete(k8sOrchestratorInstance.pvcToSnapshotsMap.items[pvcName], snapName)
-		log.With("pvc", pvcName).With("snapshot", snapName).Debug("successfully removed the snapshot from the cache")
-	}
-
-	err := k8sOrchestratorInstance.informerManager.AddSnapshotListener(ctx,
-		func(obj any) {
-			snapshotAdded(obj)
-		},
-		func(oldObj, newObj any) {
-			// TODO: implement
-			log.Info("snapshotUpdated")
-		},
-		func(obj any) {
-			snapshotDeleted(obj)
-		})
-	if err != nil {
-		return logger.LogNewErrorf(log, "failed to listen on volumesnapshots. Error: %v", err)
-	}
-
-	return nil
-}

pkg/kubernetes/informers.go

@@ -267,8 +267,10 @@ func (im *InformerManager) AddVolumeAttachmentListener(ctx context.Context, add
 }
 
 // AddSnapshotListener hooks up add, update, delete callbacks.
-func (im *InformerManager) AddSnapshotListener(ctx context.Context, add func(obj interface{}),
-	update func(oldObj, newObj interface{}), remove func(obj interface{})) error {
+func (im *InformerManager) AddSnapshotListener(ctx context.Context,
+	add func(obj interface{}),
+	update func(oldObj, newObj interface{}),
+	remove func(obj interface{})) error {
 	log := logger.GetLogger(ctx)
 	if im.snapshotInformer == nil {
 		im.snapshotInformer = im.snapshotInformerFactory.

pkg/syncer/admissionhandler/cnscsi_admissionhandler_test.go

@@ -47,6 +47,11 @@ type MockCOCommonInterface struct {
 	mock.Mock
 }
 
+func (m *MockCOCommonInterface) GetSnapshotsForPVC(ctx context.Context, pvcName, namespace string) []string {
+	//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

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