feat: add KubeflowExecutor for Kubeflow Training Operator on Kubernetes

Introduces KubeflowExecutor and a matching TorchX scheduler so users can
deploy distributed training jobs to any Kubernetes cluster running the
Kubeflow Training Operator via run.run() / run.Experiment.

Supported job kinds (toggled via job_kind field):
- PyTorchJob (Training Operator v1, kubeflow.org/v1)
- TrainJob   (Training Operator v2, trainer.kubeflow.org/v1alpha1)

Key features:
- Kubernetes config loaded automatically (local kubeconfig → in-cluster fallback)
- PyTorchJob: builds Master + Worker replica specs with nprocPerNode
- TrainJob: builds spec.trainer + merges all pod-level config (volumes,
  tolerations, affinity, imagePullSecrets, resourceClaims, etc.) into a
  single podTemplateOverrides entry targeting "node"
- env_list field supports full env var dicts (valueFrom / secretKeyRef)
- pod_spec_overrides merges arbitrary extra fields into the pod spec
- launch(wait=True) polls until RUNNING / SUCCEEDED / FAILED
- cancel(wait=True) polls until CR is gone and all pods are terminated
- TorchX scheduler persists job state in ~/.nemo_run/.kubeflow_jobs.json
  and maps KubeflowJobState → AppState (UNKNOWN/None → PENDING to avoid
  false failures on transient API errors)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: oliver könig <okoenig@nvidia.com>

Author: ko3n1g

.gitignore

@@ -183,3 +183,4 @@ _version.py
 
 # NeMo Run
 .nemo_run/
+local/

docs/guides/execution.md

@@ -53,6 +53,7 @@ The packager support matrix is described below:
 | SkypilotExecutor | run.Packager, run.GitArchivePackager, run.PatternPackager, run.HybridPackager |
 | DGXCloudExecutor | run.Packager, run.GitArchivePackager, run.PatternPackager, run.HybridPackager |
 | LeptonExecutor   | run.Packager, run.GitArchivePackager, run.PatternPackager, run.HybridPackager |
+| KubeflowExecutor | run.Packager |
 
 `run.Packager` is a passthrough base packager.
 
@@ -293,6 +294,49 @@ def your_dgx_executor(nodes: int, gpus_per_node: int, container_image: str):
 
 For a complete end-to-end example using DGX Cloud with NeMo, refer to the [NVIDIA DGX Cloud NeMo End-to-End Workflow Example](https://docs.nvidia.com/dgx-cloud/run-ai/latest/nemo-e2e-example.html).
 
+#### KubeflowExecutor
+
+The `KubeflowExecutor` integrates with the [Kubeflow Training Operator](https://github.com/kubeflow/training-operator) to run distributed training jobs on any Kubernetes cluster. It submits CRDs directly via the Kubernetes API — no `kubectl` required.
+
+Two job kinds are supported via the `job_kind` parameter:
+
+- **`"PyTorchJob"`** (default) — Training Operator v1 (`kubeflow.org/v1`)
+- **`"TrainJob"`** — Training Operator v2 (`trainer.kubeflow.org/v1alpha1`)
+
+Kubernetes configuration is loaded automatically: local kubeconfig is tried first, falling back to in-cluster config when running inside a pod.
+
+Here's an example configuration:
+
+```python
+# PyTorchJob (default)
+executor = run.KubeflowExecutor(
+    namespace="runai-nemo-ci",
+    image="nvcr.io/nvidian/nemo:nightly",
+    num_nodes=3,          # total pods: 1 Master + (num_nodes-1) Workers
+    gpus_per_node=8,      # also sets nproc_per_node unless overridden explicitly
+    cpu_requests="16",
+    memory_requests="64Gi",
+    volumes=[
+        {"name": "model-cache", "persistentVolumeClaim": {"claimName": "nemo-ci-datasets-project-nkf5l"}}
+    ],
+    volume_mounts=[{"name": "model-cache", "mountPath": "/nemo-workspace"}],
+    labels={"app": "nemo-ci-training"},
+    env_vars={"NCCL_DEBUG": "INFO"},
+)
+
+# TrainJob (Training Operator v2)
+executor = run.KubeflowExecutor(
+    job_kind="TrainJob",
+    runtime_ref="torch-distributed",  # name of the ClusterTrainingRuntime
+    namespace="runai-nemo-ci",
+    image="nvcr.io/nvidian/nemo:nightly",
+    num_nodes=3,
+    gpus_per_node=8,
+)
+```
+
+`cancel(wait=True)` polls until both the CR and all associated pods are fully terminated before returning.
+
 #### LeptonExecutor
 
 The `LeptonExecutor` integrates with an NVIDIA DGX Cloud Lepton cluster's Python SDK to launch distributed jobs. It uses API calls behind the Lepton SDK to authenticate, identify the target node group and resource shapes, and submit the job specification which will be launched as a batch job on the cluster.

nemo_run/run/torchx_backend/schedulers/kubeflow.py

@@ -194,7 +194,8 @@ def _cancel_existing(self, app_id: str) -> None:
             return None
         executor.cancel(job_name)
 
-    def list(self) -> list[ListAppResponse]: ...
+    def list(self) -> list[ListAppResponse]:
+        return []
 
     def _validate(self, app: AppDef, scheduler: str) -> None:
         pass