Refactor snapshot retrieval to use ContainerOrchestratorUtility

Simplify the getSnapshotsForPVC function by delegating snapshot retrieval
to the ContainerOrchestratorUtility interface instead of directly using
the snapshot client. This improves testability and maintains consistency
with the existing architecture.

Changes:
- Remove direct dependency on external-snapshotter client in util.go
- Update getSnapshotsForPVC to use ContainerOrchestratorUtility.GetSnapshotsForPVC
- Remove rest.Config parameter from getSnapshotsForPVC function signature
- Add comprehensive unit tests for getSnapshotsForPVC covering:
  * Uninitialized ContainerOrchestratorUtility scenario
  * PVC with no snapshots
  * PVC with multiple snapshots
- Implement GetSnapshotsForPVC in FakeK8SOrchestrator for testing
- Add TODO comment for future refactoring of pvcToSnapshotsMap key type

Benefits:
- Improved testability through dependency injection
- Reduced coupling to external snapshot client
- Consistent error handling
- Better alignment with existing orchestrator abstraction pattern

Author: kolluria

pkg/common/cns-lib/node/nodes.go

@@ -37,14 +37,23 @@ type Nodes struct {
 // Initialize helps initialize node manager and node informer manager.
 func (nodes *Nodes) Initialize(ctx context.Context) error {
 	nodes.cnsNodeManager = GetManager(ctx)
+
 	k8sclient, err := k8s.NewClient(ctx)
 	if err != nil {
 		log := logger.GetLogger(ctx)
 		log.Errorf("Creating Kubernetes client failed. Err: %v", err)
 		return err
 	}
+
+	snapshotterClient, err := k8s.NewSnapshotterClient(ctx)
+	if err != nil {
+		log := logger.GetLogger(ctx)
+		log.Errorf("Creating Snapshotter client failed. Err: %v", err)
+		return err
+	}
+
 	nodes.cnsNodeManager.SetKubernetesClient(k8sclient)
-	nodes.informMgr = k8s.NewInformer(ctx, k8sclient, true)
+	nodes.informMgr = k8s.NewInformer(ctx, k8sclient, snapshotterClient)
 	err = nodes.informMgr.AddCSINodeListener(ctx, nodes.csiNodeAdd,
 		nodes.csiNodeUpdate, nodes.csiNodeDelete)
 	if err != nil {

pkg/common/unittestcommon/utils.go

@@ -517,8 +517,11 @@ func (c *FakeK8SOrchestrator) SetCSINodeTopologyInstances(instances []interface{
 }
 
 func (c *FakeK8SOrchestrator) GetSnapshotsForPVC(ctx context.Context, pvcName, namespace string) []string {
-	//TODO implement me
-	panic("implement me")
+	if strings.Contains(pvcName, "no-snap") {
+		return []string{}
+	}
+
+	return []string{"snap1", "snap2", "snap3"}
 }
 
 // configFromVCSim starts a vcsim instance and returns config for use against the

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

@@ -240,56 +240,44 @@ func (m *volumeIDToNameMap) get(volumeID string) (string, bool) {
 	return volumeName, found
 }
 
-type namespacedName struct {
-	namespace, name string
-}
-
-func (n namespacedName) String() string {
-	return fmt.Sprintf("%s/%s", n.namespace, n.name)
-}
-
 // pvcToSnapshotsMap maps a PVC to its snapshots.
-// Key is the namespaced name of the PVC and value is a map.
-// The key of the inner map is the namespaced name of the snapshot.
+// The primary key is the namespaced name of the PVC and value is a map.
+// The key of the inner map is the name of the snapshot.
 type pvcToSnapshotsMap struct {
 	*sync.RWMutex
-	items map[namespacedName]map[namespacedName]struct{}
+	items map[k8stypes.NamespacedName]map[string]struct{}
 }
 
 func (m *pvcToSnapshotsMap) add(pvc, snapshot, namespace string) {
 	m.Lock()
 	defer m.Unlock()
 
-	pvcKey := namespacedName{
-		namespace: namespace,
-		name:      pvc,
+	pvcKey := k8stypes.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] = make(map[string]struct{})
 	}
-	m.items[pvcKey][snapKey] = struct{}{}
+	m.items[pvcKey][snapshot] = struct{}{}
 }
 
 func (m *pvcToSnapshotsMap) get(pvc, namespace string) []string {
 	m.RLock()
 	defer m.RUnlock()
 
-	pvcKey := namespacedName{
-		namespace: namespace,
-		name:      pvc,
+	pvcKey := k8stypes.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)
+	for snap := range snapMap {
+		snaps = append(snaps, snap)
 	}
 	return snaps
 }
@@ -298,20 +286,16 @@ func (m *pvcToSnapshotsMap) delete(pvc, snapshot, namespace string) {
 	m.Lock()
 	defer m.Unlock()
 
-	pvcKey := namespacedName{
-		namespace: namespace,
-		name:      pvc,
+	pvcKey := k8stypes.NamespacedName{
+		Namespace: namespace,
+		Name:      pvc,
 	}
 	snapMap, ok := m.items[pvcKey]
 	if !ok {
 		return
 	}
 
-	snapKey := namespacedName{
-		namespace: namespace,
-		name:      snapshot,
-	}
-	delete(snapMap, snapKey)
+	delete(snapMap, snapshot)
 	if len(snapMap) != 0 {
 		m.items[pvcKey] = snapMap
 		return
@@ -402,7 +386,7 @@ func Newk8sOrchestrator(ctx context.Context, controllerClusterFlavor cnstypes.Cn
 			k8sOrchestratorInstance.clusterFlavor = controllerClusterFlavor
 			k8sOrchestratorInstance.k8sClient = k8sClient
 			k8sOrchestratorInstance.snapshotterClient = snapshotterClient
-			k8sOrchestratorInstance.informerManager = k8s.NewInformer(ctx, k8sClient, true)
+			k8sOrchestratorInstance.informerManager = k8s.NewInformer(ctx, k8sClient, snapshotterClient)
 			coInstanceErr = initFSS(ctx, k8sClient, controllerClusterFlavor, params)
 			if coInstanceErr != nil {
 				log.Errorf("Failed to initialize the orchestrator. Error: %v", coInstanceErr)
@@ -2044,7 +2028,7 @@ func initPVCToSnapshotsMap(ctx context.Context, controllerClusterFlavor cnstypes
 
 	k8sOrchestratorInstance.pvcToSnapshotsMap = pvcToSnapshotsMap{
 		RWMutex: &sync.RWMutex{},
-		items:   make(map[namespacedName]map[namespacedName]struct{}),
+		items:   make(map[k8stypes.NamespacedName]map[string]struct{}),
 	}
 	snapshotAdded := func(obj any) {
 		snap, ok := obj.(*snapshotv1.VolumeSnapshot)
@@ -2085,6 +2069,8 @@ func initPVCToSnapshotsMap(ctx context.Context, controllerClusterFlavor cnstypes
 		func(obj any) {
 			snapshotAdded(obj)
 		},
+		// Since the name of PVC associated with a snapshot is immutable,
+		// update events do not have any impact on the state of the cache.
 		nil,
 		func(obj any) {
 			snapshotDeleted(obj)

pkg/internalapis/featurestates/featurestates.go

@@ -122,8 +122,15 @@ func StartSvFSSReplicationService(ctx context.Context, svFeatureStatConfigMapNam
 		return err
 	}
 
+	// Create the snapshotter client.
+	snapshotterClient, err := k8s.NewSnapshotterClient(ctx)
+	if err != nil {
+		log.Errorf("Creating Snapshotter client failed. Err: %v", err)
+		return err
+	}
+
 	// Create k8s Informer and watch on configmaps and namespaces.
-	informer := k8s.NewInformer(ctx, k8sClient, true)
+	informer := k8s.NewInformer(ctx, k8sClient, snapshotterClient)
 	// Configmap informer to watch on SV featurestate config-map.
 	err = informer.AddConfigMapListener(
 		ctx,

pkg/kubernetes/informers.go

@@ -18,10 +18,10 @@ package kubernetes
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"time"
 
+	snapclientset "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned"
 	"github.com/kubernetes-csi/external-snapshotter/client/v8/informers/externalversions"
 	"k8s.io/client-go/informers"
 	v1 "k8s.io/client-go/informers/core/v1"
@@ -46,8 +46,6 @@ const (
 )
 
 var (
-	inClusterInformerManagerInstance  *InformerManager = nil
-	inClusterInformerInstanceLock                      = &sync.Mutex{}
 	supervisorInformerManagerInstance *InformerManager = nil
 	supervisorInformerInstanceLock                     = &sync.Mutex{}
 )
@@ -56,48 +54,30 @@ func noResyncPeriodFunc() time.Duration {
 	return 0
 }
 
-// NewInformer creates a new K8S client based on a service account.
-// NOTE: This function expects caller function to pass appropriate client
+// NewInformer creates a new K8S informer manager.
+// NOTE: This function expects caller function to pass appropriate clients
 // as per config to be created Informer for.
-// This function creates shared informer factory against the client provided.
-func NewInformer(ctx context.Context, client clientset.Interface, inClusterClnt bool) *InformerManager {
+// This function creates shared informer factories against the clients provided.
+func NewInformer(ctx context.Context,
+	client clientset.Interface,
+	snapshotClient snapclientset.Interface) *InformerManager {
 	var informerInstance *InformerManager
 	log := logger.GetLogger(ctx)
 
-	if inClusterClnt {
-		inClusterInformerInstanceLock.Lock()
-		defer inClusterInformerInstanceLock.Unlock()
-
-		informerInstance = inClusterInformerManagerInstance
-	} else {
-		supervisorInformerInstanceLock.Lock()
-		defer supervisorInformerInstanceLock.Unlock()
-
-		informerInstance = supervisorInformerManagerInstance
-	}
-
-	// TODO: check if callers can pass this
-	snapClient, err := NewSnapshotterClient(ctx)
-	if err != nil {
-		// TODO: handle error appropriately
-		log.Fatalf("unable to initialise snapshot client")
-	}
+	supervisorInformerInstanceLock.Lock()
+	defer supervisorInformerInstanceLock.Unlock()
+	informerInstance = supervisorInformerManagerInstance
 
 	if informerInstance == nil {
 		informerInstance = &InformerManager{
 			client:                  client,
 			stopCh:                  signals.SetupSignalHandler().Done(),
 			informerFactory:         informers.NewSharedInformerFactory(client, noResyncPeriodFunc()),
-			snapshotInformerFactory: externalversions.NewSharedInformerFactory(snapClient, 0),
+			snapshotInformerFactory: externalversions.NewSharedInformerFactory(snapshotClient, 0),
 		}
 
-		if inClusterClnt {
-			inClusterInformerManagerInstance = informerInstance
-			log.Info("Created new informer factory for in-cluster client")
-		} else {
-			supervisorInformerManagerInstance = informerInstance
-			log.Info("Created new informer factory for supervisor client")
-		}
+		supervisorInformerManagerInstance = informerInstance
+		log.Info("Created new informer factory for supervisor client")
 	}
 
 	return informerInstance
@@ -322,8 +302,6 @@ func (im *InformerManager) Listen() (stopCh <-chan struct{}) {
 
 	go im.snapshotInformerFactory.Start(im.stopCh)
 	cacheSync := im.snapshotInformerFactory.WaitForCacheSync(im.stopCh)
-	// TODO: remove
-	fmt.Print(cacheSync)
 	for _, isSynced := range cacheSync {
 		if !isSynced {
 			return
@@ -332,3 +310,26 @@ func (im *InformerManager) Listen() (stopCh <-chan struct{}) {
 
 	return im.stopCh
 }
+
+// NewConfigMapListener creates a new configmap listener in the given namespace.
+// NOTE: This creates a NewSharedIndexInformer everytime and does not use the informer factory.
+// Only use this function when you need a configmap listener in a different namespace than the
+// one already present in the informer factory.
+func NewConfigMapListener(ctx context.Context, client clientset.Interface, namespace string,
+	add func(obj interface{}), update func(oldObj, newObj interface{}), remove func(obj interface{})) error {
+	log := logger.GetLogger(ctx)
+	configMapInformer := v1.NewFilteredConfigMapInformer(client, namespace, resyncPeriodConfigMapInformer,
+		cache.Indexers{}, nil)
+
+	_, err := configMapInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    add,
+		UpdateFunc: update,
+		DeleteFunc: remove,
+	})
+	if err != nil {
+		return logger.LogNewErrorf(log, "failed to add event handler on configmap listener. Error: %v", err)
+	}
+	stopCh := make(chan struct{})
+	go configMapInformer.Run(stopCh)
+	return nil
+}

pkg/kubernetes/types.go

@@ -51,7 +51,7 @@ type InformerManager struct {
 	client clientset.Interface
 	// main shared informer factory
 	informerFactory informers.SharedInformerFactory
-	
+	// snapshot informer factory
 	snapshotInformerFactory externalversions.SharedInformerFactory
 
 	// main signal

pkg/syncer/cnsoperator/controller/cnsunregistervolume/util.go

@@ -21,7 +21,6 @@ import (
 	"errors"
 	"strings"
 
-	snapshotclient "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned"
 	vmoperatortypes "github.com/vmware-tanzu/vm-operator/api/v1alpha2"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -111,7 +110,7 @@ func _getVolumeUsageInfo(ctx context.Context, pvcName string, pvcNamespace strin
 		return &volumeUsageInfo, nil
 	}
 
-	volumeUsageInfo.snapshots, volumeUsageInfo.isInUse, err = getSnapshotsForPVC(ctx, pvcName, pvcNamespace, *cfg)
+	volumeUsageInfo.snapshots, volumeUsageInfo.isInUse, err = getSnapshotsForPVC(ctx, pvcName, pvcNamespace)
 	if err != nil {
 		return nil, err
 	}
@@ -222,35 +221,16 @@ func getPodsForPVC(ctx context.Context, pvcName string, pvcNamespace string,
 }
 
 // getSnapshotsForPVC returns a list of snapshots that are created for the specified PVC.
-func getSnapshotsForPVC(ctx context.Context, pvcName string, pvcNamespace string,
-	cfg rest.Config) ([]string, bool, error) {
+func getSnapshotsForPVC(ctx context.Context, pvcName string, pvcNamespace string) ([]string, bool, error) {
 	log := logger.GetLogger(ctx)
-	c, err := snapshotclient.NewForConfig(&cfg)
-	if err != nil {
-		log.Warnf("Failed to create snapshot client for PVC %q in namespace %q. Error: %q",
-			pvcName, pvcNamespace, err.Error())
-		return nil, false, errors.New("failed to create snapshot client")
-	}
-
-	// TODO: check if we can use informer cache
-	list, err := c.SnapshotV1().VolumeSnapshots(pvcNamespace).List(ctx, metav1.ListOptions{})
-	if err != nil {
-		log.Warnf("Failed to list VolumeSnapshots in namespace %q for PVC %q. Error: %q",
-			pvcNamespace, pvcName, err.Error())
-		return nil, false, errors.New("failed to list VolumeSnapshots")
-	}
-
-	var snapshots []string
-	for _, snap := range list.Items {
-		if snap.Spec.Source.PersistentVolumeClaimName == nil ||
-			*snap.Spec.Source.PersistentVolumeClaimName != pvcName {
-			continue
-		}
-
-		snapshots = append(snapshots, snap.Name)
+	if commonco.ContainerOrchestratorUtility == nil {
+		err := errors.New("ContainerOrchestratorUtility is not initialized")
+		log.Warn(err)
+		return nil, false, err
 	}
 
-	return snapshots, len(snapshots) > 0, nil
+	snaps := commonco.ContainerOrchestratorUtility.GetSnapshotsForPVC(ctx, pvcName, pvcNamespace)
+	return snaps, len(snaps) > 0, nil
 }
 
 // getGuestClustersForPVC returns a list of guest clusters that are using the specified PVC.