E2E tests for PodTopologySpread

Author: Huang-Wei

pkg/features/kube_features.go

@@ -503,7 +503,7 @@ const (
 	EndpointSliceProxying featuregate.Feature = "EndpointSliceProxying"
 
 	// owner: @Huang-Wei
-	// alpha: v1.16
+	// beta: v1.18
 	//
 	// Schedule pods evenly across available topology domains.
 	EvenPodsSpread featuregate.Feature = "EvenPodsSpread"
@@ -618,7 +618,7 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 	IPv6DualStack:                                  {Default: false, PreRelease: featuregate.Alpha},
 	EndpointSlice:                                  {Default: true, PreRelease: featuregate.Beta},
 	EndpointSliceProxying:                          {Default: false, PreRelease: featuregate.Alpha},
-	EvenPodsSpread:                                 {Default: false, PreRelease: featuregate.Alpha},
+	EvenPodsSpread:                                 {Default: true, PreRelease: featuregate.Beta},
 	StartupProbe:                                   {Default: true, PreRelease: featuregate.Beta},
 	AllowInsecureBackendProxy:                      {Default: true, PreRelease: featuregate.Beta},
 	PodDisruptionBudget:                            {Default: true, PreRelease: featuregate.Beta},

pkg/scheduler/algorithmprovider/registry_test.go

@@ -56,6 +56,7 @@ func TestClusterAutoscalerProvider(t *testing.T) {
 				{Name: noderesources.FitName},
 				{Name: nodeports.Name},
 				{Name: interpodaffinity.Name},
+				{Name: podtopologyspread.Name},
 			},
 		},
 		Filter: &schedulerapi.PluginSet{
@@ -74,13 +75,15 @@ func TestClusterAutoscalerProvider(t *testing.T) {
 				{Name: volumebinding.Name},
 				{Name: volumezone.Name},
 				{Name: interpodaffinity.Name},
+				{Name: podtopologyspread.Name},
 			},
 		},
 		PreScore: &schedulerapi.PluginSet{
 			Enabled: []schedulerapi.Plugin{
 				{Name: interpodaffinity.Name},
 				{Name: defaultpodtopologyspread.Name},
 				{Name: tainttoleration.Name},
+				{Name: podtopologyspread.Name},
 			},
 		},
 		Score: &schedulerapi.PluginSet{
@@ -93,6 +96,7 @@ func TestClusterAutoscalerProvider(t *testing.T) {
 				{Name: nodepreferavoidpods.Name, Weight: 10000},
 				{Name: defaultpodtopologyspread.Name, Weight: 1},
 				{Name: tainttoleration.Name, Weight: 1},
+				{Name: podtopologyspread.Name, Weight: 1},
 			},
 		},
 		Bind: &schedulerapi.PluginSet{

pkg/scheduler/apis/config/testing/compatibility_test.go

@@ -1423,6 +1423,7 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
 			{Name: "NodeResourcesFit"},
 			{Name: "NodePorts"},
 			{Name: "InterPodAffinity"},
+			{Name: "PodTopologySpread"},
 		},
 		"FilterPlugin": {
 			{Name: "NodeUnschedulable"},
@@ -1439,11 +1440,13 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
 			{Name: "VolumeBinding"},
 			{Name: "VolumeZone"},
 			{Name: "InterPodAffinity"},
+			{Name: "PodTopologySpread"},
 		},
 		"PreScorePlugin": {
 			{Name: "InterPodAffinity"},
 			{Name: "DefaultPodTopologySpread"},
 			{Name: "TaintToleration"},
+			{Name: "PodTopologySpread"},
 		},
 		"ScorePlugin": {
 			{Name: "NodeResourcesBalancedAllocation", Weight: 1},
@@ -1454,6 +1457,7 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
 			{Name: "NodePreferAvoidPods", Weight: 10000},
 			{Name: "DefaultPodTopologySpread", Weight: 1},
 			{Name: "TaintToleration", Weight: 1},
+			{Name: "PodTopologySpread", Weight: 1},
 		},
 		"BindPlugin": {{Name: "DefaultBinder"}},
 	}
