feat(evalhub): emit Kubernetes events for operator-detected evaluation failures (RHAI-278)

config/components/evalhub/rbac/evalhub_events_role.yaml

@@ -1,8 +1,9 @@
 ---
-# ClusterRole for EvalHub event emission
-# Grants create on core events so the EvalHub server can emit Kubernetes
-# Events against backing Job resources on evaluation lifecycle transitions
-# (EvaluationStarted, EvaluationCompleted, EvaluationFailed, EvaluationThresholdViolated).
+# ClusterRole for EvalHub event emission (RHAI-277).
+# Currently bound to the operator SA; will be rebound to the EvalHub server
+# SA via a runtime RoleBinding when RHAI-277 lands.
+# Grants create/patch on core events so the EvalHub server can emit Kubernetes
+# Events against backing Job resources on evaluation lifecycle transitions.
 # Split from evalhub-jobs-writer for least-privilege.
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole

controllers/evalhub/evaluation_failed_kueue_workloads_reconciler.go

@@ -20,6 +20,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -58,14 +59,16 @@ func evaluationFailedKueueWorkloadsLogFields() []any {
 }
 
 //+kubebuilder:rbac:groups=kueue.x-k8s.io,resources=workloads,verbs=get;list;watch;patch
-//+kubebuilder:rbac:groups=batch,resources=jobs,verbs=get
+//+kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;patch
+//+kubebuilder:rbac:groups="",resources=events,verbs=create;patch
 //+kubebuilder:rbac:groups=trustyai.opendatahub.io,resources=evalhubs,verbs=get
 
 // EvalHubEvaluationFailedKueueWorkloadsReconciler POSTs a failed benchmark event to EvalHub when a Kueue
 // Workload has QuotaReserved=False with Reason=Inadmissible and is owned by an EvalHub evaluation Job.
 type EvalHubEvaluationFailedKueueWorkloadsReconciler struct {
 	client.Client
-	RESTConfig *rest.Config
+	RESTConfig    *rest.Config
+	EventRecorder record.EventRecorder
 	// tenantNS is the same instance updated by EvalHubEvaluationJobFailureReconciler's Namespace watch.
 	tenantNS *evalHubTenantNamespaces
 }
@@ -77,9 +80,10 @@ func registerEvalHubEvaluationFailedKueueWorkloadsReconciler(mgr manager.Manager
 		return fmt.Errorf("evalhub failed kueue workloads: tenantNS is nil")
 	}
 	r := &EvalHubEvaluationFailedKueueWorkloadsReconciler{
-		Client:     mgr.GetClient(),
-		RESTConfig: rest.CopyConfig(mgr.GetConfig()),
-		tenantNS:   tenantNS,
+		Client:        mgr.GetClient(),
+		RESTConfig:    rest.CopyConfig(mgr.GetConfig()),
+		tenantNS:      tenantNS,
+		EventRecorder: mgr.GetEventRecorderFor("trustyai-service-operator"),
 	}
 	return ctrl.NewControllerManagedBy(mgr).
 		Named(evalHubEvaluationFailedKueueWorkloadsControllerName).
