Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Test][RayJob] Transition to Complete if the JobStatus is STOPPED #1871

Merged
merged 2 commits into from
Jan 26, 2024
Merged

[Test][RayJob] Transition to Complete if the JobStatus is STOPPED #1871

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7d2e0fb
add tests
kevin85421 Jan 26, 2024
d47dbc4
add tests
kevin85421 Jan 26, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 41 additions & 1 deletion ray-operator/test/e2e/rayjob_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,10 @@ import (

k8serrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
corev1ac "k8s.io/client-go/applyconfigurations/core/v1"

rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
rayv1ac "github.com/ray-project/kuberay/ray-operator/pkg/client/applyconfiguration/ray/v1"
. "github.com/ray-project/kuberay/ray-operator/test/support"
)

Expand All @@ -20,7 +22,7 @@ func TestRayJob(t *testing.T) {
test.StreamKubeRayOperatorLogs()

// Job scripts
jobs := newConfigMap(namespace.Name, "jobs", files(test, "counter.py", "fail.py"))
jobs := newConfigMap(namespace.Name, "jobs", files(test, "counter.py", "fail.py", "stop.py"))
jobs, err := test.Client().Core().CoreV1().ConfigMaps(namespace.Name).Create(test.Ctx(), jobs, metav1.CreateOptions{})
test.Expect(err).NotTo(HaveOccurred())
test.T().Logf("Created ConfigMap %s/%s successfully", jobs.Namespace, jobs.Name)
Expand Down Expand Up @@ -190,4 +192,42 @@ env_vars:
test.Expect(err).NotTo(HaveOccurred())
test.T().Logf("Deleted RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)
})

test.T().Run("Should transition to 'Complete' if the Ray job has stopped.", func(t *testing.T) {
// `stop.py` will sleep for 20 seconds so that the RayJob has enough time to transition to `RUNNING`
// and then stop the Ray job. If the Ray job is stopped, the RayJob should transition to `Complete`.
rayJobAC := rayv1ac.RayJob("stop", namespace.Name).
WithSpec(rayv1ac.RayJobSpec().
WithEntrypoint("python /home/ray/jobs/stop.py").
WithSubmitterPodTemplate(jobSubmitterPodTemplateApplyConfiguration()).
WithRayClusterSpec(rayv1ac.RayClusterSpec().
WithRayVersion(GetRayVersion()).
WithHeadGroupSpec(rayv1ac.HeadGroupSpec().
WithRayStartParams(map[string]string{
"dashboard-host": "0.0.0.0",
}).
WithTemplate(podTemplateSpecApplyConfiguration(headPodTemplateApplyConfiguration(),
mountConfigMap[corev1ac.PodTemplateSpecApplyConfiguration](jobs, "/home/ray/jobs"))))))

rayJob, err := test.Client().Ray().RayV1().RayJobs(namespace.Name).Apply(test.Ctx(), rayJobAC, TestApplyOptions)
test.Expect(err).NotTo(HaveOccurred())
test.T().Logf("Created RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)

test.T().Logf("Waiting for RayJob %s/%s to be 'Running'", rayJob.Namespace, rayJob.Name)
test.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutMedium).
Should(WithTransform(RayJobDeploymentStatus, Equal(rayv1.JobDeploymentStatusRunning)))

test.T().Logf("Waiting for RayJob %s/%s to be 'Complete'", rayJob.Namespace, rayJob.Name)
test.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutMedium).
Should(WithTransform(RayJobDeploymentStatus, Equal(rayv1.JobDeploymentStatusComplete)))

// Refresh the RayJob status
rayJob = GetRayJob(test, rayJob.Namespace, rayJob.Name)
test.Expect(rayJob.Status.JobStatus).To(Equal(rayv1.JobStatusStopped))

// Delete the RayJob
err = test.Client().Ray().RayV1().RayJobs(namespace.Name).Delete(test.Ctx(), rayJob.Name, metav1.DeleteOptions{})
test.Expect(err).NotTo(HaveOccurred())
test.T().Logf("Deleted RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)
})
}
12 changes: 12 additions & 0 deletions ray-operator/test/e2e/stop.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
import time
from ray.job_submission import JobSubmissionClient

print("Sleep 20 seconds to enable the RayJob to transition to RUNNING state")
time.sleep(20)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we pass in submission ID into stop.py and poll the job status until it's RUNNING?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree. Or could we even delete the sleep entirely? Isn't it be guaranteed to be RUNNING at the time this Python code is being executed?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

RUNNING here refers to JobDeploymentStatus rather than JobStatus.

Isn't it be guaranteed to be RUNNING at the time this Python code is being executed?

JobStatus guaranteed to be RUNNING at the time this Python code is being executed. That's why I said "RayJob transition to RUNNING state" instead of "Ray job".

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, thanks for clarifying. Sleeping in a test isn't ideal but I don't know a better way right now

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, we don't have a method to know the JobDeploymentStatus without implementing some RBAC configurations.


client = JobSubmissionClient("http://127.0.0.1:8265")
job_details = client.list_jobs()
print("Number of Jobs: " + str(len(job_details)))
for job_detail in job_details:
print("Stop Job: " + job_detail.submission_id)
client.stop_job(job_detail.submission_id)
Loading