Merge pull request #113 from XenitAB/eviction-logic

Author: landerss1

.golangci.yaml

@@ -30,7 +30,7 @@ linters:
       max-complexity: 15
       package-average: 0
     dogsled:
-      max-blank-identifiers: 2
+      max-blank-identifiers: 3
     dupl:
       threshold: 100
     errcheck:

Makefile

@@ -3,11 +3,17 @@ IMG ?= ghcr.io/xenitab/node-ttl:$(TAG)
 
 all: fmt vet lint
 
+fmt:
+	go fmt ./...
+
+vet:
+	go vet ./...
+
 lint:
 	golangci-lint run ./...
 
 test:
-	go test --cover ./...
+	go test -v --cover ./...
 
 docker-build:
 	docker build -t ${IMG} .
@@ -41,17 +47,14 @@ e2e: docker-build
 	# Start node ttl
 	helm upgrade --kubeconfig $$KIND_KUBECONFIG --install --create-namespace --namespace="node-ttl" node-ttl ./charts/node-ttl --set "image.pullPolicy=Never" --set "nodeTtl.interval=10s" --set "image.tag=${TAG}"
 
-	# Run capcity check tests
-	go test ./e2e/e2e_test.go -cover -v -timeout 300s -run TestCapcityCheck
-
 	# Start pause workloads
 	kubectl --kubeconfig $$KIND_KUBECONFIG apply -f ./e2e/pause-workloads.yaml
 	kubectl --kubeconfig $$KIND_KUBECONFIG --namespace default wait --timeout=300s --for=jsonpath="{.status.active}"=1 job/pause
 	kubectl --kubeconfig $$KIND_KUBECONFIG --namespace default wait --timeout=300s --for=jsonpath="{.status.availableReplicas}"=3 deployment/pause
 	kubectl --kubeconfig $$KIND_KUBECONFIG --namespace default wait --timeout=300s --for=jsonpath="{.status.availableReplicas}"=3 statefulset/pause
 
 	# Run TTL eviction tests
-	go test ./e2e/e2e_test.go -cover -v -timeout 300s -run TestTTLEviction
+	go test ./e2e/e2e_test.go -cover -v -timeout 600s -run TestTTLEviction
 
 	# Delete cluster
 	#kind delete cluster

e2e/cluster-autoscaler.yaml

@@ -315,7 +315,7 @@ spec:
             - ./cluster-autoscaler
             - --cloud-provider=kubemark
             - --namespace=cluster-autoscaler
-            - --nodes=1:10:asg1
+            - --nodes=2:3:asg1
             - --logtostderr=true
             - --scale-down-delay-after-add=10s
             - --scale-down-delay-after-delete=10s

e2e/e2e_test.go

@@ -2,7 +2,6 @@ package e2e
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"sort"
 	"testing"
@@ -16,93 +15,56 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 )
 
-func TestCapcityCheck(t *testing.T) {
-	path := os.Getenv("KIND_KUBECONFIG")
-	cfg, err := clientcmd.BuildConfigFromFlags("", path)
-	require.NoError(t, err)
-	client, err := kubernetes.NewForConfig(cfg)
-	require.NoError(t, err)
-
-	require.Never(t, func() bool {
-		nodeList, err := client.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{LabelSelector: "xkf.xenit.io/node-ttl"})
-		require.NoError(t, err)
-		for _, node := range nodeList.Items {
-			t.Log("checking that node is not evicted", node.Name)
-			// TODO: There should be a better way to check that eviction is due to node ttl
-			return node.Spec.Unschedulable
-		}
-		return false
-	}, 1*time.Minute, 5*time.Second)
-}
-
 func TestTTLEviction(t *testing.T) {
 	path := os.Getenv("KIND_KUBECONFIG")
 	cfg, err := clientcmd.BuildConfigFromFlags("", path)
 	require.NoError(t, err)
 	client, err := kubernetes.NewForConfig(cfg)
 	require.NoError(t, err)
 
-	nodeList, err := client.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{LabelSelector: "xkf.xenit.io/node-ttl"})
+	nodeList, err := client.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{LabelSelector: "xkf.xenit.io/node-ttl,autoscaling.k8s.io/nodegroup=asg1"})
 	require.NoError(t, err)
 	nodes := nodeList.Items
 	sort.SliceStable(nodes, func(i, j int) bool {
 		return nodes[j].CreationTimestamp.After(nodes[i].CreationTimestamp.Time)
 	})
 