@@ -235,9 +239,7 @@ func (r *EvalHubEvaluationFailedKueueWorkloadsReconciler) Reconcile(ctx context.
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	if workloadFailedEventAlreadyReported(&wl) {
-		return ctrl.Result{}, nil
-	}
+	alreadyReported := workloadFailedEventAlreadyReported(&wl)
 
 	cond, ok := workloadQuotaReservedInadmissibleCondition(&wl)
 	if !ok {
@@ -281,6 +283,25 @@ func (r *EvalHubEvaluationFailedKueueWorkloadsReconciler) Reconcile(ctx context.
 	if failureAlreadyReported(&job) {
 		return ctrl.Result{}, nil
 	}
+	if serverAlreadyHandledFailure(&job) {
+		log.V(1).Info("skip: EvalHub server already set failure label on Job",
+			append(evaluationFailedKueueWorkloadsLogFields(), "action", "skip_server_handled",
+				"workload", wl.Name, "job", job.Name, "namespace", job.Namespace)...)
+		return ctrl.Result{}, nil
+	}
+
+	// Workload annotation was written but the Job labels were not (partial commit from a
+	// previous reconcile). Retry only the Job patch — skip the POST and Event.
+	if alreadyReported {
+		failureMsg, _ := classifyKueueAdmissionFailure(&job, cond)
+		if err := r.patchJobFailureLabels(ctx, &job, failureMsg); err != nil {
+			log.Error(err, "retry: failed to patch Job failure labels",
+				append(evaluationFailedKueueWorkloadsLogFields(), "action", "retry_patch_job_labels_failed",
+					"job", job.Name, "namespace", job.Namespace)...)
+			return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
+		}
+		return ctrl.Result{}, nil
+	}
 
 	jobID := strings.TrimSpace(job.Labels[evalHubJobIDLabel])
 	if jobID == "" {
@@ -325,13 +346,24 @@ func (r *EvalHubEvaluationFailedKueueWorkloadsReconciler) Reconcile(ctx context.
 		return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
 	}
 
+	// Annotate the Workload first so that retries hit workloadFailedEventAlreadyReported
+	// and never re-POST to EvalHub or re-emit the Event.
 	if err := r.annotateWorkloadReported(ctx, &wl); err != nil {
 		log.Error(err, "patch workload after EvalHub failed-workload event",
 			append(evaluationFailedKueueWorkloadsLogFields(), "action", "patch_workload_failed",
 				"workload", client.ObjectKeyFromObject(&wl))...)
 		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
 	}
 
+	r.EventRecorder.Eventf(&job, corev1.EventTypeWarning, eventReasonEvaluationFailed, "%s", failureMsg)
+
+	if err := r.patchJobFailureLabels(ctx, &job, failureMsg); err != nil {
+		log.Error(err, "failed to patch Job failure labels after Kueue workload eviction",
+			append(evaluationFailedKueueWorkloadsLogFields(), "action", "patch_job_labels_failed",
+				"job", job.Name, "namespace", job.Namespace)...)
+		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
+	}
+
 	log.Info("posted EvalHub failed status for Kueue workload (via owning Job)",
 		append(evaluationFailedKueueWorkloadsLogFields(), "action", "post_events_ok",
 			"workload", wl.Name, "workloadNamespace", wl.Namespace, "queue", wl.Spec.QueueName,
@@ -351,3 +383,18 @@ func (r *EvalHubEvaluationFailedKueueWorkloadsReconciler) annotateWorkloadReport
 	wl.Annotations[annotationKueueFailedWorkloadEventReported] = "true"
 	return r.Patch(ctx, wl, client.MergeFrom(patchBase))
 }
+
+// patchJobFailureLabels stamps the owning Job with evaluation-phase=Failed and sets the evaluation-status
+// annotation so the failure is observable on the Job resource even before it is cleaned up.
+func (r *EvalHubEvaluationFailedKueueWorkloadsReconciler) patchJobFailureLabels(ctx context.Context, job *batchv1.Job, statusMsg string) error {
+	base := job.DeepCopy()
+	if job.Labels == nil {
+		job.Labels = map[string]string{}
+	}
+	if job.Annotations == nil {
+		job.Annotations = map[string]string{}
+	}
+	job.Labels[labelEvaluationPhase] = labelEvaluationPhaseFailed
+	job.Annotations[annotationEvaluationStatus] = statusMsg
+	return r.Patch(ctx, job, client.MergeFrom(base))
+}

controllers/evalhub/evaluation_failed_kueue_workloads_reconciler_lifecycle_test.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2025.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+*/
+
+package evalhub
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	evalhubv1 "github.com/trustyai-explainability/trustyai-service-operator/api/evalhub/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+)
+
+// kueueLifecycleScheme builds the minimal scheme for Kueue failure reconciler lifecycle tests.
+func kueueLifecycleScheme(t *testing.T) *runtime.Scheme {
+	t.Helper()
+	sc := runtime.NewScheme()
+	require.NoError(t, batchv1.AddToScheme(sc))
+	require.NoError(t, corev1.AddToScheme(sc))
+	require.NoError(t, evalhubv1.AddToScheme(sc))
+	require.NoError(t, kueue.AddToScheme(sc))
+	return sc
+}
+
+// inadmissibleWorkload builds a Kueue Workload with QuotaReserved=False/Inadmissible owned by a Job.
+func inadmissibleWorkload(name, ns, jobName string, jobUID types.UID, condMsg string) *kueue.Workload {
+	return &kueue.Workload{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+			UID:       types.UID("wl-uid-" + name),
+			OwnerReferences: []metav1.OwnerReference{{
+				APIVersion: "batch/v1", Kind: "Job", Name: jobName, UID: jobUID,
+			}},
+		},
+		Spec: kueue.WorkloadSpec{
+			QueueName: "default-queue",
+		},
+		Status: kueue.WorkloadStatus{
+			Conditions: []metav1.Condition{{
+				Type:    kueue.WorkloadQuotaReserved,
+				Status:  metav1.ConditionFalse,
+				Reason:  kueueWorkloadReasonInadmissible,
+				Message: condMsg,
+			}},
+		},
+	}
+}
+
+// buildKueueReconciler returns a reconciler with a FakeRecorder and the given tenant namespace.
+func buildKueueReconciler(fc client.Client, tenantNamespace string) (*EvalHubEvaluationFailedKueueWorkloadsReconciler, *record.FakeRecorder) {
+	rec := record.NewFakeRecorder(10)
+	tn := newEvalHubTenantNamespaces()
+	tn.Add(tenantNamespace)
+	return &EvalHubEvaluationFailedKueueWorkloadsReconciler{
+		Client:        fc,
+		RESTConfig:    &rest.Config{},
+		EventRecorder: rec,
+		tenantNS:      tn,
+	}, rec
+}
+
+// TestKueueReconciler_Eviction_EmitsEventAndPatchesJobAndWorkload verifies the full lifecycle for a Kueue
+// admission failure: the reconciler POSTs to EvalHub, emits an EvaluationFailed warning event on the Job,
+// stamps evaluation-phase=Failed on the Job, and annotates the Workload as reported.
+func TestKueueReconciler_Eviction_EmitsEventAndPatchesJobAndWorkload(t *testing.T) {
+	srv := noopEvalHubServer(t)
+	sc := kueueLifecycleScheme(t)
+	ns := "tenant-ns"
+
+	job := evalHubEvaluationJob("eval-job-kueue", ns, map[string]string{
+		evalHubInstanceNameLabel:      "evalhub-1",
+		evalHubInstanceNamespaceLabel: "control-ns",
+	})
+	wl := inadmissibleWorkload("wl-1", ns, job.Name, job.UID, "insufficient quota")
+	eh := readyEvalHubCR("evalhub-1", "control-ns", srv.URL)
+
+	var patchedJobLabels map[string]string
+	var workloadAnnotated bool
+
+	fc := fake.NewClientBuilder().
+		WithScheme(sc).
+		WithObjects(eh, job, wl).
+		WithInterceptorFuncs(interceptor.Funcs{
+			Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
+				switch o := obj.(type) {
+				case *batchv1.Job:
+					if o.Labels[labelEvaluationPhase] == labelEvaluationPhaseFailed {
+						patchedJobLabels = o.Labels
+					}
+				case *kueue.Workload:
+					if o.Annotations[annotationKueueFailedWorkloadEventReported] == "true" {
+						workloadAnnotated = true
+					}
+				}
+				return c.Patch(ctx, obj, patch, opts...)
+			},
+		}).
+		Build()
+
+	r, rec := buildKueueReconciler(fc, ns)
+
+	_, err := r.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Namespace: ns, Name: wl.Name},
+	})
+	require.NoError(t, err)
+
+	select {
+	case ev := <-rec.Events:
+		assert.Contains(t, ev, corev1.EventTypeWarning)
+		assert.Contains(t, ev, eventReasonEvaluationFailed)
+	default:
+		t.Fatal("expected EvaluationFailed event but recorder is empty")
+	}
+
+	require.NotNil(t, patchedJobLabels, "evaluation-phase=Failed patch was never applied to Job")
+	assert.Equal(t, labelEvaluationPhaseFailed, patchedJobLabels[labelEvaluationPhase])
+	assert.True(t, workloadAnnotated, "Workload should be annotated as reported")
+}