Automated cherry pick of #4824: Fix getAncestorWorkload if integration disabled. (#4924)

* Fix getAncestorWorkload if integration disabled.

Co-authored-by: Michał Woźniak <[email protected]>

* With manageJobsWithoutQueueName logic.

* Rename GetAncestorJobManagedByKueue to FindAncestorJobManagedByKueue

* Add test cases when child jobs have queue-name.

* Rename getAncestorWorkload to getWorkloadForObject.

* Minor fix.

---------

Co-authored-by: Michał Woźniak <[email protected]>

Author: mbobrovskyi

pkg/controller/jobframework/interface.go

@@ -164,6 +164,14 @@ type JobWithManagedBy interface {
 	SetManagedBy(*string)
 }
 
+// TopLevelJob interface is an optional interface used to indicate
+// that the Job owns/manages the Workload object, regardless of the Job
+// owner references.
+type TopLevelJob interface {
+	// IsTopLevel returns true if the Job owns/manages the Workload.
+	IsTopLevel() bool
+}
+
 func QueueName(job GenericJob) string {
 	return QueueNameForObject(job.Object())
 }

pkg/controller/jobframework/reconciler.go

@@ -67,7 +67,7 @@ const (
 )
 
 var (
-	ErrUnknownWorkloadOwner     = errors.New("workload owner is unknown")
+	ErrCyclicOwnership          = errors.New("cyclic ownership")
 	ErrWorkloadOwnerNotFound    = errors.New("workload owner not found")
 	ErrNoMatchingWorkloads      = errors.New("no matching workloads")
 	ErrExtraWorkloads           = errors.New("extra workloads")
@@ -292,27 +292,31 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	isTopLevelJob := true
-	objectOwner := metav1.GetControllerOf(object)
-	if objectOwner != nil && IsOwnerManagedByKueue(objectOwner) {
-		isTopLevelJob = false
+	var (
+		ancestorJob   client.Object
+		isTopLevelJob bool
+	)
+
+	if topLevelJob, ok := job.(TopLevelJob); ok && topLevelJob.IsTopLevel() {
+		// Skipping traversal to top-level ancestor job because this is already a top-level job.
+		isTopLevelJob = true
+	} else {
+		ancestorJob, err = FindAncestorJobManagedByKueue(ctx, r.client, r.record, object, r.manageJobsWithoutQueueName)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		isTopLevelJob = ancestorJob == nil
 	}
 
 	// when manageJobsWithoutQueueName is disabled we only reconcile jobs that either
 	// have a queue-name label or have a kueue-managed ancestor that has a queue-name label.
 	if !r.manageJobsWithoutQueueName && QueueName(job) == "" {
 		if isTopLevelJob {
-			log.V(3).Info("queue-name label is not set, ignoring the job", "queueName", QueueName(job))
+			log.V(3).Info("queue-name label is not set, ignoring the job")
 			return ctrl.Result{}, nil
 		}
-		isAncestorJobManaged, err := r.IsAncestorJobManaged(ctx, job.Object())
-		if err != nil {
-			log.Error(err, "couldn't check whether an ancestor job is managed by kueue")
-			return ctrl.Result{}, err
-		}
-		if !isAncestorJobManaged {
-			log.V(3).Info("No kueue-managed ancestors have a queue-name label, ignoring the job",
-				"parentJob", objectOwner.Name)
+		if QueueNameForObject(ancestorJob) == "" {
+			log.V(3).Info("No kueue-managed ancestors have a queue-name label, ignoring the job")
 			return ctrl.Result{}, nil
 		}
 	}
@@ -321,7 +325,7 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 	if !isTopLevelJob {
 		_, _, finished := job.Finished()
 		if !finished && !job.IsSuspended() {
-			if ancestorWorkload, err := r.getAncestorWorkload(ctx, object); err != nil {
+			if ancestorWorkload, err := r.getWorkloadForObject(ctx, ancestorJob); err != nil {
 				log.Error(err, "couldn't get an ancestor job workload")
 				return ctrl.Result{}, err
 			} else if ancestorWorkload == nil || !workload.IsAdmitted(ancestorWorkload) {
@@ -572,66 +576,91 @@ func (r *JobReconciler) recordAdmissionCheckUpdate(wl *kueue.Workload, job Gener
 	}
 }
 
-// IsAncestorJobManaged checks whether an ancestor job is managed by kueue.
-func (r *JobReconciler) IsAncestorJobManaged(ctx context.Context, jobObj client.Object) (bool, error) {
-	ancestor, err := r.getAncestorJobManagedByKueue(ctx, jobObj)
-	if err != nil {
-		return false, err
-	}
-	return ancestor != nil, nil
-}
-
-// getAncestorWorkload returns the Workload object of the Kueue-managed ancestor job.
-func (r *JobReconciler) getAncestorWorkload(ctx context.Context, jobObj client.Object) (*kueue.Workload, error) {
-	ancestor, err := r.getAncestorJobManagedByKueue(ctx, jobObj)
-	if err != nil || ancestor == nil {
-		return nil, err
-	}
+// getWorkloadForObject returns the Workload associated with the given job.
+func (r *JobReconciler) getWorkloadForObject(ctx context.Context, jobObj client.Object) (*kueue.Workload, error) {
 	wlList := kueue.WorkloadList{}
-	if err := r.client.List(ctx, &wlList, client.InNamespace(ancestor.GetNamespace()), client.MatchingFields{indexer.OwnerReferenceUID: string(ancestor.GetUID())}); client.IgnoreNotFound(err) != nil {
+	if err := r.client.List(ctx, &wlList, client.InNamespace(jobObj.GetNamespace()), client.MatchingFields{indexer.OwnerReferenceUID: string(jobObj.GetUID())}); client.IgnoreNotFound(err) != nil {
 		return nil, err
 	}
 	if len(wlList.Items) > 0 {
 		// In theory the job can own multiple Workloads, we cannot do too much about it, maybe log it.
+		ctrl.LoggerFrom(ctx).V(2).Info(
+			"WARNING: The job has multiple associated Workloads.",
+			"job", jobObj.GetName(),
+			"workloads", klog.KObjSlice(wlList.Items),
+		)
 		return &wlList.Items[0], nil
 	}
 	return nil, nil
 }
 
-// getAncestorJobManagedByKueue traverses controllerRefs to find an ancestor job that is manged by Kueue (ie, it has a queue-name label).
-func (r *JobReconciler) getAncestorJobManagedByKueue(ctx context.Context, jobObj client.Object) (client.Object, error) {
+// FindAncestorJobManagedByKueue traverses controllerRefs to find the top-level ancestor Job managed by Kueue.
+// If manageJobsWithoutQueueName is set to false, it returns only Jobs with a queue-name.
+// If manageJobsWithoutQueueName is true, it may return a Job even if it doesn't have a queue-name.
+//
+// Examples:
+//
+// With manageJobsWithoutQueueName=false:
+// Job -> JobSet -> AppWrapper => nil
+// Job (queue-name) -> JobSet (queue-name) -> AppWrapper => JobSet
+// Job (queue-name) -> JobSet -> AppWrapper (queue-name) => AppWrapper
+// Job (queue-name) -> JobSet (queue-name) -> AppWrapper (queue-name) => AppWrapper
+// Job -> JobSet (disabled) -> AppWrapper (queue-name) => AppWrapper
+//
+// With manageJobsWithoutQueueName=true:
+// Job -> JobSet -> AppWrapper => AppWrapper
+// Job (queue-name) -> JobSet (queue-name) -> AppWrapper => AppWrapper
+// Job (queue-name) -> JobSet -> AppWrapper (queue-name) => AppWrapper
+// Job (queue-name) -> JobSet (queue-name) -> AppWrapper (queue-name) => AppWrapper
+// Job -> JobSet (disabled) -> AppWrapper => AppWrapper
+func FindAncestorJobManagedByKueue(ctx context.Context, c client.Client, record record.EventRecorder, jobObj client.Object, manageJobsWithoutQueueName bool) (client.Object, error) {
+	log := ctrl.LoggerFrom(ctx)
 	seen := sets.New[types.UID]()
-	currentJob := jobObj
+	currentObj := jobObj
+
+	var topLevelJob client.Object
 	for {
-		if seen.Has(currentJob.GetUID()) {
-			return nil, nil
+		if seen.Has(currentObj.GetUID()) {
+			log.Error(ErrCyclicOwnership,
+				"Terminated search for Kueue-managed Job because of cyclic ownership",
+				"owner", currentObj,
+			)
+			return nil, ErrCyclicOwnership
 		}
-		seen.Insert(currentJob.GetUID())
+		seen.Insert(currentObj.GetUID())
 
-		owner := metav1.GetControllerOf(currentJob)
-		if owner == nil || !IsOwnerManagedByKueue(owner) {
-			return nil, nil
+		owner := metav1.GetControllerOf(currentObj)
+		if owner == nil {
+			return topLevelJob, nil
 		}
-		parentJob := GetEmptyOwnerObject(owner)
-		if parentJob == nil {
-			return nil, fmt.Errorf("workload owner %v: %w", owner, ErrUnknownWorkloadOwner)
+
+		parentObj := GetEmptyOwnerObject(owner)
+		managed := parentObj != nil
+		if parentObj == nil {
+			parentObj = &metav1.PartialObjectMetadata{
+				TypeMeta: metav1.TypeMeta{
+					APIVersion: owner.APIVersion,
+					Kind:       owner.Kind,
+				},
+			}
 		}
-		if err := r.client.Get(ctx, client.ObjectKey{Name: owner.Name, Namespace: jobObj.GetNamespace()}, parentJob); err != nil {
+		if err := c.Get(ctx, client.ObjectKey{Name: owner.Name, Namespace: jobObj.GetNamespace()}, parentObj); err != nil {
 			return nil, errors.Join(ErrWorkloadOwnerNotFound, err)
 		}
-		if QueueNameForObject(parentJob) != "" {
-			return parentJob, nil
+		if managed && (manageJobsWithoutQueueName || QueueNameForObject(parentObj) != "") {
+			topLevelJob = parentObj
 		}
-		currentJob = parentJob
+		currentObj = parentObj
 		if len(seen) > managedOwnersChainLimit {
-			r.record.Eventf(jobObj, corev1.EventTypeWarning, ReasonJobNestingTooDeep,
-				"Terminated search for Kueue-managed Job because ancestor depth exceeded limit of %d", managedOwnersChainLimit)
-			ctrl.LoggerFrom(ctx).V(2).Info(
-				"Terminated search for Kueue-managed Job because ancestor depth exceeded managedOwnersChainlimit",
+			record.Eventf(jobObj, corev1.EventTypeWarning, ReasonJobNestingTooDeep,
+				"Terminated search for Kueue-managed Job because ancestor depth exceeded limit of %d", managedOwnersChainLimit,
+			)
+			log.V(2).Info(
+				"WARNING: Terminated search for Kueue-managed Job because ancestor depth exceeded managedOwnersChainLimit",
 				"limit ", managedOwnersChainLimit,
-				"lastParentReached", parentJob,
+				"lastParentReached", parentObj,
 			)
-			return nil, nil
+			return topLevelJob, nil
 		}
 	}
 }