[Feature][RayJob]: Generate submitter and RayCluster creation/deletion events (#2389)

Author: rueian

ray-operator/controllers/ray/rayjob_controller.go

@@ -224,7 +224,7 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 			// mode is not stuck in the `Running` status indefinitely.
 			namespacedName := common.RayJobK8sJobNamespacedName(rayJobInstance)
 			if err := r.Client.Get(ctx, namespacedName, job); err != nil {
-				logger.Error(err, "Failed to get the submitter Kubernetes Job", "NamespacedName", namespacedName)
+				logger.Error(err, "Failed to get the submitter Kubernetes Job for RayJob", "NamespacedName", namespacedName)
 				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 			}
 			if shouldUpdate := r.checkK8sJobAndUpdateStatusIfNeeded(ctx, rayJobInstance, job); shouldUpdate {
@@ -443,7 +443,7 @@ func (r *RayJobReconciler) createK8sJobIfNeed(ctx context.Context, rayJobInstanc
 		return err
 	}
 
-	logger.Info("Kubernetes Job already exists", "RayJob", rayJobInstance.Name, "Kubernetes Job", job.Name)
+	logger.Info("The submitter Kubernetes Job for RayJob already exists", "RayJob", rayJobInstance.Name, "Kubernetes Job", job.Name)
 	return nil
 }
 
@@ -528,12 +528,12 @@ func (r *RayJobReconciler) createNewK8sJob(ctx context.Context, rayJobInstance *
 
 	// Create the Kubernetes Job
 	if err := r.Client.Create(ctx, job); err != nil {
-		logger.Error(err, "Failed to create new Kubernetes Job")
-		r.Recorder.Eventf(rayJobInstance, corev1.EventTypeWarning, "k8sJobCreationFailed", "Failed to create new Kubernetes Job %s: %v", job.Name, err)
+		logger.Error(err, "Failed to create new submitter Kubernetes Job for RayJob")
+		r.Recorder.Eventf(rayJobInstance, corev1.EventTypeWarning, string(utils.FailedToCreateRayJobSubmitter), "Failed to create new Kubernetes Job %s/%s: %v", job.Namespace, job.Name, err)
 		return err
 	}
-	logger.Info("Kubernetes Job created", "RayJob", rayJobInstance.Name, "Kubernetes Job", job.Name)
-	r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, "k8sJobCreationCreated", "Created Kubernetes Job %s", job.Name)
+	logger.Info("Created submitter Kubernetes Job for RayJob", "RayJob", rayJobInstance.Name, "Kubernetes Job", job.Name)
+	r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, string(utils.CreatedRayJobSubmitter), "Created Kubernetes Job %s/%s", job.Namespace, job.Name)
 	return nil
 }
 
@@ -559,13 +559,14 @@ func (r *RayJobReconciler) deleteSubmitterJob(ctx context.Context, rayJobInstanc
 		}
 	} else {
 		if !job.DeletionTimestamp.IsZero() {
-			logger.Info("The Job deletion is ongoing.", "RayJob", rayJobInstance.Name, "Submitter K8s Job", job.Name)
+			logger.Info("The deletion of submitter Kubernetes Job for RayJob is ongoing.", "RayJob", rayJobInstance.Name, "Submitter K8s Job", job.Name)
 		} else {
 			if err := r.Client.Delete(ctx, job, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil {
+				r.Recorder.Eventf(rayJobInstance, corev1.EventTypeWarning, string(utils.FailedToDeleteRayJobSubmitter), "Failed to delete submitter K8s Job %s/%s: %v", job.Namespace, job.Name, err)
 				return false, err
 			}
-			logger.Info("The associated submitter Job is deleted", "RayJob", rayJobInstance.Name, "Submitter K8s Job", job.Name)
-			r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, "Deleted", "Deleted submitter K8s Job %s", job.Name)
+			logger.Info("The associated submitter Kubernetes Job for RayJob is deleted", "RayJob", rayJobInstance.Name, "Submitter K8s Job", job.Name)
+			r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, string(utils.DeletedRayJobSubmitter), "Deleted submitter K8s Job %s/%s", job.Namespace, job.Name)
 		}
 	}
 
