Fixed TestSyncerFlows unit test failure

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/csi/service/common/commonco/k8sorchestrator/k8sorchestrator.go

@@ -386,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)

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

@@ -21,6 +21,7 @@ import (
 	"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"
@@ -45,8 +46,6 @@ const (
 )
 
 var (
-	inClusterInformerManagerInstance  *InformerManager = nil
-	inClusterInformerInstanceLock                      = &sync.Mutex{}
 	supervisorInformerManagerInstance *InformerManager = nil
 	supervisorInformerInstanceLock                     = &sync.Mutex{}
 )
@@ -55,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

pkg/syncer/metadatasyncer.go

@@ -248,6 +248,13 @@ func InitMetadataSyncer(ctx context.Context, clusterFlavor cnstypes.CnsClusterFl
 		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
+	}
+
 	// Initialize the k8s orchestrator interface.
 	metadataSyncer.coCommonInterface, err = commonco.GetContainerOrchestratorInterface(ctx,
 		common.Kubernetes, clusterFlavor, COInitParams)
@@ -633,7 +640,7 @@ func InitMetadataSyncer(ctx context.Context, clusterFlavor cnstypes.CnsClusterFl
 	}
 
 	// Set up kubernetes resource listeners for metadata syncer.
-	metadataSyncer.k8sInformerManager = k8s.NewInformer(ctx, k8sClient, true)
+	metadataSyncer.k8sInformerManager = k8s.NewInformer(ctx, k8sClient, snapshotterClient)
 	err = metadataSyncer.k8sInformerManager.AddPVCListener(
 		ctx,
 		nil, // Add.

pkg/syncer/storagepool/service.go

@@ -120,7 +120,15 @@ func InitStoragePoolService(ctx context.Context,
 			log.Errorf("Creating Kubernetes client failed. Err: %v", err)
 			return
 		}
-		k8sInformerManager := k8s.NewInformer(ctx, k8sClient, true)
+
+		// Create the snapshotter client.
+		snapshotterClient, err := k8s.NewSnapshotterClient(ctx)
+		if err != nil {
+			log.Errorf("Creating Snapshotter client failed. Err: %v", err)
+			return
+		}
+
+		k8sInformerManager := k8s.NewInformer(ctx, k8sClient, snapshotterClient)
 		err = InitNodeAnnotationListener(ctx, k8sInformerManager, scWatchCntlr, spController)
 		if err != nil {
 			log.Errorf("InitNodeAnnotationListener failed. err: %v", err)

pkg/syncer/syncer_test.go

@@ -26,6 +26,8 @@ import (
 
 	"github.com/davecgh/go-spew/spew"
 	"github.com/google/uuid"
+	"github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned"
+	snapshotfake "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned/fake"
 	cnstypes "github.com/vmware/govmomi/cns/types"
 	"github.com/vmware/govmomi/find"
 	"github.com/vmware/govmomi/vim25/mo"
@@ -69,17 +71,18 @@ const (
 )
 
 var (
-	csiConfig        *cnsconfig.Config
-	ctx              context.Context
-	cnsVCenterConfig *cnsvsphere.VirtualCenterConfig
-	err              error
-	virtualCenter    *cnsvsphere.VirtualCenter
-	metadataSyncer   *metadataSyncInformer
-	k8sclient        clientset.Interface
-	dc               []*cnsvsphere.Datacenter
-	volumeManager    cnsvolumes.Manager
-	dsList           []vimtypes.ManagedObjectReference
-	cancel           context.CancelFunc
+	csiConfig         *cnsconfig.Config
+	ctx               context.Context
+	cnsVCenterConfig  *cnsvsphere.VirtualCenterConfig
+	err               error
+	virtualCenter     *cnsvsphere.VirtualCenter
+	metadataSyncer    *metadataSyncInformer
+	k8sclient         clientset.Interface
+	snapshotterClient versioned.Interface
+	dc                []*cnsvsphere.Datacenter
+	volumeManager     cnsvolumes.Manager
+	dsList            []vimtypes.ManagedObjectReference
+	cancel            context.CancelFunc
 )
 
 func TestSyncerWorkflows(t *testing.T) {
@@ -169,14 +172,17 @@ func TestSyncerWorkflows(t *testing.T) {
 	// Here we should use a faked client to avoid test inteference with running
 	// metadata syncer pod in real Kubernetes cluster.
 	k8sclient = testclient.NewSimpleClientset()
-	metadataSyncer.k8sInformerManager = k8s.NewInformer(ctx, k8sclient, true)
+	// Create a fake snapshot client for testing.
+	snapshotterClient = snapshotfake.NewSimpleClientset()
+	var err error
+	metadataSyncer.k8sInformerManager = k8s.NewInformer(ctx, k8sclient, snapshotterClient)
 	metadataSyncer.k8sInformerManager.GetPodLister()
 	metadataSyncer.pvLister = metadataSyncer.k8sInformerManager.GetPVLister()
 	metadataSyncer.pvcLister = metadataSyncer.k8sInformerManager.GetPVCLister()
 	metadataSyncer.podLister = metadataSyncer.k8sInformerManager.GetPodLister()
 	metadataSyncer.k8sInformerManager.Listen()
 
-	_, err := unittestcommon.GetFakeVolumeMigrationService(ctx,
+	_, err = unittestcommon.GetFakeVolumeMigrationService(ctx,
 		&metadataSyncer.volumeManager, metadataSyncer.configInfo.Cfg)
 	if err != nil {
 		t.Fatalf("failed to get migration service. Err: %v", err)