diff --git a/.github/workflows/e2e-test-train-api.yaml b/.github/workflows/e2e-test-train-api.yaml index 045c3b19e2..8d4ca0e734 100644 --- a/.github/workflows/e2e-test-train-api.yaml +++ b/.github/workflows/e2e-test-train-api.yaml @@ -53,7 +53,7 @@ jobs: - name: Run tests run: | - pip install pytest + pip install pytest "kubernetes<36" python3 -m pip install -e sdk/python[huggingface] pytest -s sdk/python/test/e2e-fine-tune-llm/test_e2e_pytorch_fine_tune_llm.py --log-cli-level=debug env: diff --git a/.github/workflows/integration-tests.yaml b/.github/workflows/integration-tests.yaml index a450a76b16..9626d8216b 100644 --- a/.github/workflows/integration-tests.yaml +++ b/.github/workflows/integration-tests.yaml @@ -80,7 +80,7 @@ jobs: - name: Run tests run: | - pip install pytest + pip install pytest "kubernetes<36" python3 -m pip install -e sdk/python; pytest -s sdk/python/test/e2e --log-cli-level=debug --namespace=default env: GANG_SCHEDULER_NAME: ${{ matrix.gang-scheduler-name }} diff --git a/.github/workflows/test-example-notebooks.yaml b/.github/workflows/test-example-notebooks.yaml index 0ee767e165..cfa2d9a8ab 100644 --- a/.github/workflows/test-example-notebooks.yaml +++ b/.github/workflows/test-example-notebooks.yaml @@ -28,7 +28,7 @@ jobs: - name: Install Python Dependencies run: | - pip install papermill==2.6.0 jupyter==1.1.1 ipykernel==6.29.5 + pip install papermill==2.6.0 jupyter==1.1.1 ipykernel==6.29.5 "kubernetes<36" "fastjsonschema<2.22" - name: Run Jupyter Notebook with Papermill shell: bash diff --git a/.github/workflows/test-python.yaml b/.github/workflows/test-python.yaml index 9a706461b7..11ebce3928 100644 --- a/.github/workflows/test-python.yaml +++ b/.github/workflows/test-python.yaml @@ -28,7 +28,12 @@ jobs: - name: Install dependencies run: | - pip install pytest python-dateutil urllib3 kubernetes + # TODO: kubernetes 36.0.0 switched to Pydantic models, breaking the + # FakeResponse deserialization hack in the SDK unit tests. To remove + # this pin, either adapt FakeResponse in + # sdk/python/kubeflow/training/utils/utils.py for Pydantic or mock + # ApiClient.deserialize directly in the test suite. + pip install pytest python-dateutil urllib3 "kubernetes<36" pip install -U './sdk/python[huggingface]' - name: Run unit test for training sdk diff --git a/cmd/training-operator.v1/main.go b/cmd/training-operator.v1/main.go index 8bbb5e1f3d..556c20682a 100644 --- a/cmd/training-operator.v1/main.go +++ b/cmd/training-operator.v1/main.go @@ -111,6 +111,9 @@ func main() { // MPI related flags flag.StringVar(&config.Config.MPIKubectlDeliveryImage, "mpi-kubectl-delivery-image", config.MPIKubectlDeliveryImageDefault, "The image for mpi launcher init container") + flag.BoolVar(&config.Config.DisableMPIRBACManagement, "disable-mpi-rbac-management", false, + "When set to true, disables the MPIJob controller's management of launcher RBAC resources (ServiceAccount, Role, RoleBinding) and ServiceAccountName injection. "+ + "Users must provide their own ServiceAccount with a Role granting get, list, watch on pods and create on pods/exec.") // Cert generation flags flag.IntVar(&webhookServerPort, "webhook-server-port", 9443, "Endpoint port for the webhook server.") diff --git a/pkg/config/config.go b/pkg/config/config.go index 2a6d26514f..fb42b4aa56 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -20,6 +20,7 @@ var Config struct { PyTorchInitContainerImage string MPIKubectlDeliveryImage string PyTorchInitContainerMaxTries int + DisableMPIRBACManagement bool } const ( diff --git a/pkg/controller.v1/mpi/mpijob_controller.go b/pkg/controller.v1/mpi/mpijob_controller.go index e85b9d6ce3..d53cab714d 100644 --- a/pkg/controller.v1/mpi/mpijob_controller.go +++ b/pkg/controller.v1/mpi/mpijob_controller.go @@ -62,8 +62,9 @@ import ( ) const ( - FailedDeleteJobReason = "FailedDeleteJob" - SuccessfulDeleteJobReason = "SuccessfulDeleteJob" + FailedDeleteJobReason = "FailedDeleteJob" + SuccessfulDeleteJobReason = "SuccessfulDeleteJob" + NoServiceAccountNameReason = "NoServiceAccountName" controllerName = "mpijob-controller" labelMPIJobName = "mpi-job-name" @@ -214,23 +215,25 @@ func (jc *MPIJobReconciler) SetupWithManager(mgr ctrl.Manager, controllerThreads util.OnDependentFuncs[*corev1.ConfigMap](jc.Scheme, jc.Expectations, &jc.JobController))); err != nil { return err } - // inject watching for job related Role - if err = c.Watch(source.Kind[*rbacv1.Role](mgr.GetCache(), &rbacv1.Role{}, - handler.TypedEnqueueRequestForOwner[*rbacv1.Role](mgr.GetScheme(), mgr.GetRESTMapper(), &kubeflowv1.MPIJob{}, handler.OnlyControllerOwner()), - util.OnDependentFuncs[*rbacv1.Role](jc.Scheme, jc.Expectations, &jc.JobController))); err != nil { - return err - } - // inject watching for job related RoleBinding - if err = c.Watch(source.Kind[*rbacv1.RoleBinding](mgr.GetCache(), &rbacv1.RoleBinding{}, - handler.TypedEnqueueRequestForOwner[*rbacv1.RoleBinding](mgr.GetScheme(), mgr.GetRESTMapper(), &kubeflowv1.MPIJob{}, handler.OnlyControllerOwner()), - util.OnDependentFuncs[*rbacv1.RoleBinding](jc.Scheme, jc.Expectations, &jc.JobController))); err != nil { - return err - } - // inject watching for job related ServiceAccount - if err = c.Watch(source.Kind[*corev1.ServiceAccount](mgr.GetCache(), &corev1.ServiceAccount{}, - handler.TypedEnqueueRequestForOwner[*corev1.ServiceAccount](mgr.GetScheme(), mgr.GetRESTMapper(), &kubeflowv1.MPIJob{}, handler.OnlyControllerOwner()), - util.OnDependentFuncs[*corev1.ServiceAccount](jc.Scheme, jc.Expectations, &jc.JobController))); err != nil { - return err + if !ctlrconfig.Config.DisableMPIRBACManagement { + // inject watching for job related Role + if err = c.Watch(source.Kind[*rbacv1.Role](mgr.GetCache(), &rbacv1.Role{}, + handler.TypedEnqueueRequestForOwner[*rbacv1.Role](mgr.GetScheme(), mgr.GetRESTMapper(), &kubeflowv1.MPIJob{}, handler.OnlyControllerOwner()), + util.OnDependentFuncs[*rbacv1.Role](jc.Scheme, jc.Expectations, &jc.JobController))); err != nil { + return err + } + // inject watching for job related RoleBinding + if err = c.Watch(source.Kind[*rbacv1.RoleBinding](mgr.GetCache(), &rbacv1.RoleBinding{}, + handler.TypedEnqueueRequestForOwner[*rbacv1.RoleBinding](mgr.GetScheme(), mgr.GetRESTMapper(), &kubeflowv1.MPIJob{}, handler.OnlyControllerOwner()), + util.OnDependentFuncs[*rbacv1.RoleBinding](jc.Scheme, jc.Expectations, &jc.JobController))); err != nil { + return err + } + // inject watching for job related ServiceAccount + if err = c.Watch(source.Kind[*corev1.ServiceAccount](mgr.GetCache(), &corev1.ServiceAccount{}, + handler.TypedEnqueueRequestForOwner[*corev1.ServiceAccount](mgr.GetScheme(), mgr.GetRESTMapper(), &kubeflowv1.MPIJob{}, handler.OnlyControllerOwner()), + util.OnDependentFuncs[*corev1.ServiceAccount](jc.Scheme, jc.Expectations, &jc.JobController))); err != nil { + return err + } } // skip watching volcano PodGroup if volcano PodGroup is not installed if _, err = mgr.GetRESTMapper().RESTMapping(schema.GroupKind{Group: v1beta1.GroupName, Kind: "PodGroup"}, @@ -367,24 +370,25 @@ func (jc *MPIJobReconciler) ReconcilePods( } isGPULauncher := isGPULauncher(mpiJob) - // Get the launcher ServiceAccount for this MPIJob. - if sa, err := jc.getOrCreateLauncherServiceAccount(mpiJob); sa == nil || err != nil { - return err - } - // Get the ConfigMap for this MPIJob. if config, err := jc.getOrCreateConfigMap(mpiJob, workerReplicas, isGPULauncher); config == nil || err != nil { return err } - // Get the launcher Role for this MPIJob. - if r, err := jc.getOrCreateLauncherRole(mpiJob, workerReplicas); r == nil || err != nil { - return err - } + if !ctlrconfig.Config.DisableMPIRBACManagement { + // Get the launcher ServiceAccount for this MPIJob. + if sa, err := jc.getOrCreateLauncherServiceAccount(mpiJob); sa == nil || err != nil { + return err + } + // Get the launcher Role for this MPIJob. + if r, err := jc.getOrCreateLauncherRole(mpiJob, workerReplicas); r == nil || err != nil { + return err + } - // Get the launcher RoleBinding for this MPIJob. - if rb, err := jc.getLauncherRoleBinding(mpiJob); rb == nil || err != nil { - return err + // Get the launcher RoleBinding for this MPIJob. + if rb, err := jc.getLauncherRoleBinding(mpiJob); rb == nil || err != nil { + return err + } } worker, err = jc.getOrCreateWorker(mpiJob) @@ -1034,8 +1038,13 @@ func (jc *MPIJobReconciler) newLauncher(mpiJob *kubeflowv1.MPIJob, kubectlDelive jc.PodGroupControl.DecoratePodTemplateSpec(podSpec, mpiJob, rt) } - if len(mpiJob.Spec.MPIReplicaSpecs[kubeflowv1.MPIJobReplicaTypeLauncher].Template.Spec.ServiceAccountName) == 0 { - podSpec.Spec.ServiceAccountName = launcherName + if !ctlrconfig.Config.DisableMPIRBACManagement { + if len(mpiJob.Spec.MPIReplicaSpecs[kubeflowv1.MPIJobReplicaTypeLauncher].Template.Spec.ServiceAccountName) == 0 { + podSpec.Spec.ServiceAccountName = launcherName + } + } else if len(mpiJob.Spec.MPIReplicaSpecs[kubeflowv1.MPIJobReplicaTypeLauncher].Template.Spec.ServiceAccountName) == 0 { + jc.Recorder.Eventf(mpiJob, corev1.EventTypeWarning, NoServiceAccountNameReason, + "Launcher has no ServiceAccountName; ensure the namespace default ServiceAccount has get, list, watch pods and create pods/exec permissions") } podSpec.Spec.InitContainers = append(podSpec.Spec.InitContainers, corev1.Container{ diff --git a/pkg/controller.v1/mpi/mpijob_controller_test.go b/pkg/controller.v1/mpi/mpijob_controller_test.go index 024cbaaec1..e078f8ecea 100644 --- a/pkg/controller.v1/mpi/mpijob_controller_test.go +++ b/pkg/controller.v1/mpi/mpijob_controller_test.go @@ -17,21 +17,29 @@ package mpi import ( "context" "fmt" + "path/filepath" "strings" common "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/scheme" "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + ctrlconfig "sigs.k8s.io/controller-runtime/pkg/config" + "sigs.k8s.io/controller-runtime/pkg/envtest" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" kubeflowv1 "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1" + ctlrconfig "github.com/kubeflow/training-operator/pkg/config" + ctrlcommon "github.com/kubeflow/training-operator/pkg/controller.v1/common" commonutil "github.com/kubeflow/training-operator/pkg/util" "github.com/kubeflow/training-operator/pkg/util/testutil" ) @@ -573,6 +581,349 @@ var _ = Describe("MPIJob controller", func() { }) }) + Context("MPIJob with disabled RBAC management", func() { + It("Should not create ServiceAccount, Role, or RoleBinding", func() { + By("Setting DisableMPIRBACManagement to true") + oldDisableMPIRBACManagement := ctlrconfig.Config.DisableMPIRBACManagement + ctlrconfig.Config.DisableMPIRBACManagement = true + defer func() { + ctlrconfig.Config.DisableMPIRBACManagement = oldDisableMPIRBACManagement + }() + + By("Creating an MPIJob with a user-provided ServiceAccount") + jobName := "test-disable-rbac" + launcherSaName := "custom-launcher-sa" + + ctx := context.Background() + startTime := metav1.Now() + completionTime := metav1.Now() + + mpiJob := newMPIJob(jobName, ptr.To[int32](1), 1, gpuResourceName, &startTime, &completionTime) + mpiJob.Spec.MPIReplicaSpecs[kubeflowv1.MPIJobReplicaTypeLauncher].Template.Spec.ServiceAccountName = launcherSaName + + // Pre-create the ServiceAccount since the operator won't create it + sa := &corev1.ServiceAccount{ + ObjectMeta: metav1.ObjectMeta{ + Name: launcherSaName, + Namespace: metav1.NamespaceDefault, + }, + } + Expect(testK8sClient.Create(ctx, sa)).Should(Succeed()) + Expect(testK8sClient.Create(ctx, mpiJob)).Should(Succeed()) + + By("Reconciling until the launcher pod is created") + Eventually(func() error { + req := ctrl.Request{NamespacedName: types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: mpiJob.GetName(), + }} + _, err := reconciler.Reconcile(ctx, req) + return err + }, testutil.Timeout, testutil.Interval).Should(BeNil()) + + By("Verifying no Role was created") + Eventually(func() bool { + role := &rbacv1.Role{} + err := testK8sClient.Get(ctx, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, role) + return errors.IsNotFound(err) + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying no RoleBinding was created") + Eventually(func() bool { + rb := &rbacv1.RoleBinding{} + err := testK8sClient.Get(ctx, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, rb) + return errors.IsNotFound(err) + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying no operator-created ServiceAccount was created") + Eventually(func() bool { + operatorSa := &corev1.ServiceAccount{} + err := testK8sClient.Get(ctx, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, operatorSa) + return errors.IsNotFound(err) + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying ConfigMap was created") + Eventually(func() error { + cm := &corev1.ConfigMap{} + return testK8sClient.Get(ctx, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + configSuffix, + }, cm) + }, testutil.Timeout, testutil.Interval).Should(Succeed()) + + By("Verifying kubectl delivery init container is present") + Eventually(func() bool { + launcher := &corev1.Pod{} + err := testK8sClient.Get(ctx, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, launcher) + if err != nil { + return false + } + for _, ic := range launcher.Spec.InitContainers { + if ic.Name == kubectlDeliveryName { + return true + } + } + return false + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying launcher pod uses the user-provided ServiceAccount") + Eventually(func() string { + launcher := &corev1.Pod{} + err := testK8sClient.Get(ctx, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, launcher) + if err != nil { + return "" + } + return launcher.Spec.ServiceAccountName + }, testutil.Timeout, testutil.Interval).Should(Equal(launcherSaName)) + }) + + It("Should warn when no ServiceAccount is provided on the launcher", func() { + By("Setting DisableMPIRBACManagement to true") + oldDisableMPIRBACManagement := ctlrconfig.Config.DisableMPIRBACManagement + ctlrconfig.Config.DisableMPIRBACManagement = true + defer func() { + ctlrconfig.Config.DisableMPIRBACManagement = oldDisableMPIRBACManagement + }() + + By("Creating an MPIJob without a user-provided ServiceAccount") + jobName := "test-disable-rbac-no-sa" + + ctx := context.Background() + startTime := metav1.Now() + completionTime := metav1.Now() + + mpiJob := newMPIJob(jobName, ptr.To[int32](1), 1, gpuResourceName, &startTime, &completionTime) + Expect(testK8sClient.Create(ctx, mpiJob)).Should(Succeed()) + + By("Reconciling until the launcher pod is created") + Eventually(func() error { + req := ctrl.Request{NamespacedName: types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: mpiJob.GetName(), + }} + _, err := reconciler.Reconcile(ctx, req) + return err + }, testutil.Timeout, testutil.Interval).Should(BeNil()) + + By("Verifying a warning event was recorded") + Eventually(func() bool { + eventList := &corev1.EventList{} + err := testK8sClient.List(ctx, eventList, client.InNamespace(metav1.NamespaceDefault)) + if err != nil { + return false + } + for _, event := range eventList.Items { + if event.Reason == NoServiceAccountNameReason && event.InvolvedObject.Name == jobName { + return true + } + } + return false + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + }) + }) + + Context("SetupWithManager with disabled RBAC management", func() { + It("Should not watch RBAC resources when flag is true", func() { + By("Starting a second envtest instance for isolation") + testEnv2 := &envtest.Environment{ + CRDDirectoryPaths: []string{filepath.Join("..", "..", "..", "manifests", "base", "crds")}, + ErrorIfCRDPathMissing: true, + } + cfg2, err := testEnv2.Start() + Expect(err).NotTo(HaveOccurred()) + Expect(cfg2).NotTo(BeNil()) + defer func() { + Expect(testEnv2.Stop()).NotTo(HaveOccurred()) + }() + + By("Setting DisableMPIRBACManagement to true") + oldDisableMPIRBACManagement := ctlrconfig.Config.DisableMPIRBACManagement + ctlrconfig.Config.DisableMPIRBACManagement = true + defer func() { + ctlrconfig.Config.DisableMPIRBACManagement = oldDisableMPIRBACManagement + }() + + By("Creating a second manager with disabled RBAC management") + mgr2, err := ctrl.NewManager(cfg2, ctrl.Options{ + Metrics: metricsserver.Options{ + BindAddress: "0", + }, + Controller: ctrlconfig.Controller{ + SkipNameValidation: ptr.To(true), + }, + }) + Expect(err).NotTo(HaveOccurred()) + + gangSchedulingSetupFunc := ctrlcommon.GenNonGangSchedulerSetupFunc() + reconciler2 := NewReconciler(mgr2, gangSchedulingSetupFunc) + Expect(reconciler2.SetupWithManager(mgr2, 1)).NotTo(HaveOccurred()) + + ctx2, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + defer GinkgoRecover() + err := mgr2.Start(ctx2) + Expect(err).ToNot(HaveOccurred(), "failed to run second manager") + }() + + By("Creating a client for the second envtest") + client2, err := client.New(cfg2, client.Options{Scheme: scheme.Scheme}) + Expect(err).NotTo(HaveOccurred()) + Expect(client2).NotTo(BeNil()) + + By("Creating an MPIJob with a user-provided ServiceAccount") + jobName := "test-watch-suppression" + launcherSaName := "custom-watch-sa" + + mpiJob := newMPIJob(jobName, ptr.To[int32](1), 1, gpuResourceName, nil, nil) + mpiJob.Spec.MPIReplicaSpecs[kubeflowv1.MPIJobReplicaTypeLauncher].Template.Spec.ServiceAccountName = launcherSaName + + sa := &corev1.ServiceAccount{ + ObjectMeta: metav1.ObjectMeta{ + Name: launcherSaName, + Namespace: metav1.NamespaceDefault, + }, + } + Expect(client2.Create(ctx2, sa)).Should(Succeed()) + Expect(client2.Create(ctx2, mpiJob)).Should(Succeed()) + + launcherKey := types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + } + + By("Waiting for the launcher pod to be created") + Eventually(func() error { + return client2.Get(ctx2, launcherKey, &corev1.Pod{}) + }, testutil.Timeout, testutil.Interval).Should(Succeed()) + + By("Verifying no Role was created") + Eventually(func() bool { + role := &rbacv1.Role{} + err := client2.Get(ctx2, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, role) + return errors.IsNotFound(err) + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying no RoleBinding was created") + Eventually(func() bool { + rb := &rbacv1.RoleBinding{} + err := client2.Get(ctx2, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, rb) + return errors.IsNotFound(err) + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying no operator-created ServiceAccount was created") + Eventually(func() bool { + operatorSa := &corev1.ServiceAccount{} + err := client2.Get(ctx2, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + launcherSuffix, + }, operatorSa) + return errors.IsNotFound(err) + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying ConfigMap was created") + Eventually(func() error { + return client2.Get(ctx2, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + configSuffix, + }, &corev1.ConfigMap{}) + }, testutil.Timeout, testutil.Interval).Should(Succeed()) + + By("Verifying kubectl delivery init container is present") + Eventually(func() bool { + launcher := &corev1.Pod{} + err := client2.Get(ctx2, launcherKey, launcher) + if err != nil { + return false + } + for _, ic := range launcher.Spec.InitContainers { + if ic.Name == kubectlDeliveryName { + return true + } + } + return false + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + By("Verifying ConfigMap watch is active via ConfigMap update") + cm := &corev1.ConfigMap{} + Expect(client2.Get(ctx2, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + configSuffix, + }, cm)).Should(Succeed()) + + // Corrupt the ConfigMap data + cm.Data = map[string]string{"corrupted": "true"} + Expect(client2.Update(ctx2, cm)).Should(Succeed()) + + // The ConfigMap update should trigger a reconcile via the ConfigMap watch. + // The reconciler should restore the original data. + Eventually(func() bool { + restoredCm := &corev1.ConfigMap{} + err := client2.Get(ctx2, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + configSuffix, + }, restoredCm) + if err != nil { + return false + } + // Check that the corrupted key is gone and original data is restored + _, hasCorrupt := restoredCm.Data["corrupted"] + return !hasCorrupt && len(restoredCm.Data) > 0 + }, testutil.Timeout, testutil.Interval).Should(BeTrue()) + + // Note: RBAC watch suppression at the SetupWithManager level (the + // `if !ctlrconfig.Config.DisableMPIRBACManagement` guard at line ~217) + // is verified structurally by code inspection, not behaviorally. A + // behavioral negative control is infeasible in envtest because: + // 1. The Pod watch is always active (not gated by the flag), so pod + // state cannot be used as a signal for RBAC watch activity. + // 2. No-op reconciles don't change MPIJob status in envtest (pods stay + // Pending, reflect.DeepEqual guard prevents status writes), so + // resourceVersion stability doesn't prove watch suppression. + // The test verifies: SetupWithManager succeeds with flag=true, no RBAC + // resources are created, ConfigMap/kubectl delivery remain unconditional, + // ConfigMap and Pod watches remain active. + By("Deleting launcher pod to verify reconciler is still functional") + launcher := &corev1.Pod{} + Expect(client2.Get(ctx2, launcherKey, launcher)).Should(Succeed()) + Expect(client2.Delete(ctx2, launcher)).Should(Succeed()) + + By("Verifying launcher pod is recreated by the reconciler") + Eventually(func() error { + return client2.Get(ctx2, launcherKey, &corev1.Pod{}) + }, testutil.Timeout, testutil.Interval).Should(Succeed()) + + By("Verifying ConfigMap still exists after reconcile") + Eventually(func() error { + return client2.Get(ctx2, types.NamespacedName{ + Namespace: metav1.NamespaceDefault, + Name: jobName + configSuffix, + }, &corev1.ConfigMap{}) + }, testutil.Timeout, testutil.Interval).Should(Succeed()) + }) + }) + Context("MPIJob with launcher Pod not controlled by itself", func() { It("Should return error", func() { By("Calling Reconcile method") diff --git a/sdk/python/test/e2e/test_e2e_jaxjob.py b/sdk/python/test/e2e/test_e2e_jaxjob.py index 7471f67338..9ddb18cfd4 100644 --- a/sdk/python/test/e2e/test_e2e_jaxjob.py +++ b/sdk/python/test/e2e/test_e2e_jaxjob.py @@ -156,5 +156,5 @@ def generate_container() -> V1Container: return V1Container( name=CONTAINER_NAME, image=os.getenv("JAX_JOB_IMAGE", "docker.io/kubeflow/jaxjob-dist-spmd-mnist:latest"), - resources=V1ResourceRequirements(limits={"memory": "3Gi", "cpu": "1.2"}), + resources=V1ResourceRequirements(limits={"memory": "5Gi", "cpu": "1.2"}), )