add integration test for job failure event delay and remove the unit test

Author: kmala

pkg/controller/job/job_controller_test.go

@@ -56,7 +56,6 @@ import (
 	_ "k8s.io/kubernetes/pkg/apis/core/install"
 	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/pkg/controller/job/metrics"
-	"k8s.io/kubernetes/pkg/controller/job/util"
 	"k8s.io/kubernetes/pkg/controller/testutil"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/utils/clock"
@@ -2880,114 +2879,6 @@ func TestSingleJobFailedCondition(t *testing.T) {
 
 }
 
-func TestJobControllerMissingJobSucceedEvent(t *testing.T) {
-	t.Cleanup(setDurationDuringTest(&SyncJobBatchPeriod, fastSyncJobBatchPeriod))
-	logger, ctx := ktesting.NewTestContext(t)
-	job1 := newJob(1, 1, 6, batch.NonIndexedCompletion)
-	job1.Name = "job1"
-	clientSet := fake.NewSimpleClientset(job1)
-	fakeClock := clocktesting.NewFakeClock(time.Now())
-	jm, informer := newControllerFromClientWithClock(ctx, t, clientSet, controller.NoResyncPeriodFunc, fakeClock)
-	jm.podControl = &controller.RealPodControl{
-		KubeClient: clientSet,
-		Recorder:   testutil.NewFakeRecorder(),
-	}
-	jm.podStoreSynced = alwaysReady
-	jm.jobStoreSynced = alwaysReady
-
-	err := informer.Batch().V1().Jobs().Informer().GetIndexer().Add(job1)
-	if err != nil {
-		t.Fatalf("Unexpected error when adding job to indexer %v", err)
-	}
-	// 1st reconcile should create a new pod
-	err = jm.syncJob(ctx, testutil.GetKey(job1, t))
-	if err != nil {
-		t.Fatalf("Unexpected error when syncing jobs %v", err)
-	}
-
-	podIndexer := informer.Core().V1().Pods().Informer().GetIndexer()
-	podList, err := clientSet.Tracker().List(
-		schema.GroupVersionResource{Version: "v1", Resource: "pods"},
-		schema.GroupVersionKind{Version: "v1", Kind: "Pod"},
-		"default")
-	if err != nil {
-		t.Fatalf("Unexpected error when fetching pods %v", err)
-	}
-	// manually adding the just-created pod from fake clientset memory to informer cache because informer is not started.
-	// we are updating the pod status to succeeded which should update the job status to succeeded and remove the finalizer of the pod.
-	justCreatedPod := podList.(*v1.PodList).Items[0]
-	t.Logf("pod is %v\n", podList.(*v1.PodList).Items[0])
-	justCreatedPod.Status.Phase = v1.PodSucceeded
-	err = podIndexer.Add(&justCreatedPod)
-	if err != nil {
-		t.Fatalf("Unexpected error when adding pod to indexer %v", err)
-	}
-	jm.addPod(logger, &justCreatedPod)
-	err = jm.syncJob(ctx, testutil.GetKey(job1, t))
-	if err != nil {
-		t.Fatalf("Unexpected error when syncing jobs %v", err)
-	}
-
-	// Verify that the job is updated as succeeded in the client set. However this status is not updated yet in the
-	// informer is not started
-	jobList, err := clientSet.Tracker().List(
-		schema.GroupVersionResource{Group: "batch", Version: "v1", Resource: "jobs"},
-		schema.GroupVersionKind{Group: "batch", Version: "v1", Kind: "Job"},
-		"default")
-	if err != nil {
-		t.Fatalf("Unexpected error when trying to get job from the store: %v", err)
-	}
-	updatedJob := jobList.(*batch.JobList).Items[0]
-	if !util.IsJobSucceeded(&updatedJob) {
-		t.Fatalf("job status is not succeeded: %v", updatedJob)
-	}
-
-	// add the updated pod from the fake clientset memory to informer cache because informer is not started. This is to make
-	// sure the job controller informer cache has the latest pod status.
-	podList, err = clientSet.Tracker().List(
-		schema.GroupVersionResource{Version: "v1", Resource: "pods"},
-		schema.GroupVersionKind{Version: "v1", Kind: "Pod"},
-		"default")
-	if err != nil {
-		t.Fatalf("Unexpected error when fetching pods %v", err)
-	}
-	t.Logf("pod is %v\n", podList.(*v1.PodList).Items[0])
-	updatedPod := podList.(*v1.PodList).Items[0]
-	updatedPod.Status.Phase = v1.PodSucceeded
-	err = podIndexer.Add(&updatedPod)
-	if err != nil {
-		t.Fatalf("Unexpected error when adding pod to indexer %v", err)
-	}
-
-	// removing the just created pod from fake clientset memory but the pod will remain inside informer cache
-	// of the job controller. We are removing from the client set because in case of a bug if the job controller
-	// is trying to create the pod again it can succeed because it creates using the same name again.
-	err = clientSet.Tracker().Delete(
-		schema.GroupVersionResource{Version: "v1", Resource: "pods"},
-		"default", "")
-	if err != nil {
-		t.Fatalf("Unexpected error when deleting pod to indexer %v", err)
-	}
-
-	err = jm.syncJob(ctx, testutil.GetKey(job1, t))
-	if err != nil {
-		t.Fatalf("Unexpected error when syncing jobs %v", err)
-	}
-	time.Sleep(100 * time.Millisecond)
-
-	podList, err = clientSet.Tracker().List(
-		schema.GroupVersionResource{Version: "v1", Resource: "pods"},
-		schema.GroupVersionKind{Version: "v1", Kind: "Pod"},
-		"default")
-	if err != nil {
-		t.Fatalf("Unexpected error when syncing jobs %v", err)
-	}
-	// no pod should be created. Here it is 0 because we had deleted the pod from the client set.
-	if len(podList.(*v1.PodList).Items) != 0 {
-		t.Errorf("expect no pods to be created but %v pods are created", len(podList.(*v1.PodList).Items))
-	}
-}
-
 func TestSyncJobComplete(t *testing.T) {
 	_, ctx := ktesting.NewTestContext(t)
 	clientset := clientset.NewForConfigOrDie(&restclient.Config{Host: "", ContentConfig: restclient.ContentConfig{GroupVersion: &schema.GroupVersion{Group: "", Version: "v1"}}})

test/integration/job/job_test.go

@@ -4050,11 +4050,11 @@ func TestNodeSelectorUpdate(t *testing.T) {
 
 }
 
-// TestDelayedJobUpdateEvent tests that a Job that only executes one Pod even when
-// the job events are delayed. This test verfies the finishedJobStore is working
-// correctly and preventing from job controller creating a new pod if the job complete
+// TestDelayedJobSucceededUpdateEvent tests that a Job only creates one Pod even when
+// the job success events are delayed. This test verfies the finishedJobStore is working
+// correctly and preventing from job controller creating a new pod if the job success
 // even is delayed.
-func TestDelayedJobUpdateEvent(t *testing.T) {
+func TestDelayedJobSucceededUpdateEvent(t *testing.T) {
 	t.Cleanup(setDurationDuringTest(&jobcontroller.DefaultJobPodFailureBackOff, fastPodFailureBackoff))
 	t.Cleanup(setDurationDuringTest(&jobcontroller.SyncJobBatchPeriod, fastSyncJobBatchPeriod))
 	closeFn, restConfig, clientSet, ns := setup(t, "simple")
@@ -4113,6 +4113,94 @@ func TestDelayedJobUpdateEvent(t *testing.T) {
 	}
 }
 
+// TestDelayedJobFailedUpdateEvent tests that a Job only creates one Pod even when
+// the job failed events are delayed. This test verfies the finishedJobStore is working
+// correctly and preventing from job controller creating a new pod if the job failed
+// event is delayed.
+func TestDelayedJobFailedUpdateEvent(t *testing.T) {
+	t.Cleanup(setDurationDuringTest(&jobcontroller.DefaultJobPodFailureBackOff, fastPodFailureBackoff))
+	t.Cleanup(setDurationDuringTest(&jobcontroller.SyncJobBatchPeriod, fastSyncJobBatchPeriod))
+	closeFn, restConfig, clientSet, ns := setup(t, "pod-failure-policy")
+	t.Cleanup(closeFn)
+	// the transform is used to introduce a delay for the job events. Since all the object have to go through
+	// transform func first before being added to the informer cache, this would serve as an indirect way to
+	// introduce watch event delay.
+	transformOpt := informers.WithTransform(cache.TransformFunc(func(obj interface{}) (interface{}, error) {
+		_, ok := obj.(*batchv1.Job)
+		if ok {
+			// This will make sure pod events are processed before the job events occur.
+			time.Sleep(2 * fastSyncJobBatchPeriod)
+		}
+		return obj, nil
+	}))
+	ctx, cancel := startJobControllerAndWaitForCaches(t, restConfig, transformOpt)
+	t.Cleanup(func() {
+		cancel()
+	})
+	resetMetrics()
+
+	jobObj, err := createJobWithDefaults(ctx, clientSet, ns.Name, &batchv1.Job{
+		Spec: batchv1.JobSpec{
+			Template: v1.PodTemplateSpec{
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
+						{
+							Name:                     "main-container",
+							Image:                    "foo",
+							ImagePullPolicy:          v1.PullIfNotPresent,
+							TerminationMessagePolicy: v1.TerminationMessageFallbackToLogsOnError,
+						},
+					},
+				},
+			},
+			BackoffLimit: ptr.To[int32](0),
+		},
+	})
+	if err != nil {
+		t.Fatalf("Failed to create Job: %v", err)
+	}
+	validateJobPodsStatus(ctx, t, clientSet, jobObj, podsByStatus{
+		Active:      1,
+		Ready:       ptr.To[int32](0),
+		Terminating: ptr.To[int32](0),
+	})
+
+	op := func(p *v1.Pod) bool {
+		p.Status = v1.PodStatus{
+			Phase: v1.PodFailed,
+			ContainerStatuses: []v1.ContainerStatus{
+				{
+					Name: "main-container",
+					State: v1.ContainerState{
+						Terminated: &v1.ContainerStateTerminated{
+							ExitCode: 5,
+						},
+					},
+				},
+			},
+		}
+		return true
+	}
+	if _, err := updateJobPodsStatus(ctx, clientSet, jobObj, op, 1); err != nil {
+		t.Fatalf("Error %q while updating pod status for Job: %v", err, jobObj.Name)
+	}
+	validateJobsPodsStatusOnly(ctx, t, clientSet, jobObj, podsByStatus{
+		Active:      0,
+		Failed:      1,
+		Ready:       ptr.To[int32](0),
+		Terminating: ptr.To[int32](0),
+	})
+
+	validateJobFailed(ctx, t, clientSet, jobObj)
+	jobPods, err := getJobPods(ctx, t, clientSet, jobObj, func(ps v1.PodStatus) bool { return true })
+	if err != nil {
+		t.Fatalf("Error %v getting the list of pods for job %q", err, klog.KObj(jobObj))
+	}
+	if len(jobPods) != 1 {
+		t.Errorf("Found %d Pods for the job %q, want 1", len(jobPods), klog.KObj(jobObj))
+	}
+}
+
 type podsByStatus struct {
 	Active      int
 	Ready       *int32