diff --git a/config/components/evalhub/rbac/evalhub_events_role.yaml b/config/components/evalhub/rbac/evalhub_events_role.yaml index d56ebd404..551d13d27 100644 --- a/config/components/evalhub/rbac/evalhub_events_role.yaml +++ b/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 diff --git a/controllers/evalhub/evaluation_failed_kueue_workloads_reconciler.go b/controllers/evalhub/evaluation_failed_kueue_workloads_reconciler.go index f758d05b4..5ca2b49ea 100644 --- a/controllers/evalhub/evaluation_failed_kueue_workloads_reconciler.go +++ b/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,6 +346,8 @@ 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", @@ -332,6 +355,15 @@ func (r *EvalHubEvaluationFailedKueueWorkloadsReconciler) Reconcile(ctx context. 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)) +} diff --git a/controllers/evalhub/evaluation_failed_kueue_workloads_reconciler_lifecycle_test.go b/controllers/evalhub/evaluation_failed_kueue_workloads_reconciler_lifecycle_test.go new file mode 100644 index 000000000..9771e4a56 --- /dev/null +++ b/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") +} diff --git a/controllers/evalhub/evaluation_job_failure_reconciler.go b/controllers/evalhub/evaluation_job_failure_reconciler.go index 5a93ea66e..256b117f3 100644 --- a/controllers/evalhub/evaluation_job_failure_reconciler.go +++ b/controllers/evalhub/evaluation_job_failure_reconciler.go @@ -27,6 +27,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" "k8s.io/client-go/transport" "k8s.io/client-go/util/workqueue" ctrl "sigs.k8s.io/controller-runtime" @@ -73,6 +74,17 @@ const ( sidecarContainerName = "sidecar" ) +// Job lifecycle label and annotation stamped by the operator on infrastructure failures. +// The same label is also set by the EvalHub server; the operator checks it to prevent duplicate Events. +const ( + labelEvaluationPhase = "trustyai.opendatahub.io/evaluation-phase" + labelEvaluationPhaseFailed = "Failed" + annotationEvaluationStatus = "trustyai.opendatahub.io/evaluation-status" + // eventReasonEvaluationFailed matches the reason used by server-emitted Events; source.component + // distinguishes operator vs server (set automatically by the EventRecorder). + eventReasonEvaluationFailed = "EvaluationFailed" +) + // evalHubEvaluationJobFailureControllerName matches ctrl.NewControllerManagedBy(mgr).Named(...) for logs and registration. const evalHubEvaluationJobFailureControllerName = "evalhub-evaluation-job-failure" @@ -89,6 +101,7 @@ func failureWatcherLogFields() []any { //+kubebuilder:rbac:groups="",resources=namespaces,verbs=get;list;watch //+kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch //+kubebuilder:rbac:groups="",resources=pods/log,verbs=get +//+kubebuilder:rbac:groups="",resources=events,verbs=create;patch //+kubebuilder:rbac:groups=trustyai.opendatahub.io,resources=evalhubs,verbs=get;list;watch // EvalHubEvaluationJobFailureReconciler POSTs a failed benchmark event to EvalHub when init, adapter, @@ -101,7 +114,8 @@ func failureWatcherLogFields() []any { type EvalHubEvaluationJobFailureReconciler struct { client.Client // RESTConfig is used to build an HTTP transport that authenticates like the operator (SA token + cluster CA). - RESTConfig *rest.Config + RESTConfig *rest.Config + EventRecorder record.EventRecorder // tenantNS tracks namespaces labelled evalhub.trustyai.opendatahub.io/tenant (shared with the Kueue workload failure reconciler). tenantNS *evalHubTenantNamespaces @@ -168,9 +182,10 @@ func registerEvalHubEvaluationJobFailureController(mgr manager.Manager, tenantNS } r := &EvalHubEvaluationJobFailureReconciler{ - 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"), } labelPred, err := predicate.LabelSelectorPredicate(evalHubJobPodLabelSelector()) @@ -328,6 +343,15 @@ func (r *EvalHubEvaluationJobFailureReconciler) Reconcile(ctx context.Context, r } return ctrl.Result{}, nil } + if serverAlreadyHandledFailure(&job) { + log.Info("skip: EvalHub server already set failure label", + append(failureWatcherLogFields(), "action", "skip_server_handled", + "job", job.Name, "namespace", job.Namespace)...) + if err := r.deleteEvalHubFailureSyncedJob(ctx, &job); err != nil { + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil + } + return ctrl.Result{}, nil + } failed, msg, err := r.detectFailure(ctx, &job) if err != nil { @@ -381,6 +405,8 @@ func (r *EvalHubEvaluationJobFailureReconciler) Reconcile(ctx context.Context, r benchmarkIndex := benchmarkIndexFromJob(&job) + // If failure-pending is already set, the POST succeeded in a prior reconcile but the + // promote patch failed. Skip the POST and go straight to the promote patch. if !failurePendingReport(&job) { pendingPatch := client.MergeFrom(job.DeepCopy()) if job.Annotations == nil { @@ -392,35 +418,42 @@ func (r *EvalHubEvaluationJobFailureReconciler) Reconcile(ctx context.Context, r append(failureWatcherLogFields(), "action", "patch_pending_failed", "job", job.Name)...) return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } - } - if err := postEvalHubBenchmarkFailed(ctx, r.RESTConfig, baseURL, job.Namespace, jobID, providerID, benchmarkID, benchmarkIndex, msg, ""); err != nil { - log.Error(err, "failed to post EvalHub benchmark failure event", - append(failureWatcherLogFields(), "action", "post_events_failed", "job", job.Name, "evalJobID", jobID)...) - revert := client.MergeFrom(job.DeepCopy()) - delete(job.Annotations, annotationFailurePending) - if len(job.Annotations) == 0 { - job.Annotations = nil - } - if err2 := r.Patch(ctx, &job, revert); err2 != nil { - log.Error(err2, "failed to revert failure-pending annotation after POST failure", - append(failureWatcherLogFields(), "action", "revert_pending_failed", "job", job.Name)...) + if err := postEvalHubBenchmarkFailed(ctx, r.RESTConfig, baseURL, job.Namespace, jobID, providerID, benchmarkID, benchmarkIndex, msg, ""); err != nil { + log.Error(err, "failed to post EvalHub benchmark failure event", + append(failureWatcherLogFields(), "action", "post_events_failed", "job", job.Name, "evalJobID", jobID)...) + revert := client.MergeFrom(job.DeepCopy()) + delete(job.Annotations, annotationFailurePending) + if len(job.Annotations) == 0 { + job.Annotations = nil + } + if err2 := r.Patch(ctx, &job, revert); err2 != nil { + log.Error(err2, "failed to revert failure-pending annotation after POST failure", + append(failureWatcherLogFields(), "action", "revert_pending_failed", "job", job.Name)...) + } + return ctrl.Result{RequeueAfter: 30 * time.Second}, nil } - return ctrl.Result{RequeueAfter: 30 * time.Second}, nil } promotePatch := client.MergeFrom(job.DeepCopy()) if job.Annotations == nil { job.Annotations = map[string]string{} } + if job.Labels == nil { + job.Labels = map[string]string{} + } delete(job.Annotations, annotationFailurePending) job.Annotations[annotationFailureReported] = "true" + job.Annotations[annotationEvaluationStatus] = msg + job.Labels[labelEvaluationPhase] = labelEvaluationPhaseFailed if err := r.Patch(ctx, &job, promotePatch); err != nil { log.Error(err, "failed to promote failure-reported annotation after successful POST", append(failureWatcherLogFields(), "action", "promote_reported_failed", "job", job.Name)...) return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } + r.EventRecorder.Eventf(&job, corev1.EventTypeWarning, eventReasonEvaluationFailed, "%s", msg) + if err := r.deleteEvalHubFailureSyncedJob(ctx, &job); err != nil { return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } @@ -729,6 +762,16 @@ func failureAlreadyReported(job *batchv1.Job) bool { return job.Annotations[annotationFailureReported] == "true" } +// serverAlreadyHandledFailure returns true when the EvalHub server has already set evaluation-phase=Failed +// on the Job, indicating the server reported the failure via its own event path. Prevents duplicate +// Events when both the server and operator can detect the same failure. +func serverAlreadyHandledFailure(job *batchv1.Job) bool { + if job.Labels == nil { + return false + } + return job.Labels[labelEvaluationPhase] == labelEvaluationPhaseFailed +} + func failurePendingReport(job *batchv1.Job) bool { if job.Annotations == nil { return false diff --git a/controllers/evalhub/evaluation_job_failure_reconciler_lifecycle_test.go b/controllers/evalhub/evaluation_job_failure_reconciler_lifecycle_test.go new file mode 100644 index 000000000..0f08fc835 --- /dev/null +++ b/controllers/evalhub/evaluation_job_failure_reconciler_lifecycle_test.go @@ -0,0 +1,278 @@ +/* +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" + "net/http" + "net/http/httptest" + "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" + apierrors "k8s.io/apimachinery/pkg/api/errors" + 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" +) + +// jobFailureLifecycleScheme builds the minimal scheme for job failure reconciler lifecycle tests. +func jobFailureLifecycleScheme(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)) + return sc +} + +// noopEvalHubServer starts a local HTTP server that accepts any request and returns 204. +func noopEvalHubServer(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(srv.Close) + return srv +} + +// readyEvalHubCR returns a minimal EvalHub CR that IsReady() and points at the given URL. +func readyEvalHubCR(name, ns, url string) *evalhubv1.EvalHub { + return &evalhubv1.EvalHub{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}, + Status: evalhubv1.EvalHubStatus{ + URL: url, + Ready: corev1.ConditionTrue, + }, + } +} + +// evalHubEvaluationJob builds a failed Job with the mandatory EvalHub labels. +// extra is merged into the label map (use it to add instance/phase labels per test). +func evalHubEvaluationJob(name, ns string, extra map[string]string) *batchv1.Job { + labels := map[string]string{ + evalHubAppLabel: evalHubAppValue, + evalHubComponentLabel: evalHubComponentValue, + evalHubJobIDLabel: "jid-" + name, + evalHubProviderIDLabel: "provider-1", + evalHubBenchmarkIDLabel: "bench-1", + } + for k, v := range extra { + labels[k] = v + } + return &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: ns, + UID: types.UID("uid-" + name), + Labels: labels, + }, + Status: batchv1.JobStatus{ + Conditions: []batchv1.JobCondition{{ + Type: batchv1.JobFailed, Status: corev1.ConditionTrue, + }}, + }, + } +} + +// oomKilledAdapterPod builds a Pod owned by the given Job where the adapter container is OOMKilled. +func oomKilledAdapterPod(jobName, ns string, jobUID types.UID) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: jobName + "-pod", + Namespace: ns, + Labels: map[string]string{"batch.kubernetes.io/job-name": jobName}, + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "batch/v1", Kind: "Job", Name: jobName, UID: jobUID, + }}, + }, + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{{ + Name: adapterContainerName, + State: corev1.ContainerState{ + Terminated: &corev1.ContainerStateTerminated{Reason: "OOMKilled", ExitCode: 137}, + }, + }}, + }, + } +} + +// errImagePullInitPod builds a Pod owned by the given Job where the init container has ErrImagePull. +func errImagePullInitPod(jobName, ns string, jobUID types.UID) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: jobName + "-pod", + Namespace: ns, + Labels: map[string]string{"batch.kubernetes.io/job-name": jobName}, + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "batch/v1", Kind: "Job", Name: jobName, UID: jobUID, + }}, + }, + Status: corev1.PodStatus{ + InitContainerStatuses: []corev1.ContainerStatus{{ + Name: initContainerName, + State: corev1.ContainerState{ + Waiting: &corev1.ContainerStateWaiting{Reason: waitingReasonErrImagePull, Message: "pull failed"}, + }, + }}, + }, + } +} + +// buildJobFailureReconciler returns a reconciler with a FakeRecorder and the given tenant namespace. +func buildJobFailureReconciler(fc client.Client, tenantNamespace string) (*EvalHubEvaluationJobFailureReconciler, *record.FakeRecorder) { + rec := record.NewFakeRecorder(10) + tn := newEvalHubTenantNamespaces() + tn.Add(tenantNamespace) + return &EvalHubEvaluationJobFailureReconciler{ + Client: fc, + RESTConfig: &rest.Config{}, + EventRecorder: rec, + tenantNS: tn, + }, rec +} + +// TestJobFailureReconciler_ServerAlreadyHandled_NoEvent verifies that when the EvalHub server has +// already set evaluation-phase=Failed on the Job, the operator emits no duplicate event. +func TestJobFailureReconciler_ServerAlreadyHandled_NoEvent(t *testing.T) { + sc := jobFailureLifecycleScheme(t) + ns := "tenant-ns" + + job := evalHubEvaluationJob("eval-job-dedup", ns, map[string]string{ + labelEvaluationPhase: labelEvaluationPhaseFailed, + }) + + fc := fake.NewClientBuilder().WithScheme(sc).WithObjects(job).Build() + r, rec := buildJobFailureReconciler(fc, ns) + + result, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Namespace: ns, Name: job.Name}, + }) + require.NoError(t, err) + assert.Equal(t, ctrl.Result{}, result) + + select { + case ev := <-rec.Events: + t.Fatalf("expected no event after dedup (server already handled), got: %s", ev) + default: + } +} + +// TestJobFailureReconciler_OOMKill_EmitsEventAndPatchesJob verifies the full lifecycle for an OOM-killed +// adapter container: the reconciler emits an EvaluationFailed event and stamps evaluation-phase=Failed. +func TestJobFailureReconciler_OOMKill_EmitsEventAndPatchesJob(t *testing.T) { + srv := noopEvalHubServer(t) + sc := jobFailureLifecycleScheme(t) + ns := "tenant-ns" + + job := evalHubEvaluationJob("eval-job-oom", ns, map[string]string{ + evalHubInstanceNameLabel: "evalhub-1", + evalHubInstanceNamespaceLabel: "control-ns", + }) + pod := oomKilledAdapterPod(job.Name, ns, job.UID) + eh := readyEvalHubCR("evalhub-1", "control-ns", srv.URL) + + var patchedLabels map[string]string + var patchedAnnotations map[string]string + + fc := fake.NewClientBuilder(). + WithScheme(sc). + WithObjects(eh, job, pod). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if j, ok := obj.(*batchv1.Job); ok && j.Labels[labelEvaluationPhase] == labelEvaluationPhaseFailed { + patchedLabels = j.Labels + patchedAnnotations = j.Annotations + } + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + r, rec := buildJobFailureReconciler(fc, ns) + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Namespace: ns, Name: job.Name}, + }) + require.NoError(t, err) + + select { + case ev := <-rec.Events: + assert.Contains(t, ev, corev1.EventTypeWarning) + assert.Contains(t, ev, eventReasonEvaluationFailed) + assert.Contains(t, ev, "OOMKilled") + default: + t.Fatal("expected EvaluationFailed event but recorder is empty") + } + + require.NotNil(t, patchedLabels, "evaluation-phase=Failed patch was never applied to Job") + assert.Equal(t, labelEvaluationPhaseFailed, patchedLabels[labelEvaluationPhase]) + assert.NotEmpty(t, patchedAnnotations[annotationEvaluationStatus]) + + // After a successful sync the job is deleted. + err = fc.Get(context.Background(), types.NamespacedName{Namespace: ns, Name: job.Name}, &batchv1.Job{}) + assert.True(t, apierrors.IsNotFound(err), "job should be deleted after successful failure sync") +} + +// TestJobFailureReconciler_ImagePullError_EmitsEventAndPatchesJob verifies the full lifecycle for an init +// container stuck on ErrImagePull: the reconciler emits an EvaluationFailed event and stamps the Job. +func TestJobFailureReconciler_ImagePullError_EmitsEventAndPatchesJob(t *testing.T) { + srv := noopEvalHubServer(t) + sc := jobFailureLifecycleScheme(t) + ns := "tenant-ns" + + job := evalHubEvaluationJob("eval-job-imagepull", ns, map[string]string{ + evalHubInstanceNameLabel: "evalhub-1", + evalHubInstanceNamespaceLabel: "control-ns", + }) + pod := errImagePullInitPod(job.Name, ns, job.UID) + eh := readyEvalHubCR("evalhub-1", "control-ns", srv.URL) + + var patchedLabels map[string]string + + fc := fake.NewClientBuilder(). + WithScheme(sc). + WithObjects(eh, job, pod). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if j, ok := obj.(*batchv1.Job); ok && j.Labels[labelEvaluationPhase] == labelEvaluationPhaseFailed { + patchedLabels = j.Labels + } + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + r, rec := buildJobFailureReconciler(fc, ns) + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Namespace: ns, Name: job.Name}, + }) + require.NoError(t, err) + + select { + case ev := <-rec.Events: + assert.Contains(t, ev, corev1.EventTypeWarning) + assert.Contains(t, ev, eventReasonEvaluationFailed) + assert.Contains(t, ev, waitingReasonErrImagePull) + default: + t.Fatal("expected EvaluationFailed event but recorder is empty") + } + + require.NotNil(t, patchedLabels, "evaluation-phase=Failed patch was never applied to Job") + assert.Equal(t, labelEvaluationPhaseFailed, patchedLabels[labelEvaluationPhase]) +}