-	nodeNames := []string{}
-	for _, node := range nodes {
-		nodeNames = append(nodeNames, node.Name)
-	}
+	nodeMap := getNodesMap(nodes)
+	nodeNames := getNodeKeys(nodeMap)
 	t.Log("checking eviction of nodes", nodeNames)
 
-	for _, node := range nodeList.Items {
-		t.Log("waiting for node to be evicted due to TTL", node.Name)
-
-		require.Eventually(t, func() bool {
-			getNode, err := client.CoreV1().Nodes().Get(context.TODO(), node.Name, metav1.GetOptions{})
-			require.NoError(t, err)
-			// TODO: There should be a better way to check that eviction is due to node ttl
-			if !getNode.Spec.Unschedulable {
-				return false
-			}
-			return true
-		}, 2*time.Minute, 1*time.Second, "node should be evicted due to TTL")
-		t.Log("node has been marked unschedulable by node ttl", node.Name)
-
-		require.Eventually(t, func() bool {
-			podList, err := client.CoreV1().Pods("").List(context.TODO(), metav1.ListOptions{FieldSelector: fmt.Sprintf("spec.nodeName=%s", node.Name)})
-			require.NoError(t, err)
-			pods := testFilterDaemonset(podList.Items)
-			if len(pods) != 0 {
-				return false
+	// What we want to test now is that all the nodes eventually get replaced by new ones
+	require.Eventually(t, func() bool {
+		for _, name := range nodeMap {
+			_, err := client.CoreV1().Nodes().Get(context.TODO(), name, metav1.GetOptions{})
+			if apierrors.IsNotFound(err) {
+				delete(nodeMap, name)
+				t.Logf("node %s doesn't exist anymore, continuing with next one", name)
+				continue
 			}
-			return true
-		}, 30*time.Second, 1*time.Second, "node should be drained")
-		t.Log("node has been drained", node.Name)
+		}
+		return len(nodeMap) == 0
+	}, 5*time.Minute, 5*time.Second, "all nodes should have been evicted and replaced by new nodes")
 
-		// TODO: Make sure only one node is beeing evicted at once
+	nodeList, err = client.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{LabelSelector: "xkf.xenit.io/node-ttl,autoscaling.k8s.io/nodegroup=asg1"})
+	require.NoError(t, err)
+	nodeMap = getNodesMap(nodeList.Items)
+	nodeNames = getNodeKeys(nodeMap)
+	t.Log("nodes after all nodes have been evicted", nodeNames)
+}
 
-		require.Eventually(t, func() bool {
-			_, err := client.CoreV1().Nodes().Get(context.TODO(), node.Name, metav1.GetOptions{})
-			if !apierrors.IsNotFound(err) {
-				return false
-			}
-			return true
-		}, 2*time.Minute, 1*time.Second, "node should be delted")
-		t.Log("underutilized node has been deleted", node.Name)
+func getNodesMap(nodes []corev1.Node) map[string]string {
+	nodeNames := make(map[string]string)
+	for _, node := range nodes {
+		nodeNames[node.Name] = node.Name
 	}
+	return nodeNames
 }
 
