[feat][kubectl-plugin] Implement kubectl ray job submit with Deletion Policy API for RayJob Cleanup #3064

win5923 · 2025-02-14T13:14:35Z

Why are these changes needed?

Adds the shutdown-after-job-finishes, deletion-policy and ttl-seconds-after-finished flag to the kubectl ray job submit to allow users to manage RayJob CRs cleanup more easily by configuring deletion policies.

This behavior relies on ray-operator enable RayJobDeletionPolicy feature gate.

If the user has not enabled the RayJobDeletionPolicy

$ kubectl ray job submit --name ray-job-sample --deletion-policy DeleteCluster --working-dir ~/workdir --runtime-env ~/workdir/runtimeEnv.yaml -- python sample_code.py
Submitted RayJob ray-job-sample.
Deleting RayJob...
Cleaned Up RayJob: ray-job-sample
2025/02/16 15:06:14 The RayJob spec is invalid default/ray-job-sample: RayJobDeletionPolicy feature gate must be enabled to use the DeletionPolicy feature

Set ttl-seconds-after-finished but not set shutdown-after-job-finishes

$ kubectl ray job submit --name ray-job-sample --ttl-seconds-after-finished 10  --working-dir ~/workdir --runtime-env ~/workdir/runtimeEnv.yaml  -- python sample_code.py
Error: TTLSecondsAfterFinished only works when shutdown-after-job-finishes is set to true

Set ttl-seconds-after-finished and shutdown-after-job-finishes

$ kubectl ray job submit --name ray-job-sample --shutdown-after-job-finishes --ttl-seconds-after-finished 10  --working-dir ~/workdir --runtime-env ~/workdir/runtimeEnv.yaml -- python sample_code.py
Submitted RayJob ray-job-sample.

$ k get rayjob
NAME             JOB STATUS   DEPLOYMENT STATUS   RAY CLUSTER NAME                  START TIME             END TIME               AGE
ray-job-sample   SUCCEEDED    Complete            ray-job-sample-raycluster-5lvzq   2025-02-15T04:06:52Z   2025-02-15T04:07:36Z   62s

$ k get raycluster
No resources found in default namespace.

Set deletion-policy to DeleteSelf

$ kubectl ray job submit --name ray-job-sample --deletion-policy DeleteSelf  --working-dir ~/workdir --runtime-env ~/workdir/runtimeEnv.yaml -- python sample_code.py
Submitted RayJob ray-job-sample.

$ k get raycluster
No resources found in default namespace.

$ k get rayjob
No resources found in default namespace.

Set deletion-policy to DeleteCluster

kubectl ray job submit --name ray-job-sample --deletion-policy DeleteCluster  --working-dir ~/workdir --runtime-env ~/workdir/runtimeEnv.yaml -- python sample_code.py
Submitted RayJob ray-job-sample.

$ k get rayjob
NAME             JOB STATUS   DEPLOYMENT STATUS   RAY CLUSTER NAME                  START TIME             END TIME               AGE
ray-job-sample   SUCCEEDED    Complete            ray-job-sample-raycluster-krkv5   2025-02-15T04:14:04Z   2025-02-15T04:14:51Z   66s

$ k get raycluster
No resources found in default namespace.

Set deletion-policy to DeleteWorkers

$ kubectl ray job submit --name ray-job-sample --deletion-policy DeleteWorkers  --working-dir ~/workdir --runtime-env ~/workdir/runtimeEnv.yaml -- python sample_code.py
Submitted RayJob ray-job-sample.

$ k get rayjob
NAME             JOB STATUS   DEPLOYMENT STATUS   RAY CLUSTER NAME                  START TIME             END TIME               AGE
ray-job-sample   SUCCEEDED    Complete            ray-job-sample-raycluster-w5j8m   2025-02-15T04:16:23Z   2025-02-15T04:17:11Z   80s

$ k get raycluster
NAME                              DESIRED WORKERS   AVAILABLE WORKERS   CPUS   MEMORY   GPUS   STATUS   AGE
ray-job-sample-raycluster-w5j8m                                         2      4Gi      0      ready    66s

$ k get pod
NAME                                         READY   STATUS    RESTARTS   AGE
kuberay-operator-f95d56cc6-txxwd             1/1     Running   0          101m
ray-job-sample-raycluster-w5j8m-head-qhdn7   1/1     Running   0          97s

Set deletion-policy to DeleteNone

