Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Trigger the workloads eviction on admission check rejection. #1562

Merged
merged 8 commits into from
Jan 12, 2024
Merged

Trigger the workloads eviction on admission check rejection. #1562

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
838fa1a
Trigger the workloads eviction on admission check rejection.
trasc Jan 9, 2024
b1a07d4
Wait for the eviction to and, add integration test.
trasc Jan 10, 2024
8725651
Review remarks.
trasc Jan 11, 2024
535444a
Review Remarks
trasc Jan 11, 2024
8a05225
Review Remarks
trasc Jan 11, 2024
d277388
Review Remarks
trasc Jan 11, 2024
b10409e
Fix after rebase.
trasc Jan 12, 2024
fbca3bf
Update pkg/controller/core/workload_controller.go
trasc Jan 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 25 additions & 26 deletions pkg/controller/core/workload_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,45 +142,22 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
return ctrl.Result{}, nil
}

if rejectedChecks := workload.GetRejectedChecks(&wl); len(rejectedChecks) > 0 {
// Finish the workload
log.V(3).Info("Workload has Rejected admission checks, Finish with failure")
err := workload.UpdateStatus(ctx, r.client, &wl, kueue.WorkloadFinished,
metav1.ConditionTrue,
"AdmissionChecksRejected",
fmt.Sprintf("Admission checks %v are rejected", rejectedChecks),
constants.KueueName)

if err == nil {
for _, owner := range wl.OwnerReferences {
uowner := unstructured.Unstructured{}
uowner.SetKind(owner.Kind)
uowner.SetAPIVersion(owner.APIVersion)
uowner.SetName(owner.Name)
uowner.SetNamespace(wl.Namespace)
uowner.SetUID(owner.UID)
r.recorder.Eventf(&uowner, corev1.EventTypeNormal, "WorkloadFinished", "Admission checks %v are rejected", rejectedChecks)
}
}

return ctrl.Result{}, err
}

cqName, cqOk := r.queues.ClusterQueueForWorkload(&wl)
if cqOk {
if updated, err := r.reconcileSyncAdmissionChecks(ctx, &wl, cqName); updated || err != nil {
return ctrl.Result{}, err
}
}

if workload.SyncAdmittedCondition(&wl) {
// If the workload is admitted, updating the status here would set the Admitted condition to
// false before the workloads eviction.
if !workload.IsAdmitted(&wl) && workload.SyncAdmittedCondition(&wl) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this change needed, or just a performance optimization? If needed, does the order matter?

Also, is it correct, I'm asking because if the workload was already admitted, then (IIUC) we don't enter the branch, we also skip if workload.HasQuotaReservation(&wl) {, then we enter if rejectedChecks := workload.GetRejectedChecks(&wl); len(rejectedChecks) > 0 {. So we would mark the workload as finished.
I think the idea was to go via eviction rather than marking the workload as finished? Can you explain what is the code path leading to workload eviction in that scenario now?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is needed, and the order is important since SyncAdmittedCondition can update the Admitted cind. in the wl.

If the workload is admitted, it will skip this and just check if the eviction is needed.

Copy link
Contributor

@mimowo mimowo Jan 11, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I checked that the integration test passes when the check !workload.IsAdmitted(&wl) is removed. Can we add / extend the integration test to demonstrate its importance?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

(If we don't keep this the Admitted condition is cleared before the eviction starts)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(If we don't keep this the Admitted condition is cleared before the eviction starts)

Why does that matter?

To be clear I'm fine with having Admitted=True and Evicted=True, I'm also fine with clearing the Admitted before adding Evicted=True. Trying to understand what is the difference, to avoid unnecessary code complications.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

comment updated.

To be clear I'm fine with having Admitted=True and Evicted=True, I'm also fine with clearing the Admitted before adding Evicted=True. Trying to understand what is the difference, to avoid unnecessary code complications.

It's more logical in my opinion and is the same flow as in case of premonition based eviction.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sgtm, I buy the argument of consistent flow with preemption-based eviction

if err := workload.ApplyAdmissionStatus(ctx, r.client, &wl, true); err != nil {
return ctrl.Result{}, err
}
if workload.IsAdmitted(&wl) {
c := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadQuotaReserved)
r.recorder.Eventf(&wl, corev1.EventTypeNormal, "Admitted", "Admitted by ClusterQueue %v, wait time since reservation was %.0fs", wl.Status.Admission.ClusterQueue, time.Since(c.LastTransitionTime.Time).Seconds())

}
return ctrl.Result{}, nil
}
Expand All @@ -197,6 +174,28 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
return r.reconcileNotReadyTimeout(ctx, req, &wl)
}

