[WIP] [Feat] Re-do mk admission after eviction in worker cluster #8477
+282
−8
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -349,22 +349,53 @@ func (w *wlReconciler) reconcileGroup(ctx context.Context, group *wlGroup) (reco | |
| return reconcile.Result{}, workload.Finish(ctx, w.client, group.local, remoteFinishedCond.Reason, remoteFinishedCond.Message, w.clock) | ||
| } | ||
|
|
||
| // 4. Handle workload eviction | ||
| remoteEvictCond, evictedRemote := group.bestMatchByCondition(kueue.WorkloadEvicted) | ||
| if remoteEvictCond != nil { | ||
| remoteCl := group.remoteClients[evictedRemote].client | ||
| remoteWl := group.remotes[evictedRemote] | ||
|
|
||
| log = log.WithValues("remote", evictedRemote, "remoteWorkload", klog.KObj(remoteWl)) | ||
| ctx = ctrl.LoggerInto(ctx, log) | ||
|
|
||
| // workload evicted on manager cluster | ||
| if remoteEvictCond.Reason == workload.ReasonWithCause(kueue.WorkloadDeactivated, kueue.WorkloadEvictedOnManagerCluster) { | ||
| if err := group.jobAdapter.SyncJob(ctx, w.client, remoteCl, group.controllerKey, group.local.Name, w.origin); err != nil { | ||
| log.Error(err, "Syncing remote controller object") | ||
| // We'll retry this in the next reconciling. | ||
| return reconcile.Result{}, err | ||
| } | ||
| return reconcile.Result{}, nil | ||
| } | ||
|
|
||
| // workload eviction on worker cluster | ||
| log.V(5).Info("Workload gets evicted in the remote cluster", "cluster", evictedRemote) | ||
| needsACUpdate := acs.State == kueue.CheckStateReady | ||
| if err := workload.PatchAdmissionStatus(ctx, w.client, group.local, w.clock, func(wl *kueue.Workload) (bool, error) { | ||
|
Contributor
There was a problem hiding this comment. Why would we do it when |
||
| if needsACUpdate { | ||
| acs.State = kueue.CheckStatePending | ||
| acs.Message = fmt.Sprintf("Workload evicted on worker cluster: %q, resetting for re-admission", *group.local.Status.ClusterName) | ||
| acs.LastTransitionTime = metav1.NewTime(w.clock.Now()) | ||
| workload.SetAdmissionCheckState(&wl.Status.AdmissionChecks, *acs, w.clock) | ||
| wl.Status.ClusterName = nil | ||
| wl.Status.NominatedClusterNames = nil | ||
|
Contributor
There was a problem hiding this comment. Out of curiosity - can this have an effect, given this rule? |
||
| } | ||
| return true, nil | ||
| }); err != nil { | ||
| log.Error(err, "Failed to patch workload status") | ||
| return reconcile.Result{}, err | ||
| } | ||
|
|
||
| if needsACUpdate { | ||
| w.recorder.Eventf(group.local, corev1.EventTypeNormal, "MultiKueue", acs.Message) | ||
| } | ||
|
|
||
| for cluster := range group.remotes { | ||
| if err := client.IgnoreNotFound(group.RemoveRemoteObjects(ctx, cluster)); err != nil { | ||
|
Contributor
There was a problem hiding this comment. In most cases, other calls to |
||
| log.Error(err, "Failed to remove cluster remote objects", "cluster", cluster) | ||
| return reconcile.Result{}, err | ||
| } | ||
| } | ||
| return reconcile.Result{}, nil | ||
| } | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -34,6 +34,7 @@ import ( | |
| batchv1 "k8s.io/api/batch/v1" | ||
| corev1 "k8s.io/api/core/v1" | ||
| apimeta "k8s.io/apimachinery/pkg/api/meta" | ||
| "k8s.io/apimachinery/pkg/api/resource" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/apimachinery/pkg/types" | ||
| "k8s.io/utils/ptr" | ||
|
|
@@ -603,6 +604,93 @@ var _ = ginkgo.Describe("MultiKueue", func() { | |
| }) | ||
| }) | ||
|
|
||
| ginkgo.It("Should re-do admission process when workload gets evicted in the worker", func() { | ||
| job := testingjob.MakeJob("job", managerNs.Name). | ||
| WorkloadPriorityClass(managerLowWPC.Name). | ||
| Queue(kueue.LocalQueueName(managerLq.Name)). | ||
| RequestAndLimit(corev1.ResourceCPU, "0.9"). | ||
| RequestAndLimit(corev1.ResourceMemory, "0.5G"). | ||
| Obj() | ||
| util.MustCreate(ctx, k8sManagerClient, job) | ||
|
|
||
| wlKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job.Name, job.UID), Namespace: managerNs.Name} | ||
| managerWl := &kueue.Workload{} | ||
| workerWorkload := &kueue.Workload{} | ||
|
|
||
| ginkgo.By("Checking that the workload is created and admitted in the manager cluster", func() { | ||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| g.Expect(k8sManagerClient.Get(ctx, wlKey, managerWl)).To(gomega.Succeed()) | ||
| g.Expect(workload.IsAdmitted(managerWl)).To(gomega.BeTrue()) | ||
| }, util.Timeout, util.Interval).Should(gomega.Succeed()) | ||
| }) | ||
|
|
||
| createdAtWorker := "" | ||
|
|
||
| ginkgo.By("Checking that the workload is created in one of the workers", func() { | ||
|
Contributor
There was a problem hiding this comment. Below, you'll modify worker's CQ limits to manipulate what it can fit. AFAICS the initial CPU quotas are: 2 at worker1 and 1 at worker2. Compared to what you have now, +1 CQ update but -4 |
||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| if err := k8sWorker1Client.Get(ctx, wlKey, workerWorkload); err == nil { | ||
| createdAtWorker = "worker1" | ||
| } else { | ||
| g.Expect(k8sWorker2Client.Get(ctx, wlKey, workerWorkload)).To(gomega.Succeed()) | ||
| createdAtWorker = "worker2" | ||
| } | ||
| g.Expect(workload.IsAdmitted(workerWorkload)).To(gomega.BeTrue()) | ||
| g.Expect(workerWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec)) | ||
| }, util.Timeout, util.Interval).Should(gomega.Succeed()) | ||
| }) | ||
|
|
||
| ginkgo.GinkgoLogr.Info("Workload created at", "cluster", createdAtWorker) | ||
|
|
||
| ginkgo.By("Modifying worker cluster queue to not have enough resources", func() { | ||
| if createdAtWorker == "worker1" { | ||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(worker1Cq), worker1Cq)).To(gomega.Succeed()) | ||
| worker1Cq.Spec.ResourceGroups[0].Flavors[0].Resources[0].NominalQuota = resource.MustParse("0.5") | ||
| g.Expect(k8sWorker1Client.Update(ctx, worker1Cq)).To(gomega.Succeed()) | ||
| }, util.Timeout, util.Interval).Should(gomega.Succeed()) | ||
| } else { | ||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(worker2Cq), worker2Cq)).To(gomega.Succeed()) | ||
| worker2Cq.Spec.ResourceGroups[0].Flavors[0].Resources[0].NominalQuota = resource.MustParse("0.5") | ||
| g.Expect(k8sWorker2Client.Update(ctx, worker2Cq)).To(gomega.Succeed()) | ||
| }, util.Timeout, util.Interval).Should(gomega.Succeed()) | ||
| } | ||
| }) | ||
|
|
||
| ginkgo.By("Triggering eviction in worker", func() { | ||
| if createdAtWorker == "worker1" { | ||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| g.Expect(k8sWorker1Client.Get(ctx, wlKey, workerWorkload)).To(gomega.Succeed()) | ||
| g.Expect(workload.SetConditionAndUpdate(ctx, k8sWorker1Client, workerWorkload, kueue.WorkloadEvicted, metav1.ConditionTrue, kueue.WorkloadEvictedByPreemption, "By test", "evict", util.RealClock)).To(gomega.Succeed()) | ||
| }, util.Timeout, util.Interval).Should(gomega.Succeed()) | ||
| } else { | ||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| g.Expect(k8sWorker2Client.Get(ctx, wlKey, workerWorkload)).To(gomega.Succeed()) | ||
| g.Expect(workload.SetConditionAndUpdate(ctx, k8sWorker2Client, workerWorkload, kueue.WorkloadEvicted, metav1.ConditionTrue, kueue.WorkloadEvictedByPreemption, "By test", "evict", util.RealClock)).To(gomega.Succeed()) | ||
| }, util.Timeout, util.Interval).Should(gomega.Succeed()) | ||
| } | ||
| }) | ||
|
|
||
| ginkgo.By("Checking that the workload is re-admitted in the other worker cluster", func() { | ||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| g.Expect(k8sManagerClient.Get(ctx, wlKey, managerWl)).To(gomega.Succeed()) | ||
| g.Expect(managerWl.Status.ClusterName).NotTo(gomega.HaveValue(gomega.Equal(createdAtWorker))) | ||
|
Contributor
There was a problem hiding this comment. You could also check that it's not empty. |
||
| }, util.LongTimeout, util.Interval).Should(gomega.Succeed()) | ||
| }) | ||
|
|
||
| ginkgo.By("Checking that the workload is created in the other worker", func() { | ||
| gomega.Eventually(func(g gomega.Gomega) { | ||
| if createdAtWorker == "worker1" { | ||
| g.Expect(k8sWorker2Client.Get(ctx, wlKey, workerWorkload)).To(gomega.Succeed()) | ||
| } else { | ||
| g.Expect(k8sWorker1Client.Get(ctx, wlKey, workerWorkload)).To(gomega.Succeed()) | ||
| } | ||
| g.Expect(workload.IsAdmitted(workerWorkload)).To(gomega.BeTrue()) | ||
| g.Expect(workerWorkload.Spec).To(gomega.BeComparableTo(managerWl.Spec)) | ||
| }, util.Timeout, util.Interval).Should(gomega.Succeed()) | ||
| }) | ||
| }) | ||
|
|
||
| ginkgo.It("Should preempt a running low-priority workload when a high-priority workload is admitted (other workers)", func() { | ||
| lowJob := testingjob.MakeJob("low-job", managerNs.Name). | ||
| WorkloadPriorityClass(managerLowWPC.Name). | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nit: this present tense feels slightly confusing; IIUC the workload already got evicted.