@@ -1483,6 +1487,7 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
 					{Name: "NodeResourcesFit"},
 					{Name: "NodePorts"},
 					{Name: "InterPodAffinity"},
+					{Name: "PodTopologySpread"},
 				},
 				"FilterPlugin": {
 					{Name: "NodeUnschedulable"},
@@ -1499,11 +1504,13 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
 					{Name: "VolumeBinding"},
 					{Name: "VolumeZone"},
 					{Name: "InterPodAffinity"},
+					{Name: "PodTopologySpread"},
 				},
 				"PreScorePlugin": {
 					{Name: "InterPodAffinity"},
 					{Name: "DefaultPodTopologySpread"},
 					{Name: "TaintToleration"},
+					{Name: "PodTopologySpread"},
 				},
 				"ScorePlugin": {
 					{Name: "NodeResourcesBalancedAllocation", Weight: 1},
@@ -1514,6 +1521,7 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
 					{Name: "NodePreferAvoidPods", Weight: 10000},
 					{Name: "DefaultPodTopologySpread", Weight: 1},
 					{Name: "TaintToleration", Weight: 1},
+					{Name: "PodTopologySpread", Weight: 1},
 				},
 				"BindPlugin": {{Name: "DefaultBinder"}},
 			},

test/e2e/scheduling/predicates.go

