Repository: kubernetes-sigs/kueue

Issue:
Support KubeRay in MultiKueue via managedBy
**What would you like to be added**:

Support for KubeRay via managedBy in MultiKueue.

The relevant support for the managedBy field has been recently merged in KubeRay, see https://github.com/ray-project/kuberay/issues/2544, and will be released most likely in 1.3.
Until then we can use the main branch of kuberay in Kueue to test it all works. Once KubeRay is released we can switch to the released version and merge to Kueue.

**Why is this needed**:

Support of KubeRay via managedBy in MultiKueue, allowing for:
- ease of setup where full KubeRay can be installed on the management cluster (not just CRDs)
- hybrid deployments, where some RayJobs are executed via MultiKueue and some locally on the management cluster


Hints from issue comments:
@mimowo: /assign @mszadkow 

cc @mbobrovskyi @andrewsykim @mwielgus @mwysokin @mszadkow: This ticket requires to first provide RayJob and RayCluster Multikueue adapter and setup e2e tests,
as we don't have them yet.
First PR will be about adapter and multikueue e2e tests with ray-operator v1.2.2- no managedBy yet, as this requires `latest`.
Second PR will be the attempt to work with ray-operator `latest` and use of managedBy.
   @mimowo: sgtm, we can start with tests using 1.2.2 but without managedBy field and merge it as a starter. Or alternatively already use latest (main) of KubeRay just for testing purposes - we will merge once KubeRay is released.@mszadkow: Another problem that I have right now is that Kuberay clusters startup time is huge for e2e tests.
It's about 5 minutes, maybe it's due to requirement to use rayproject image, but setting anything else as image results in job to stall at initialisation.
Normally we use `sleep` image just to call anything, but it won't work with ray cluster.@mimowo: Ok, in that case we may need to have a separate CI for Ray. 

However, let me first understand what you exactly mean by "cluster startup time" - is this the installation of the KubeRay, or time to run the first RayJob? Does it also take long to run follow-up Jobs?

Also, please make sure you are rebased against the main branch, because recently we increased CPU limits for Kueue to 2000m which might be relevant here too.@andrewsykim: > Normally we use sleep image just to call anything, but it won't work with ray cluster.

It may be possible to construct a dummy RayCluster that just calls `sleep` with a lighter image, but I haven't tried this myself. Assuming the start-up time issue is due to pulling the default Ray images @mimowo: thank you @andrewsykim for the suggestion. @mszadkow can we try that? maybe you already did and hit some complications?@mszadkow: Well I have tried to load the image up-front to the test cluster (kind), but then I had space limitation issues, at least locally.
Will have to verify what happens in CI.
@andrewsykim did you mean I could build lighter version of `rayproject/ray` ?@andrewsykim: > @andrewsykim did you mean I could build lighter version of rayproject/ray ?

No, when you constuct a RayJob or RayCluster, you can pass arbitrary images for the Head and Worker pods. So you can try using much smaller images used in other tests that just need to run `sleep`. @mszadkow: I found a way to run other type of image, just with bash and calling sleep.
However it seems that kuberay overrides KUBERAY_GEN_RAY_START_CMD and never mind what I put there it's ` ray start --head  --dashboard-host=0.0.0.0  --metrics-export-port=8080  --block  --dashboard-agent-listen-port=52365`.
Then I would have to install ray to the docker...
Let's first go with `rayproject` image@mszadkow: I think, I will start early integration of v1.3.0-rc.0 within days
https://github.com/ray-project/kuberay/releases/tag/v1.3.0-rc.0
quay.io/kuberay/operator:v1.3.0-rc.0
@mszadkow: @andrewsykim @mimowo 
I have encountered an issue with kuberay v1.3.0-rc.0 upgrade.

