[RayJob] Add Failure Feedback (log and event) for Failed k8s Creation Task (#2306)

Author: tinaxfwu

ray-operator/controllers/ray/rayjob_controller.go

@@ -528,10 +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)
 		return err
 	}
 	logger.Info("Kubernetes Job created", "RayJob", rayJobInstance.Name, "Kubernetes Job", job.Name)
-	r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, "Created", "Created Kubernetes Job %s", job.Name)
+	r.Recorder.Eventf(rayJobInstance, corev1.EventTypeNormal, "k8sJobCreationCreated", "Created Kubernetes Job %s", job.Name)
 	return nil
 }
 

ray-operator/controllers/ray/rayjob_controller_unit_test.go

@@ -2,6 +2,7 @@ package ray
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -12,10 +13,13 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	utils "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
+	"github.com/ray-project/kuberay/ray-operator/pkg/client/clientset/versioned/scheme"
 )
 
 func TestCreateK8sJobIfNeed(t *testing.T) {
@@ -370,3 +374,58 @@ 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) {
+	rayJob := &rayv1.RayJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-rayjob",
+			Namespace: "default",
+		},
+	}
+
+	submitterTemplate := corev1.PodTemplateSpec{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-submit-pod",
+			Namespace: "default",
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:  "ray-submit",
+					Image: "rayproject/ray:latest",
+				},
+			},
+		},
+	}
+
+	fakeClient := clientFake.NewClientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+		Create: func(_ context.Context, _ client.WithWatch, _ client.Object, _ ...client.CreateOption) error {
+			return utils.ErrFailedCreateWorkerPod
+		},
+	}).WithScheme(scheme.Scheme).Build()
+
+	recorder := record.NewFakeRecorder(100)
+
+	reconciler := &RayJobReconciler{
+		Client:   fakeClient,
+		Recorder: recorder,
+		Scheme:   scheme.Scheme,
+	}
+
+	err := reconciler.createNewK8sJob(context.Background(), rayJob, submitterTemplate)
+
+	assert.NotNil(t, err, "Expected error due to simulated job creation failure")
+
+	var foundFailureEvent bool
+	events := []string{}
+	for len(recorder.Events) > 0 {
+		event := <-recorder.Events
+		if strings.Contains(event, "Failed to create new Kubernetes Job") {
+			foundFailureEvent = true
+			break
+		}
+		events = append(events, event)
+	}
+
+	assert.Truef(t, foundFailureEvent, "Expected event to be generated for job creation failure, got events: %s", strings.Join(events, "\n"))
+}