adjust e2e test so that RayCluster really is managed by Kueue

Author: dgrove-oss

.github/workflows/e2e_tests.yaml

@@ -55,10 +55,6 @@ jobs:
       - name: Setup and start KinD cluster
         uses: ./common/github-actions/kind
 
-      - name: Deploy Kueue
-        run: |
-          make kueue-e2e
-
       - name: Deploy CodeFlare stack
         id: deploy
         run: |

Makefile

@@ -389,13 +389,9 @@ test-e2e: manifests fmt vet ## Run e2e tests.
 kind-e2e: ## Set up e2e KinD cluster
 	test/e2e/kind.sh
 
-.PHONY: kueue-e2e
-kueue-e2e: ## Deploy Kueue for e2e tests
-	KUEUE_VERSION=$(KUEUE_VERSION) test/e2e/kueue.sh
-
 .PHONY: setup-e2e
 setup-e2e: ## Set up e2e tests.
-	KUBERAY_VERSION=$(KUBERAY_VERSION) test/e2e/setup.sh
+	KUBERAY_VERSION=$(KUBERAY_VERSION) KUEUE_VERSION=$(KUEUE_VERSION) test/e2e/setup.sh
 
 .PHONY: imports
 imports: openshift-goimports ## Organize imports in go files using openshift-goimports. Example: make imports

README.md

@@ -33,8 +33,6 @@ The e2e tests can be executed locally by running the following commands:
     ```bash
     # Create a KinD cluster
     make kind-e2e
-    # Deploy Kueue
-    make kueue-e2e
     # Install the CRDs
     make install
     ```