@@ -585,19 +586,20 @@ func (r *RayJobReconciler) deleteClusterResources(ctx context.Context, rayJobIns
 			// If the cluster is not found, it means the cluster has been already deleted.
 			// Don't return error to make this function idempotent.
 			isClusterDeleted = true
-			logger.Info("The associated cluster has been already deleted and it can not be found", "RayCluster", clusterIdentifier)
+			logger.Info("The associated RayCluster for RayJob has been already deleted and it can not be found", "RayCluster", clusterIdentifier, "RayJob", rayJobInstance.Name)
 		} else {
 			return false, err
 		}
 	} else {
 		if !cluster.DeletionTimestamp.IsZero() {
-			logger.Info("The cluster deletion is ongoing.", "rayjob", rayJobInstance.Name, "raycluster", cluster.Name)
+			logger.Info("The deletion of the associated RayCluster for RayJob is ongoing.", "RayJob", rayJobInstance.Name, "RayCluster", cluster.Name)
 		} else {
 			if err := r.Delete(ctx, &cluster); err != nil {
+				r.Recorder.Eventf(rayJobInstance, corev1.EventTypeWarning, string(utils.FailedToDeleteRayCluster), "Failed to delete cluster %s/%s: %v", cluster.Namespace, cluster.Name, err)
 				return false, err
 			}
-			logger.Info("The associated cluster is deleted", "RayCluster", clusterIdentifier)
-			r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, "Deleted", "Deleted cluster %s", rayJobInstance.Status.RayClusterName)
+			logger.Info("The associated RayCluster for RayJob is deleted", "RayCluster", clusterIdentifier, "RayJob", rayJobInstance.Name)
+			r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, string(utils.DeletedRayCluster), "Deleted cluster %s/%s", cluster.Namespace, cluster.Name)
 		}
 	}
 
@@ -712,14 +714,15 @@ func (r *RayJobReconciler) getOrCreateRayClusterInstance(ctx context.Context, ra
 				return nil, err
 			}
 			if err := r.Create(ctx, rayClusterInstance); err != nil {
+				r.Recorder.Eventf(rayJobInstance, corev1.EventTypeWarning, string(utils.FailedToCreateRayCluster), "Failed to create RayCluster %s/%s: %v", rayClusterInstance.Namespace, rayClusterInstance.Name, err)
 				return nil, err
 			}
-			r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, "Created", "Created RayCluster %s", rayJobInstance.Status.RayClusterName)
+			r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, string(utils.CreatedRayCluster), "Created RayCluster %s/%s", rayClusterInstance.Namespace, rayClusterInstance.Name)
 		} else {
 			return nil, err
 		}
 	}
-	logger.Info("Found associated RayCluster for RayJob", "RayJob", rayJobInstance.Name, "RayCluster", rayClusterNamespacedName)
+	logger.Info("Found the associated RayCluster for RayJob", "RayJob", rayJobInstance.Name, "RayCluster", rayClusterNamespacedName)
 
 	// Verify that RayJob is not in cluster selector mode first to avoid nil pointer dereference error during spec comparison.
 	// This is checked by ensuring len(rayJobInstance.Spec.ClusterSelector) equals 0.

ray-operator/controllers/ray/rayjob_controller_test.go

@@ -312,6 +312,67 @@ var _ = Context("RayJob with different submission modes", func() {
 			})
 		})
 
