From c8c1cc022c017d45503b06644019966e912044d8 Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Fri, 24 Jul 2026 16:31:04 +0800 Subject: [PATCH 1/9] fix(ci): pin kubernetes<36 in all Python CI workflows kubernetes 36.0.0 regenerated the sync client with Pydantic models, breaking the FakeResponse deserialization hack used in SDK unit tests. Pin kubernetes<36 in all CI workflows that install the Python SDK (test-python, integration-tests, e2e-test-train-api, test-example-notebooks) to avoid impacting SDK release. Add a TODO for future Pydantic work. Signed-off-by: Vico Chu --- .github/workflows/e2e-test-train-api.yaml | 2 +- .github/workflows/integration-tests.yaml | 2 +- .github/workflows/test-example-notebooks.yaml | 2 +- .github/workflows/test-python.yaml | 7 ++++++- 4 files changed, 9 insertions(+), 4 deletions(-) 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..61338ef2ea 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" - 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 From 82b2b71ee3b18ec16ff36a28c73b68fe165753a7 Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Fri, 24 Jul 2026 19:42:09 +0800 Subject: [PATCH 2/9] fix(test): increase JAX e2e test memory limit to avoid OOMKilled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit JAX worker pods were OOMKilled (exit 137) with 3Gi memory limit. Increase to 5Gi to accommodate XLA compilation and training memory usage. GitHub-hosted runners have 16GB RAM, so 2 replicas × 5Gi is feasible. Signed-off-by: Vico Chu --- sdk/python/test/e2e/test_e2e_jaxjob.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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"}), ) From 421f35581c0c184159c091e83db7ff88e9e9daa1 Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Mon, 27 Jul 2026 11:06:59 +0800 Subject: [PATCH 3/9] fix(ci): pin fastjsonschema<2.22 to fix Python 3.9 compatibility fastjsonschema 2.22.0 introduced PEP 604 union type syntax (dict | bool) in type annotations, which requires Python 3.10+. This breaks papermill imports on Python 3.9 in the notebook test workflow. Pin to <2.22 until Python 3.9 support is dropped from the test matrix. Signed-off-by: Vico Chu --- .github/workflows/test-example-notebooks.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test-example-notebooks.yaml b/.github/workflows/test-example-notebooks.yaml index 61338ef2ea..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 "kubernetes<36" + 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 From cf4382d80490b01afd7c59f738862894a8a06628 Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Tue, 14 Jul 2026 19:32:30 +0800 Subject: [PATCH 4/9] feat(mpi): add --disable-mpi-rbac-management flag When the flag is set to true, the MPIJob controller skips creating ServiceAccount, Role, and RoleBinding, skips injecting ServiceAccountName into the launcher pod, and skips watching RBAC resources. Users must pre-provide a ServiceAccount with sufficient permissions via the launcher pod template's spec.serviceAccountName. ConfigMap creation and kubectl delivery remain unconditional. Signed-off-by: Vico Chu --- cmd/training-operator.v1/main.go | 2 + pkg/config/config.go | 1 + pkg/controller.v1/mpi/mpijob_controller.go | 67 +++++++------ .../mpi/mpijob_controller_test.go | 95 +++++++++++++++++++ 4 files changed, 134 insertions(+), 31 deletions(-) diff --git a/cmd/training-operator.v1/main.go b/cmd/training-operator.v1/main.go index 8bbb5e1f3d..f2b7754ae1 100644 --- a/cmd/training-operator.v1/main.go +++ b/cmd/training-operator.v1/main.go @@ -111,6 +111,8 @@ 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.") // 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..fb08736bb0 100644 --- a/pkg/controller.v1/mpi/mpijob_controller.go +++ b/pkg/controller.v1/mpi/mpijob_controller.go @@ -214,23 +214,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 +369,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 +1037,10 @@ 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 + } } 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..499fae28ba 100644 --- a/pkg/controller.v1/mpi/mpijob_controller_test.go +++ b/pkg/controller.v1/mpi/mpijob_controller_test.go @@ -23,6 +23,7 @@ import ( . "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" @@ -32,6 +33,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" kubeflowv1 "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1" + ctlrconfig "github.com/kubeflow/training-operator/pkg/config" commonutil "github.com/kubeflow/training-operator/pkg/util" "github.com/kubeflow/training-operator/pkg/util/testutil" ) @@ -573,6 +575,99 @@ 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") + ctlrconfig.Config.DisableMPIRBACManagement = true + defer func() { + ctlrconfig.Config.DisableMPIRBACManagement = false + }() + + 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 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)) + }) + }) + Context("MPIJob with launcher Pod not controlled by itself", func() { It("Should return error", func() { By("Calling Reconcile method") From 0861b7faee4e00c6594bf65ec277d228b02fba8f Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Wed, 22 Jul 2026 13:46:25 +0800 Subject: [PATCH 5/9] test(mpi): add SetupWithManager watch suppression test Add integration test that starts a second envtest instance with DisableMPIRBACManagement=true and verifies: - No Role/RoleBinding/ServiceAccount created by the operator - ConfigMap and kubectl delivery init container still created - Role changes do not trigger reconcile (watch suppressed) - ConfigMap changes still trigger reconcile (positive control) - Launcher pod recreated only after ConfigMap-triggered reconcile Also adds kubectl delivery init container assertion to pin the spec invariant that kubectl delivery remains unconditional. Signed-off-by: Vico Chu --- .../mpi/mpijob_controller_test.go | 190 ++++++++++++++++++ 1 file changed, 190 insertions(+) diff --git a/pkg/controller.v1/mpi/mpijob_controller_test.go b/pkg/controller.v1/mpi/mpijob_controller_test.go index 499fae28ba..f1cee14c47 100644 --- a/pkg/controller.v1/mpi/mpijob_controller_test.go +++ b/pkg/controller.v1/mpi/mpijob_controller_test.go @@ -17,6 +17,7 @@ package mpi import ( "context" "fmt" + "path/filepath" "strings" common "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1" @@ -28,12 +29,17 @@ import ( "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" ) @@ -668,6 +674,190 @@ var _ = Describe("MPIJob controller", func() { }) }) + 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 testEnv2.Stop() + + By("Setting DisableMPIRBACManagement to true") + ctlrconfig.Config.DisableMPIRBACManagement = true + defer func() { + ctlrconfig.Config.DisableMPIRBACManagement = false + }() + + 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") From abdbdc46095d5794096acdc9d937c7b74beab8cb Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Wed, 22 Jul 2026 18:19:12 +0800 Subject: [PATCH 6/9] test(mpi): add kubectl delivery assertion to disabled RBAC test Add kubectl delivery init container check to the first disabled RBAC test context, mirroring the assertion in the SetupWithManager test. Both ConfigMap creation and kubectl delivery are unconditional (not gated by the flag), so both should be verified when the flag is true. Signed-off-by: Vico Chu --- .../mpi/mpijob_controller_test.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/pkg/controller.v1/mpi/mpijob_controller_test.go b/pkg/controller.v1/mpi/mpijob_controller_test.go index f1cee14c47..f7c4caf2f0 100644 --- a/pkg/controller.v1/mpi/mpijob_controller_test.go +++ b/pkg/controller.v1/mpi/mpijob_controller_test.go @@ -659,6 +659,24 @@ var _ = Describe("MPIJob controller", func() { }, 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{} From 18573ed6b23458dc197309d203fa00fc2a28cbba Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Thu, 23 Jul 2026 20:02:31 +0800 Subject: [PATCH 7/9] fix(mpi): save/restore prior config value and assert testEnv2.Stop succeeds Signed-off-by: Vico Chu --- pkg/controller.v1/mpi/mpijob_controller_test.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/pkg/controller.v1/mpi/mpijob_controller_test.go b/pkg/controller.v1/mpi/mpijob_controller_test.go index f7c4caf2f0..60780e4551 100644 --- a/pkg/controller.v1/mpi/mpijob_controller_test.go +++ b/pkg/controller.v1/mpi/mpijob_controller_test.go @@ -584,9 +584,10 @@ 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 = false + ctlrconfig.Config.DisableMPIRBACManagement = oldDisableMPIRBACManagement }() By("Creating an MPIJob with a user-provided ServiceAccount") @@ -702,12 +703,15 @@ var _ = Describe("MPIJob controller", func() { cfg2, err := testEnv2.Start() Expect(err).NotTo(HaveOccurred()) Expect(cfg2).NotTo(BeNil()) - defer testEnv2.Stop() + 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 = false + ctlrconfig.Config.DisableMPIRBACManagement = oldDisableMPIRBACManagement }() By("Creating a second manager with disabled RBAC management") From 2cc0c48136b3db5cf2085bbb0877827405b4910b Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Thu, 23 Jul 2026 20:15:00 +0800 Subject: [PATCH 8/9] docs(mpi): document required permissions in --disable-mpi-rbac-management help text Specify that the user-provided ServiceAccount needs a Role granting get, list, watch on pods and create on pods/exec, matching the permissions that newLauncherRole would have granted. Signed-off-by: Vico Chu --- cmd/training-operator.v1/main.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/cmd/training-operator.v1/main.go b/cmd/training-operator.v1/main.go index f2b7754ae1..556c20682a 100644 --- a/cmd/training-operator.v1/main.go +++ b/cmd/training-operator.v1/main.go @@ -112,7 +112,8 @@ func main() { 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.") + "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.") From 1200c717474f00c9748e09d46ee7d4d7a4b59a66 Mon Sep 17 00:00:00 2001 From: Vico Chu Date: Thu, 23 Jul 2026 20:28:38 +0800 Subject: [PATCH 9/9] feat(mpi): add warning event if there is no serviceAccountName in launcher template Signed-off-by: Vico Chu --- pkg/controller.v1/mpi/mpijob_controller.go | 8 +++- .../mpi/mpijob_controller_test.go | 44 +++++++++++++++++++ 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/pkg/controller.v1/mpi/mpijob_controller.go b/pkg/controller.v1/mpi/mpijob_controller.go index fb08736bb0..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" @@ -1041,6 +1042,9 @@ func (jc *MPIJobReconciler) newLauncher(mpiJob *kubeflowv1.MPIJob, kubectlDelive 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 60780e4551..e078f8ecea 100644 --- a/pkg/controller.v1/mpi/mpijob_controller_test.go +++ b/pkg/controller.v1/mpi/mpijob_controller_test.go @@ -691,6 +691,50 @@ var _ = Describe("MPIJob controller", func() { 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() {