Kuberay has [k8s.io/component-base](http://k8s.io/component-base) v0.29.6
Kueue has [k8s.io/component-base](http://k8s.io/component-base) v0.32.1

There is a small change that affect us https://github.com/kubernetes/component-base/blob/264c1fd30132a3b36b7588e50ac54eb0ff75f26a/featuregate/testing/feature_gate.go#L47
this function now returns nothing, while before it returned a cleanup function.

This result in:
vendor/github.com/ray-project/kuberay/ray-operator/pkg/features/features.go:40:9: featuregatetesting.SetFeatureGateDuringTest(tb, utilfeature.DefaultFeatureGate, f, value) (no value) used as value
make: *** [build] Error 1

Question remains if it's ok to upgrade Kuberay's component base to v0.32.1 for 1.3.0 release?@andrewsykim: No objections to upgrade component base to v0.32.1 -- do you mind opening a PR for that?@mszadkow: I am in the process of doing that, though I am afraid it won't just this package.
But let's see the results first ;) @mszadkow: @andrewsykim @mimowo 
After lengthy battle here is the PR: https://github.com/ray-project/kuberay/pull/3004@mszadkow: @andrewsykim is there any slack channel on `kubernetes.slack.com` dedicated to kuberay?
I only found dedicated workspace: https://github.com/ray-project/kuberay/issues/1059
@tenzen-y: > [@andrewsykim](https://github.com/andrewsykim) is there any slack channel on `kubernetes.slack.com` dedicated to kuberay? I only found dedicated workspace: [ray-project/kuberay#1059](https://github.com/ray-project/kuberay/issues/1059)

IIUC, Ray slack is only ray.slack.com
@mszadkow: @mimowo @tenzen-y @andrewsykim 
I have tested locally that [kuberay release-1.3 with backport](https://github.com/ray-project/kuberay/pull/3163) works with Kueue.
I even run MultiKueue e2e tests with managedBy and kuberay-operator running on manager cluster and it passes, thus I think it is safe to proceed whenever Kuberay releases next 1.3.X version.@tenzen-y: > [@mimowo](https://github.com/mimowo) [@tenzen-y](https://github.com/tenzen-y) [@andrewsykim](https://github.com/andrewsykim) I have tested locally that [kuberay release-1.3 with backport](https://github.com/ray-project/kuberay/pull/3163) works with Kueue. I even run MultiKueue e2e tests with managedBy and kuberay-operator running on manager cluster and it passes, thus I think it is safe to proceed whenever Kuberay releases next 1.3.X version.

Thank you for your effort!
@mimowo: /reopen
this is not done yet, only KubeRay is bumped so far. which is the prerequisite@k8s-ci-robot: @mimowo: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/3822#issuecomment-2733406446):

>/reopen
>this is not done yet, only KubeRay is bumped so far. which is the prerequisite


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@mszadkow: This https://github.com/kubernetes-sigs/kueue/pull/4677 should conclude the task.@mimowo: We ahould also follow up with docs update. but this can be post release@mimowo: /release-note-edit
```release-note
Support KubeRay (RayJob and RayCluster integrations) in MultiKueue via managedBy. 
This allows using the integrations and installing the KubeRay operator on the management cluster.
```
@mszadkow: @mimowo Docs have beed updated - https://github.com/kubernetes-sigs/kueue/pull/4686.
Are there any other places to be updated or we could close this one?@mimowo: /close@k8s-ci-robot: @mimowo: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/3822#issuecomment-2809330402):

>/close


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

Solution patch:
diff --git a/Makefile-test.mk b/Makefile-test.mk
index 7fbfd60ff..71c145927 100644
--- a/Makefile-test.mk
+++ b/Makefile-test.mk
@@ -83,10 +83,7 @@ test: gotestsum ## Run tests.
 	$(GOTESTSUM) --junitfile $(ARTIFACTS)/junit.xml -- $(GOFLAGS) $(GO_TEST_FLAGS) $(shell $(GO_CMD) list ./... | grep -v '/test/') -coverpkg=./... -coverprofile $(ARTIFACTS)/cover.out
 
 .PHONY: test-integration
-test-integration: test-singlecluster-integration test-multikueue-integration
-
-.PHONY: test-singlecluster-integration
-test-singlecluster-integration: gomod-download envtest ginkgo dep-crds kueuectl ginkgo-top ## Run integration tests for all singlecluster suites.
+test-integration: gomod-download envtest ginkgo dep-crds kueuectl ginkgo-top ## Run integration tests for all singlecluster suites.
 	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) -p path)" \
 	PROJECT_DIR=$(PROJECT_DIR)/ \
 	KUEUE_BIN=$(PROJECT_DIR)/bin \
diff --git a/charts/kueue/templates/rbac/role.yaml b/charts/kueue/templates/rbac/role.yaml
index fc8b59ce6..f9efc3dbd 100644
--- a/charts/kueue/templates/rbac/role.yaml
+++ b/charts/kueue/templates/rbac/role.yaml
@@ -304,7 +304,6 @@ rules:
       - ray.io
     resources:
       - rayclusters/finalizers
-      - rayclusters/status
       - rayjobs/finalizers
     verbs:
       - get
@@ -312,6 +311,7 @@ rules:
   - apiGroups:
       - ray.io
     resources:
+      - rayclusters/status
       - rayjobs/status
     verbs:
       - get
diff --git a/config/components/rbac/role.yaml b/config/components/rbac/role.yaml
index 15483e27d..5f15d3d52 100644
--- a/config/components/rbac/role.yaml
+++ b/config/components/rbac/role.yaml
@@ -303,7 +303,6 @@ rules:
   - ray.io
   resources:
   - rayclusters/finalizers
-  - rayclusters/status
   - rayjobs/finalizers
   verbs:
   - get
@@ -311,6 +310,7 @@ rules:
 - apiGroups:
   - ray.io
   resources:
+  - rayclusters/status
   - rayjobs/status
   verbs:
   - get
diff --git a/pkg/controller/jobframework/validation.go b/pkg/controller/jobframework/validation.go
index c6b237074..bae1b1701 100644
--- a/pkg/controller/jobframework/validation.go
+++ b/pkg/controller/jobframework/validation.go
@@ -52,7 +52,8 @@ var (
 		kftraining.SchemeGroupVersion.WithKind(kftraining.XGBoostJobKind).String(),
 		kfmpi.SchemeGroupVersion.WithKind(kfmpi.Kind).String(),
 		rayv1.SchemeGroupVersion.WithKind("RayJob").String(),
-		corev1.SchemeGroupVersion.WithKind("Pod").String())
+		corev1.SchemeGroupVersion.WithKind("Pod").String(),
+		rayv1.SchemeGroupVersion.WithKind("RayCluster").String())
 )
 
 // ValidateJobOnCreate encapsulates all GenericJob validations that must be performed on a Create operation
diff --git a/pkg/controller/jobs/raycluster/raycluster_controller.go b/pkg/controller/jobs/raycluster/raycluster_controller.go
index c17edeb73..690dba337 100644
--- a/pkg/controller/jobs/raycluster/raycluster_controller.go
+++ b/pkg/controller/jobs/raycluster/raycluster_controller.go
@@ -51,12 +51,13 @@ func init() {
 		JobType:                &rayv1.RayCluster{},
 		AddToScheme:            rayv1.AddToScheme,
 		IsManagingObjectsOwner: isRayCluster,
+		MultiKueueAdapter:      &multikueueAdapter{},
 	}))
 }
 
 // +kubebuilder:rbac:groups="",resources=events,verbs=create;watch;update
 // +kubebuilder:rbac:groups=ray.io,resources=rayclusters,verbs=get;list;watch;update;patch
