Introduce ManagedBy field in RunPolicy

deploy/v2beta1/mpi-operator.yaml

@@ -7757,6 +7757,17 @@ spec:
                       CleanPodPolicy defines the policy to kill pods after the job completes.
                       Default to Running.
                     type: string
+                  managedBy:
+                    description: |-
+                      ManagedBy is used to indicate the controller or entity that manages a MPIJob.
+                      The value must be either an empty, 'kubeflow.org/mpi-operator' or
+                      'kueue.x-k8s.io/multikueue'.
+                      The mpi-operator reconciles a MPIJob which doesn't have this
+                      field at all or the field value is the reserved string
+                      'kubeflow.org/mpi-operator', but delegates reconciling the MPIJob
+                      with 'kueue.x-k8s.io/multikueue' to the Kueue.
+                      The field is immutable.
+                    type: string
                   schedulingPolicy:
                     description: SchedulingPolicy defines the policy related to scheduling,
                       e.g. gang-scheduling

manifests/base/kubeflow.org_mpijobs.yaml

@@ -7734,6 +7734,17 @@ spec:
                       CleanPodPolicy defines the policy to kill pods after the job completes.
                       Default to Running.
                     type: string
+                  managedBy:
+                    description: |-
+                      ManagedBy is used to indicate the controller or entity that manages a MPIJob.
+                      The value must be either an empty, 'kubeflow.org/mpi-operator' or
+                      'kueue.x-k8s.io/multikueue'.
+                      The mpi-operator reconciles a MPIJob which doesn't have this
+                      field at all or the field value is the reserved string
+                      'kubeflow.org/mpi-operator', but delegates reconciling the MPIJob
+                      with 'kueue.x-k8s.io/multikueue' to the Kueue.
+                      The field is immutable.
+                    type: string
                   schedulingPolicy:
                     description: SchedulingPolicy defines the policy related to scheduling,
                       e.g. gang-scheduling

pkg/apis/kubeflow/v2beta1/swagger.json

@@ -241,6 +241,10 @@
           "description": "CleanPodPolicy defines the policy to kill pods after the job completes. Default to Running.",
           "type": "string"
         },
+        "managedBy": {
+          "description": "ManagedBy is used to indicate the controller or entity that manages a MPIJob. The value must be either an empty, 'kubeflow.org/mpi-operator' or 'kueue.x-k8s.io/multikueue'. The mpi-operator reconciles a MPIJob which doesn't have this field at all or the field value is the reserved string 'kubeflow.org/mpi-operator', but delegates reconciling the MPIJob with 'kueue.x-k8s.io/multikueue' to the Kueue. The field is immutable.",
+          "type": "string"
+        },
         "schedulingPolicy": {
           "description": "SchedulingPolicy defines the policy related to scheduling, e.g. gang-scheduling",
           "$ref": "#/definitions/v2beta1.SchedulingPolicy"

pkg/apis/kubeflow/v2beta1/types.go

@@ -93,6 +93,14 @@ type SchedulingPolicy struct {
 	ScheduleTimeoutSeconds *int32 `json:"scheduleTimeoutSeconds,omitempty"`
 }
 
+const (
+	// KubeflowJobController represents the value of the default job controller
+	KubeflowJobController = "kubeflow.org/mpi-operator"
+
+	// MultiKueueController represents the vaue of the MultiKueue controller
+	MultiKueueController = "kueue.x-k8s.io/multikueue"
+)
+
 // RunPolicy encapsulates various runtime policies of the distributed training
 // job, for example how to clean up resources and how long the job can stay
 // active.
@@ -131,6 +139,17 @@ type RunPolicy struct {
 	// Defaults to false.
 	// +kubebuilder:default:=false
 	Suspend *bool `json:"suspend,omitempty"`
+
+	// ManagedBy is used to indicate the controller or entity that manages a MPIJob.
+	// The value must be either an empty, 'kubeflow.org/mpi-operator' or
+	// 'kueue.x-k8s.io/multikueue'.
+	// The mpi-operator reconciles a MPIJob which doesn't have this
+	// field at all or the field value is the reserved string
+	// 'kubeflow.org/mpi-operator', but delegates reconciling the MPIJob
+	// with 'kueue.x-k8s.io/multikueue' to the Kueue.
+	// The field is immutable.
+	// +optional
+	ManagedBy *string `json:"managedBy,omitempty"`
 }
 
 type LauncherCreationPolicy string

pkg/apis/kubeflow/validation/validation.go

@@ -39,8 +39,11 @@ var (
 
 	validRestartPolicies = sets.NewString(
 		string(kubeflow.RestartPolicyNever),
-		string(kubeflow.RestartPolicyOnFailure),
-	)
+		string(kubeflow.RestartPolicyOnFailure))
+
+	validManagedBy = sets.NewString(
+		string(kubeflow.MultiKueueController),
+		string(kubeflow.KubeflowJobController))
 )
 
 func ValidateMPIJob(job *kubeflow.MPIJob) field.ErrorList {
@@ -98,6 +101,11 @@ func validateRunPolicy(policy *kubeflow.RunPolicy, path *field.Path) field.Error
 	if policy.BackoffLimit != nil {
 		errs = append(errs, apivalidation.ValidateNonnegativeField(int64(*policy.BackoffLimit), path.Child("backoffLimit"))...)
 	}
+	if policy.ManagedBy != nil {
+		if !validManagedBy.Has(*policy.ManagedBy) {
+			errs = append(errs, field.NotSupported(path.Child("managedBy"), *policy.ManagedBy, validManagedBy.List()))
+		}
+	}
 	return errs
 }
 