-func testFilterDaemonset(pods []corev1.Pod) []corev1.Pod {
-	filteredPods := []corev1.Pod{}
-OUTER:
-	for _, pod := range pods {
-		for _, ownerRef := range pod.OwnerReferences {
-			if ownerRef.APIVersion == "apps/v1" && ownerRef.Kind == "DaemonSet" {
-				continue OUTER
-			}
-		}
-		filteredPods = append(filteredPods, pod)
+func getNodeKeys(m map[string]string) []string {
+	nodeKeys := []string{}
+	for _, node := range m {
+		nodeKeys = append(nodeKeys, node)
 	}
-	return filteredPods
+	return nodeKeys
 }

internal/status/status.go

@@ -72,17 +72,20 @@ type ClusterAutoscalerStatusConfigMap struct {
 	NodeGroups       []*NodeGroupsType `yaml:"nodeGroups"`
 }
 
-func HasScaleDownCapacity(status string, node *corev1.Node) (bool, error) {
+func CanEvictNode(status string, node *corev1.Node) (bool, error) {
 	nodePoolName, err := getNodePoolName(node)
 	if err != nil {
 		return false, err
 	}
 
-	ready, min, err := getNodePoolReadyAndMinCount(node.Status.NodeInfo.KubeletVersion, status, nodePoolName)
+	ready, min, max, err := getNodePoolCounts(node.Status.NodeInfo.KubeletVersion, status, nodePoolName)
 	if err != nil {
 		return false, err
 	}
-	if ready <= min {
+	// We evict the node if we can temporarily scale down or add a new node
+	//nolint:staticcheck // QF1001: this is exactly what the above comment states
+	if !(ready-1 >= min || ready+1 <= max) {
+		log.Printf("not safe to evict node (ready: %d, min: %d, max: %d)", ready, min, max)
 		return false, nil
 	}
 	return true, nil
@@ -119,29 +122,29 @@ func getNodePoolName(node *corev1.Node) (string, error) {
 	return "", fmt.Errorf("could not find node pool label in node: %s", node.Name)
 }
 
-func getNodePoolReadyAndMinCount(kubeletVersion, status, nodePoolName string) (int, int, error) {
+func getNodePoolCounts(kubeletVersion, status, nodePoolName string) (int, int, int, error) {
 	// Assume we are running at least v1.2.X
 	preV130 := strings.Contains(kubeletVersion, "v1.2")
 	if preV130 {
 		health, err := getNodePoolHealthPreV130(status, nodePoolName)
 		if err != nil {
-			return 0, 0, err
+			return 0, 0, 0, err
 		}
-		ready, min, err := getReadyAndMinCountPreV130(health)
-		return ready, min, err
+		ready, min, max, err := getNodePoolCountsPreV130(health)
+		return ready, min, max, err
 	}
 
 	// v1.3.X or later
 	health, err := getNodePoolHealth(status, nodePoolName)
 	if err != nil {
 		fmt.Printf("Error: %s", err.Error())
-		return 0, 0, err
+		return 0, 0, 0, err
 	}
 
 	if health.NodeCounts != nil && health.NodeCounts.Registered != nil {
-		return health.NodeCounts.Registered.Ready, health.MinSize, nil
+		return health.NodeCounts.Registered.Ready, health.MinSize, health.MaxSize, nil
 	}
-	return 0, 0, nil
+	return 0, 0, 0, nil
 }
 
 func getNodePoolHealthPreV130(status string, nodePoolName string) (string, error) {
@@ -179,21 +182,26 @@ func getNodePoolHealth(s string, nodePoolName string) (*HealthType, error) {
 	return nil, fmt.Errorf("could not find status for node pool: %s", nodePoolName)
 }
 
-func getReadyAndMinCountPreV130(health string) (int, int, error) {
-	reg := regexp.MustCompile(`Healthy \(ready=(\d+).*minSize=(\d+)`)
+func getNodePoolCountsPreV130(health string) (int, int, int, error) {
+	reg := regexp.MustCompile(`Healthy \(ready=(\d+).*minSize=(\d+).*maxSize=(\d+)`)
 	matches := reg.FindStringSubmatch(health)
 	if len(matches) != 3 {
-		return 0, 0, fmt.Errorf("expected match list to be of length 3: %d", len(matches))
+		return 0, 0, 0, fmt.Errorf("expected match list to be of length 3: %d", len(matches))
 	}
 
 	ready, err := strconv.Atoi(matches[1])
 	if err != nil {
-		return 0, 0, fmt.Errorf("could not convert ready count to int: %w", err)
+		return 0, 0, 0, fmt.Errorf("could not convert ready count to int: %w", err)
 	}
 
 	min, err := strconv.Atoi(matches[2])
 	if err != nil {
-		return 0, 0, fmt.Errorf("could not convert min count to int: %w", err)
+		return 0, 0, 0, fmt.Errorf("could not convert min count to int: %w", err)
 	}
-	return ready, min, nil
+
+	max, err := strconv.Atoi(matches[3])
+	if err != nil {
+		return 0, 0, 0, fmt.Errorf("could not convert min count to int: %w", err)
+	}
+	return ready, min, max, nil
 }

internal/status/status_test.go

@@ -9,35 +9,40 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func TestGetNodePoolReadyAndMinCountExisting(t *testing.T) {
+func TestGetNodePoolReadyAndMinMaxCountExisting(t *testing.T) {
 	nodePools := []testNodePool{
 		{
 			name:  "foo",
 			ready: 10,
 			min:   22,
+			max:   30,
 		},
 		{
 			name:  "bar",
 			ready: 25,
 			min:   1,
+			max:   30,
 		},
 		{
 			name:  "nodePool",
 			ready: 2,
 			min:   100,
+			max:   500,
 		},
 		{
 			name:  "baz",
 			ready: 87,
 			min:   35,
+			max:   100,
 		},
 	}
 	status := mockClusterAutoscalerStatus(t, nodePools)
 	for _, nodePool := range nodePools {
-		ready, min, err := getNodePoolReadyAndMinCount("v1.31.2", status, nodePool.name)
+		ready, min, max, err := getNodePoolCounts("v1.31.2", status, nodePool.name)
 		require.NoError(t, err)
 		require.Equal(t, nodePool.ready, ready)
 		require.Equal(t, nodePool.min, min)
+		require.Equal(t, nodePool.max, max)
 	}
 }
 
@@ -50,29 +55,32 @@ func TestGetNodePoolReadyAndMinCountNotFound(t *testing.T) {
 		},
 	}
 	status := mockClusterAutoscalerStatus(t, nodePools)
-	_, _, err := getNodePoolReadyAndMinCount("v1.31.2", status, "bar")
+	_, _, _, err := getNodePoolCounts("v1.31.2", status, "bar")
 	require.EqualError(t, err, "could not find status for node pool: bar")
 }
 
-func TestHasScaleDownCapacity(t *testing.T) {
+func TestCanEvictNode(t *testing.T) {
 	type test struct {
 		name   string
 		ready  int
 		min    int
+		max    int
 		isSafe bool
 	}
 
 	tests := []test{
 		{
-			name:   "safe to scale down node pool foo",
+			name:   "safe to evict node in pool foo",
 			ready:  2,
 			min:    1,
+			max:    3,
 			isSafe: true,
 		},
 		{
-			name:   "not safe to scale down node pool bar",
+			name:   "not safe to evict node in pool bar",
 			ready:  1,
 			min:    1,
+			max:    1,
 			isSafe: false,
 		},
 	}
@@ -85,9 +93,10 @@ func TestHasScaleDownCapacity(t *testing.T) {
 					name:  nodePoolName,
 					ready: tt.ready,
 					min:   tt.min,
+					max:   tt.max,
 				}
 				status := mockClusterAutoscalerStatus(t, []testNodePool{nodePool})
-				ok, err := HasScaleDownCapacity(status, node)
+				ok, err := CanEvictNode(status, node)
 				require.NoError(t, err)
 				require.Equal(t, tt.isSafe, ok)
 			}
@@ -99,6 +108,7 @@ type testNodePool struct {
 	name  string
 	ready int
 	min   int
+	max   int
 }
 
 func mockClusterAutoscalerStatus(t *testing.T, nodePools []testNodePool) string {
@@ -143,9 +153,9 @@ nodeGroups:`
       unregistered: 0
     cloudProviderTarget: %[3]d
     minSize: %[4]d
-    maxSize: 10
+    maxSize: %[5]d
     lastProbeTime: "2025-04-22T14:29:08.360891242Z"
-    lastTransitionTime: "2025-04-17T23:46:40.655271485Z"`, status, nodePool.name, nodePool.ready, nodePool.min)
+    lastTransitionTime: "2025-04-17T23:46:40.655271485Z"`, status, nodePool.name, nodePool.ready, nodePool.min, nodePool.max)
 	}
 
 	return status