$ kubectl ray job submit --name ray-job-sample --deletion-policy DeleteNone  --working-dir ~/workdir --runtime-env ~/workdir/runtimeEnv.yaml -- python sample_code.py
Submitted RayJob ray-job-sample.

$ k get rayjob
NAME             JOB STATUS   DEPLOYMENT STATUS   RAY CLUSTER NAME                  START TIME             END TIME               AGE
ray-job-sample   SUCCEEDED    Complete            ray-job-sample-raycluster-rdx6h   2025-02-15T05:06:04Z   2025-02-15T05:06:47Z   52s

$ k get rayclusters
NAME                              DESIRED WORKERS   AVAILABLE WORKERS   CPUS   MEMORY   GPUS   STATUS   AGE
ray-job-sample-raycluster-rdx6h   1                 1                   4      8Gi      0      ready    56s

$ k get pod
NAME                                                         READY   STATUS    RESTARTS   AGE
kuberay-operator-f95d56cc6-l8b26                             1/1     Running   0          3m12s
ray-job-sample-raycluster-rdx6h-default-group-worker-xx9qt   1/1     Running   0          59s
ray-job-sample-raycluster-rdx6h-head-2h4l9                   1/1     Running   0          59s

Related issue number

Checks

I've made sure the tests are passing.
Testing Strategy
- Unit tests
- Manual tests
- This PR is not tested :(

andrewsykim · 2025-02-14T16:08:11Z

kubectl-plugin/pkg/cmd/job/job_submit.go

@@ -161,6 +162,7 @@ func NewJobSubmitCommand(streams genericclioptions.IOStreams) *cobra.Command {
 	cmd.Flags().StringVar(&options.workerMemory, "worker-memory", "4Gi", "amount of memory in each worker group replica")
 	cmd.Flags().StringVar(&options.workerGPU, "worker-gpu", "0", "number of GPUs in each worker group replica")
 	cmd.Flags().BoolVar(&options.dryRun, "dry-run", false, "print the generated YAML instead of creating the cluster. Only works when filename is not provided")
+	cmd.Flags().BoolVar(&options.cleanupJob, "cleanup", false, "Delete the Ray job after job completion")


Alternatively, we can have this be a --deletion-policy that maps to this: https://github.com/ray-project/kuberay/blob/master/ray-operator/apis/ray/v1/rayjob_types.go#L112

For compatibility maybe we need a flag for both shutdownAfterJobFinishes and deletionPolicy

I have rewrited the code and introduced the shutdown-after-job-finishes, deletion-policy, and ttl-seconds-after-finished flags for cleanup.

win5923 · 2025-02-17T14:51:00Z

kubectl-plugin/pkg/cmd/job/job_submit.go

+	dashboardAddr         = "http://localhost:8265"
+	clusterTimeout        = 120.0
+	portforwardtimeout    = 60.0
+	rayjobDeletionTimeout = 30.0


Judge within 30 seconds whether the event RayJobDeletionPolicy feature gate must be enabled to use the DeletionPolicy feature is received.

…olicy flag Signed-off-by: win5923 <[email protected]>

Signed-off-by: win5923 <[email protected]>

win5923 · 2025-02-19T07:14:35Z

kubectl-plugin/pkg/util/client/client.go

+func (c *k8sClient) WaitRayJobDeletionPolicyEnabled(ctx context.Context, namespace, name string, startTime time.Time, timeout time.Duration) error {
+	timeoutCtx, cancel := context.WithTimeout(ctx, timeout*time.Second)
+	defer cancel()
+
+	watcher, err := c.KubernetesClient().CoreV1().Events(namespace).Watch(ctx, metav1.ListOptions{
+		FieldSelector: fmt.Sprintf("involvedObject.name=%s", name),
+	})
+	if err != nil {
+		return fmt.Errorf("failed to watch events for RayJob %s in namespace %s: %w", name, namespace, err)
+	}
+	defer watcher.Stop()
+
+	for {
+		select {
+		case <-timeoutCtx.Done():
+			// If the RayJobDeletionPolicy feature event is not received within the timeout period, it is considered enabled.
+			return nil
+		case event := <-watcher.ResultChan():
+			if event.Type == watch.Error {
+				return fmt.Errorf("error watching events: %v", event.Object)
+			}
+
+			e, ok := event.Object.(*corev1.Event)
+			if !ok {
+				continue
+			}
+
+			if strings.Contains(e.Message, "RayJobDeletionPolicy feature gate must be enabled to use the DeletionPolicy feature") {
+				if e.FirstTimestamp.Time.After(startTime) || e.LastTimestamp.Time.After(startTime) {
+					return fmt.Errorf("%s", e.Message)
+				}
+			}
+		}
+	}
+}


This will block for 30 seconds until the RayJobDeletionPolicy feature event is found. If the event is not found within 30 seconds, it assume that the user has enabled RayJobDeletionPolicy by default.

There might be a better way, but I haven't thought of one yet.

win5923 changed the title ~~[feat][kubectl-plugin] add --cleanup flag to kubectl plugin for RayJob cleanup~~ [feat][kubectl-plugin] add --cleanup flag to ray job submit for RayJob cleanup Feb 14, 2025

andrewsykim reviewed Feb 14, 2025

View reviewed changes

win5923 marked this pull request as draft February 14, 2025 17:27

kevin85421 assigned andrewsykim Feb 14, 2025

win5923 force-pushed the plugin/clean-job branch from 3c14973 to c5f33b3 Compare February 15, 2025 03:50

win5923 changed the title ~~[feat][kubectl-plugin] add --cleanup flag to ray job submit for RayJob cleanup~~ [feat][kubectl-plugin] add shutdown-after-job-finishes and deletion-policy flag to ray job submit for cleanup Feb 15, 2025

win5923 changed the title ~~[feat][kubectl-plugin] add shutdown-after-job-finishes and deletion-policy flag to ray job submit for cleanup~~ [feat][kubectl-plugin] Implement kubectl ray job submit with Deletion Policy API for RayJob Cleanup Feb 15, 2025

win5923 marked this pull request as ready for review February 15, 2025 04:59

win5923 force-pushed the plugin/clean-job branch from 7aeff99 to 70ddaa8 Compare February 15, 2025 05:05

win5923 requested a review from andrewsykim February 15, 2025 09:27

win5923 force-pushed the plugin/clean-job branch from ec4a3ca to a673e44 Compare February 17, 2025 13:28

win5923 marked this pull request as draft February 17, 2025 14:22

win5923 force-pushed the plugin/clean-job branch from 73c9099 to 748812f Compare February 17, 2025 14:46

win5923 commented Feb 17, 2025

View reviewed changes

win5923 marked this pull request as ready for review February 17, 2025 16:22

win5923 added 7 commits February 19, 2025 06:59

[feat][kubectl-plugin] add shutdown-after-job-finishes and deletion-p…

1199aec

…olicy flag Signed-off-by: win5923 <[email protected]>

Add TTLSecondsAfterFinished fla

a0e3148

Signed-off-by: win5923 <[email protected]>

Check for events related to RayJobDeletionPolicy feature gate

a53709f

Signed-off-by: win5923 <[email protected]>

Add comments

8331f63

Signed-off-by: win5923 <[email protected]>

add unit test

e9cf5d3

Signed-off-by: win5923 <[email protected]>

Refactor

8cec9fa

Signed-off-by: win5923 <[email protected]>

add unit test

188d52b

Signed-off-by: win5923 <[email protected]>

win5923 force-pushed the plugin/clean-job branch from d94a1ec to 188d52b Compare February 19, 2025 07:08

win5923 commented Feb 19, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[feat][kubectl-plugin] Implement kubectl ray job submit with Deletion Policy API for RayJob Cleanup #3064

[feat][kubectl-plugin] Implement kubectl ray job submit with Deletion Policy API for RayJob Cleanup #3064

win5923 commented Feb 14, 2025 •

edited

Loading

andrewsykim Feb 14, 2025

win5923 Feb 15, 2025 •

edited

Loading

win5923 Feb 17, 2025

win5923 Feb 19, 2025

[feat][kubectl-plugin] Implement kubectl ray job submit with Deletion Policy API for RayJob Cleanup #3064

Are you sure you want to change the base?

[feat][kubectl-plugin] Implement kubectl ray job submit with Deletion Policy API for RayJob Cleanup #3064

Conversation

win5923 commented Feb 14, 2025 • edited Loading

Why are these changes needed?

Related issue number

Checks

andrewsykim Feb 14, 2025

Choose a reason for hiding this comment

win5923 Feb 15, 2025 • edited Loading

Choose a reason for hiding this comment

win5923 Feb 17, 2025

Choose a reason for hiding this comment

win5923 Feb 19, 2025

Choose a reason for hiding this comment

win5923 commented Feb 14, 2025 •

edited

Loading

win5923 Feb 15, 2025 •

edited

Loading