pkg/apis/kubeflow/validation/validation_test.go

@@ -193,6 +193,7 @@ func TestValidateMPIJob(t *testing.T) {
 						TTLSecondsAfterFinished: ptr.To[int32](-1),
 						ActiveDeadlineSeconds:   ptr.To[int64](-1),
 						BackoffLimit:            ptr.To[int32](-1),
+						ManagedBy:               ptr.To("invalid.com/controller"),
 					},
 					SSHAuthMountPath:  "/root/.ssh",
 					MPIImplementation: kubeflow.MPIImplementation("Unknown"),
@@ -239,6 +240,10 @@ func TestValidateMPIJob(t *testing.T) {
 					Type:  field.ErrorTypeInvalid,
 					Field: "spec.runPolicy.backoffLimit",
 				},
+				{
+					Type:  field.ErrorTypeNotSupported,
+					Field: "spec.runPolicy.managedBy",
+				},
 				{
 					Type:  field.ErrorTypeNotSupported,
 					Field: "spec.mpiImplementation",

pkg/controller/mpi_job_controller.go

@@ -582,6 +582,11 @@ func (c *MPIJobController) syncHandler(key string) error {
 	// Set default for the new mpiJob.
 	scheme.Scheme.Default(mpiJob)
 
+	if manager := managedByExternalController(mpiJob.Spec.RunPolicy.ManagedBy); manager != nil {
+		klog.V(2).Info("Skipping MPIJob managed by a custom controller", "managed-by", manager)
+		return nil
+	}
+
 	// for mpi job that is terminating, just return.
 	if mpiJob.DeletionTimestamp != nil {
 		return nil
@@ -1722,3 +1727,10 @@ func truncateMessage(message string) string {
 	suffix := "..."
 	return message[:eventMessageLimit-len(suffix)] + suffix
 }
+
+func managedByExternalController(controllerName *string) *string {
+	if controllerName != nil && *controllerName != kubeflow.KubeflowJobController {
+		return controllerName
+	}
+	return nil
+}

pkg/controller/mpi_job_controller_test.go

@@ -418,6 +418,11 @@ func (f *fixture) expectCreateSecretAction(d *corev1.Secret) {
 	f.kubeActions = append(f.kubeActions, core.NewCreateAction(schema.GroupVersionResource{Resource: "secrets"}, d.Namespace, d))
 }
 
+func (f *fixture) expectNoKubeActions() bool {
+	k8sActions := filterInformerActions(f.kubeClient.Actions())
+	return len(k8sActions) == 0
+}
+
 func (f *fixture) expectUpdateMPIJobStatusAction(mpiJob *kubeflow.MPIJob) {
 	action := core.NewUpdateAction(schema.GroupVersionResource{Resource: "mpijobs"}, mpiJob.Namespace, mpiJob)
 	action.Subresource = "status"
@@ -504,6 +509,21 @@ func TestDoNothingWithInvalidMPIJob(t *testing.T) {
 	f.run(getKey(mpiJob, t))
 }
 
+func TestDoNothingWithMPIJobManagedExternally(t *testing.T) {
+	f := newFixture(t, "")
+	var replicas int32 = 1
+	startTime := metav1.Now()
+	completionTime := metav1.Now()
+	mpiJob := newMPIJob("test", &replicas, &startTime, &completionTime)
+	mpiJob.Spec.MPIImplementation = kubeflow.MPIImplementationOpenMPI
+	mpiJob.Spec.RunPolicy.ManagedBy = ptr.To(kubeflow.MultiKueueController)
+	f.setUpMPIJob(mpiJob)
+	f.run(getKey(mpiJob, t))
+	if !f.expectNoKubeActions() {
+		t.Fatalf("Expected no kubeActions (secrets, pods, services etc.)")
+	}
+}
+
 func TestAllResourcesCreated(t *testing.T) {
 	impls := []kubeflow.MPIImplementation{kubeflow.MPIImplementationOpenMPI, kubeflow.MPIImplementationIntel, kubeflow.MPIImplementationMPICH}
 	for _, implementation := range impls {