[YUNIKORN-2068] E2E Test for Preemption

test/e2e/preemption/preemption_test.go

@@ -49,6 +49,8 @@ var Worker = ""
 var WorkerMemRes int64
 var sleepPodMemLimit int64
 var sleepPodMemLimit2 int64
+var sleepPodMemOverLimit int64
+var nodeName string
 var taintKey = "e2e_test_preemption"
 var nodesToTaint []string
 
@@ -102,12 +104,14 @@ var _ = ginkgo.BeforeSuite(func() {
 	for _, node := range *nodesDAOInfo {
 		if node.NodeID == Worker {
 			WorkerMemRes = node.Available["memory"]
+			nodeName = node.HostName
 		}
 	}
 	WorkerMemRes /= (1000 * 1000) // change to M
 	fmt.Fprintf(ginkgo.GinkgoWriter, "Worker node %s available memory %dM\n", Worker, WorkerMemRes)
 
 	sleepPodMemLimit = int64(float64(WorkerMemRes) / 3)
+	sleepPodMemOverLimit = int64(float64(WorkerMemRes) * 1.5)
 	Ω(sleepPodMemLimit).NotTo(gomega.BeZero(), "Sleep pod memory limit cannot be zero")
 	fmt.Fprintf(ginkgo.GinkgoWriter, "Sleep pod limit memory %dM\n", sleepPodMemLimit)
 
@@ -175,7 +179,7 @@ var _ = ginkgo.Describe("Preemption", func() {
 			// Wait for pod to move to running state
 			podErr = kClient.WaitForPodBySelectorRunning(dev,
 				fmt.Sprintf("app=%s", sleepRespPod.ObjectMeta.Labels["app"]),
-				60)
+				120)
 			gomega.Ω(podErr).NotTo(gomega.HaveOccurred())
 		}
 
@@ -546,6 +550,74 @@ var _ = ginkgo.Describe("Preemption", func() {
 		gomega.Ω(err).ShouldNot(HaveOccurred())
 	})
 
+	ginkgo.It("Verify_preemption_on_specific_node", func() {
+		/*
+		 1. Create Two Queue (High and Low Guaranteed Limit)
+		 2. Select a schedulable node from the cluster
+		 3. Schedule a number of small, Low priority pause tasks on Low Guaranteed queue (Enough to fill the node)
+		 4. Schedule a large task in High Priority queue with same node
+		 5. Wait for few seconds to schedule the task
+		 6. This should trigger preemption on low-priority queue and remove or preempt task from low priority queue
+		 7. Do cleanup once test is done either passed or failed
+		*/
+
+		ginkgo.By("Create Two Queue High and Low Guaranteed Limit")
+		annotation = "ann-" + common.RandSeq(10)
+		yunikorn.UpdateCustomConfigMapWrapper(oldConfigMap, "", annotation, func(sc *configs.SchedulerConfig) error {
+			// remove placement rules so we can control queue
+			sc.Partitions[0].PlacementRules = nil
+			var err error
+			if err = common.AddQueue(sc, "default", "root", configs.QueueConfig{
+				Name:       "sandbox2",
+				Resources:  configs.Resources{Guaranteed: map[string]string{"memory": fmt.Sprintf("%dM", sleepPodMemOverLimit)}},
+				Properties: map[string]string{"preemption.delay": "1s", "priority.offset": "100"},
+			}); err != nil {
+				return err
+			}
+			if err = common.AddQueue(sc, "default", "root", configs.QueueConfig{
+				Name:       "sandbox1",
+				Resources:  configs.Resources{Guaranteed: map[string]string{"memory": fmt.Sprintf("%dM", sleepPodMemLimit)}},
+				Properties: map[string]string{"preemption.delay": "1s", "priority.offset": "-100"},
+			}); err != nil {
+				return err
+			}
+			return nil
+		})
+
+		ginkgo.By("Schedule a number of small, Low priority pause tasks on Low Guaranteed queue (Enough to fill the node)")
+
+		sandbox1SleepPodConfigs := createSandbox1SleepPodCofigsWithStaticNode(3, 600)
+		sleepPod4Config := k8s.SleepPodConfig{Name: "sleepjob4", NS: dev, Mem: sleepPodMemLimit, Time: 600, Optedout: k8s.Allow, Labels: map[string]string{"queue": "root.sandbox2"}, RequiredNode: nodeName}
+		sandbox1SleepPodConfigs = append(sandbox1SleepPodConfigs, sleepPod4Config)
+		for _, config := range sandbox1SleepPodConfigs {
+			ginkgo.By("Deploy the sleep pod " + config.Name + " to the development namespace")
+			sleepObj, podErr := k8s.InitSleepPod(config)
+			Ω(podErr).NotTo(gomega.HaveOccurred())
+			sleepRespPod, podErr := kClient.CreatePod(sleepObj, dev)
+			gomega.Ω(podErr).NotTo(gomega.HaveOccurred())
+
+			// Wait for pod to move to running state
+			podErr = kClient.WaitForPodBySelectorRunning(dev,
+				fmt.Sprintf("app=%s", sleepRespPod.ObjectMeta.Labels["app"]),
+				60)
+			gomega.Ω(podErr).NotTo(gomega.HaveOccurred())
+		}
+
+		ginkgo.By("Two pods in root.sandbox1 queue are preempted")
+		sandbox1RunningPodsCnt := 0
+		pods, err := kClient.ListPodsByLabelSelector(dev, "queue=root.sandbox1")
+		gomega.Ω(err).NotTo(gomega.HaveOccurred())
+		for _, pod := range pods.Items {
+			if pod.DeletionTimestamp != nil {
+				continue
+			}
+			if pod.Status.Phase == v1.PodRunning {
+				sandbox1RunningPodsCnt++
+			}
+		}
+		Ω(sandbox1RunningPodsCnt).To(gomega.Equal(2), "two pods in root.sandbox1 queue should be preempted")
+	})
+
 	ginkgo.AfterEach(func() {
 		testDescription := ginkgo.CurrentSpecReport()
 		if testDescription.Failed() {
@@ -572,3 +644,11 @@ func createSandbox1SleepPodCofigs(cnt, time int) []k8s.SleepPodConfig {
 	}
 	return sandbox1Configs
 }
+
+func createSandbox1SleepPodCofigsWithStaticNode(cnt, time int) []k8s.SleepPodConfig {
+	sandbox1Configs := make([]k8s.SleepPodConfig, 0, cnt)
+	for i := 0; i < cnt; i++ {
+		sandbox1Configs = append(sandbox1Configs, k8s.SleepPodConfig{Name: fmt.Sprintf("sleepjob%d", i+1), NS: dev, Mem: sleepPodMemLimit2, Time: time, Optedout: k8s.Allow, Labels: map[string]string{"queue": "root.sandbox1"}, RequiredNode: nodeName})
+	}
+	return sandbox1Configs
+}