Merge pull request kubernetes#8952 from pravk03/ndf

feat: Add tests for NodeDeclaredFeatures

Author: k8s-ci-robot

cluster-autoscaler/core/static_autoscaler_test.go

@@ -80,6 +80,12 @@ import (
 	v1appslister "k8s.io/client-go/listers/apps/v1"
 	kube_record "k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
+
+	"k8s.io/apimachinery/pkg/util/version"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	ndf "k8s.io/component-helpers/nodedeclaredfeatures"
+	ndffeatures "k8s.io/component-helpers/nodedeclaredfeatures/features"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 type podListerMock struct {
@@ -343,7 +349,6 @@ func setupAutoscaler(config *autoscalerSetupConfig) (*StaticAutoscaler, error) {
 }
 
 // TODO: Refactor tests to use setupAutoscaler
-
 func TestStaticAutoscalerRunOnce(t *testing.T) {
 	readyNodeLister := kubernetes.NewTestNodeLister(nil)
 	allNodeLister := kubernetes.NewTestNodeLister(nil)
@@ -3343,3 +3348,267 @@ func assertNodesSoftTaintsStatus(t *testing.T, fakeClient *fake.Clientset, nodes
 		assert.Equal(t, tainted, taints.HasDeletionCandidateTaint(newNode))
 	}
 }
+
+// mockFeature is a mock implementation of the Feature interface for testing.
+type mockFeature struct {
+	name       string
+	maxVersion *version.Version
+}
+
+func (f *mockFeature) Name() string {
+	return f.name
+}
+
+func (f *mockFeature) Discover(cfg *ndf.NodeConfiguration) bool {
+	return true
+}
+
+func (f *mockFeature) InferForScheduling(podInfo *ndf.PodInfo) bool {
+	// Check if any container has an env var matching the feature name.
+	if podInfo.Spec == nil {
+		return false
+	}
+	for _, container := range podInfo.Spec.Containers {
+		for _, envVar := range container.Env {
+			if envVar.Value == f.name {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func (f *mockFeature) InferForUpdate(oldPodInfo, newPodInfo *ndf.PodInfo) bool {
+	return false
+}
+
+func (f *mockFeature) MaxVersion() *version.Version {
+	return f.maxVersion
+}
+
+func createMockFeature(name string, maxVersionStr string) ndf.Feature {
+	var v *version.Version
+	if maxVersionStr != "" {
+		v = version.MustParseSemantic(maxVersionStr)
+	}
+	return &mockFeature{
+		name:       name,
+		maxVersion: v,
+	}
+}
+
+func setupMockDeclaredFeatures(features ...string) func() {
+	nodeFeatures := make([]ndf.Feature, 0, len(features))
+	for _, feature := range features {
+		nodeFeatures = append(nodeFeatures, createMockFeature(feature, ""))
+	}
+	originalAllFeatures := ndffeatures.AllFeatures
+	ndffeatures.AllFeatures = nodeFeatures
+	return func() {
+		ndffeatures.AllFeatures = originalAllFeatures
+	}
+}
+
+func TestStaticAutoscalerWithNodeDeclaredFeatures(t *testing.T) {
+	nodeInfoTmplA := framework.NewTestNodeInfo(BuildTestNode("templateA", 1000, 1000))
+	nodeInfoTmplB := framework.NewTestNodeInfo(BuildTestNode("templateB", 1000, 1000))
+
+	// stableTime is needed to make sure the existing nodes from a group are sampled to create template NodeInfo
+	stableTime := time.Now().Add(-2 * time.Minute)
+	nodeA := BuildTestNode("initialNodeA", 1000, 1000)
+	nodeA.Status.DeclaredFeatures = []string{"FeatureA"}
+	SetNodeReadyState(nodeA, true, stableTime)
+
+	nodeB := BuildTestNode("initialNodeB", 1000, 1000)
+	nodeB.Status.DeclaredFeatures = []string{"FeatureB"}
+	SetNodeReadyState(nodeB, true, stableTime)
+
+	fillerPodOnNodeA := BuildTestPod("fillerPodA", 900, 900)
+	fillerPodOnNodeA.Spec.NodeName = "initialNodeA"
+	fillerPodOnNodeB := BuildTestPod("fillerPodB", 900, 900)
+	fillerPodOnNodeB.Spec.NodeName = "initialNodeB"
+
+	podRequiresFeatureA := BuildTestPod("podA", 900, 900, MarkUnschedulable())
+	podRequiresFeatureA.Spec.Containers[0].Env = []apiv1.EnvVar{{Name: "REQUIRE_FEATURE", Value: "FeatureA"}}
+	podRequiresFeatureB := BuildTestPod("podB", 900, 900, MarkUnschedulable())
+	podRequiresFeatureB.Spec.Containers[0].Env = []apiv1.EnvVar{{Name: "REQUIRE_FEATURE", Value: "FeatureB"}}
+	podRequiresFeatureC := BuildTestPod("podC", 900, 900, MarkUnschedulable())
+	podRequiresFeatureC.Spec.Containers[0].Env = []apiv1.EnvVar{{Name: "REQUIRE_FEATURE", Value: "FeatureC"}}
+	anotherPodRequiresFeatureA := BuildTestPod("podA2", 900, 900, MarkUnschedulable())
+	anotherPodRequiresFeatureA.Spec.Containers[0].Env = []apiv1.EnvVar{{Name: "REQUIRE_FEATURE", Value: "FeatureA"}}
+
+	options := config.AutoscalingOptions{
+		NodeGroupDefaults: config.NodeGroupAutoscalingOptions{
+			ScaleDownUnneededTime:         time.Minute,
+			ScaleDownUnreadyTime:          time.Minute,
+			ScaleDownUtilizationThreshold: 0.5,
+			MaxNodeProvisionTime:          10 * time.Second,
+		},
+		EstimatorName:                  estimator.BinpackingEstimatorName,
+		EnforceNodeGroupMinSize:        true,
+		ScaleDownEnabled:               false,
+		MaxNodesTotal:                  10,
+		MaxCoresTotal:                  10,
+		MaxMemoryTotal:                 1000,
+		MaxNodeGroupBinpackingDuration: 1 * time.Second,
+	}
+
+	type testCase struct {
+		name                        string
+		nodeDeclaredFeaturesEnabled bool
+		declaredFeatures            []string
+		initialNodes                []*apiv1.Node
+		pods                        []*apiv1.Pod
+		expectedScaleUps            []scaleCall
+	}
+	testCases := []testCase{
+		{
+			name:                        "Scale up node group A due to pod requiring FeatureA",
+			nodeDeclaredFeaturesEnabled: true,
+			declaredFeatures:            []string{"FeatureA", "FeatureB"},
+			initialNodes:                []*apiv1.Node{nodeA, nodeB},
+			pods:                        []*apiv1.Pod{fillerPodOnNodeA, podRequiresFeatureA},
+			expectedScaleUps:            []scaleCall{{ng: "nodeGroupA", delta: 1}},
+		},
+		{
+			name:                        "Feature gate disabled - No scale up as pod requiring FeatureA can be scheduled on other nodes",
+			nodeDeclaredFeaturesEnabled: false,
+			declaredFeatures:            []string{"FeatureA", "FeatureB"},
+			initialNodes:                []*apiv1.Node{nodeA, nodeB},
+			pods:                        []*apiv1.Pod{fillerPodOnNodeA, podRequiresFeatureA},
+			expectedScaleUps:            nil,
+		},
+		{
+			name:                        "Two pods require FeatureA - Scale up by 2",
+			nodeDeclaredFeaturesEnabled: true,
+			declaredFeatures:            []string{"FeatureA", "FeatureB"},
+			initialNodes:                []*apiv1.Node{nodeA, nodeB},
+			pods:                        []*apiv1.Pod{fillerPodOnNodeA, podRequiresFeatureA, anotherPodRequiresFeatureA},
+			expectedScaleUps:            []scaleCall{{ng: "nodeGroupA", delta: 2}},
+		},
+		{
+			name:                        "No scale up - sufficient capacity on existing node",
+			nodeDeclaredFeaturesEnabled: true,
+			declaredFeatures:            []string{"FeatureA", "FeatureB"},
+			initialNodes:                []*apiv1.Node{nodeA, nodeB},
+			pods:                        []*apiv1.Pod{podRequiresFeatureB},
+			expectedScaleUps:            nil,
+		},
+		{
+			name:                        "Scale up node group B due to pod requiring FeatureB",
+			nodeDeclaredFeaturesEnabled: true,
+			declaredFeatures:            []string{"FeatureA", "FeatureB"},
+			initialNodes:                []*apiv1.Node{nodeA, nodeB},
+			pods:                        []*apiv1.Pod{fillerPodOnNodeB, podRequiresFeatureB},
+			expectedScaleUps:            []scaleCall{{ng: "nodeGroupB", delta: 1}},
+		},
+		{
+			name:                        "No scale up when pod requiring feature is not present on any node group",
+			nodeDeclaredFeaturesEnabled: false,
+			declaredFeatures:            []string{"FeatureA", "FeatureB"},
+			initialNodes:                []*apiv1.Node{nodeA, nodeB},
+			pods:                        []*apiv1.Pod{podRequiresFeatureC},
+			expectedScaleUps:            nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			allPodListerMock := &podListerMock{}
+			podDisruptionBudgetListerMock := &podDisruptionBudgetListerMock{}
+			daemonSetListerMock := &daemonSetListerMock{}
+			onScaleUpMock := &onScaleUpMock{}
+			onScaleDownMock := &onScaleDownMock{}
+
+			// Feature gate setup
+			utilfeature.DefaultMutableFeatureGate.Set(fmt.Sprintf("%s=%v", features.NodeDeclaredFeatures, tc.nodeDeclaredFeaturesEnabled))
+			cleanup := setupMockDeclaredFeatures(tc.declaredFeatures...)
+			defer cleanup()
+
+			readyNodeLister := kubernetes.NewTestNodeLister(tc.initialNodes)
+			allNodeLister := kubernetes.NewTestNodeLister(tc.initialNodes)
+
+			provider := testprovider.NewTestCloudProviderBuilder().WithOnScaleUp(func(id string, delta int) error {
+				return onScaleUpMock.ScaleUp(id, delta)
+			}).WithMachineTemplates(map[string]*framework.NodeInfo{
+				"nodeGroupA": nodeInfoTmplA,
+				"nodeGroupB": nodeInfoTmplB,
+			}).Build()
+
+			// Group nodes by features to add to an appropriate node group
+			nodesByFeature := make(map[string][]*apiv1.Node)
+			for _, node := range tc.initialNodes {
+				if len(node.Status.DeclaredFeatures) > 0 {
+					feature := node.Status.DeclaredFeatures[0]
+					nodesByFeature[feature] = append(nodesByFeature[feature], node)
+				}
+			}
+
+			if nodes, ok := nodesByFeature["FeatureA"]; ok {
+				provider.AddNodeGroup("nodeGroupA", 1, 10, len(nodes))
+				for _, node := range nodes {
+					provider.AddNode("nodeGroupA", node)
+				}
+			}
+			if nodes, ok := nodesByFeature["FeatureB"]; ok {
+				provider.AddNodeGroup("nodeGroupB", 1, 10, len(nodes))
+				for _, node := range nodes {
+					provider.AddNode("nodeGroupB", node)
+				}
+			}
+
+			processorCallbacks := newStaticAutoscalerProcessorCallbacks()
+			listerRegistry := kube_util.NewListerRegistry(allNodeLister, readyNodeLister, allPodListerMock, podDisruptionBudgetListerMock, daemonSetListerMock, nil, nil, nil, nil)
+			autoscalingCtx, err := NewScaleTestAutoscalingContext(options, &fake.Clientset{}, listerRegistry, provider, processorCallbacks, nil)
+			assert.NoError(t, err)
+
+			for _, node := range tc.initialNodes {
+				nodeInfo := framework.NewNodeInfo(node, nil)
+				err = autoscalingCtx.ClusterSnapshot.AddNodeInfo(nodeInfo)
+				assert.NoError(t, err)
+			}
+
+			processors := processorstest.NewTestProcessors(&autoscalingCtx)
+			clusterStateConfig := clusterstate.ClusterStateRegistryConfig{OkTotalUnreadyCount: 2}
+			clusterState := clusterstate.NewClusterStateRegistry(provider, clusterStateConfig, autoscalingCtx.LogRecorder, NewBackoff(), processors.NodeGroupConfigProcessor, processors.AsyncNodeGroupStateChecker)
+
+			sdPlanner, sdActuator := newScaleDownPlannerAndActuator(&autoscalingCtx, processors, clusterState, nil)
+			autoscalingCtx.ScaleDownActuator = sdActuator
+			quotasTrackerFactory := newQuotasTrackerFactory(&autoscalingCtx, processors)
+			suOrchestrator := orchestrator.New()
+			suOrchestrator.Initialize(&autoscalingCtx, processors, clusterState, newEstimatorBuilder(), taints.TaintConfig{}, quotasTrackerFactory)
+
+			autoscaler := &StaticAutoscaler{
+				AutoscalingContext:    &autoscalingCtx,
+				clusterStateRegistry:  clusterState,
+				lastScaleUpTime:       time.Now().Add(-time.Hour),
+				lastScaleDownFailTime: time.Now().Add(-time.Hour),
+				scaleUpOrchestrator:   suOrchestrator,
+				scaleDownPlanner:      sdPlanner,
+				scaleDownActuator:     sdActuator,
+				processors:            processors,
+				loopStartNotifier:     loopstart.NewObserversList(nil),
+				processorCallbacks:    processorCallbacks,
+				initialized:           true,
+			}
+
+			// Update ClusterStateRegistry with initial nodes
+			err = clusterState.UpdateNodes(tc.initialNodes, make(map[string]*framework.NodeInfo), time.Now())
+			assert.NoError(t, err)
+
+			allPodListerMock.On("List").Return(tc.pods, nil).Once()
+			daemonSetListerMock.On("List", labels.Everything()).Return([]*appsv1.DaemonSet{}, nil).Once()
+			podDisruptionBudgetListerMock.On("List").Return([]*policyv1.PodDisruptionBudget{}, nil).Once()
+
+			// Expected scale up calls
+			for _, scaleUp := range tc.expectedScaleUps {
+				onScaleUpMock.On("ScaleUp", scaleUp.ng, scaleUp.delta).Return(nil).Once()
+			}
+
+			// Run the autoscaler
+			err = autoscaler.RunOnce(time.Now())
+			assert.NoError(t, err)
+			mock.AssertExpectationsForObjects(t, allPodListerMock, podDisruptionBudgetListerMock, daemonSetListerMock, onScaleUpMock, onScaleDownMock)
+		})
+	}
+}