Merge pull request #7874 from norbertcyran/threadsafe-add-taints

Make AddTaints and CleanTaints thread safe

Author: k8s-ci-robot

cluster-autoscaler/core/scaledown/actuation/actuator.go

@@ -391,7 +391,7 @@ func (a *Actuator) scaleDownNodeToReport(node *apiv1.Node, drain bool) (*status.
 
 // taintNode taints the node with NoSchedule to prevent new pods scheduling on it.
 func (a *Actuator) taintNode(node *apiv1.Node) error {
-	if err := taints.MarkToBeDeleted(node, a.ctx.ClientSet, a.ctx.CordonNodeBeforeTerminate); err != nil {
+	if _, err := taints.MarkToBeDeleted(node, a.ctx.ClientSet, a.ctx.CordonNodeBeforeTerminate); err != nil {
 		a.ctx.Recorder.Eventf(node, apiv1.EventTypeWarning, "ScaleDownFailed", "failed to mark the node as toBeDeleted/unschedulable: %v", err)
 		return errors.ToAutoscalerError(errors.ApiCallError, err)
 	}

cluster-autoscaler/core/scaledown/actuation/actuator_test.go

@@ -1302,9 +1302,7 @@ func runStartDeletionTest(t *testing.T, tc startDeletionTestCase, force bool) {
 	// Verify ScaleDownNodes looks as expected.
 	ignoreSdNodeOrder := cmpopts.SortSlices(func(a, b *status.ScaleDownNode) bool { return a.Node.Name < b.Node.Name })
 	cmpNg := cmp.Comparer(func(a, b *testprovider.TestNodeGroup) bool { return a.Id() == b.Id() })
-	// Nodes will have deletion taints, skipping them here since we check them later
-	ignoreTaints := cmpopts.IgnoreFields(apiv1.NodeSpec{}, "Taints")
-	statusCmpOpts := cmp.Options{ignoreSdNodeOrder, cmpNg, cmpopts.EquateEmpty(), ignoreTaints}
+	statusCmpOpts := cmp.Options{ignoreSdNodeOrder, cmpNg, cmpopts.EquateEmpty()}
 	if diff := cmp.Diff(wantScaleDownNodes, gotScaleDownNodes, statusCmpOpts); diff != "" {
 		t.Errorf("StartDeletion scaled down nodes diff (-want +got):\n%s", diff)
 	}

cluster-autoscaler/core/scaledown/actuation/delete_in_batch.go

@@ -206,7 +206,7 @@ func CleanUpAndRecordFailedScaleDownEvent(ctx *context.AutoscalingContext, node
 		klog.Errorf("Scale-down: "+logMsgFormat+", %v, status error: %v", node.Name, errMsg, status.Err)
 	}
 	ctx.Recorder.Eventf(node, apiv1.EventTypeWarning, "ScaleDownFailed", eventMsgFormat+": %v", status.Err)
-	taints.CleanToBeDeleted(node.DeepCopy(), ctx.ClientSet, ctx.CordonNodeBeforeTerminate)
+	taints.CleanToBeDeleted(node, ctx.ClientSet, ctx.CordonNodeBeforeTerminate)
 	nodeDeletionTracker.EndDeletion(nodeGroupId, node.Name, status)
 }
 

cluster-autoscaler/core/scaledown/actuation/softtaint.go

@@ -60,7 +60,7 @@ func UpdateSoftDeletionTaints(context *context.AutoscalingContext, uneededNodes,
 			continue
 		}
 		b.processWithinBudget(func() {
-			err := taints.MarkDeletionCandidate(node, context.ClientSet)
+			_, err := taints.MarkDeletionCandidate(node, context.ClientSet)
 			if err != nil {
 				errors = append(errors, err)
 				klog.Warningf("Soft taint on %s adding error %v", node.Name, err)

cluster-autoscaler/utils/taints/taints.go

@@ -19,6 +19,7 @@ package taints
 import (
 	"context"
 	"fmt"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -160,7 +161,7 @@ func taintKeys(taints []apiv1.Taint) []string {
 }
 
 // MarkToBeDeleted sets a taint that makes the node unschedulable.
-func MarkToBeDeleted(node *apiv1.Node, client kube_client.Interface, cordonNode bool) error {
+func MarkToBeDeleted(node *apiv1.Node, client kube_client.Interface, cordonNode bool) (*apiv1.Node, error) {
 	taint := apiv1.Taint{
 		Key:    ToBeDeletedTaint,
 		Value:  fmt.Sprint(time.Now().Unix()),
@@ -170,7 +171,7 @@ func MarkToBeDeleted(node *apiv1.Node, client kube_client.Interface, cordonNode
 }
 
 // MarkDeletionCandidate sets a soft taint that makes the node preferably unschedulable.
-func MarkDeletionCandidate(node *apiv1.Node, client kube_client.Interface) error {
+func MarkDeletionCandidate(node *apiv1.Node, client kube_client.Interface) (*apiv1.Node, error) {
 	taint := apiv1.Taint{
 		Key:    DeletionCandidateTaint,
 		Value:  fmt.Sprint(time.Now().Unix()),
@@ -179,8 +180,8 @@ func MarkDeletionCandidate(node *apiv1.Node, client kube_client.Interface) error
 	return AddTaints(node, client, []apiv1.Taint{taint}, false)
 }
 
-// AddTaints sets the specified taints on the node.
-func AddTaints(node *apiv1.Node, client kube_client.Interface, taints []apiv1.Taint, cordonNode bool) error {
+// AddTaints sets the specified taints on the node and returns an updated copy of the node.
+func AddTaints(node *apiv1.Node, client kube_client.Interface, taints []apiv1.Taint, cordonNode bool) (*apiv1.Node, error) {
 	retryDeadline := time.Now().Add(maxRetryDeadline)
 	freshNode := node.DeepCopy()
 	var err error
@@ -191,7 +192,7 @@ func AddTaints(node *apiv1.Node, client kube_client.Interface, taints []apiv1.Ta
 			freshNode, err = client.CoreV1().Nodes().Get(context.TODO(), node.Name, metav1.GetOptions{})
 			if err != nil || freshNode == nil {
 				klog.Warningf("Error while adding %v taints on node %v: %v", strings.Join(taintKeys(taints), ","), node.Name, err)
-				return fmt.Errorf("failed to get node %v: %v", node.Name, err)
+				return nil, fmt.Errorf("failed to get node %v: %v", node.Name, err)
 			}
 		}
 
@@ -201,7 +202,7 @@ func AddTaints(node *apiv1.Node, client kube_client.Interface, taints []apiv1.Ta
 				refresh = true
 				continue
 			}
-			return nil
+			return freshNode, nil
 		}
 		_, err = client.CoreV1().Nodes().Update(context.TODO(), freshNode, metav1.UpdateOptions{})
 		if err != nil && errors.IsConflict(err) && time.Now().Before(retryDeadline) {
@@ -212,11 +213,10 @@ func AddTaints(node *apiv1.Node, client kube_client.Interface, taints []apiv1.Ta
 
 		if err != nil {
 			klog.Warningf("Error while adding %v taints on node %v: %v", strings.Join(taintKeys(taints), ","), node.Name, err)
-			return err
+			return nil, err
 		}
 		klog.V(1).Infof("Successfully added %v on node %v", strings.Join(taintKeys(taints), ","), node.Name)
-		node.Spec.Taints = append([]apiv1.Taint{}, freshNode.Spec.Taints...)
-		return nil
+		return freshNode, nil
 	}
 }
 
@@ -286,17 +286,17 @@ func GetTaintTime(node *apiv1.Node, taintKey string) (*time.Time, error) {
 }
 
 // CleanToBeDeleted cleans CA's NoSchedule taint from a node.
-func CleanToBeDeleted(node *apiv1.Node, client kube_client.Interface, cordonNode bool) (bool, error) {
+func CleanToBeDeleted(node *apiv1.Node, client kube_client.Interface, cordonNode bool) (*apiv1.Node, error) {
 	return CleanTaints(node, client, []string{ToBeDeletedTaint}, cordonNode)
 }
 
 // CleanDeletionCandidate cleans CA's soft NoSchedule taint from a node.
-func CleanDeletionCandidate(node *apiv1.Node, client kube_client.Interface) (bool, error) {
+func CleanDeletionCandidate(node *apiv1.Node, client kube_client.Interface) (*apiv1.Node, error) {
 	return CleanTaints(node, client, []string{DeletionCandidateTaint}, false)
 }
 
-// CleanTaints cleans the specified taints from a node.
-func CleanTaints(node *apiv1.Node, client kube_client.Interface, taintKeys []string, cordonNode bool) (bool, error) {
+// CleanTaints cleans the specified taints from a node and returns an updated copy of the node.
+func CleanTaints(node *apiv1.Node, client kube_client.Interface, taintKeys []string, cordonNode bool) (*apiv1.Node, error) {
 	retryDeadline := time.Now().Add(maxRetryDeadline)
 	freshNode := node.DeepCopy()
 	var err error
@@ -307,7 +307,7 @@ func CleanTaints(node *apiv1.Node, client kube_client.Interface, taintKeys []str
 			freshNode, err = client.CoreV1().Nodes().Get(context.TODO(), node.Name, metav1.GetOptions{})
 			if err != nil || freshNode == nil {
 				klog.Warningf("Error while removing %v taints from node %v: %v", strings.Join(taintKeys, ","), node.Name, err)
-				return false, fmt.Errorf("failed to get node %v: %v", node.Name, err)
+				return nil, fmt.Errorf("failed to get node %v: %v", node.Name, err)
 			}
 		}
 		newTaints := make([]apiv1.Taint, 0)
@@ -330,7 +330,7 @@ func CleanTaints(node *apiv1.Node, client kube_client.Interface, taintKeys []str
 				refresh = true
 				continue
 			}
-			return false, nil
+			return freshNode, nil
 		}
 
 		freshNode.Spec.Taints = newTaints
@@ -348,11 +348,10 @@ func CleanTaints(node *apiv1.Node, client kube_client.Interface, taintKeys []str
 
 		if err != nil {
 			klog.Warningf("Error while releasing %v taints on node %v: %v", strings.Join(taintKeys, ","), node.Name, err)
-			return false, err
+			return nil, err
 		}
 		klog.V(1).Infof("Successfully released %v on node %v", strings.Join(taintKeys, ","), node.Name)
-		node.Spec.Taints = append([]apiv1.Taint{}, freshNode.Spec.Taints...)
-		return true, nil
+		return freshNode, nil
 	}
 }
 
@@ -376,11 +375,11 @@ func CleanAllTaints(nodes []*apiv1.Node, client kube_client.Interface, recorder
 		if !taintsPresent {
 			continue
 		}
-		cleaned, err := CleanTaints(node, client, taintKeys, cordonNode)
+		updatedNode, err := CleanTaints(node, client, taintKeys, cordonNode)
 		if err != nil {
 			recorder.Eventf(node, apiv1.EventTypeWarning, "ClusterAutoscalerCleanup",
 				"failed to clean %v on node %v: %v", strings.Join(taintKeys, ","), node.Name, err)
-		} else if cleaned {
+		} else if node != nil && updatedNode != nil && !slices.Equal(updatedNode.Spec.Taints, node.Spec.Taints) {
 			recorder.Eventf(node, apiv1.EventTypeNormal, "ClusterAutoscalerCleanup",
 				"removed %v taints from node %v", strings.Join(taintKeys, ","), node.Name)
 		}

cluster-autoscaler/utils/taints/taints_test.go

@@ -19,6 +19,7 @@ package taints
 import (
 	"context"
 	"fmt"
+	"slices"
 	"strconv"
 	"sync/atomic"
 	"testing"
@@ -44,7 +45,7 @@ func TestMarkNodes(t *testing.T) {
 	defer setConflictRetryInterval(setConflictRetryInterval(time.Millisecond))
 	node := BuildTestNode("node", 1000, 1000)
 	fakeClient := buildFakeClientWithConflicts(t, node)
-	err := MarkToBeDeleted(node, fakeClient, false)
+	_, err := MarkToBeDeleted(node, fakeClient, false)
 	assert.NoError(t, err)
 
 	updatedNode := getNode(t, fakeClient, "node")
@@ -56,7 +57,7 @@ func TestSoftMarkNodes(t *testing.T) {
 	defer setConflictRetryInterval(setConflictRetryInterval(time.Millisecond))
 	node := BuildTestNode("node", 1000, 1000)
 	fakeClient := buildFakeClientWithConflicts(t, node)
-	err := MarkDeletionCandidate(node, fakeClient)
+	_, err := MarkDeletionCandidate(node, fakeClient)
 	assert.NoError(t, err)
 
 	updatedNode := getNode(t, fakeClient, "node")
@@ -115,7 +116,7 @@ func TestQueryNodes(t *testing.T) {
 	defer setConflictRetryInterval(setConflictRetryInterval(time.Millisecond))
 	node := BuildTestNode("node", 1000, 1000)
 	fakeClient := buildFakeClientWithConflicts(t, node)
-	err := MarkToBeDeleted(node, fakeClient, false)
+	_, err := MarkToBeDeleted(node, fakeClient, false)
 	assert.NoError(t, err)
 
 	updatedNode := getNode(t, fakeClient, "node")
@@ -131,7 +132,7 @@ func TestSoftQueryNodes(t *testing.T) {
 	defer setConflictRetryInterval(setConflictRetryInterval(time.Millisecond))
 	node := BuildTestNode("node", 1000, 1000)
 	fakeClient := buildFakeClientWithConflicts(t, node)
-	err := MarkDeletionCandidate(node, fakeClient)
+	_, err := MarkDeletionCandidate(node, fakeClient)
 	assert.NoError(t, err)
 
 	updatedNode := getNode(t, fakeClient, "node")
@@ -161,20 +162,21 @@ func TestCleanNodes(t *testing.T) {
 	addTaintsToSpec(node, taints, false)
 	fakeClient := buildFakeClientWithConflicts(t, node)
 
-	updatedNode := getNode(t, fakeClient, "node")
-	assert.True(t, HasToBeDeletedTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
-	assert.False(t, updatedNode.Spec.Unschedulable)
+	apiNode := getNode(t, fakeClient, "node")
+	assert.True(t, HasToBeDeletedTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
+	assert.False(t, apiNode.Spec.Unschedulable)
 
-	cleaned, err := CleanToBeDeleted(node, fakeClient, false)
+	updatedNode, err := CleanToBeDeleted(node, fakeClient, false)
+	cleaned := !slices.Equal(updatedNode.Spec.Taints, node.Spec.Taints)
 	assert.True(t, cleaned)
 	assert.NoError(t, err)
 
-	updatedNode = getNode(t, fakeClient, "node")
+	apiNode = getNode(t, fakeClient, "node")
 	assert.NoError(t, err)
-	assert.False(t, HasToBeDeletedTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
-	assert.False(t, updatedNode.Spec.Unschedulable)
+	assert.False(t, HasToBeDeletedTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
+	assert.False(t, apiNode.Spec.Unschedulable)
 }
 
 func TestCleanNodesWithCordon(t *testing.T) {
@@ -195,20 +197,21 @@ func TestCleanNodesWithCordon(t *testing.T) {
 	addTaintsToSpec(node, taints, true)
 	fakeClient := buildFakeClientWithConflicts(t, node)
 
-	updatedNode := getNode(t, fakeClient, "node")
-	assert.True(t, HasToBeDeletedTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
-	assert.True(t, updatedNode.Spec.Unschedulable)
+	apiNode := getNode(t, fakeClient, "node")
+	assert.True(t, HasToBeDeletedTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
+	assert.True(t, apiNode.Spec.Unschedulable)
 
-	cleaned, err := CleanToBeDeleted(node, fakeClient, true)
+	updatedNode, err := CleanToBeDeleted(node, fakeClient, true)
+	cleaned := !slices.Equal(updatedNode.Spec.Taints, node.Spec.Taints)
 	assert.True(t, cleaned)
 	assert.NoError(t, err)
 
-	updatedNode = getNode(t, fakeClient, "node")
+	apiNode = getNode(t, fakeClient, "node")
 	assert.NoError(t, err)
-	assert.False(t, HasToBeDeletedTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
-	assert.False(t, updatedNode.Spec.Unschedulable)
+	assert.False(t, HasToBeDeletedTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
+	assert.False(t, apiNode.Spec.Unschedulable)
 }
 
 func TestCleanNodesWithCordonOnOff(t *testing.T) {
@@ -229,20 +232,21 @@ func TestCleanNodesWithCordonOnOff(t *testing.T) {
 	addTaintsToSpec(node, taints, true)
 	fakeClient := buildFakeClientWithConflicts(t, node)
 
-	updatedNode := getNode(t, fakeClient, "node")
-	assert.True(t, HasToBeDeletedTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
-	assert.True(t, updatedNode.Spec.Unschedulable)
+	apiNode := getNode(t, fakeClient, "node")
+	assert.True(t, HasToBeDeletedTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
+	assert.True(t, apiNode.Spec.Unschedulable)
 
-	cleaned, err := CleanToBeDeleted(node, fakeClient, false)
+	updatedNode, err := CleanToBeDeleted(node, fakeClient, false)
+	cleaned := !slices.Equal(updatedNode.Spec.Taints, node.Spec.Taints)
 	assert.True(t, cleaned)
 	assert.NoError(t, err)
 
-	updatedNode = getNode(t, fakeClient, "node")
+	apiNode = getNode(t, fakeClient, "node")
 	assert.NoError(t, err)
-	assert.False(t, HasToBeDeletedTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
-	assert.True(t, updatedNode.Spec.Unschedulable)
+	assert.False(t, HasToBeDeletedTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
+	assert.True(t, apiNode.Spec.Unschedulable)
 }
 
 func TestSoftCleanNodes(t *testing.T) {
@@ -263,18 +267,19 @@ func TestSoftCleanNodes(t *testing.T) {
 	addTaintsToSpec(node, taints, false)
 	fakeClient := buildFakeClientWithConflicts(t, node)
 
-	updatedNode := getNode(t, fakeClient, "node")
-	assert.True(t, HasDeletionCandidateTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
+	apiNode := getNode(t, fakeClient, "node")
+	assert.True(t, HasDeletionCandidateTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
 
-	cleaned, err := CleanDeletionCandidate(node, fakeClient)
+	updatedNode, err := CleanDeletionCandidate(node, fakeClient)
+	cleaned := !slices.Equal(updatedNode.Spec.Taints, node.Spec.Taints)
 	assert.True(t, cleaned)
 	assert.NoError(t, err)
 
-	updatedNode = getNode(t, fakeClient, "node")
+	apiNode = getNode(t, fakeClient, "node")
 	assert.NoError(t, err)
-	assert.False(t, HasDeletionCandidateTaint(updatedNode))
-	assert.True(t, HasTaint(updatedNode, "other-taint"))
+	assert.False(t, HasDeletionCandidateTaint(apiNode))
+	assert.True(t, HasTaint(apiNode, "other-taint"))
 }
 
 func TestCleanAllToBeDeleted(t *testing.T) {
@@ -768,11 +773,11 @@ func TestAddTaints(t *testing.T) {
 					Effect: apiv1.TaintEffectNoSchedule,
 				}
 			}
-			err := AddTaints(n, fakeClient, newTaints, false)
+			updatedNode, err := AddTaints(n, fakeClient, newTaints, false)
 			assert.NoError(t, err)
 			apiNode := getNode(t, fakeClient, "node")
 			for _, want := range tc.wantTaints {
-				assert.True(t, HasTaint(n, want))
+				assert.True(t, HasTaint(updatedNode, want))
 				assert.True(t, HasTaint(apiNode, want))
 			}
 		})
@@ -829,7 +834,8 @@ func TestCleanTaints(t *testing.T) {
 			n.Spec.Taints = append([]apiv1.Taint{}, existingTaints...)
 			fakeClient := buildFakeClient(t, n)
 
-			modified, err := CleanTaints(n, fakeClient, tc.taintsToRemove, false)
+			updatedNode, err := CleanTaints(n, fakeClient, tc.taintsToRemove, false)
+			modified := !slices.Equal(updatedNode.Spec.Taints, n.Spec.Taints)
 			assert.NoError(t, err)
 			assert.Equal(t, tc.wantModified, modified)
 
@@ -842,7 +848,7 @@ func TestCleanTaints(t *testing.T) {
 
 			for _, removed := range tc.taintsToRemove {
 				assert.False(t, HasTaint(apiNode, removed))
-				assert.False(t, HasTaint(n, removed), "Taint %s should have been removed from local node object", removed)
+				assert.False(t, HasTaint(updatedNode, removed), "Taint %s should have been removed from local node object", removed)
 			}
 		})
 	}