Fix kubevirt failover

Fixes an auto-failover issue with kubevirt.

Author: jharrod

frontend/crd/controller/crd_controller.go

@@ -228,7 +228,10 @@ func NewTridentCrdController(
 
 	indexers := indexers.NewIndexers(clients.KubeClient)
 	indexers.Activate()
-	remediationUtils := NewNodeRemediationUtils(clients.KubeClient, orchestrator, indexers)
+
+	pvcGetter := getPVCGetter(ctx, orchestrator)
+
+	remediationUtils := NewNodeRemediationUtils(clients.KubeClient, orchestrator, indexers, pvcGetter)
 
 	return newTridentCrdControllerImpl(orchestrator, clients.Namespace, clients.KubeClient,
 		clients.SnapshotClient, clients.TridentClient, indexers, remediationUtils)
@@ -1249,13 +1252,16 @@ func (c *TridentCrdController) SetForceDetach(enableForceDetach bool) {
 
 // getPVCGetter returns the Kubernetes helper frontend from the orchestrator (by name) for cache-first
 // PVC lookup, or nil if not available (e.g. plain CSI helper or not yet registered).
-func (c *TridentCrdController) getPVCGetter(ctx context.Context) k8shelper.K8SControllerHelperPlugin {
-	frontend, err := c.orchestrator.GetFrontend(ctx, controllerhelpers.KubernetesHelper)
+func getPVCGetter(ctx context.Context, orchestrator core.Orchestrator) k8shelper.K8SControllerHelperPlugin {
+	if orchestrator == nil {
+		return nil
+	}
+	frontend, err := orchestrator.GetFrontend(ctx, controllerhelpers.KubernetesHelper)
 	if err != nil {
 		return nil
 	}
-	if k8s, ok := frontend.(k8shelper.K8SControllerHelperPlugin); ok {
-		return k8s
+	if helper, ok := frontend.(k8shelper.K8SControllerHelperPlugin); ok {
+		return helper
 	}
 	return nil
 }

frontend/crd/controller/crd_controller_test.go

@@ -678,20 +678,14 @@ func TestDeleteCRHandler(t *testing.T) {
 // TestGetPVCGetter exercises getPVCGetter: GetFrontend(KubernetesHelper) success, error, and wrong type.
 func TestGetPVCGetter(t *testing.T) {
 	ctx := context.Background()
-	tridentNamespace := "trident"
-	kubeClient := GetTestKubernetesClientset()
-	snapClient := GetTestSnapshotClientset()
-	crdClient := GetTestCrdClientset()
 
 	t.Run("returns_K8s_helper_when_GetFrontend_returns_it", func(t *testing.T) {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
 		orchestrator := mockcore.NewMockOrchestrator(mockCtrl)
 		mockK8s := mockk8s.NewMockK8SControllerHelperPlugin(mockCtrl)
 		orchestrator.EXPECT().GetFrontend(gomock.Any(), controllerhelpers.KubernetesHelper).Return(mockK8s, nil)
-		controller, err := newTridentCrdControllerImpl(orchestrator, tridentNamespace, kubeClient, snapClient, crdClient, nil, nil)
-		require.NoError(t, err)
-		got := controller.getPVCGetter(ctx)
+		got := getPVCGetter(ctx, orchestrator)
 		assert.Same(t, mockK8s, got)
 	})
 
@@ -700,9 +694,7 @@ func TestGetPVCGetter(t *testing.T) {
 		defer mockCtrl.Finish()
 		orchestrator := mockcore.NewMockOrchestrator(mockCtrl)
 		orchestrator.EXPECT().GetFrontend(gomock.Any(), controllerhelpers.KubernetesHelper).Return(nil, fmt.Errorf("not found"))
-		controller, err := newTridentCrdControllerImpl(orchestrator, tridentNamespace, kubeClient, snapClient, crdClient, nil, nil)
-		require.NoError(t, err)
-		got := controller.getPVCGetter(ctx)
+		got := getPVCGetter(ctx, orchestrator)
 		assert.Nil(t, got)
 	})
 
@@ -712,9 +704,7 @@ func TestGetPVCGetter(t *testing.T) {
 		orchestrator := mockcore.NewMockOrchestrator(mockCtrl)
 		// Frontend is nil (e.g. not yet registered); type assertion in getPVCGetter yields nil.
 		orchestrator.EXPECT().GetFrontend(gomock.Any(), controllerhelpers.KubernetesHelper).Return(nil, nil)
-		controller, err := newTridentCrdControllerImpl(orchestrator, tridentNamespace, kubeClient, snapClient, crdClient, nil, nil)
-		require.NoError(t, err)
-		got := controller.getPVCGetter(ctx)
+		got := getPVCGetter(ctx, orchestrator)
 		assert.Nil(t, got)
 	})
 }

frontend/crd/controller/indexers/indexers.go

@@ -34,27 +34,27 @@ type VolumeAttachmentIndexer interface {
 	GetCachedVolumeAttachmentsByNode(ctx context.Context, nodeName string) ([]*k8sstoragev1.VolumeAttachment, error)
 }
 
-type k8sIndexers struct {
+type K8sIndexers struct {
 	Indexer
 	vaIndexer VolumeAttachmentIndexer
 }
 
-func NewIndexers(kubeClient kubernetes.Interface) *k8sIndexers {
-	indexers := &k8sIndexers{}
+func NewIndexers(kubeClient kubernetes.Interface) *K8sIndexers {
+	indexers := &K8sIndexers{}
 	indexers.vaIndexer = vaindexer.NewVolumeAttachmentIndexer(kubeClient)
 	return indexers
 }
 
 // Activate starts all the indexers
-func (k *k8sIndexers) Activate() {
+func (k *K8sIndexers) Activate() {
 	k.vaIndexer.Activate()
 }
 
 // Deactivate stops all the indexers
-func (k *k8sIndexers) Deactivate() {
+func (k *K8sIndexers) Deactivate() {
 	k.vaIndexer.Deactivate()
 }
 
-func (k *k8sIndexers) VolumeAttachmentIndexer() VolumeAttachmentIndexer {
+func (k *K8sIndexers) VolumeAttachmentIndexer() VolumeAttachmentIndexer {
 	return k.vaIndexer
 }

frontend/crd/controller/indexers/indexers_test.go

@@ -13,9 +13,22 @@ import (
 	mock_indexer "github.com/netapp/trident/mocks/mock_frontend/crd/controller/indexers/indexer"
 )
 
+func GetMockK8sIndexer(ctrl *gomock.Controller) (
+	*K8sIndexers, *mock_indexer.MockVolumeAttachmentIndexer,
+) {
+	// Create mock VolumeAttachmentIndexer
+	mockVAIndexer := mock_indexer.NewMockVolumeAttachmentIndexer(ctrl)
+
+	// Create k8sIndexers with mock
+	indexers := &K8sIndexers{
+		vaIndexer: mockVAIndexer,
+	}
+	return indexers, mockVAIndexer
+}
+
 func TestNewIndexers(t *testing.T) {
 	// Create a fake Kubernetes client
-	kubeClient := fake.NewSimpleClientset()
+	kubeClient := fake.NewClientset()
 
 	// Test creating new indexers
 	indexers := NewIndexers(kubeClient)
@@ -36,13 +49,8 @@ func TestK8sIndexers_Activate(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
-	// Create mock VolumeAttachmentIndexer
-	mockVAIndexer := mock_indexer.NewMockVolumeAttachmentIndexer(ctrl)
-
-	// Create k8sIndexers with mock
-	indexers := &k8sIndexers{
-		vaIndexer: mockVAIndexer,
-	}
+	// Create mock indexers
+	indexers, mockVAIndexer := GetMockK8sIndexer(ctrl)
 
 	// Set expectation for Activate call
 	mockVAIndexer.EXPECT().Activate().Times(1)
@@ -55,13 +63,8 @@ func TestK8sIndexers_Deactivate(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
-	// Create mock VolumeAttachmentIndexer
-	mockVAIndexer := mock_indexer.NewMockVolumeAttachmentIndexer(ctrl)
-
-	// Create k8sIndexers with mock
-	indexers := &k8sIndexers{
-		vaIndexer: mockVAIndexer,
-	}
+	// Create mock indexers
+	indexers, mockVAIndexer := GetMockK8sIndexer(ctrl)
 
 	// Set expectation for Deactivate call
 	mockVAIndexer.EXPECT().Deactivate().Times(1)
@@ -74,13 +77,8 @@ func TestK8sIndexers_VolumeAttachmentIndexer(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
-	// Create mock VolumeAttachmentIndexer
-	mockVAIndexer := mock_indexer.NewMockVolumeAttachmentIndexer(ctrl)
-
-	// Create k8sIndexers with mock
-	indexers := &k8sIndexers{
-		vaIndexer: mockVAIndexer,
-	}
+	// Create mock indexers
+	indexers, mockVAIndexer := GetMockK8sIndexer(ctrl)
 
 	// Test VolumeAttachmentIndexer returns correct indexer
 	result := indexers.VolumeAttachmentIndexer()

frontend/crd/controller/indexers/vaindexer/vaindexer.go

@@ -17,7 +17,7 @@ import (
 // Indexer keys
 const vaByNodeIndex = "vaByNode"
 
-type vaIndexer struct {
+type VaIndexer struct {
 	// Volume Attachment Indexer
 	vaIndexer            cache.Indexer
 	vaController         cache.SharedIndexInformer
@@ -26,8 +26,8 @@ type vaIndexer struct {
 	vaSynced             cache.InformerSynced
 }
 
-func NewVolumeAttachmentIndexer(kubeClient kubernetes.Interface) *vaIndexer {
-	va := &vaIndexer{}
+func NewVolumeAttachmentIndexer(kubeClient kubernetes.Interface) *VaIndexer {
+	va := &VaIndexer{}
 
 	// Init stop channels
 	va.vaControllerStopChan = make(chan struct{})
@@ -40,7 +40,7 @@ func NewVolumeAttachmentIndexer(kubeClient kubernetes.Interface) *vaIndexer {
 		fields.Everything(),
 	)
 
-	// Set cachSyncPeriod to 0 to disable resync. This is to avoid scaling concens with large number of VAs.
+	// Set cacheSyncPeriod to 0 to disable resync. This is to avoid scaling concerns with large number of VAs.
 	const cacheSyncPeriod = 0
 
 	// Set up the VolumeAttachments indexing controller
@@ -55,21 +55,21 @@ func NewVolumeAttachmentIndexer(kubeClient kubernetes.Interface) *vaIndexer {
 	return va
 }
 
-func (v *vaIndexer) Activate() {
+func (v *VaIndexer) Activate() {
 	go v.vaController.Run(v.vaControllerStopChan)
 }
 
-func (v *vaIndexer) Deactivate() {
+func (v *VaIndexer) Deactivate() {
 	close(v.vaControllerStopChan)
 }
 
-// getCachedVolumeAttachmentsByNode returns a VA list from the client's cache, or an error if not found.
-func (v *vaIndexer) GetCachedVolumeAttachmentsByNode(
+// GetCachedVolumeAttachmentsByNode returns a VA list for a node from the client's cache.
+func (v *VaIndexer) GetCachedVolumeAttachmentsByNode(
 	ctx context.Context, nodeName string,
 ) ([]*k8sstoragev1.VolumeAttachment, error) {
 	logFields := LogFields{"nodeName": nodeName}
-	Logc(ctx).WithFields(logFields).Trace(">>>> getCachedVolumeAttachmentsByNode")
-	defer Logc(ctx).Trace("<<<< getCachedVolumeAttachmentsByNode")
+	Logc(ctx).WithFields(logFields).Trace(">>>> GetCachedVolumeAttachmentsByNode")
+	defer Logc(ctx).Trace("<<<< GetCachedVolumeAttachmentsByNode")
 
 	items, err := v.vaIndexer.ByIndex(vaByNodeIndex, nodeName)
 	if err != nil {
@@ -94,14 +94,12 @@ func (v *vaIndexer) GetCachedVolumeAttachmentsByNode(
 	return attachments, nil
 }
 
-func (v *vaIndexer) WaitForCacheSync(ctx context.Context) bool {
-	successful := true
-	if ok := cache.WaitForCacheSync(v.vaControllerStopChan, v.vaSynced); !ok {
-		successful = false
-		err := fmt.Errorf("failed to wait for vaController cache to sync")
-		Logc(ctx).Errorf("Error: %v", err)
+func (v *VaIndexer) WaitForCacheSync(ctx context.Context) bool {
+	var ok bool
+	if ok = cache.WaitForCacheSync(v.vaControllerStopChan, v.vaSynced); !ok {
+		Logc(ctx).Errorf("failed to wait for vaController cache to sync")
 	}
-	return successful
+	return ok
 }
 
 // volumeAttachmentsByNodeKeyFunc is a indexer KeyFunc which knows how to make keys for VolumeAttachment by node.

frontend/crd/controller/trident_autogrow_request_internal.go

@@ -907,7 +907,7 @@ func (c *TridentCrdController) getPVCForVolume(ctx context.Context, volExternal
 		return nil, fmt.Errorf("invalid volume name")
 	}
 
-	getter := c.getPVCGetter(ctx)
+	getter := getPVCGetter(ctx, c.orchestrator)
 	if getter != nil {
 		pvc, err := getter.GetPVCForPV(ctx, volumeName)
 		if err == nil {

frontend/crd/controller/trident_node_remediation.go

@@ -313,11 +313,8 @@ func (c *TridentCrdController) failoverDetach(
 		return err
 	}
 
-	pvcToTvolMap, err := c.nodeRemediationUtils.GetPvcToTvolMap(ctx, nodeName)
-	if err != nil {
-		Logc(ctx).WithError(err).Error("Failed to get PVC to Tvol map.")
-		return err
-	}
+	nodePvcs := c.nodeRemediationUtils.GetPvcsForPods(ctx, nodePods)
+	pvcToTvolMap := c.nodeRemediationUtils.GetPvcToTvolMap(ctx, nodePvcs)
 
 	// Get pods to delete
 	Logc(ctx).WithField("node", nodeName).Info("Finding pods to remove from failed node.")