Initial implementation of using informer caches for snapshots

Author: kolluria

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/go-version v1.6.0
 	github.com/kubernetes-csi/csi-proxy/v2 v2.0.0-alpha.1
-	github.com/kubernetes-csi/external-snapshotter/client/v8 v8.2.0
+	github.com/kubernetes-csi/external-snapshotter/client/v8 v8.4.0
 	github.com/onsi/ginkgo/v2 v2.22.0
 	github.com/onsi/gomega v1.36.0
 	github.com/pkg/sftp v1.13.6

go.sum

@@ -178,8 +178,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kubernetes-csi/csi-proxy/v2 v2.0.0-alpha.1 h1:tVPvlL5N5X598hrO3g9rhyoi6h0LP4RpSJlGHItsbEE=
 github.com/kubernetes-csi/csi-proxy/v2 v2.0.0-alpha.1/go.mod h1:pacx+PW7lLlu6kAvpr8Lgq/5fdiAsKxOtXXFHMaLMb8=
-github.com/kubernetes-csi/external-snapshotter/client/v8 v8.2.0 h1:Q3jQ1NkFqv5o+F8dMmHd8SfEmlcwNeo1immFApntEwE=
-github.com/kubernetes-csi/external-snapshotter/client/v8 v8.2.0/go.mod h1:E3vdYxHj2C2q6qo8/Da4g7P+IcwqRZyy3gJBzYybV9Y=
+github.com/kubernetes-csi/external-snapshotter/client/v8 v8.4.0 h1:bMqrb3UHgHbP+PW9VwiejfDJU1R0PpXVZNMdeH8WYKI=
+github.com/kubernetes-csi/external-snapshotter/client/v8 v8.4.0/go.mod h1:E3vdYxHj2C2q6qo8/Da4g7P+IcwqRZyy3gJBzYybV9Y=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de h1:9TO3cAIGXtEhnIaL+V+BEER86oLrvS+kWobKpbJuye0=

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

@@ -29,6 +29,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	"google.golang.org/grpc/codes"
 	"k8s.io/client-go/util/retry"
 
@@ -239,6 +240,31 @@ 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)
+}
+
+type snapshotInfo struct {
+	snapshotName        string
+	snapshotContentName string
+}
+
+func (s snapshotInfo) String() string {
+	return fmt.Sprintf("name: %s, content name: %s", s.snapshotName, s.snapshotContentName)
+}
+
+// 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.
+type pvcToSnapshotsMap struct {
+	*sync.RWMutex
+	items map[namespacedName]map[namespacedName]snapshotInfo
+}
+
 // K8sOrchestrator defines set of properties specific to K8s.
 type K8sOrchestrator struct {
 	supervisorFSS        FSSConfigMapInfo
@@ -251,6 +277,7 @@ type K8sOrchestrator struct {
 	nodeIDToNameMap      *nodeIDToNameMap
 	volumeNameToNodesMap *volumeNameToNodesMap // used when ListVolume FSS is enabled
 	volumeIDToNameMap    *volumeIDToNameMap    // used when ListVolume FSS is enabled
+	pvcToSnapshotsMap    *pvcToSnapshotsMap
 	k8sClient            clientset.Interface
 	snapshotterClient    snapshotterClientSet.Interface
 	// pvcUIDCache maps PVC UID to its namespaced name (namespace/name).
@@ -381,6 +408,12 @@ func Newk8sOrchestrator(ctx context.Context, controllerClusterFlavor cnstypes.Cn
 				}
 			}
 
+			// Initialize the map for pvc to snapshots
+			err := initPVCToSnapshotsMap(ctx, controllerClusterFlavor)
+			if err != nil {
+				return nil, fmt.Errorf("failed to create PVC to snapshots map. Error: %v", err)
+			}
+
 			k8sOrchestratorInstance.informerManager.Listen()
 			atomic.StoreUint32(&k8sOrchestratorInstanceInitialized, 1)
 			log.Info("k8sOrchestratorInstance initialized")
@@ -2286,3 +2319,46 @@ 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 mapping is only required for WCP.
+		return nil
+	}
+
+	log.Debugf("Initializing pvc namespaced name to volumesnapshot names map")
+	k8sOrchestratorInstance.pvcToSnapshotsMap = &pvcToSnapshotsMap{
+		RWMutex: &sync.RWMutex{},
+		items:   make(map[namespacedName]map[namespacedName]snapshotInfo),
+	}
+
+	snapshotAdded := func(obj interface{}) {
+		snap, ok := obj.(*snapshotv1.VolumeSnapshot)
+		if !ok || snap == nil {
+			log.Warnf("snapshotAdded: unrecognized object %+v", obj)
+			return
+		}
+
+		// TODO: implement
+		log.Infof("snapshotAdded: snapshot=%v", snap)
+	}
+
+	err := k8sOrchestratorInstance.informerManager.AddSnapshotListener(ctx,
+		func(obj interface{}) {
+			snapshotAdded(obj)
+		},
+		func(oldObj, newObj interface{}) {
+			// TODO: implement
+			log.Info("snapshotUpdated")
+		},
+		func(obj interface{}) {
+			// TODO: implement
+			log.Info("snapshotDeleted")
+		})
+	if err != nil {
+		return logger.LogNewErrorf(log, "failed to listen on volumesnapshots. Error: %v", err)
+	}
+
+	return nil
+}