// At this point the workload is not Admitted, if it has rejected admission checks mark it as finished.
if rejectedChecks := workload.GetRejectedChecks(&wl); len(rejectedChecks) > 0 {
log.V(3).Info("Workload has Rejected admission checks, Finish with failure")
err := workload.UpdateStatus(ctx, r.client, &wl, kueue.WorkloadFinished,
metav1.ConditionTrue,
"AdmissionChecksRejected",
fmt.Sprintf("Admission checks %v are rejected", rejectedChecks),
constants.KueueName)
if err == nil {
for _, owner := range wl.OwnerReferences {
uowner := unstructured.Unstructured{}
uowner.SetKind(owner.Kind)
uowner.SetAPIVersion(owner.APIVersion)
uowner.SetName(owner.Name)
uowner.SetNamespace(wl.Namespace)
uowner.SetUID(owner.UID)
r.recorder.Eventf(&uowner, corev1.EventTypeNormal, "WorkloadFinished", "Admission checks %v are rejected", rejectedChecks)
}
}
return ctrl.Result{}, err
}

if !r.queues.QueueForWorkloadExists(&wl) {
log.V(3).Info("Workload is inadmissible because of missing LocalQueue", "localQueue", klog.KRef(wl.Namespace, wl.Spec.QueueName))
workload.UnsetQuotaReservationWithCondition(&wl, "Inadmissible", fmt.Sprintf("LocalQueue %s doesn't exist", wl.Spec.QueueName))
Expand Down
32 changes: 27 additions & 5 deletions pkg/controller/core/workload_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,19 +394,15 @@ func TestReconcile(t *testing.T) {
DeletionTimestamp(testStartTime).
Obj(),
},
"workload with rejected checks": {
"unadmitted workload with rejected checks": {
workload: utiltesting.MakeWorkload("wl", "ns").
ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
Admitted(true).
OwnerReference("ownerapi", "ownerkind", "ownername", "owneruid", true, true).
AdmissionCheck(kueue.AdmissionCheckState{
Name: "check",
State: kueue.CheckStateRejected,
}).
Obj(),
wantWorkload: utiltesting.MakeWorkload("wl", "ns").
ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
Admitted(true).
OwnerReference("ownerapi", "ownerkind", "ownername", "owneruid", true, true).
AdmissionCheck(kueue.AdmissionCheckState{
Name: "check",
Expand All @@ -427,6 +423,32 @@ func TestReconcile(t *testing.T) {
},
},
},
"admitted workload with rejected checks": {
workload: utiltesting.MakeWorkload("wl", "ns").
ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
Admitted(true).
OwnerReference("ownerapi", "ownerkind", "ownername", "owneruid", true, true).
AdmissionCheck(kueue.AdmissionCheckState{
Name: "check",
State: kueue.CheckStateRejected,
}).
Obj(),
wantWorkload: utiltesting.MakeWorkload("wl", "ns").
ReserveQuota(utiltesting.MakeAdmission("q1").Obj()).
Admitted(true).
OwnerReference("ownerapi", "ownerkind", "ownername", "owneruid", true, true).
AdmissionCheck(kueue.AdmissionCheckState{
Name: "check",
State: kueue.CheckStateRejected,
}).
Condition(metav1.Condition{
Type: "Evicted",
Status: "True",
Reason: "AdmissionCheck",
Message: "At least one admission check is false",
}).
Obj(),
},
}
for name, tc := range cases {
t.Run(name, func(t *testing.T) {
Expand Down
3 changes: 3 additions & 0 deletions pkg/workload/workload.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,6 +310,9 @@ func UnsetQuotaReservationWithCondition(wl *kueue.Workload, reason, message stri
}
apimeta.SetStatusCondition(&wl.Status.Conditions, condition)
wl.Status.Admission = nil

// Reset the admitted condition if necessary.
_ = SyncAdmittedCondition(wl)
trasc marked this conversation as resolved.
Show resolved Hide resolved
}

// BaseSSAWorkload creates a new object based on the input workload that
Expand Down
116 changes: 114 additions & 2 deletions test/integration/controller/core/workload_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,11 +150,24 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
})

ginkgo.When("the queue has admission checks", func() {
var flavor *kueue.ResourceFlavor
var (
flavor *kueue.ResourceFlavor
check1 *kueue.AdmissionCheck
check2 *kueue.AdmissionCheck
)

ginkgo.BeforeEach(func() {
flavor = testing.MakeResourceFlavor(flavorOnDemand).Obj()
gomega.Expect(k8sClient.Create(ctx, flavor)).Should(gomega.Succeed())

check1 = testing.MakeAdmissionCheck("check1").ControllerName("ctrl").Obj()
gomega.Expect(k8sClient.Create(ctx, check1)).Should(gomega.Succeed())
util.SetAdmissionCheckActive(ctx, k8sClient, check1, metav1.ConditionTrue)

check2 = testing.MakeAdmissionCheck("check2").ControllerName("ctrl").Obj()
gomega.Expect(k8sClient.Create(ctx, check2)).Should(gomega.Succeed())
util.SetAdmissionCheckActive(ctx, k8sClient, check2, metav1.ConditionTrue)

clusterQueue = testing.MakeClusterQueue("cluster-queue").
ResourceGroup(*testing.MakeFlavorQuotas(flavorOnDemand).
Resource(resourceGPU, "5", "5").Obj()).
Expand All @@ -168,6 +181,8 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
ginkgo.AfterEach(func() {
gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
util.ExpectClusterQueueToBeDeleted(ctx, k8sClient, clusterQueue, true)
util.ExpectAdmissionCheckToBeDeleted(ctx, k8sClient, check2, true)
util.ExpectAdmissionCheckToBeDeleted(ctx, k8sClient, check1, true)
util.ExpectResourceFlavorToBeDeleted(ctx, k8sClient, flavor, true)
})

Expand Down Expand Up @@ -224,7 +239,7 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
gomega.Expect(check2Cond).To(gomega.Equal(oldCheck2Cond))
})
})
ginkgo.It("should finish the workload with failure when a check is rejected", func() {
ginkgo.It("should finish an unadmitted workload with failure when a check is rejected", func() {
wl := testing.MakeWorkload("wl", ns.Name).Queue("queue").Obj()
wlKey := client.ObjectKeyFromObject(wl)
createdWl := kueue.Workload{}
Expand Down Expand Up @@ -261,6 +276,103 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
}, cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")))
})
})

ginkgo.It("should evict then finish with failure an admitted workload when a check is rejected", func() {
wl := testing.MakeWorkload("wl", ns.Name).Queue("queue").Obj()
wlKey := client.ObjectKeyFromObject(wl)
createdWl := kueue.Workload{}
ginkgo.By("creating the workload, the check conditions should be added", func() {
gomega.Expect(k8sClient.Create(ctx, wl)).To(gomega.Succeed())

gomega.Eventually(func() []string {
gomega.Expect(k8sClient.Get(ctx, wlKey, &createdWl)).To(gomega.Succeed())
return slices.Map(createdWl.Status.AdmissionChecks, func(c *kueue.AdmissionCheckState) string { return c.Name })
}, util.Timeout, util.Interval).Should(gomega.ConsistOf("check1", "check2"))
})

ginkgo.By("setting quota reservation and the checks ready, should admit the workload", func() {
gomega.Eventually(func(g gomega.Gomega) {
g.Expect(k8sClient.Get(ctx, wlKey, &createdWl)).To(gomega.Succeed())
g.Expect(util.SetQuotaReservation(ctx, k8sClient, &createdWl, testing.MakeAdmission(clusterQueue.Name).Obj())).To(gomega.Succeed())
}, util.Timeout, util.Interval).Should(gomega.Succeed())

gomega.Eventually(func() error {
gomega.Expect(k8sClient.Get(ctx, wlKey, &createdWl)).To(gomega.Succeed())
workload.SetAdmissionCheckState(&createdWl.Status.AdmissionChecks, kueue.AdmissionCheckState{
Name: "check1",
State: kueue.CheckStateReady,
Message: "check ready",
})
workload.SetAdmissionCheckState(&createdWl.Status.AdmissionChecks, kueue.AdmissionCheckState{
Name: "check2",
State: kueue.CheckStateReady,
Message: "check ready",
})
return k8sClient.Status().Update(ctx, &createdWl)
}, util.Timeout, util.Interval).Should(gomega.Succeed())

gomega.Eventually(func(g gomega.Gomega) {
g.Expect(k8sClient.Get(ctx, wlKey, &createdWl)).To(gomega.Succeed())
g.Expect(createdWl.Status.Conditions).To(gomega.ContainElement(gomega.BeComparableTo(metav1.Condition{
Type: kueue.WorkloadAdmitted,
Status: metav1.ConditionTrue,
Reason: "Admitted",
Message: "The workload is admitted",
}, cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime"))))
}, util.Timeout, util.Interval).Should(gomega.Succeed())
})

ginkgo.By("setting a rejected check conditions the workload should be evicted and admitted condition kept", func() {
gomega.Eventually(func() error {
gomega.Expect(k8sClient.Get(ctx, wlKey, &createdWl)).To(gomega.Succeed())
workload.SetAdmissionCheckState(&createdWl.Status.AdmissionChecks, kueue.AdmissionCheckState{
Name: "check1",
State: kueue.CheckStateRejected,
Message: "check rejected",
})
return k8sClient.Status().Update(ctx, &createdWl)
}, util.Timeout, util.Interval).Should(gomega.Succeed())

gomega.Eventually(func(g gomega.Gomega) {
g.Expect(k8sClient.Get(ctx, wlKey, &createdWl)).To(gomega.Succeed())
g.Expect(createdWl.Status.Conditions).To(gomega.ContainElements(
gomega.BeComparableTo(metav1.Condition{
Type: kueue.WorkloadEvicted,
Status: metav1.ConditionTrue,
Reason: "AdmissionCheck",
Message: "At least one admission check is false",
}, cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")),
gomega.BeComparableTo(metav1.Condition{
Type: kueue.WorkloadAdmitted,
Status: metav1.ConditionTrue,
Reason: "Admitted",
Message: "The workload is admitted",
}, cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")),
))
}, util.Timeout, util.Interval).Should(gomega.Succeed())
})

ginkgo.By("finishing the eviction the finish condition should be set and admitted condition false", func() {
util.FinishEvictionForWorkloads(ctx, k8sClient, &createdWl)
gomega.Eventually(func(g gomega.Gomega) {
g.Expect(k8sClient.Get(ctx, wlKey, &createdWl)).To(gomega.Succeed())
g.Expect(createdWl.Status.Conditions).To(gomega.ContainElements(
gomega.BeComparableTo(metav1.Condition{
Type: kueue.WorkloadFinished,
Status: metav1.ConditionTrue,
Reason: "AdmissionChecksRejected",
Message: "Admission checks [check1] are rejected",
}, cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")),
gomega.BeComparableTo(metav1.Condition{
Type: kueue.WorkloadAdmitted,
Status: metav1.ConditionFalse,
Reason: "NoReservationNoChecks",
Message: "The workload has no reservation and not all checks ready",
}, cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")),
))
}, util.Timeout, util.Interval).Should(gomega.Succeed())
})
})
})

ginkgo.When("changing the priority value of PriorityClass doesn't affect the priority of the workload", func() {
Expand Down