Merge branch 'slice-main' into evict_if_not_tolerated

Author: pajakd

slice/internal/controller/workload_controller.go

@@ -186,8 +186,10 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 
 	if ac.State == kueue.CheckStateReady && (len(grouped.deleted) > 0 || len(slices) != totalDesiredSlices(wl, nodes)) {
 		log.V(3).Info("Slice has been deleted, evicting workload")
-		err := r.evictWorkload(ctx, wl, ac, "Slice has been deleted")
-		return ctrl.Result{}, err
+		if err := r.evictWorkload(ctx, wl, ac, "Slice has been deleted"); err != nil {
+			return ctrl.Result{}, err
+		}
+		return ctrl.Result{}, r.deleteSlicesForEvictedWorkload(ctx, grouped)
 	}
 
 	if len(grouped.deleted) > 0 {
@@ -217,6 +219,10 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
+	if ac.State == kueue.CheckStateRetry {
+		return ctrl.Result{}, r.deleteSlicesForEvictedWorkload(ctx, grouped)
+	}
+
 	// Delete any Slices that are in a failed or stale state.
 	if len(grouped.toDelete) > 0 {
 		log.V(2).Info(
@@ -393,6 +399,20 @@ func (r *WorkloadReconciler) deleteSlices(ctx context.Context, slices []*v1beta1
 	return nil
 }
 
+func (r *WorkloadReconciler) deleteSlicesForEvictedWorkload(ctx context.Context, grouped groupedSlices) error {
+	numSlicesToDelete := len(grouped.active) + len(grouped.initializing) + len(grouped.toDelete)
+	if numSlicesToDelete == 0 {
+		return nil
+	}
+	log := ctrl.LoggerFrom(ctx)
+	toDelete := make([]*v1beta1.Slice, 0, numSlicesToDelete)
+	toDelete = append(toDelete, grouped.active...)
+	toDelete = append(toDelete, grouped.initializing...)
+	toDelete = append(toDelete, grouped.toDelete...)
+	log.V(3).Info("AdmissionCheck is Retry, deleting all Slices")
+	return r.deleteSlices(ctx, toDelete)
+}
+
 func (r *WorkloadReconciler) ownerPodsFinished(ctx context.Context, wl *kueue.Workload) (bool, error) {
 	if isJobSetOwner(wl) {
 		return r.jobSetPodsFinished(ctx, wl)
@@ -551,6 +571,11 @@ func (r *WorkloadReconciler) syncSlices(
 	slices []v1beta1.Slice,
 	nodes map[string]corev1.Node,
 ) ([]v1beta1.Slice, error) {
+	// this is to prevent from creating slices when AC is Retry
+	// and the workload still has the old Admission
+	if ac.State != kueue.CheckStatePending {
+		return nil, nil
+	}
 	existingSlicesByName := make(map[string]*v1beta1.Slice, len(slices))
 	for _, slice := range slices {
 		existingSlicesByName[slice.Name] = &slice
@@ -729,7 +754,7 @@ func (r *WorkloadReconciler) validatePartitionCount(
 
 func (r *WorkloadReconciler) evictWorkload(ctx context.Context, wl *kueue.Workload, ac *kueue.AdmissionCheckState, message string) error {
 	ac.State = kueue.CheckStateRetry
-	ac.RequeueAfterSeconds = ptr.To(int32(10))
+	ac.RequeueAfterSeconds = ptr.To(int32(r.retryDelayOnSliceFailure.Round(time.Second).Seconds()))
 	ac.Message = api.TruncateConditionMessage(message)
 	return r.updateWorkloadAdmissionCheckStatus(ctx, wl, ac)
 }
@@ -881,6 +906,10 @@ func buildAdmissionCheckMessage(slicesByState map[core.SliceState][]v1beta1.Slic
 }
 
 func (r *WorkloadReconciler) prepareAdmissionCheckStatus(wl *kueue.Workload, ac *kueue.AdmissionCheckState, slices []v1beta1.Slice) {
+	// wait for Kueue to reset check to Pending after eviction
+	if ac.State == kueue.CheckStateRetry {
+		return
+	}
 	slicesByState := groupSlicesByState(slices, r.activationTimeout)
 
 	podSetRequiresHealthy := make(map[string]bool)

slice/internal/controller/workload_controller_test.go

@@ -738,6 +738,32 @@ func TestWorkloadReconciler(t *testing.T) {
 					Obj(),
 			},
 		},
+		"should not create Slices (and delete existing) if AdmissionCheck is in Retry state": {
+			request: baseRequest,
+			objs: []client.Object{
+				worker1Node.DeepCopy(),
+				worker2Node.DeepCopy(),
+				baseAdmissionCheckWrapper.DeepCopy(),
+				baseWorkloadWrapper.Clone().
+					PodSets(basePodSets...).
+					ReserveQuota(baseAdmission, now).
+					ControllerReference(jobSetGVK, baseJobSetName, baseJobSetName).
+					Finalizers(SliceControllerName).
+					AdmissionCheck(buildAdmissionCheckStateWithRequeue(kueue.CheckStateRetry, "retrying", ptr.To(int32(10)))).
+					Obj(),
+				baseSlice1Wrapper.DeepCopy(),
+				baseSlice2Wrapper.DeepCopy(),
+			},
+			wantWorkloads: []kueue.Workload{
+				*baseWorkloadWrapper.Clone().
+					PodSets(basePodSets...).
+					ReserveQuota(baseAdmission, now).
+					ControllerReference(jobSetGVK, baseJobSetName, baseJobSetName).
+					Finalizers(SliceControllerName).
+					AdmissionCheck(buildAdmissionCheckStateWithRequeue(kueue.CheckStateRetry, "retrying", ptr.To(int32(10)))).
+					Obj(),
+			},
+		},
 		"shouldn't create a Slice because there’s no AdmissionCheck": {
 			request: baseRequest,
 			objs: []client.Object{
@@ -1680,9 +1706,6 @@ func TestWorkloadReconciler(t *testing.T) {
 						`Slices are in states: 1 ACTIVE, 1 FAILED. Errors: Error by test`, ptr.To(int32(10)))).
 					Obj(),
 			},
-			wantSlices: []slice.Slice{
-				*baseSlice1Wrapper.Clone().Active().Obj(),
-			},
 			wantEvents: []utiltesting.EventRecord{
 				buildEventRecord(corev1.NamespaceDefault, corev1.EventTypeNormal, AdmissionCheckUpdatedEventType,
 					fmt.Sprintf(`Admission check %q updated state from "Pending" to "Retry"`, baseACName)),
@@ -1741,9 +1764,6 @@ func TestWorkloadReconciler(t *testing.T) {
 						"Slice has been deleted", ptr.To(int32(10)))).
 					Obj(),
 			},
-			wantSlices: []slice.Slice{
-				*baseSlice2Wrapper.Clone().Active().Obj(),
-			},
 		},
 		"should evict workload if slice is deleted unexpectedly": {
 			request: baseRequest,
@@ -1773,7 +1793,6 @@ func TestWorkloadReconciler(t *testing.T) {
 			},
 			wantSlices: []slice.Slice{
 				*baseSlice1Wrapper.Clone().Active().DeletionTimestamp(now).Finalizers("accelerator.gke.io/slice-finalizer").Obj(),
-				*baseSlice2Wrapper.Clone().Active().Obj(),
 			},
 		},
 		"should use the first AdmissionCheck if more than one is found": {