fix(dataprocess): use finished job condition for status updates

Signed-off-by: Harsh <harshmastic@gmail.com>

Author: hxrshxz

pkg/controllers/v1alpha1/dataprocess/status_handler.go

@@ -78,7 +78,7 @@ func (handler *OnceStatusHandler) GetOperationStatus(ctx runtime.ReconcileReques
 	}
 
 	// job either failed or complete, update DataLoad's phase status
-	jobCondition := job.Status.Conditions[0]
+	jobCondition := *finishedJobCondition
 
 	result.Conditions = []datav1alpha1.Condition{
 		{

pkg/controllers/v1alpha1/dataprocess/status_handler_test.go

@@ -20,11 +20,14 @@ import (
 	"testing"
 	"time"
 
+	"github.com/agiledragon/gomonkey/v2"
 	"github.com/fluid-cloudnative/fluid/api/v1alpha1"
 	"github.com/fluid-cloudnative/fluid/pkg/common"
 	cruntime "github.com/fluid-cloudnative/fluid/pkg/runtime"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/fake"
+	"github.com/fluid-cloudnative/fluid/pkg/utils/helm"
 	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -111,3 +114,116 @@ func TestOnceGetOperationStatus(t *testing.T) {
 		}
 	}
 }
+
+func TestOnceGetOperationStatusIgnoresMissingJobAfterHelmCleanup(t *testing.T) {
+	testScheme := runtime.NewScheme()
+	_ = v1alpha1.AddToScheme(testScheme)
+	_ = batchv1.AddToScheme(testScheme)
+
+	mockDataProcess := v1alpha1.DataProcess{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "test",
+			Namespace: "default",
+		},
+		Spec: v1alpha1.DataProcessSpec{},
+	}
+
+	patch := gomonkey.ApplyFunc(helm.DeleteReleaseIfExists, func(_ string, _ string) error { return nil })
+	defer patch.Reset()
+
+	client := fake.NewFakeClientWithScheme(testScheme, &mockDataProcess)
+	handler := &OnceStatusHandler{Client: client, dataProcess: &mockDataProcess}
+	ctx := cruntime.ReconcileRequestContext{
+		NamespacedName: types.NamespacedName{Namespace: "default", Name: "test"},
+		Log:            fake.NullLogger(),
+	}
+
+	opStatus := &v1alpha1.OperationStatus{Phase: common.PhaseExecuting}
+	result, err := handler.GetOperationStatus(ctx, opStatus)
+	if err != nil {
+		t.Fatalf("expected missing job to be handled without error, got %v", err)
+	}
+	if result == nil {
+		t.Fatal("expected non-nil result")
+	}
+	if result == opStatus {
+		t.Fatal("expected deep-copied status result")
+	}
+	if result.Phase != common.PhaseExecuting {
+		t.Fatalf("expected phase %q, got %q", common.PhaseExecuting, result.Phase)
+	}
+}
+
+func TestOnceGetOperationStatusUsesFinishedConditionWhenConditionsOutOfOrder(t *testing.T) {
+	testScheme := runtime.NewScheme()
+	_ = v1alpha1.AddToScheme(testScheme)
+	_ = batchv1.AddToScheme(testScheme)
+
+	createdAt := time.Now().Add(-2 * time.Minute)
+	suspendedAt := createdAt.Add(30 * time.Second)
+	failedAt := createdAt.Add(90 * time.Second)
+
+	mockDataProcess := v1alpha1.DataProcess{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "test",
+			Namespace: "default",
+		},
+		Spec: v1alpha1.DataProcessSpec{},
+	}
+
+	mockJob := batchv1.Job{
+		ObjectMeta: v1.ObjectMeta{
+			Name:              "test-processor-job",
+			Namespace:         "default",
+			CreationTimestamp: v1.NewTime(createdAt),
+		},
+		Status: batchv1.JobStatus{
+			Conditions: []batchv1.JobCondition{
+				{
+					Type:               batchv1.JobSuspended,
+					Status:             corev1.ConditionTrue,
+					LastProbeTime:      v1.NewTime(suspendedAt),
+					LastTransitionTime: v1.NewTime(suspendedAt),
+				},
+				{
+					Type:               batchv1.JobFailed,
+					Status:             corev1.ConditionTrue,
+					Reason:             "FailedReason",
+					Message:            "failed after resume",
+					LastProbeTime:      v1.NewTime(failedAt),
+					LastTransitionTime: v1.NewTime(failedAt),
+				},
+			},
+		},
+	}
+
+	client := fake.NewFakeClientWithScheme(testScheme, &mockDataProcess, &mockJob)
+	handler := &OnceStatusHandler{Client: client, dataProcess: &mockDataProcess}
+	ctx := cruntime.ReconcileRequestContext{
+		NamespacedName: types.NamespacedName{Namespace: "default", Name: "test"},
+		Log:            fake.NullLogger(),
+	}
+
+	result, err := handler.GetOperationStatus(ctx, &v1alpha1.OperationStatus{})
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if result == nil {
+		t.Fatal("expected non-nil result")
+	}
+	if result.Phase != common.PhaseFailed {
+		t.Fatalf("expected phase %q, got %q", common.PhaseFailed, result.Phase)
+	}
+	if len(result.Conditions) != 1 {
+		t.Fatalf("expected 1 condition, got %d", len(result.Conditions))
+	}
+	if result.Conditions[0].Type != common.ConditionType(batchv1.JobFailed) {
+		t.Fatalf("expected failed condition type, got %q", result.Conditions[0].Type)
+	}
+	if result.Conditions[0].Reason != "FailedReason" {
+		t.Fatalf("expected failed reason, got %q", result.Conditions[0].Reason)
+	}
+	if result.Duration != failedAt.Sub(createdAt).String() {
+		t.Fatalf("expected duration %q, got %q", failedAt.Sub(createdAt).String(), result.Duration)
+	}
+}