@@ -67,6 +67,7 @@ type pausePodConfig struct {
 	OwnerReferences                   []metav1.OwnerReference
 	PriorityClassName                 string
 	DeletionGracePeriodSeconds        *int64
+	TopologySpreadConstraints         []v1.TopologySpreadConstraint
 }
 
 var _ = SIGDescribe("SchedulerPredicates [Serial]", func() {
@@ -604,6 +605,84 @@ var _ = SIGDescribe("SchedulerPredicates [Serial]", func() {
 		ginkgo.By(fmt.Sprintf("Trying to create another pod(pod5) with hostport %v but hostIP 127.0.0.1 on the node which pod4 resides and expect not scheduled", port))
 		createHostPortPodOnNode(f, "pod5", ns, "127.0.0.1", port, v1.ProtocolTCP, nodeSelector, false)
 	})
+
+	ginkgo.Context("PodTopologySpread Filtering", func() {
+		var nodeNames []string
+		topologyKey := "kubernetes.io/e2e-pts-filter"
+
+		ginkgo.BeforeEach(func() {
+			ginkgo.By("Trying to get 2 available nodes which can run pod")
+			nodeNames = Get2NodesThatCanRunPod(f)
+			ginkgo.By(fmt.Sprintf("Apply dedicated topologyKey %v for this test on the 2 nodes.", topologyKey))
+			for _, nodeName := range nodeNames {
+				framework.AddOrUpdateLabelOnNode(cs, nodeName, topologyKey, nodeName)
+			}
+		})
+		ginkgo.AfterEach(func() {
+			for _, nodeName := range nodeNames {
+				framework.RemoveLabelOffNode(cs, nodeName, topologyKey)
+			}
+		})
+
+		ginkgo.It("validates 4 pods with MaxSkew=1 are evenly distributed into 2 nodes", func() {
+			podLabel := "e2e-pts-filter"
+			replicas := 4
+			rsConfig := pauseRSConfig{
+				Replicas: int32(replicas),
+				PodConfig: pausePodConfig{
+					Name:      podLabel,
+					Namespace: ns,
+					Labels:    map[string]string{podLabel: ""},
+					Affinity: &v1.Affinity{
+						NodeAffinity: &v1.NodeAffinity{
+							RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
+								NodeSelectorTerms: []v1.NodeSelectorTerm{
+									{
+										MatchExpressions: []v1.NodeSelectorRequirement{
+											{
+												Key:      topologyKey,
+												Operator: v1.NodeSelectorOpIn,
+												Values:   nodeNames,
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+					TopologySpreadConstraints: []v1.TopologySpreadConstraint{
+						{
+							MaxSkew:           1,
+							TopologyKey:       topologyKey,
+							WhenUnsatisfiable: v1.DoNotSchedule,
+							LabelSelector: &metav1.LabelSelector{
+								MatchExpressions: []metav1.LabelSelectorRequirement{
+									{
+										Key:      podLabel,
+										Operator: metav1.LabelSelectorOpExists,
+									},
+								},
+							},
+						},
+					},
+				},
+			}
+			runPauseRS(f, rsConfig)
+			podList, err := cs.CoreV1().Pods(ns).List(context.TODO(), metav1.ListOptions{})
+			framework.ExpectNoError(err)
+			numInNode1, numInNode2 := 0, 0
+			for _, pod := range podList.Items {
+				if pod.Spec.NodeName == nodeNames[0] {
+					numInNode1++
+				} else if pod.Spec.NodeName == nodeNames[1] {
+					numInNode2++
+				}
+			}
+			expected := replicas / len(nodeNames)
+			framework.ExpectEqual(numInNode1, expected, fmt.Sprintf("Pods are not distributed as expected on node %q", nodeNames[0]))
+			framework.ExpectEqual(numInNode2, expected, fmt.Sprintf("Pods are not distributed as expected on node %q", nodeNames[1]))
+		})
+	})
 })
 
 // printAllKubeletPods outputs status of all kubelet pods into log.
@@ -633,8 +712,9 @@ func initPausePod(f *framework.Framework, conf pausePodConfig) *v1.Pod {
 			OwnerReferences: conf.OwnerReferences,
 		},
 		Spec: v1.PodSpec{
-			NodeSelector: conf.NodeSelector,
-			Affinity:     conf.Affinity,
+			NodeSelector:              conf.NodeSelector,
+			Affinity:                  conf.Affinity,
+			TopologySpreadConstraints: conf.TopologySpreadConstraints,
 			Containers: []v1.Container{
 				{
 					Name:  conf.Name,
@@ -669,7 +749,7 @@ func createPausePod(f *framework.Framework, conf pausePodConfig) *v1.Pod {
 
 func runPausePod(f *framework.Framework, conf pausePodConfig) *v1.Pod {
 	pod := createPausePod(f, conf)
-	framework.ExpectNoError(e2epod.WaitForPodRunningInNamespace(f.ClientSet, pod))
+	framework.ExpectNoError(e2epod.WaitTimeoutForPodRunningInNamespace(f.ClientSet, pod.Name, pod.Namespace, framework.PollShortTimeout))
 	pod, err := f.ClientSet.CoreV1().Pods(pod.Namespace).Get(context.TODO(), conf.Name, metav1.GetOptions{})
 	framework.ExpectNoError(err)
 	return pod
@@ -750,6 +830,30 @@ func GetNodeThatCanRunPod(f *framework.Framework) string {
 	return runPodAndGetNodeName(f, pausePodConfig{Name: "without-label"})
 }
 
+// Get2NodesThatCanRunPod return a 2-node slice where can run pod.
+func Get2NodesThatCanRunPod(f *framework.Framework) []string {
+	firstNode := GetNodeThatCanRunPod(f)
+	ginkgo.By("Trying to launch a pod without a label to get a node which can launch it.")
+	pod := pausePodConfig{
+		Name: "without-label",
+		Affinity: &v1.Affinity{
+			NodeAffinity: &v1.NodeAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
+					NodeSelectorTerms: []v1.NodeSelectorTerm{
+						{
+							MatchFields: []v1.NodeSelectorRequirement{
+								{Key: "metadata.name", Operator: v1.NodeSelectorOpNotIn, Values: []string{firstNode}},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	secondNode := runPodAndGetNodeName(f, pod)
+	return []string{firstNode, secondNode}
+}
+
 func getNodeThatCanRunPodWithoutToleration(f *framework.Framework) string {
 	ginkgo.By("Trying to launch a pod without a toleration to get a node which can launch it.")
 	return runPodAndGetNodeName(f, pausePodConfig{Name: "without-toleration"})