pkg/kubernetes/informers.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/kubernetes-csi/external-snapshotter/client/v8/informers/externalversions"
 	"k8s.io/client-go/informers"
 	v1 "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
@@ -74,11 +75,19 @@ func NewInformer(ctx context.Context, client clientset.Interface, inClusterClnt
 		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")
+	}
+
 	if informerInstance == nil {
 		informerInstance = &InformerManager{
-			client:          client,
-			stopCh:          signals.SetupSignalHandler().Done(),
-			informerFactory: informers.NewSharedInformerFactory(client, noResyncPeriodFunc()),
+			client:                  client,
+			stopCh:                  signals.SetupSignalHandler().Done(),
+			informerFactory:         informers.NewSharedInformerFactory(client, noResyncPeriodFunc()),
+			snapshotInformerFactory: externalversions.NewSharedInformerFactory(snapClient, 0),
 		}
 
 		if inClusterClnt {
@@ -256,6 +265,28 @@ func (im *InformerManager) AddVolumeAttachmentListener(ctx context.Context, add
 	return nil
 }
 
+// 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 {
+	log := logger.GetLogger(ctx)
+	if im.snapshotInformer == nil {
+		im.snapshotInformer = im.snapshotInformerFactory.
+			Snapshot().V1().VolumeSnapshots().Informer()
+	}
+
+	_, err := im.snapshotInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    add,
+		UpdateFunc: update,
+		DeleteFunc: remove,
+	})
+	if err != nil {
+		return logger.LogNewErrorf(
+			log, "failed to add event handler on snapshot listener. Error: %v", err)
+	}
+
+	return nil
+}
+
 // GetPVLister returns PV Lister for the calling informer manager.
 func (im *InformerManager) GetPVLister() corelisters.PersistentVolumeLister {
 	return im.informerFactory.Core().V1().PersistentVolumes().Lister()
@@ -287,26 +318,3 @@ 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

@@ -17,6 +17,7 @@ limitations under the License.
 package kubernetes
 
 import (
+	"github.com/kubernetes-csi/external-snapshotter/client/v8/informers/externalversions"
 	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
@@ -50,6 +51,10 @@ type InformerManager struct {
 	client clientset.Interface
 	// main shared informer factory
 	informerFactory informers.SharedInformerFactory
+
+	// snapshotInformerFactory - TODO: update this
+	snapshotInformerFactory externalversions.SharedInformerFactory
+
 	// main signal
 	stopCh (<-chan struct{})
 
@@ -83,4 +88,7 @@ type InformerManager struct {
 
 	// volume attachment informer
 	volumeAttachmentInformer cache.SharedInformer
+
+	// snapshot informer
+	snapshotInformer cache.SharedInformer
 }