+		Describe("Invalid RayJob in K8sJobMode", Ordered, func() {
+			ctx := context.Background()
+			namespace := "default"
+			rayJob := rayJobTemplate("rayjob-invalid-test", namespace)
+			rayCluster := &rayv1.RayCluster{Spec: *rayJob.Spec.RayClusterSpec}
+			template := common.GetDefaultSubmitterTemplate(rayCluster)
+			template.Spec.RestartPolicy = "" // Make it invalid to create a submitter. Ref: https://github.com/ray-project/kuberay/pull/2389#issuecomment-2359564334
+			rayJob.Spec.SubmitterPodTemplate = &template
+
+			It("Verify RayJob spec", func() {
+				Expect(rayJob.Spec.SubmissionMode).To(Equal(rayv1.K8sJobMode))
+			})
+
+			It("Create a RayJob custom resource", func() {
+				err := k8sClient.Create(ctx, rayJob)
+				Expect(err).NotTo(HaveOccurred(), "Failed to create RayJob")
+				Eventually(
+					getResourceFunc(ctx, client.ObjectKey{Name: rayJob.Name, Namespace: namespace}, rayJob),
+					time.Second*3, time.Millisecond*500).Should(BeNil(), "Should be able to see RayJob: %v", rayJob.Name)
+			})
+
+			It("RayJobs's JobDeploymentStatus transitions from New to Initializing.", func() {
+				Eventually(
+					getRayJobDeploymentStatus(ctx, rayJob),
+					time.Second*3, time.Millisecond*500).Should(Equal(rayv1.JobDeploymentStatusInitializing), "JobDeploymentStatus = %v", rayJob.Status.JobDeploymentStatus)
+			})
+
+			It("In Initializing state, the RayCluster should eventually be created.", func() {
+				Eventually(
+					getResourceFunc(ctx, client.ObjectKey{Name: rayJob.Status.RayClusterName, Namespace: namespace}, rayCluster),
+					time.Second*3, time.Millisecond*500).Should(BeNil(), "RayCluster %v not found", rayJob.Status.RayClusterName)
+			})
+
+			It("Make RayCluster.Status.State to be rayv1.Ready", func() {
+				// The RayCluster is not 'Ready' yet because Pods are not running and ready.
+				Expect(rayCluster.Status.State).NotTo(Equal(rayv1.Ready)) //nolint:staticcheck // https://github.com/ray-project/kuberay/pull/2288
+
+				updateHeadPodToRunningAndReady(ctx, rayJob.Status.RayClusterName, namespace)
+				updateWorkerPodsToRunningAndReady(ctx, rayJob.Status.RayClusterName, namespace)
+
+				// The RayCluster.Status.State should be Ready.
+				Eventually(
+					getClusterState(ctx, namespace, rayCluster.Name),
+					time.Second*3, time.Millisecond*500).Should(Equal(rayv1.Ready))
+			})
+
+			It("RayJobs's JobDeploymentStatus transitions from Initializing to Running.", func() {
+				Eventually(
+					func() ([]corev1.Event, error) {
+						events := &corev1.EventList{}
+						if err := k8sClient.List(ctx, events, client.InNamespace(rayJob.Namespace)); err != nil {
+							return nil, err
+						}
+						return events.Items, nil
+					},
+					time.Second*3, time.Millisecond*500).Should(ContainElement(HaveField("Message", ContainSubstring("Failed to create new Kubernetes Job default/rayjob-invalid-test"))))
+
+				_ = k8sClient.Delete(ctx, rayJob)
+			})
+		})
+
 		Describe("Successful RayJob in K8sjobMode with DELETE_RAYJOB_CR_AFTER_JOB_FINISHES", Ordered, func() {
 			ctx := context.Background()
 			namespace := "default"

ray-operator/controllers/ray/rayjob_controller_unit_test.go

@@ -2,6 +2,7 @@ package ray
 
 import (
 	"context"
+	"errors"
 	"strings"
 	"testing"
 
@@ -22,7 +23,7 @@ import (
 	"github.com/ray-project/kuberay/ray-operator/pkg/client/clientset/versioned/scheme"
 )
 
-func TestCreateK8sJobIfNeed(t *testing.T) {
+func TestCreateRayJobSubmitterIfNeed(t *testing.T) {
 	newScheme := runtime.NewScheme()
 	_ = rayv1.AddToScheme(newScheme)
 	_ = batchv1.AddToScheme(newScheme)
@@ -375,7 +376,7 @@ func TestValidateRayJobSpec(t *testing.T) {
 	assert.Error(t, err, "The RayJob is invalid because the backoffLimit must be a positive integer.")
 }
 
-func TestFailedCreatek8sJob(t *testing.T) {
+func TestFailedToCreateRayJobSubmitterEvent(t *testing.T) {
 	rayJob := &rayv1.RayJob{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-rayjob",
@@ -400,7 +401,7 @@ func TestFailedCreatek8sJob(t *testing.T) {
 
 	fakeClient := clientFake.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
 		Create: func(_ context.Context, _ client.WithWatch, _ client.Object, _ ...client.CreateOption) error {
-			return utils.ErrFailedCreateWorkerPod
+			return errors.New("random")
 		},
 	}).WithScheme(scheme.Scheme).Build()
 
@@ -429,3 +430,148 @@ func TestFailedCreatek8sJob(t *testing.T) {
 
 	assert.Truef(t, foundFailureEvent, "Expected event to be generated for job creation failure, got events: %s", strings.Join(events, "\n"))
 }
+
+func TestFailedCreateRayClusterEvent(t *testing.T) {
+	rayJob := &rayv1.RayJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-rayjob",
+			Namespace: "default",
+		},
+		Spec: rayv1.RayJobSpec{
+			RayClusterSpec: &rayv1.RayClusterSpec{},
+		},
+	}
+
+	fakeClient := clientFake.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+		Create: func(_ context.Context, _ client.WithWatch, _ client.Object, _ ...client.CreateOption) error {
+			return errors.New("random")
+		},
+	}).WithScheme(scheme.Scheme).Build()
+
+	recorder := record.NewFakeRecorder(100)
+
+	reconciler := &RayJobReconciler{
+		Client:   fakeClient,
+		Recorder: recorder,
+		Scheme:   scheme.Scheme,
+	}
+
+	_, err := reconciler.getOrCreateRayClusterInstance(context.Background(), rayJob)
+
+	assert.NotNil(t, err, "Expected error due to cluster creation failure")
+
+	var foundFailureEvent bool
+	events := []string{}
+	for len(recorder.Events) > 0 {
+		event := <-recorder.Events
+		if strings.Contains(event, "Failed to create RayCluster") {
+			foundFailureEvent = true
+			break
+		}
+		events = append(events, event)
+	}
+
+	assert.Truef(t, foundFailureEvent, "Expected event to be generated for cluster creation failure, got events: %s", strings.Join(events, "\n"))
+}
+
+func TestFailedDeleteRayJobSubmitterEvent(t *testing.T) {
+	newScheme := runtime.NewScheme()
+	_ = batchv1.AddToScheme(newScheme)
+
+	rayJob := &rayv1.RayJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-rayjob",
+			Namespace: "default",
+		},
+	}
+	submitter := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-rayjob",
+			Namespace: "default",
+		},
+	}
+
+	fakeClient := clientFake.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+		Delete: func(_ context.Context, _ client.WithWatch, _ client.Object, _ ...client.DeleteOption) error {
+			return errors.New("random")
+		},
+	}).WithScheme(newScheme).WithRuntimeObjects(submitter).Build()
+
+	recorder := record.NewFakeRecorder(100)
+
+	reconciler := &RayJobReconciler{
+		Client:   fakeClient,
+		Recorder: recorder,
+		Scheme:   scheme.Scheme,
+	}
+
+	_, err := reconciler.deleteSubmitterJob(context.Background(), rayJob)
+
+	assert.NotNil(t, err, "Expected error due to job deletion failure")
+
+	var foundFailureEvent bool
+	events := []string{}
+	for len(recorder.Events) > 0 {
+		event := <-recorder.Events
+		if strings.Contains(event, "Failed to delete submitter K8s Job") {
+			foundFailureEvent = true
+			break
+		}
+		events = append(events, event)
+	}
+
+	assert.Truef(t, foundFailureEvent, "Expected event to be generated for cluster deletion failure, got events: %s", strings.Join(events, "\n"))
+}
+
+func TestFailedDeleteRayClusterEvent(t *testing.T) {
+	newScheme := runtime.NewScheme()
+	_ = rayv1.AddToScheme(newScheme)
+
+	rayCluster := &rayv1.RayCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-raycluster",
+			Namespace: "default",
+		},
+	}
+
+	rayJob := &rayv1.RayJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-rayjob",
+			Namespace: "default",
+		},
+		Status: rayv1.RayJobStatus{
+			RayClusterName: "test-raycluster",
+		},
+	}
+
+	fakeClient := clientFake.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+		Delete: func(_ context.Context, _ client.WithWatch, _ client.Object, _ ...client.DeleteOption) error {
+			return errors.New("random")
+		},
+	}).WithScheme(newScheme).WithRuntimeObjects(rayCluster).Build()
+
+	recorder := record.NewFakeRecorder(100)
+
+	reconciler := &RayJobReconciler{
+		Client:   fakeClient,
+		Recorder: recorder,
+		Scheme:   scheme.Scheme,
+	}
+
+	_, err := reconciler.deleteClusterResources(context.Background(), rayJob)
+
+	assert.NotNil(t, err, "Expected error due to cluster deletion failure")
+
+	var foundFailureEvent bool
+	events := []string{}
+	for len(recorder.Events) > 0 {
+		event := <-recorder.Events
+		if strings.Contains(event, "Failed to delete cluster") {
+			foundFailureEvent = true
+			break
+		}
+		events = append(events, event)
+	}
+
+	assert.Truef(t, foundFailureEvent, "Expected event to be generated for cluster deletion failure, got events: %s", strings.Join(events, "\n"))
+}

ray-operator/controllers/ray/utils/constant.go

@@ -248,6 +248,16 @@ const (
 	CreatedRedisCleanupJob        K8sEventType = "CreatedRedisCleanupJob"
 	FailedToCreateRedisCleanupJob K8sEventType = "FailedToCreateRedisCleanupJob"
 
+	// RayJob event list
+	CreatedRayJobSubmitter        K8sEventType = "CreatedRayJobSubmitter"
+	DeletedRayJobSubmitter        K8sEventType = "DeletedRayJobSubmitter"
+	FailedToCreateRayJobSubmitter K8sEventType = "FailedToCreateRayJobSubmitter"
+	FailedToDeleteRayJobSubmitter K8sEventType = "FailedToDeleteRayJobSubmitter"
+	CreatedRayCluster             K8sEventType = "CreatedRayCluster"
+	DeletedRayCluster             K8sEventType = "DeletedRayCluster"
+	FailedToCreateRayCluster      K8sEventType = "FailedToCreateRayCluster"
+	FailedToDeleteRayCluster      K8sEventType = "FailedToDeleteRayCluster"
+
 	// Generic Pod event list
 	DeletedPod        K8sEventType = "DeletedPod"
 	FailedToDeletePod K8sEventType = "FailedToDeletePod"