@@ -43,20 +41,15 @@ The e2e tests can be executed locally by running the following commands:
    Some e2e tests cover the access to services via Ingresses, as end-users would do, which requires access to the Ingress controller load balancer by its IP.
    For it to work on macOS, this requires installing [docker-mac-net-connect](https://github.com/chipmk/docker-mac-net-connect).
 
-2. Start the operator locally:
-
-    ```bash
-    NAMESPACE=default make run
-    ```
-
-   Alternatively, You can run the operator from your IDE / debugger.
-
-3. Set up the test CodeFlare stack:
+2. Setup the rest of the CodeFlare stack.
 
    ```bash
    make setup-e2e
    ```
 
+   [!NOTE]
+   Kueue will only activate its Ray integration if KubeRay is installed before Kueue (as done by this make target).
+
    [!NOTE]
    In OpenShift the KubeRay operator pod gets random user assigned. This user is then used to run Ray cluster.
    However the random user assigned by OpenShift doesn't have rights to store dataset downloaded as part of test execution, causing tests to fail.
@@ -76,6 +69,14 @@ The e2e tests can be executed locally by running the following commands:
       - 'system:serviceaccount:$(namespace):kuberay-operator'
     ```
 
+3. Start the operator locally:
+
+    ```bash
+    NAMESPACE=default make run
+    ```
+
+   Alternatively, You can run the operator from your IDE / debugger.
+
 4.  In a separate terminal, set your output directory for test files, and run the e2e suite:
     ```bash
     export CODEFLARE_TEST_OUTPUT_DIR=<your_output_directory>

test/e2e/kueue.sh

@@ -35,13 +35,57 @@ import (
 // directly managed by Kueue, and asserts successful completion of the training job.
 func TestMNISTRayJobRayCluster(t *testing.T) {
 	test := With(t)
-	test.T().Parallel()
 
 	// Create a namespace and localqueue in that namespace
 	namespace := test.NewTestNamespace()
+	localQueue := CreateKueueLocalQueue(test, namespace.Name, "e2e-cluster-queue")
 
-	// MNIST training script
-	mnist := &corev1.ConfigMap{
+	// Create MNIST training script
+	mnist := constructMNISTConfigMap(test, namespace)
+	mnist, err := test.Client().Core().CoreV1().ConfigMaps(namespace.Name).Create(test.Ctx(), mnist, metav1.CreateOptions{})
+	test.Expect(err).NotTo(HaveOccurred())
+	test.T().Logf("Created ConfigMap %s/%s successfully", mnist.Namespace, mnist.Name)
+
+	// Create RayCluster and assign it to the localqueue
+	rayCluster := constructRayCluster(test, namespace, mnist)
+	AssignToLocalQueue(rayCluster, localQueue)
+	rayCluster, err = test.Client().Ray().RayV1().RayClusters(namespace.Name).Create(test.Ctx(), rayCluster, metav1.CreateOptions{})
+	test.Expect(err).NotTo(HaveOccurred())
+	test.T().Logf("Created RayCluster %s/%s successfully", rayCluster.Namespace, rayCluster.Name)
+
+	test.T().Logf("Waiting for RayCluster %s/%s to be running", rayCluster.Namespace, rayCluster.Name)
+	test.Eventually(RayCluster(test, namespace.Name, rayCluster.Name), TestTimeoutMedium).
+		Should(WithTransform(RayClusterState, Equal(rayv1.Ready)))
+
+	// Create RayJob
+	rayJob := constructRayJob(test, namespace, rayCluster)
+	rayJob, err = test.Client().Ray().RayV1().RayJobs(namespace.Name).Create(test.Ctx(), rayJob, metav1.CreateOptions{})
+	test.Expect(err).NotTo(HaveOccurred())
+	test.T().Logf("Created RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)
+
+	rayDashboardURL := getRayDashboardURL(test, rayCluster.Namespace, rayCluster.Name)
+
+	test.T().Logf("Connecting to Ray cluster at: %s", rayDashboardURL.String())
+	rayClient := NewRayClusterClient(rayDashboardURL)
+
+	// Wait for Ray job id to be available, this value is needed for writing logs in defer
+	test.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutShort).
+		Should(WithTransform(RayJobId, Not(BeEmpty())))
+
+	// Retrieving the job logs once it has completed or timed out
+	defer WriteRayJobAPILogs(test, rayClient, GetRayJobId(test, rayJob.Namespace, rayJob.Name))
+
+	test.T().Logf("Waiting for RayJob %s/%s to complete", rayJob.Namespace, rayJob.Name)
+	test.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutLong).
+		Should(WithTransform(RayJobStatus, Satisfy(rayv1.IsJobTerminal)))
+
+	// Assert the Ray job has completed successfully
+	test.Expect(GetRayJob(test, rayJob.Namespace, rayJob.Name)).
+		To(WithTransform(RayJobStatus, Equal(rayv1.JobStatusSucceeded)))
+}
+
+func constructMNISTConfigMap(test Test, namespace *corev1.Namespace) *corev1.ConfigMap {
+	return &corev1.ConfigMap{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: corev1.SchemeGroupVersion.String(),
 			Kind:       "ConfigMap",
@@ -55,12 +99,10 @@ func TestMNISTRayJobRayCluster(t *testing.T) {
 		},
 		Immutable: Ptr(true),
 	}
-	mnist, err := test.Client().Core().CoreV1().ConfigMaps(namespace.Name).Create(test.Ctx(), mnist, metav1.CreateOptions{})
-	test.Expect(err).NotTo(HaveOccurred())
-	test.T().Logf("Created ConfigMap %s/%s successfully", mnist.Namespace, mnist.Name)
+}
 
-	// RayCluster
-	rayCluster := &rayv1.RayCluster{
+func constructRayCluster(_ Test, namespace *corev1.Namespace, mnist *corev1.ConfigMap) *rayv1.RayCluster {
+	return &rayv1.RayCluster{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: rayv1.GroupVersion.String(),
 			Kind:       "RayCluster",
@@ -173,16 +215,10 @@ func TestMNISTRayJobRayCluster(t *testing.T) {
 			},
 		},
 	}
+}
 
-	_, err = test.Client().Ray().RayV1().RayClusters(namespace.Name).Create(test.Ctx(), rayCluster, metav1.CreateOptions{})
-	test.Expect(err).NotTo(HaveOccurred())
-	test.T().Logf("Created RayCluster %s/%s successfully", rayCluster.Namespace, rayCluster.Name)
-
-	test.T().Logf("Waiting for RayCluster %s/%s to be running", rayCluster.Namespace, rayCluster.Name)
-	test.Eventually(RayCluster(test, namespace.Name, rayCluster.Name), TestTimeoutMedium).
-		Should(WithTransform(RayClusterState, Equal(rayv1.Ready)))
-
-	rayJob := &rayv1.RayJob{
+func constructRayJob(_ Test, namespace *corev1.Namespace, rayCluster *rayv1.RayCluster) *rayv1.RayJob {
+	return &rayv1.RayJob{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: rayv1.GroupVersion.String(),
 			Kind:       "RayJob",
@@ -220,29 +256,6 @@ func TestMNISTRayJobRayCluster(t *testing.T) {
 			},
 		},
 	}
-	rayJob, err = test.Client().Ray().RayV1().RayJobs(namespace.Name).Create(test.Ctx(), rayJob, metav1.CreateOptions{})
-	test.Expect(err).NotTo(HaveOccurred())
-	test.T().Logf("Created RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)
-
-	rayDashboardURL := getRayDashboardURL(test, rayCluster.Namespace, rayCluster.Name)
-
-	test.T().Logf("Connecting to Ray cluster at: %s", rayDashboardURL.String())
-	rayClient := NewRayClusterClient(rayDashboardURL)
-
-	// Wait for Ray job id to be available, this value is needed for writing logs in defer
-	test.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutShort).
-		Should(WithTransform(RayJobId, Not(BeEmpty())))
-
-	// Retrieving the job logs once it has completed or timed out
-	defer WriteRayJobAPILogs(test, rayClient, GetRayJobId(test, rayJob.Namespace, rayJob.Name))
-
-	test.T().Logf("Waiting for RayJob %s/%s to complete", rayJob.Namespace, rayJob.Name)
-	test.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutLong).
-		Should(WithTransform(RayJobStatus, Satisfy(rayv1.IsJobTerminal)))
-
-	// Assert the Ray job has completed successfully
-	test.Expect(GetRayJob(test, rayJob.Namespace, rayJob.Name)).
-		To(WithTransform(RayJobStatus, Equal(rayv1.JobStatusSucceeded)))
 }
 
 func getRayDashboardURL(test Test, namespace, rayClusterName string) url.URL {

test/e2e/mnist_rayjob_raycluster_test.go

@@ -57,6 +57,17 @@ roleRef:
   name: e2e-controller-rayclusters
 EOF
 
+echo "Deploying Kueue $KUEUE_VERSION"
+kubectl apply --server-side -f https://github.com/kubernetes-sigs/kueue/releases/download/${KUEUE_VERSION}/manifests.yaml
+
+# Sleep until the kueue manager is running
+echo "Waiting for pods in the kueue-system namespace to become ready"
+while [[ $(kubectl get pods -n kueue-system -o 'jsonpath={..status.conditions[?(@.type=="Ready")].status}' | tr ' ' '\n' | sort -u) != "True" ]]
+do
+    echo -n "." && sleep 1;
+done
+echo ""
+
 echo Creating Kueue ResourceFlavor and ClusterQueue
 cat <<EOF | kubectl apply -f -
 apiVersion: kueue.x-k8s.io/v1beta1

test/e2e/setup.sh

@@ -21,6 +21,8 @@ import (
 
 	"github.com/onsi/gomega"
 	"github.com/project-codeflare/codeflare-common/support"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	kueuev1beta1 "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 )
 
 //go:embed *.py *.txt *.sh
@@ -32,3 +34,12 @@ func ReadFile(t support.Test, fileName string) []byte {
 	t.Expect(err).NotTo(gomega.HaveOccurred())
 	return file
 }
+
+func AssignToLocalQueue(object client.Object, localqueue *kueuev1beta1.LocalQueue) {
+	labels := object.GetLabels()
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	labels["kueue.x-k8s.io/queue-name"] = localqueue.Name
+	object.SetLabels(labels)
+}