-// +kubebuilder:rbac:groups=ray.io,resources=rayclusters/status,verbs=get;update
+// +kubebuilder:rbac:groups=ray.io,resources=rayclusters/status,verbs=get;patch;update
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=workloads,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=workloads/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=workloads/finalizers,verbs=update
diff --git a/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter.go b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter.go
new file mode 100644
index 000000000..0841ee248
--- /dev/null
+++ b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter.go
@@ -0,0 +1,124 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package raycluster
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	"sigs.k8s.io/kueue/pkg/util/api"
+	clientutil "sigs.k8s.io/kueue/pkg/util/client"
+)
+
+type multikueueAdapter struct{}
+
+var _ jobframework.MultiKueueAdapter = (*multikueueAdapter)(nil)
+
+func (b *multikueueAdapter) SyncJob(ctx context.Context, localClient client.Client, remoteClient client.Client, key types.NamespacedName, workloadName, origin string) error {
+	log := ctrl.LoggerFrom(ctx)
+
+	localJob := rayv1.RayCluster{}
+	err := localClient.Get(ctx, key, &localJob)
+	if err != nil {
+		return err
+	}
+
+	remoteJob := rayv1.RayCluster{}
+	err = remoteClient.Get(ctx, key, &remoteJob)
+	if client.IgnoreNotFound(err) != nil {
+		return err
+	}
+
+	// if the remote exists, just copy the status
+	if err == nil {
+		if fromObject(&localJob).IsSuspended() {
+			// Ensure the job is unsuspended before updating its status; otherwise, it will fail when patching the spec.
+			log.V(2).Info("Skipping the sync since the local job is still suspended")
+			return nil
+		}
+
+		return clientutil.PatchStatus(ctx, localClient, &localJob, func() (bool, error) {
+			localJob.Status = remoteJob.Status
+			return true, nil
+		})
+	}
+
+	remoteJob = rayv1.RayCluster{
+		ObjectMeta: api.CloneObjectMetaForCreation(&localJob.ObjectMeta),
+		Spec:       *localJob.Spec.DeepCopy(),
+	}
+
+	// add the prebuilt workload
+	if remoteJob.Labels == nil {
+		remoteJob.Labels = make(map[string]string, 2)
+	}
+	remoteJob.Labels[constants.PrebuiltWorkloadLabel] = workloadName
+	remoteJob.Labels[kueue.MultiKueueOriginLabel] = origin
+
+	return remoteClient.Create(ctx, &remoteJob)
+}
+
+func (b *multikueueAdapter) DeleteRemoteObject(ctx context.Context, remoteClient client.Client, key types.NamespacedName) error {
+	job := rayv1.RayCluster{}
+	job.SetName(key.Name)
+	job.SetNamespace(key.Namespace)
+	return client.IgnoreNotFound(remoteClient.Delete(ctx, &job))
+}
+
+func (b *multikueueAdapter) KeepAdmissionCheckPending() bool {
+	return false
+}
+
+func (b *multikueueAdapter) IsJobManagedByKueue(ctx context.Context, c client.Client, key types.NamespacedName) (bool, string, error) {
+	return true, "", nil
+}
+
+func (b *multikueueAdapter) GVK() schema.GroupVersionKind {
+	return gvk
+}
+
+var _ jobframework.MultiKueueWatcher = (*multikueueAdapter)(nil)
+
+func (*multikueueAdapter) GetEmptyList() client.ObjectList {
+	return &rayv1.RayClusterList{}
+}
+
+func (*multikueueAdapter) WorkloadKeyFor(o runtime.Object) (types.NamespacedName, error) {
+	job, isJob := o.(*rayv1.RayCluster)
+	if !isJob {
+		return types.NamespacedName{}, errors.New("not a raycluster")
+	}
+
+	prebuiltWl, hasPrebuiltWorkload := job.Labels[constants.PrebuiltWorkloadLabel]
+	if !hasPrebuiltWorkload {
+		return types.NamespacedName{}, fmt.Errorf("no prebuilt workload found for raycluster: %s", klog.KObj(job))
+	}
+
+	return types.NamespacedName{Name: prebuiltWl, Namespace: job.Namespace}, nil
+}
diff --git a/site/content/en/docs/overview/_index.md b/site/content/en/docs/overview/_index.md
index 55e537733..d374e5722 100644
--- a/site/content/en/docs/overview/_index.md
+++ b/site/content/en/docs/overview/_index.md
@@ -41,7 +41,7 @@ A core design principle for Kueue is to avoid duplicating mature functionality i
 | Feature                                                                                                         | Batch&nbsp;Job | JobSet | MXJob | PaddleJob | PytorchJob | TFJob | XGBoostJob | MPIJob | Pod | RayCluster | RayJob | AppWrapper |
 |-----------------------------------------------------------------------------------------------------------------|----------------|--------|-------|-----------|------------|-------|------------|--------|-----|------------|--------|------------|
 | [Dynamic Reclaim](/docs/concepts/workload/#dynamic-reclaim)                                                     | +              | +      |       |           |            |       |            |        | +   |            |        |            |
-| [MultiKueue](/docs/concepts/multikueue/)                                                                        | +              | +      |       | +         | +          | +     | +          | +      |     |            |        |            |
+| [MultiKueue](/docs/concepts/multikueue/)                                                                        | +              | +      |       | +         | +          | +     | +          | +      |     | +          | +      |            |
 | [MultiKueueBatchJobWithManagedBy](/docs/concepts/multikueue/#multikueuebatchjobwithmanagedby-enabled)           | +              |        |       |           |            |       |            |        |     |            |        |            |
 | [PartialAdmission](/docs/tasks/run/jobs/#partial-admission)                                                     | +              |        |       |           |            |       |            |        |     |            |        |            |
 | [Workload Priority Class](/docs/concepts/workload_priority_class/)                                              | +              | +      | +     | +         | +          | +     | +          | +      | +   | +          | +      | +          |



Test patch:
diff --git a/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter_test.go b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter_test.go
new file mode 100644
index 000000000..088e53524
--- /dev/null
+++ b/pkg/controller/jobs/raycluster/raycluster_multikueue_adapter_test.go
@@ -0,0 +1,159 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package raycluster
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/util/slices"
+	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	utiltestingraycluster "sigs.k8s.io/kueue/pkg/util/testingjobs/raycluster"
+)
+
+const (
+	TestNamespace = "ns"
+)
+
+func TestMultikueueAdapter(t *testing.T) {
+	objCheckOpts := []cmp.Option{
+		cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
+		cmpopts.EquateEmpty(),
+	}
+
+	rayClusterBuilder := utiltestingraycluster.MakeCluster("raycluster1", TestNamespace).Suspend(false)
+
+	cases := map[string]struct {
+		managersRayClusters []rayv1.RayCluster
+		workerRayClusters   []rayv1.RayCluster
+
+		operation func(ctx context.Context, adapter *multikueueAdapter, managerClient, workerClient client.Client) error
+
+		wantError               error
+		wantManagersRayClusters []rayv1.RayCluster
+		wantWorkerRayClusters   []rayv1.RayCluster
+	}{
+		"sync creates missing remote raycluster": {
+			managersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.DeepCopy(),
+			},
+			operation: func(ctx context.Context, adapter *multikueueAdapter, managerClient, workerClient client.Client) error {
+				return adapter.SyncJob(ctx, managerClient, workerClient, types.NamespacedName{Name: "raycluster1", Namespace: TestNamespace}, "wl1", "origin1")
+			},
+
+			wantManagersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.DeepCopy(),
+			},
+			wantWorkerRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					Obj(),
+			},
+		},
+		"sync status from remote raycluster": {
+			managersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.DeepCopy(),
+			},
+			workerRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					StatusConditions(metav1.Condition{Type: string(rayv1.HeadPodReady), Status: metav1.ConditionStatus(corev1.ConditionTrue)}).
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multikueueAdapter, managerClient, workerClient client.Client) error {
+				return adapter.SyncJob(ctx, managerClient, workerClient, types.NamespacedName{Name: "raycluster1", Namespace: TestNamespace}, "wl1", "origin1")
+			},
+
+			wantManagersRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					StatusConditions(metav1.Condition{Type: string(rayv1.HeadPodReady), Status: metav1.ConditionStatus(corev1.ConditionTrue)}).
+					Obj(),
+			},
+			wantWorkerRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					StatusConditions(metav1.Condition{Type: string(rayv1.HeadPodReady), Status: metav1.ConditionStatus(corev1.ConditionTrue)}).
+					Obj(),
+			},
+		},
+		"remote raycluster is deleted": {
+			workerRayClusters: []rayv1.RayCluster{
+				*rayClusterBuilder.Clone().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multikueueAdapter, managerClient, workerClient client.Client) error {
+				return adapter.DeleteRemoteObject(ctx, workerClient, types.NamespacedName{Name: "raycluster1", Namespace: TestNamespace})
+			},
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			managerBuilder := utiltesting.NewClientBuilder(rayv1.AddToScheme).WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
+			managerBuilder = managerBuilder.WithLists(&rayv1.RayClusterList{Items: tc.managersRayClusters})
+			managerBuilder = managerBuilder.WithStatusSubresource(slices.Map(tc.managersRayClusters, func(w *rayv1.RayCluster) client.Object { return w })...)
+			managerClient := managerBuilder.Build()
+
+			workerBuilder := utiltesting.NewClientBuilder(rayv1.AddToScheme).WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
+			workerBuilder = workerBuilder.WithLists(&rayv1.RayClusterList{Items: tc.workerRayClusters})
+			workerClient := workerBuilder.Build()
+
+			ctx, _ := utiltesting.ContextWithLog(t)
+
+			adapter := &multikueueAdapter{}
+
+			gotErr := tc.operation(ctx, adapter, managerClient, workerClient)
+
+			if diff := cmp.Diff(tc.wantError, gotErr, cmpopts.EquateErrors()); diff != "" {
+				t.Errorf("unexpected error (-want/+got):\n%s", diff)
+			}
+
+			gotManagersRayClusters := &rayv1.RayClusterList{}
+			if err := managerClient.List(ctx, gotManagersRayClusters); err != nil {
+				t.Errorf("unexpected list manager's rayclusters error %s", err)
+			} else {
+				if diff := cmp.Diff(tc.wantManagersRayClusters, gotManagersRayClusters.Items, objCheckOpts...); diff != "" {
+					t.Errorf("unexpected manager's rayclusters (-want/+got):\n%s", diff)
+				}
+			}
+
+			gotWorkerRayClusters := &rayv1.RayClusterList{}
+			if err := workerClient.List(ctx, gotWorkerRayClusters); err != nil {
+				t.Errorf("unexpected list worker's rayclusters error %s", err)
+			} else {
+				if diff := cmp.Diff(tc.wantWorkerRayClusters, gotWorkerRayClusters.Items, objCheckOpts...); diff != "" {
+					t.Errorf("unexpected worker's rayclusters (-want/+got):\n%s", diff)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/util/testingjobs/raycluster/wrappers.go b/pkg/util/testingjobs/raycluster/wrappers.go
index db05f1e1b..86992f461 100644
--- a/pkg/util/testingjobs/raycluster/wrappers.go
+++ b/pkg/util/testingjobs/raycluster/wrappers.go
@@ -39,13 +39,18 @@ func MakeCluster(name, ns string) *ClusterWrapper {
 		},
 		Spec: rayv1.RayClusterSpec{
 			HeadGroupSpec: rayv1.HeadGroupSpec{
-				RayStartParams: map[string]string{"p1": "v1"},
+				RayStartParams: map[string]string{},
 				Template: corev1.PodTemplateSpec{
 					Spec: corev1.PodSpec{
 						NodeSelector: map[string]string{},
 						Containers: []corev1.Container{
 							{
-								Name: "head-container",
+								Name:    "head-container",
+								Command: []string{},
+								Resources: corev1.ResourceRequirements{
+									Requests: corev1.ResourceList{},
+									Limits:   corev1.ResourceList{},
+								},
 							},
 						},
 					},
@@ -57,12 +62,18 @@ func MakeCluster(name, ns string) *ClusterWrapper {
 					Replicas:       ptr.To[int32](1),
 					MinReplicas:    ptr.To[int32](0),
 					MaxReplicas:    ptr.To[int32](10),
-					RayStartParams: map[string]string{"p1": "v1"},
+					RayStartParams: map[string]string{},
 					Template: corev1.PodTemplateSpec{
 						Spec: corev1.PodSpec{
+							NodeSelector: map[string]string{},
 							Containers: []corev1.Container{
 								{
-									Name: "worker-container",
+									Name:    "worker-container",
+									Command: []string{},
+									Resources: corev1.ResourceRequirements{
+										Requests: corev1.ResourceList{},
+										Limits:   corev1.ResourceList{},
+									},
 								},
 							},
 						},
@@ -167,3 +178,50 @@ func (j *ClusterWrapper) WorkloadPriorityClass(wpc string) *ClusterWrapper {
 	j.Labels[constants.WorkloadPriorityClassLabel] = wpc
 	return j
 }
+
+// Label sets the label key and value
+func (j *ClusterWrapper) Label(key, value string) *ClusterWrapper {
+	if j.Labels == nil {
+		j.Labels = make(map[string]string)
+	}
+	j.Labels[key] = value
+	return j
+}
+
+// StatusConditions adds a condition
+func (j *ClusterWrapper) StatusConditions(c metav1.Condition) *ClusterWrapper {
+	j.Status.Conditions = append(j.Status.Conditions, c)
+	return j
+}
+
+// ManagedBy adds a managedby.
+func (j *ClusterWrapper) ManagedBy(_ string) *ClusterWrapper {
+	// j.Spec.ManagedBy = &c
+	return j
+}
+
+// Request adds a resource request to the default container.
+func (j *ClusterWrapper) Request(rayType rayv1.RayNodeType, r corev1.ResourceName, v string) *ClusterWrapper {
+	if rayType == rayv1.HeadNode {
+		j.Spec.HeadGroupSpec.Template.Spec.Containers[0].Resources.Requests[r] = resource.MustParse(v)
+	} else if rayType == rayv1.WorkerNode {
+		j.Spec.WorkerGroupSpecs[0].Template.Spec.Containers[0].Resources.Requests[r] = resource.MustParse(v)
+	}
+	return j
+}
+
+func (j *ClusterWrapper) Image(rayType rayv1.RayNodeType, image string, args []string) *ClusterWrapper {
+	if rayType == rayv1.HeadNode {
+		j.Spec.HeadGroupSpec.Template.Spec.Containers[0].Image = image
+		j.Spec.HeadGroupSpec.Template.Spec.Containers[0].Args = args
+	} else if rayType == rayv1.WorkerNode {
+		j.Spec.WorkerGroupSpecs[0].Template.Spec.Containers[0].Image = image
+		j.Spec.WorkerGroupSpecs[0].Template.Spec.Containers[0].Args = args
+	}
+	return j
+}
+
+func (j *ClusterWrapper) RayVersion(rv string) *ClusterWrapper {
+	j.Spec.RayVersion = rv
+	return j
+}
diff --git a/test/e2e/config/default/controller_manager_config.yaml b/test/e2e/config/default/controller_manager_config.yaml
index 1cbefc06e..29c9a9c92 100644
--- a/test/e2e/config/default/controller_manager_config.yaml
+++ b/test/e2e/config/default/controller_manager_config.yaml
@@ -20,6 +20,7 @@ integrations:
   - "batch/job"
   - "kubeflow.org/mpijob"
   - "ray.io/rayjob"
+  - "ray.io/raycluster"
   - "jobset.x-k8s.io/jobset"
   - "kubeflow.org/mxjob"
   - "kubeflow.org/paddlejob"
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index 35f0925ad..2e47294b5 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -44,6 +44,7 @@ import (
 	workloadpytorchjob "sigs.k8s.io/kueue/pkg/controller/jobs/kubeflow/jobs/pytorchjob"
 	workloadmpijob "sigs.k8s.io/kueue/pkg/controller/jobs/mpijob"
 	workloadpod "sigs.k8s.io/kueue/pkg/controller/jobs/pod"
+	workloadraycluster "sigs.k8s.io/kueue/pkg/controller/jobs/raycluster"
 	workloadrayjob "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
@@ -52,6 +53,7 @@ import (
 	testingmpijob "sigs.k8s.io/kueue/pkg/util/testingjobs/mpijob"
 	testingpod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
 	testingpytorchjob "sigs.k8s.io/kueue/pkg/util/testingjobs/pytorchjob"
+	testingraycluster "sigs.k8s.io/kueue/pkg/util/testingjobs/raycluster"
 	testingrayjob "sigs.k8s.io/kueue/pkg/util/testingjobs/rayjob"
 	"sigs.k8s.io/kueue/pkg/workload"
 	"sigs.k8s.io/kueue/test/util"
@@ -574,14 +576,7 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 		})
 
 		ginkgo.It("Should run a RayJob on worker if admitted", func() {
-			var found bool
-			E2eKuberayTestImage, found := os.LookupEnv("KUBERAY_RAY_IMAGE")
-			gomega.Expect(found).To(gomega.BeTrue())
-			if runtime.GOOS == "darwin" {
-				E2eKuberayTestImage, found = os.LookupEnv("KUBERAY_RAY_IMAGE_ARM")
-				gomega.Expect(found).To(gomega.BeTrue())
-			}
-
+			kuberayTestImage := getKuberayTestImage()
 			// Since it requires 1.5 CPU, this job can only be admitted in worker 1.
 			rayjob := testingrayjob.MakeJob("rayjob1", managerNs.Name).
 				Suspend(true).
@@ -590,8 +585,8 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 				Request(rayv1.HeadNode, corev1.ResourceCPU, "1").
 				Request(rayv1.WorkerNode, corev1.ResourceCPU, "0.5").
 				Entrypoint("python -c \"import ray; ray.init(); print(ray.cluster_resources())\"").
-				Image(rayv1.HeadNode, E2eKuberayTestImage, []string{}).
-				Image(rayv1.WorkerNode, E2eKuberayTestImage, []string{}).
+				Image(rayv1.HeadNode, kuberayTestImage, []string{}).
+				Image(rayv1.WorkerNode, kuberayTestImage, []string{}).
 				Obj()
 
 			ginkgo.By("Creating the RayJob", func() {
@@ -623,6 +618,57 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 		})
+
+		ginkgo.It("Should run a RayCluster on worker if admitted", func() {
+			kuberayTestImage := getKuberayTestImage()
+			// Since it requires 1.5 CPU, this job can only be admitted in worker 1.
+			raycluster := testingraycluster.MakeCluster("raycluster1", managerNs.Name).
+				Suspend(true).
+				Queue(managerLq.Name).
+				Request(rayv1.HeadNode, corev1.ResourceCPU, "1").
+				Request(rayv1.WorkerNode, corev1.ResourceCPU, "0.5").
+				Image(rayv1.HeadNode, kuberayTestImage, []string{}).
+				Image(rayv1.WorkerNode, kuberayTestImage, []string{}).
+				Obj()
+
+			ginkgo.By("Creating the RayCluster", func() {
+				gomega.Expect(k8sManagerClient.Create(ctx, raycluster)).Should(gomega.Succeed())
+			})
+
+			wlLookupKey := types.NamespacedName{Name: workloadraycluster.GetWorkloadNameForRayCluster(raycluster.Name, raycluster.UID), Namespace: managerNs.Name}
+			// the execution should be given to the worker1
+			waitForJobAdmitted(wlLookupKey, multiKueueAc.Name, "worker1")
+
+			ginkgo.By("Checking the RayCluster is ready", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdRayCluster := &rayv1.RayCluster{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(raycluster), createdRayCluster)).To(gomega.Succeed())
+					g.Expect(createdRayCluster.Status.DesiredWorkerReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdRayCluster.Status.ReadyWorkerReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdRayCluster.Status.AvailableWorkerReplicas).To(gomega.Equal(int32(1)))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Setting the RayCluster to suspended", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					updatedCluster := &rayv1.RayCluster{}
+					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(raycluster), updatedCluster)).To(gomega.Succeed())
+					updatedCluster.Spec.Suspend = ptr.To(true)
+					g.Expect(k8sWorker1Client.Update(ctx, updatedCluster)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking the RayCluster got suspended", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdRayCluster := &rayv1.RayCluster{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(raycluster), createdRayCluster)).To(gomega.Succeed())
+					// Suspended RayCluster manifests with removed pods
+					g.Expect(createdRayCluster.Status.DesiredWorkerReplicas).To(gomega.Equal(int32(1)))
+					g.Expect(createdRayCluster.Status.ReadyWorkerReplicas).To(gomega.Equal(int32(0)))
+					g.Expect(createdRayCluster.Status.AvailableWorkerReplicas).To(gomega.Equal(int32(0)))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
 	})
 	ginkgo.When("The connection to a worker cluster is unreliable", func() {
 		ginkgo.It("Should update the cluster status to reflect the connection state", func() {
@@ -731,3 +777,17 @@ func checkFinishStatusCondition(g gomega.Gomega, wlLookupKey types.NamespacedNam
 		Message: finishReasonMessage,
 	}, util.IgnoreConditionTimestampsAndObservedGeneration))
 }
+
+func getKuberayTestImage() string {
+	var (
+		kuberayTestImage string
+		found            bool
+	)
+	if runtime.GOOS == "darwin" {
+		kuberayTestImage, found = os.LookupEnv("KUBERAY_RAY_IMAGE_ARM")
+	} else {
+		kuberayTestImage, found = os.LookupEnv("KUBERAY_RAY_IMAGE")
+	}
+	gomega.Expect(found).To(gomega.BeTrue())
+	return kuberayTestImage
+}
diff --git a/test/e2e/multikueue/suite_test.go b/test/e2e/multikueue/suite_test.go
index c5cc4e777..0bb818e69 100644
--- a/test/e2e/multikueue/suite_test.go
+++ b/test/e2e/multikueue/suite_test.go
@@ -103,6 +103,8 @@ func kubeconfigForMultiKueueSA(ctx context.Context, c client.Client, restConfig
 			policyRule(rayv1.SchemeGroupVersion.Group, "rayjobs/status", "get"),
 			policyRule(corev1.SchemeGroupVersion.Group, "pods", resourceVerbs...),
 			policyRule(corev1.SchemeGroupVersion.Group, "pods/status", "get"),
+			policyRule(rayv1.SchemeGroupVersion.Group, "rayclusters", resourceVerbs...),
+			policyRule(rayv1.SchemeGroupVersion.Group, "rayclusters/status", "get"),
 		},
 	}
 	err := c.Create(ctx, cr)