diff --git a/test/integration/job/job_test.go b/test/integration/job/job_test.go index b7e2cff1163..8e05d7d9352 100644 --- a/test/integration/job/job_test.go +++ b/test/integration/job/job_test.go @@ -5293,3 +5293,195 @@ func failTerminatingPods(ctx context.Context, t *testing.T, clientSet clientset. t.Fatalf("Failed to update pod statuses: %v", err) } } + +// TestMutablePodResourcesWithPodReplacementPolicyFailed tests that when a job with +// PodReplacementPolicy=Failed is suspended and pods are terminating, resource updates +// can be made to the job but new pods are only created after terminating pods are removed. +// The new pods should have the updated resources. +func TestMutablePodResourcesWithPodReplacementPolicyFailed(t *testing.T) { + const blockDeletionFinalizerForTest string = "fake.example.com/blockDeletion" + + // Enable MutablePodResourcesForSuspendedJobs feature gate + featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.MutablePodResourcesForSuspendedJobs, true) + + closeFn, restConfig, cs, ns := setup(t, "mutable-resources-pod-replacement") + defer closeFn() + + ctx, cancel := startJobControllerAndWaitForCaches(t, restConfig) + defer cancel() + + // Create a job with PodReplacementPolicy=Failed and a finalizer to block pod deletion + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-job", + Namespace: ns.Name, + }, + Spec: batchv1.JobSpec{ + Parallelism: ptr.To[int32](2), + Completions: ptr.To[int32](2), + PodReplacementPolicy: ptr.To(batchv1.Failed), + Template: v1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Finalizers: []string{blockDeletionFinalizerForTest}, + }, + Spec: v1.PodSpec{ + Containers: []v1.Container{{ + Name: "test-container", + Image: "busybox", + Resources: v1.ResourceRequirements{ + Requests: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("100m"), + v1.ResourceMemory: resource.MustParse("128Mi"), + }, + Limits: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("200m"), + v1.ResourceMemory: resource.MustParse("256Mi"), + }, + }, + }}, + RestartPolicy: v1.RestartPolicyNever, + }, + }, + }, + } + + job, err := createJobWithDefaults(ctx, cs, ns.Name, job) + if err != nil { + t.Fatalf("Failed to create Job: %v", err) + } + + jobClient := cs.BatchV1().Jobs(ns.Name) + + // Wait for pods to be active + t.Logf("Waiting for 2 pods to be active") + waitForPodsToBeActive(ctx, t, jobClient, 2, job) + + // Suspend the job + t.Logf("Suspending the job") + _, err = updateJob(ctx, jobClient, job.Name, func(j *batchv1.Job) { + j.Spec.Suspend = ptr.To(true) + }) + if err != nil { + t.Fatalf("Failed to suspend Job: %v", err) + } + + // Wait for the job to be suspended + err = wait.PollUntilContextTimeout(ctx, time.Second, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) { + j, err := jobClient.Get(ctx, job.Name, metav1.GetOptions{}) + if err != nil { + return false, err + } + return j.Spec.Suspend != nil && *j.Spec.Suspend, nil + }) + if err != nil { + t.Fatalf("Failed to wait for job to be suspended: %v", err) + } + t.Logf("Job is suspended") + + // Delete the pods to make them go to Terminating state + t.Logf("Deleting pods to make them go to Terminating state") + deletePods(ctx, t, cs, ns.Name) + + // Verify pods are in Terminating state (because of the finalizer) + validateJobsPodsStatusOnly(ctx, t, cs, job, podsByStatus{ + Terminating: ptr.To[int32](2), + Active: 0, + Ready: ptr.To[int32](0), + }) + t.Logf("Pods are in Terminating state") + + // Update the job's pod template with new resources + t.Logf("Updating job resources") + _, err = updateJob(ctx, jobClient, job.Name, func(j *batchv1.Job) { + for i := range j.Spec.Template.Spec.Containers { + j.Spec.Template.Spec.Containers[i].Resources = v1.ResourceRequirements{ + Requests: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("300m"), + v1.ResourceMemory: resource.MustParse("384Mi"), + }, + Limits: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("600m"), + v1.ResourceMemory: resource.MustParse("768Mi"), + }, + } + } + }) + if err != nil { + t.Fatalf("Failed to update Job resources: %v", err) + } + t.Logf("Job resources updated") + + // Wait a bit to ensure the controller has time to process the update + // With PodReplacementPolicy=Failed, no new pods should be created while old pods are terminating + time.Sleep(sleepDurationForControllerLatency) + + // Verify that no new pods are created while old pods are terminating + validateJobsPodsStatusOnly(ctx, t, cs, job, podsByStatus{ + Terminating: ptr.To[int32](2), + Active: 0, + Ready: ptr.To[int32](0), + }) + t.Logf("Verified: No new pods created while old pods are terminating") + + // Remove the finalizers to allow pods to finish terminating + t.Logf("Removing finalizers from terminating pods") + err = removePodsFinalizers(ctx, cs, ns.Name, []string{blockDeletionFinalizerForTest}) + if err != nil { + t.Fatalf("Failed to remove finalizers: %v", err) + } + + // Wait for pods to be deleted and verify terminating count drops to 0 + err = wait.PollUntilContextTimeout(ctx, time.Second, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) { + updatedJob, err := jobClient.Get(ctx, job.Name, metav1.GetOptions{}) + if err != nil { + return false, err + } + return updatedJob.Status.Terminating == nil || *updatedJob.Status.Terminating == 0, nil + }) + if err != nil { + t.Fatalf("Failed to wait for terminating pods to be removed: %v", err) + } + t.Logf("Terminating pods have been removed") + + // Unsuspend the job + t.Logf("Unsuspending the job") + _, err = updateJob(ctx, jobClient, job.Name, func(j *batchv1.Job) { + j.Spec.Suspend = ptr.To(false) + }) + if err != nil { + t.Fatalf("Failed to unsuspend Job: %v", err) + } + + // Wait for new pods to be created + waitForPodsToBeActive(ctx, t, jobClient, 2, job) + t.Logf("New pods are active") + + // Verify that the new pods have the updated resources + pods, err := getJobPods(ctx, t, cs, job, func(s v1.PodStatus) bool { return true }) + if err != nil { + t.Fatalf("Failed to get Job pods: %v", err) + } + if len(pods) != 2 { + t.Fatalf("Expected 2 pods, got %d", len(pods)) + } + + expectedResources := v1.ResourceRequirements{ + Requests: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("300m"), + v1.ResourceMemory: resource.MustParse("384Mi"), + }, + Limits: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("600m"), + v1.ResourceMemory: resource.MustParse("768Mi"), + }, + } + + for _, pod := range pods { + for i := range pod.Spec.Containers { + if diff := cmp.Diff(expectedResources, pod.Spec.Containers[i].Resources); diff != "" { + t.Errorf("Unexpected resources in pod %s for container %q (-want +got):\n%s", pod.Name, pod.Spec.Containers[i].Name, diff) + } + } + } + t.Logf("Verified: New pods have the updated resources") +}