CC refactor iteration 4 (#224)

Author: cirisked

internal/actions/delete_node_handler.go

@@ -51,6 +51,7 @@ type DeleteNodeHandler struct {
 	cfg       deleteNodeConfig
 }
 
+// nolint: gocognit
 func (h *DeleteNodeHandler) Handle(ctx context.Context, action *castai.ClusterAction) error {
 	if action == nil {
 		return fmt.Errorf("action is nil %w", k8s.ErrAction)
@@ -115,7 +116,7 @@ func (h *DeleteNodeHandler) Handle(ctx context.Context, action *castai.ClusterAc
 	}
 
 	podsListingBackoff := waitext.NewConstantBackoff(h.cfg.podsTerminationWait)
-	var pods []v1.Pod
+	var pods []*v1.Pod
 	err = waitext.Retry(
 		ctx,
 		podsListingBackoff,
@@ -127,7 +128,10 @@ func (h *DeleteNodeHandler) Handle(ctx context.Context, action *castai.ClusterAc
 			if err != nil {
 				return true, err
 			}
-			pods = podList.Items
+			pods = make([]*v1.Pod, len(podList.Items))
+			for i := range podList.Items {
+				pods[i] = &podList.Items[i]
+			}
 			return false, nil
 		},
 		func(err error) {

internal/actions/drain_node_handler.go

@@ -64,8 +64,8 @@ type DrainNodeHandler struct {
 }
 
 type nodePods struct {
-	toEvict      []v1.Pod
-	nonEvictable []v1.Pod
+	toEvict      []*v1.Pod
+	nonEvictable []*v1.Pod
 }
 
 func (h *DrainNodeHandler) Handle(ctx context.Context, action *castai.ClusterAction) error {
@@ -103,7 +103,7 @@ func (h *DrainNodeHandler) Handle(ctx context.Context, action *castai.ClusterAct
 
 	log.Info("cordoning node for draining")
 
-	if err := k8s.CordonNode(ctx, h.log, h.clientset, node); err != nil {
+	if err = k8s.CordonNode(ctx, h.log, h.clientset, node); err != nil {
 		return fmt.Errorf("cordoning node %q: %w", req.NodeName, err)
 	}
 
@@ -177,7 +177,7 @@ func (h *DrainNodeHandler) Handle(ctx context.Context, action *castai.ClusterAct
 // waitForVolumeDetachIfEnabled waits for VolumeAttachments to be deleted if the feature is enabled.
 // This is called after successful drain to give CSI drivers time to clean up volumes.
 // nonEvictablePods are pods that won't be evicted (DaemonSet, static) - their was are excluded from waiting.
-func (h *DrainNodeHandler) waitForVolumeDetachIfEnabled(ctx context.Context, log logrus.FieldLogger, nodeName string, req *castai.ActionDrainNode, nonEvictablePods []v1.Pod) {
+func (h *DrainNodeHandler) waitForVolumeDetachIfEnabled(ctx context.Context, log logrus.FieldLogger, nodeName string, req *castai.ActionDrainNode, nonEvictablePods []*v1.Pod) {
 	if !ShouldWaitForVolumeDetach(req) || h.vaWaiter == nil {
 		return
 	}
@@ -204,7 +204,7 @@ func (h *DrainNodeHandler) waitForVolumeDetachIfEnabled(ctx context.Context, log
 // The method will still wait for termination of other evicted pods first.
 // Returns non-evictable pods (DaemonSet, static).
 // A return value of (pods, nil) means all evictable pods on the node should be evicted and terminated.
-func (h *DrainNodeHandler) evictNodePods(ctx context.Context, log logrus.FieldLogger, node *v1.Node) ([]v1.Pod, error) {
+func (h *DrainNodeHandler) evictNodePods(ctx context.Context, log logrus.FieldLogger, node *v1.Node) ([]*v1.Pod, error) {
 	nodePods, err := h.listNodePods(ctx, log, node)
 	if err != nil {
 		return nil, err
@@ -257,7 +257,7 @@ func (h *DrainNodeHandler) evictNodePods(ctx context.Context, log logrus.FieldLo
 // The method will still wait for termination of other deleted pods first.
 // Returns non-evictable pods (DaemonSet, static).
 // A return value of (pods, nil) means all evictable pods on the node should be deleted and terminated.
-func (h *DrainNodeHandler) deleteNodePods(ctx context.Context, log logrus.FieldLogger, node *v1.Node, options metav1.DeleteOptions) ([]v1.Pod, error) {
+func (h *DrainNodeHandler) deleteNodePods(ctx context.Context, log logrus.FieldLogger, node *v1.Node, options metav1.DeleteOptions) ([]*v1.Pod, error) {
 	nodePods, err := h.listNodePods(ctx, log, node)
 	if err != nil {
 		return nil, err
@@ -338,11 +338,16 @@ func (h *DrainNodeHandler) listNodePods(ctx context.Context, log logrus.FieldLog
 		return nil, fmt.Errorf("listing node %v pods: %w", node.Name, err)
 	}
 
-	partitioned := k8s.PartitionPodsForEviction(pods.Items, h.cfg.castNamespace, h.cfg.skipDeletedTimeoutSeconds)
+	podPtrs := make([]*v1.Pod, len(pods.Items))
+	for i := range pods.Items {
+		podPtrs[i] = &pods.Items[i]
+	}
+
+	partitioned := k8s.PartitionPodsForEviction(podPtrs, h.cfg.castNamespace, h.cfg.skipDeletedTimeoutSeconds)
 
 	result := &nodePods{
-		toEvict:      make([]v1.Pod, 0, len(partitioned.Evictable)+len(partitioned.CastPods)),
-		nonEvictable: make([]v1.Pod, 0, len(partitioned.NonEvictable)),
+		toEvict:      make([]*v1.Pod, 0, len(partitioned.Evictable)+len(partitioned.CastPods)),
+		nonEvictable: make([]*v1.Pod, 0, len(partitioned.NonEvictable)),
 	}
 
 	logCastPodsToEvict(log, partitioned.CastPods)
@@ -381,7 +386,7 @@ func (h *DrainNodeHandler) waitNodePodsTerminated(ctx context.Context, log logru
 				return true, fmt.Errorf("listing %q pods to be terminated: %w", node.Name, err)
 			}
 
-			podsNames := lo.Map(nodePods.toEvict, func(p v1.Pod, _ int) string {
+			podsNames := lo.Map(nodePods.toEvict, func(p *v1.Pod, _ int) string {
 				return fmt.Sprintf("%s/%s", p.Namespace, p.Name)
 			})
 
@@ -403,7 +408,7 @@ func (h *DrainNodeHandler) waitNodePodsTerminated(ctx context.Context, log logru
 	)
 }
 
-func logCastPodsToEvict(log logrus.FieldLogger, castPods []v1.Pod) {
+func logCastPodsToEvict(log logrus.FieldLogger, castPods []*v1.Pod) {
 	if len(castPods) == 0 {
 		return
 	}

internal/actions/drain_node_handler_test.go

@@ -83,7 +83,7 @@ func TestLogCastPodsToEvict(t *testing.T) {
 	t.Run("should log pods to evict", func(t *testing.T) {
 		r := require.New(t)
 		log, hook := test.NewNullLogger()
-		pods := []v1.Pod{
+		pods := []*v1.Pod{
 			{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "ns1"}},
 			{ObjectMeta: metav1.ObjectMeta{Name: "pod2", Namespace: "ns2"}},
 		}
@@ -97,7 +97,7 @@ func TestLogCastPodsToEvict(t *testing.T) {
 		r := require.New(t)
 		log, hook := test.NewNullLogger()
 
-		var pods []v1.Pod
+		var pods []*v1.Pod
 		logCastPodsToEvict(log, pods)
 
 		r.Len(hook.Entries, 0)

internal/k8s/kubernetes.go

@@ -33,10 +33,11 @@ const (
 )
 
 var (
-	ErrAction            = errors.New("not valid action")
-	ErrNodeNotFound      = errors.New("node not found")
-	ErrNodeDoesNotMatch  = fmt.Errorf("node does not match")
-	ErrNodeWatcherClosed = fmt.Errorf("node watcher closed, no more events will be received")
+	ErrAction             = errors.New("not valid action")
+	ErrNodeNotFound       = errors.New("node not found")
+	ErrNodeDoesNotMatch   = fmt.Errorf("node does not match")
+	ErrProviderIDMismatch = errors.New("provider ID mismatch")
+	ErrNodeWatcherClosed  = fmt.Errorf("node watcher closed, no more events will be received")
 )
 
 const (
@@ -179,7 +180,7 @@ func (c *Client) GetNodeByIDs(ctx context.Context, nodeName, nodeID, providerID
 // actionName might be used to distinguish what is the operation (for logs, debugging, etc.) but is optional.
 func (c *Client) ExecuteBatchPodActions(
 	ctx context.Context,
-	pods []v1.Pod,
+	pods []*v1.Pod,
 	action func(context.Context, v1.Pod) error,
 	actionName string,
 ) ([]*v1.Pod, []PodActionFailure) {
@@ -201,7 +202,7 @@ func (c *Client) ExecuteBatchPodActions(
 		wg                 sync.WaitGroup
 	)
 
-	logger.Debugf("Starting %d parallel tasks for %d pods: [%v]", parallelTasks, len(pods), lo.Map(pods, func(t v1.Pod, i int) string {
+	logger.Debugf("Starting %d parallel tasks for %d pods: [%v]", parallelTasks, len(pods), lo.Map(pods, func(t *v1.Pod, i int) string {
 		return fmt.Sprintf("%s/%s", t.Namespace, t.Name)
 	}))
 
@@ -226,7 +227,7 @@ func (c *Client) ExecuteBatchPodActions(
 	}
 
 	for _, pod := range pods {
-		taskChan <- pod
+		taskChan <- *pod
 	}
 
 	close(taskChan)
@@ -256,7 +257,6 @@ func (c *Client) EvictPod(ctx context.Context, pod v1.Pod, podEvictRetryDelay ti
 	action := func(ctx context.Context) (bool, error) {
 		var err error
 
-		c.log.Debugf("requesting eviction for pod %s/%s", pod.Namespace, pod.Name)
 		if version == policyv1.SchemeGroupVersion {
 			err = c.clientset.PolicyV1().Evictions(pod.Namespace).Evict(ctx, &policyv1.Eviction{
 				ObjectMeta: metav1.ObjectMeta{
@@ -402,15 +402,15 @@ type PodActionFailure struct {
 }
 
 type PartitionResult struct {
-	Evictable    []v1.Pod
-	NonEvictable []v1.Pod
-	CastPods     []v1.Pod
+	Evictable    []*v1.Pod
+	NonEvictable []*v1.Pod
+	CastPods     []*v1.Pod
 }
 
-func PartitionPodsForEviction(pods []v1.Pod, castNamespace string, skipDeletedTimeoutSeconds int) *PartitionResult {
-	castPods := make([]v1.Pod, 0)
-	evictable := make([]v1.Pod, 0)
-	nonEvictable := make([]v1.Pod, 0)
+func PartitionPodsForEviction(pods []*v1.Pod, castNamespace string, skipDeletedTimeoutSeconds int) *PartitionResult {
+	castPods := make([]*v1.Pod, 0)
+	evictable := make([]*v1.Pod, 0)
+	nonEvictable := make([]*v1.Pod, 0)
 
 	for _, p := range pods {
 		// Skip pods that have been recently removed.
@@ -424,7 +424,7 @@ func PartitionPodsForEviction(pods []v1.Pod, castNamespace string, skipDeletedTi
 			continue
 		}
 
-		if IsDaemonSetPod(&p) || IsStaticPod(&p) {
+		if IsDaemonSetPod(p) || IsStaticPod(p) {
 			nonEvictable = append(nonEvictable, p)
 			continue
 		}
@@ -502,7 +502,7 @@ func GetNodeByIDs(ctx context.Context, clientSet corev1client.NodeInterface, nod
 func ExecuteBatchPodActions(
 	ctx context.Context,
 	log logrus.FieldLogger,
-	pods []v1.Pod,
+	pods []*v1.Pod,
 	action func(context.Context, v1.Pod) error,
 	actionName string,
 ) ([]*v1.Pod, []PodActionFailure) {

internal/k8s/kubernetes_test.go

@@ -380,7 +380,7 @@ func TestClient_ExecuteBatchPodActions(t *testing.T) {
 
 	tests := []struct {
 		name              string
-		pods              []v1.Pod
+		pods              []*v1.Pod
 		action            func(context.Context, v1.Pod) error
 		actionName        string
 		wantSuccessCount  int
@@ -389,15 +389,15 @@ func TestClient_ExecuteBatchPodActions(t *testing.T) {
 	}{
 		{
 			name:             "empty pod list",
-			pods:             []v1.Pod{},
+			pods:             []*v1.Pod{},
 			action:           func(ctx context.Context, pod v1.Pod) error { return nil },
 			actionName:       "test-action",
 			wantSuccessCount: 0,
 			wantFailureCount: 0,
 		},
 		{
 			name: "all pods succeed",
-			pods: []v1.Pod{
+			pods: []*v1.Pod{
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "default"}},
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod2", Namespace: "default"}},
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod3", Namespace: "default"}},
@@ -409,7 +409,7 @@ func TestClient_ExecuteBatchPodActions(t *testing.T) {
 		},
 		{
 			name: "all pods fail",
-			pods: []v1.Pod{
+			pods: []*v1.Pod{
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "default"}},
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod2", Namespace: "default"}},
 			},
@@ -421,7 +421,7 @@ func TestClient_ExecuteBatchPodActions(t *testing.T) {
 		},
 		{
 			name: "mixed success and failure",
-			pods: []v1.Pod{
+			pods: []*v1.Pod{
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "default"}},
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod-fail", Namespace: "default"}},
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod2", Namespace: "default"}},
@@ -439,7 +439,7 @@ func TestClient_ExecuteBatchPodActions(t *testing.T) {
 		},
 		{
 			name: "default action name when empty",
-			pods: []v1.Pod{
+			pods: []*v1.Pod{
 				{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "default"}},
 			},
 			action:            func(ctx context.Context, pod v1.Pod) error { return fmt.Errorf("fail") },

internal/volume/detachment_waiter.go

@@ -29,7 +29,7 @@ type DetachmentWaitOptions struct {
 
 	// PodsToExclude are pods whose VolumeAttachments should not be waited for
 	// (e.g., DaemonSet pods, static pods that won't be evicted).
-	PodsToExclude []v1.Pod
+	PodsToExclude []*v1.Pod
 }
 
 // DetachmentWaiter waits for VolumeAttachments to be detached from a node.
@@ -99,15 +99,15 @@ func (w *detachmentWaiter) Wait(
 
 // getExcludedPVsForNode returns PV names that should NOT be waited for.
 // These are PVs used by podsToExclude (e.g., DaemonSets, static pods) since
-// those pods won't be evicted and would cause a deadlock waiting for their VAs.
+// those pods won't be evicted and would cause a deadlock waiting for their VolumeAttachments.
 func (w *detachmentWaiter) getExcludedPVsForNode(
 	ctx context.Context,
 	log logrus.FieldLogger,
-	podsToExclude []v1.Pod,
+	podsToExclude []*v1.Pod,
 ) map[string]struct{} {
 	excludedPVs := make(map[string]struct{})
 	for i := range podsToExclude {
-		pod := &podsToExclude[i]
+		pod := podsToExclude[i]
 		for _, vol := range pod.Spec.Volumes {
 			if vol.PersistentVolumeClaim == nil {
 				continue
@@ -126,8 +126,8 @@ func (w *detachmentWaiter) getExcludedPVsForNode(
 
 // waitForVolumeDetach waits for all VolumeAttachments on the node to be deleted,
 // except those belonging to excludedPVs (e.g., DaemonSet pods).
-// Returns nil when all VAs are deleted.
-// Returns DetachmentError on timeout with the list of remaining VAs.
+// Returns nil when all VolumeAttachments are deleted.
+// Returns DetachmentError on timeout with the list of remaining VolumeAttachments.
 // Respects context cancellation.
 func (w *detachmentWaiter) waitForVolumeDetach(
 	ctx context.Context,

internal/volume/detachment_waiter_test.go

@@ -20,11 +20,11 @@ import (
 	"github.com/castai/cluster-controller/internal/informer"
 )
 
-func newTestWaiterVAInformer(t *testing.T, vas []*storagev1.VolumeAttachment, additionalObjs ...runtime.Object) (cache.Indexer, kubernetes.Interface) {
+func newTestWaiterVAInformer(t *testing.T, volumeAttachments []*storagev1.VolumeAttachment, additionalObjs ...runtime.Object) (cache.Indexer, kubernetes.Interface) {
 	t.Helper()
 
-	objs := make([]runtime.Object, 0, len(vas)+len(additionalObjs))
-	for _, va := range vas {
+	objs := make([]runtime.Object, 0, len(volumeAttachments)+len(additionalObjs))
+	for _, va := range volumeAttachments {
 		objs = append(objs, va)
 	}
 	objs = append(objs, additionalObjs...)
@@ -63,7 +63,7 @@ func vaStrPtr(s string) *string {
 func TestDetachmentWaiter_Wait(t *testing.T) {
 	t.Parallel()
 
-	t.Run("should return immediately when no VAs on node", func(t *testing.T) {
+	t.Run("should return immediately when no VolumeAttachments on node", func(t *testing.T) {
 		synctest.Test(t, func(t *testing.T) {
 			r := require.New(t)
 			log := logrus.New()
@@ -79,7 +79,7 @@ func TestDetachmentWaiter_Wait(t *testing.T) {
 		})
 	})
 
-	t.Run("should complete when VAs are deleted", func(t *testing.T) {
+	t.Run("should complete when VolumeAttachments are deleted", func(t *testing.T) {
 		synctest.Test(t, func(t *testing.T) {
 			r := require.New(t)
 			log := logrus.New()
@@ -163,7 +163,7 @@ func TestDetachmentWaiter_Wait(t *testing.T) {
 		})
 	})
 
-	t.Run("should only wait for VAs on specified node", func(t *testing.T) {
+	t.Run("should only wait for VolumeAttachments on specified node", func(t *testing.T) {
 		synctest.Test(t, func(t *testing.T) {
 			r := require.New(t)
 			log := logrus.New()
@@ -199,7 +199,7 @@ func TestDetachmentWaiter_Wait(t *testing.T) {
 		})
 	})
 
-	t.Run("should exclude VAs from excluded pods", func(t *testing.T) {
+	t.Run("should exclude VolumeAttachments from excluded pods", func(t *testing.T) {
 		synctest.Test(t, func(t *testing.T) {
 			r := require.New(t)
 			log := logrus.New()
@@ -256,13 +256,13 @@ func TestDetachmentWaiter_Wait(t *testing.T) {
 			err := waiter.Wait(context.Background(), log, DetachmentWaitOptions{
 				NodeName:      "node1",
 				Timeout:       2 * time.Second,
-				PodsToExclude: []v1.Pod{dsPod},
+				PodsToExclude: []*v1.Pod{&dsPod},
 			})
 			r.NoError(err)
 		})
 	})
 
-	t.Run("should exclude VAs from static pods", func(t *testing.T) {
+	t.Run("should exclude VolumeAttachments from static pods", func(t *testing.T) {
 		synctest.Test(t, func(t *testing.T) {
 			r := require.New(t)
 			log := logrus.New()
@@ -319,7 +319,7 @@ func TestDetachmentWaiter_Wait(t *testing.T) {
 			err := waiter.Wait(context.Background(), log, DetachmentWaitOptions{
 				NodeName:      "node1",
 				Timeout:       2 * time.Second,
-				PodsToExclude: []v1.Pod{staticPod},
+				PodsToExclude: []*v1.Pod{&staticPod},
 			})
 			r.NoError(err)
 		})