Add UpdateThreshold method to ndlt and use it during RemovableAt

Author: ttetyanka

cluster-autoscaler/core/scaledown/latencytracker/node_latency_tracker.go

@@ -11,6 +11,7 @@ import (
 type LatencyTracker interface {
 	ObserveDeletion(nodeName string, timestamp time.Time)
 	UpdateStateWithUnneededList(list []*apiv1.Node, currentlyInDeletion map[string]bool, timestamp time.Time)
+	UpdateThreshold(nodeName string, threshold time.Duration)
 }
 type NodeInfo struct {
 	UnneededSince time.Time
@@ -75,6 +76,17 @@ func (t *NodeLatencyTracker) ObserveDeletion(nodeName string, timestamp time.Tim
 	}
 }
 
+// UpdateThreshold updates the scale-down threshold for a tracked node.
+func (t *NodeLatencyTracker) UpdateThreshold(nodeName string, threshold time.Duration) {
+	if info, exists := t.nodes[nodeName]; exists {
+		info.Threshold = threshold
+		t.nodes[nodeName] = info
+		klog.V(2).Infof("Updated threshold for node %q to %s", nodeName, threshold)
+	} else {
+		klog.Warningf("Attempted to update threshold for unknown node %q", nodeName)
+	}
+}
+
 // GetTrackedNodes returns the names of all nodes currently tracked as unneeded.
 func (t *NodeLatencyTracker) GetTrackedNodes() []string {
 	names := make([]string, 0, len(t.nodes))

cluster-autoscaler/core/scaledown/planner/planner.go

@@ -96,7 +96,7 @@ func New(context *context.AutoscalingContext, processors *processors.Autoscaling
 	return &Planner{
 		context:               context,
 		unremovableNodes:      unremovable.NewNodes(),
-		unneededNodes:         unneededNodes,
+		unneededNodes:         unneeded.NewNodes(processors.NodeGroupConfigProcessor, resourceLimitsFinder, nlt),
 		rs:                    simulator.NewRemovalSimulator(context.ListerRegistry, context.ClusterSnapshot, deleteOptions, drainabilityRules, true),
 		actuationInjector:     scheduling.NewHintingSimulator(),
 		eligibilityChecker:    eligibility.NewChecker(processors.NodeGroupConfigProcessor),

cluster-autoscaler/core/scaledown/unneeded/nodes.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/context"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/eligibility"
+	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/latencytracker"
 	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/resource"
 	"k8s.io/autoscaler/cluster-autoscaler/metrics"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodes"
@@ -39,11 +40,12 @@ import (
 
 // Nodes tracks the state of cluster nodes that are not needed.
 type Nodes struct {
-	sdtg              scaleDownTimeGetter
-	limitsFinder      *resource.LimitsFinder
-	cachedList        []*apiv1.Node
-	byName            map[string]*node
-	unneededTimeCache map[string]time.Duration
+	sdtg               scaleDownTimeGetter
+	limitsFinder       *resource.LimitsFinder
+	nodeLatencyTracker latencytracker.LatencyTracker
+	cachedList         []*apiv1.Node
+	byName             map[string]*node
+	unneededTimeCache  map[string]time.Duration
 }
 
 type node struct {
@@ -59,11 +61,12 @@ type scaleDownTimeGetter interface {
 }
 
 // NewNodes returns a new initialized Nodes object.
-func NewNodes(sdtg scaleDownTimeGetter, limitsFinder *resource.LimitsFinder) *Nodes {
+func NewNodes(sdtg scaleDownTimeGetter, limitsFinder *resource.LimitsFinder, nlt latencytracker.LatencyTracker) *Nodes {
 	return &Nodes{
-		sdtg:              sdtg,
-		limitsFinder:      limitsFinder,
-		unneededTimeCache: make(map[string]time.Duration),
+		sdtg:               sdtg,
+		limitsFinder:       limitsFinder,
+		unneededTimeCache:  make(map[string]time.Duration),
+		nodeLatencyTracker: nlt,
 	}
 }
 
@@ -268,6 +271,9 @@ func (n *Nodes) unremovableReason(context *context.AutoscalingContext, scaleDown
 	if ready {
 		// Check how long a ready node was underutilized.
 		unneededTime, err := n.sdtg.GetScaleDownUnneededTime(nodeGroup)
+		if n.nodeLatencyTracker != nil {
+			n.nodeLatencyTracker.UpdateThreshold(node.Name, unneededTime)
+		}
 		if err != nil {
 			klog.Errorf("Error trying to get ScaleDownUnneededTime for node %s (in group: %s)", node.Name, nodeGroup.Id())
 			return simulator.UnexpectedError
@@ -278,6 +284,9 @@ func (n *Nodes) unremovableReason(context *context.AutoscalingContext, scaleDown
 	} else {
 		// Unready nodes may be deleted after a different time than underutilized nodes.
 		unreadyTime, err := n.sdtg.GetScaleDownUnreadyTime(nodeGroup)
+		if n.nodeLatencyTracker != nil {
+			n.nodeLatencyTracker.UpdateThreshold(node.Name, unreadyTime)
+		}
 		if err != nil {
 			klog.Errorf("Error trying to get ScaleDownUnreadyTime for node %s (in group: %s)", node.Name, nodeGroup.Id())
 			return simulator.UnexpectedError

cluster-autoscaler/core/scaledown/unneeded/nodes_test.go

@@ -101,7 +101,7 @@ func TestUpdate(t *testing.T) {
 		tc := tc
 		t.Run(tc.desc, func(t *testing.T) {
 			t.Parallel()
-			nodes := NewNodes(nil, nil)
+			nodes := NewNodes(nil, nil, nil)
 			nodes.Update(tc.initialNodes, initialTimestamp)
 			nodes.Update(tc.finalNodes, finalTimestamp)
 			wantNodes := len(tc.wantTimestamps)
@@ -203,7 +203,8 @@ func TestRemovableAt(t *testing.T) {
 			ctx, err := NewScaleTestAutoscalingContext(config.AutoscalingOptions{ScaleDownSimulationTimeout: 5 * time.Minute}, &fake.Clientset{}, registry, provider, nil, nil)
 			assert.NoError(t, err)
 
-			n := NewNodes(&fakeScaleDownTimeGetter{}, &resource.LimitsFinder{})
+			fakeTracker := NewFakeLatencyTracker()
+			n := NewNodes(&fakeScaleDownTimeGetter{}, &resource.LimitsFinder{}, fakeTracker)
 			n.Update(removableNodes, time.Now())
 			gotEmptyToRemove, gotDrainToRemove, _ := n.RemovableAt(&ctx, nodes.ScaleDownContext{
 				ActuationStatus:     as,
@@ -213,6 +214,16 @@ func TestRemovableAt(t *testing.T) {
 			if len(gotDrainToRemove) != tc.numDrainToRemove || len(gotEmptyToRemove) != tc.numEmptyToRemove {
 				t.Errorf("%s: getNodesToRemove() return %d, %d, want %d, %d", tc.name, len(gotEmptyToRemove), len(gotDrainToRemove), tc.numEmptyToRemove, tc.numDrainToRemove)
 			}
+			expectedThreshold := 0 * time.Second // matches fakeScaleDownTimeGetter
+			for _, node := range removableNodes {
+				nodeName := node.Node.Name
+				got, ok := fakeTracker.Observed[nodeName]
+				if !ok {
+					t.Errorf("NodeLatencyTracker not called for node %s", nodeName)
+				} else if got != expectedThreshold {
+					t.Errorf("NodeLatencyTracker called with %v for node %s, want %v", got, nodeName, expectedThreshold)
+				}
+			}
 		})
 	}
 }
@@ -342,3 +353,19 @@ func (f *fakeScaleDownTimeGetter) GetScaleDownUnneededTime(cloudprovider.NodeGro
 func (f *fakeScaleDownTimeGetter) GetScaleDownUnreadyTime(cloudprovider.NodeGroup) (time.Duration, error) {
 	return 0 * time.Second, nil
 }
+
+type fakeLatencyTracker struct {
+	Observed map[string]time.Duration
+}
+
+func NewFakeLatencyTracker() *fakeLatencyTracker {
+	return &fakeLatencyTracker{Observed: make(map[string]time.Duration)}
+}
+
+func (f *fakeLatencyTracker) UpdateThreshold(nodeName string, threshold time.Duration) {
+	f.Observed[nodeName] = threshold
+}
+func (f *fakeLatencyTracker) ObserveDeletion(nodeName string, timestamp time.Time) {
+}
+func (f *fakeLatencyTracker) UpdateStateWithUnneededList(list []*apiv1.Node, currentlyInDeletion map[string]bool, timestamp time.Time) {
+}